conn_not_found(struct conntrack *ct, struct dp_packet *pkt,
               struct conn_lookup_ctx *ctx, bool commit, long long now,
               const struct nat_action_info_t *nat_action_info,
               struct conn *conn_for_un_nat_copy,
               const char *helper,
               const struct alg_exp_node *alg_exp,
               enum ct_alg_ctl_type ct_alg_ctl)
{
    struct conn *nc = NULL;
    struct conn connl;

    if (!valid_new(pkt, &ctx->key)) {
        pkt->md.ct_state = CS_INVALID;
        return nc;
    }

    pkt->md.ct_state = CS_NEW;

    if (alg_exp) {
        pkt->md.ct_state |= CS_RELATED;
    }

    if (commit) {
        unsigned int n_conn_limit;
        atomic_read_relaxed(&ct->n_conn_limit, &n_conn_limit);

        if (atomic_count_get(&ct->n_conn) >= n_conn_limit) {
            COVERAGE_INC(conntrack_full);
            return nc;
        }

        unsigned bucket = hash_to_bucket(ctx->hash);
        nc = &connl;
        memset(nc, 0, sizeof *nc);
        memcpy(&nc->key, &ctx->key, sizeof nc->key);
        memcpy(&nc->rev_key, &nc->key, sizeof nc->rev_key);
        conn_key_reverse(&nc->rev_key);

        if (ct_verify_helper(helper, ct_alg_ctl)) {
            nc->alg = nullable_xstrdup(helper);
        }

        if (alg_exp) {
            nc->alg_related = true;
            nc->mark = alg_exp->master_mark;
            nc->label = alg_exp->master_label;
            nc->master_key = alg_exp->master_key;
        }

        if (nat_action_info) {
            nc->nat_info = xmemdup(nat_action_info, sizeof *nc->nat_info);

            if (alg_exp) {
                if (alg_exp->nat_rpl_dst) {
                    nc->rev_key.dst.addr = alg_exp->alg_nat_repl_addr;
                    nc->nat_info->nat_action = NAT_ACTION_SRC;
                } else {
                    nc->rev_key.src.addr = alg_exp->alg_nat_repl_addr;
                    nc->nat_info->nat_action = NAT_ACTION_DST;
                }
                *conn_for_un_nat_copy = *nc;
                ct_rwlock_wrlock(&ct->resources_lock);
                bool new_insert = nat_conn_keys_insert(&ct->nat_conn_keys,
                                                       conn_for_un_nat_copy,
                                                       ct->hash_basis);
                ct_rwlock_unlock(&ct->resources_lock);
                if (!new_insert) {
                    char *log_msg = xasprintf("Pre-existing alg "
                                              "nat_conn_key");
                    ct_print_conn_info(conn_for_un_nat_copy, log_msg, VLL_INFO,
                                       true, false);
                    free(log_msg);
                }
            } else {
                *conn_for_un_nat_copy = *nc;
                ct_rwlock_wrlock(&ct->resources_lock);
                bool nat_res = nat_select_range_tuple(ct, nc,
                                                      conn_for_un_nat_copy);
                ct_rwlock_unlock(&ct->resources_lock);

                if (!nat_res) {
                    goto nat_res_exhaustion;
                }

                /* Update nc with nat adjustments made to
                 * conn_for_un_nat_copy by nat_select_range_tuple(). */
                memcpy(nc, conn_for_un_nat_copy, sizeof *nc);
            }
            conn_for_un_nat_copy->conn_type = CT_CONN_TYPE_UN_NAT;
            conn_for_un_nat_copy->nat_info = NULL;
            conn_for_un_nat_copy->alg = NULL;
            nat_packet(pkt, nc, ctx->icmp_related);
        }
        struct conn *nconn = new_conn(&ct->buckets[bucket], pkt, &ctx->key,
                                      now);
        memcpy(&nconn->key, &nc->key, sizeof nconn->key);
        memcpy(&nconn->rev_key, &nc->rev_key, sizeof nconn->rev_key);
        memcpy(&nconn->master_key, &nc->master_key, sizeof nconn->master_key);
        nconn->alg_related = nc->alg_related;
        nconn->alg = nc->alg;
        nconn->mark = nc->mark;
        nconn->label = nc->label;
        nconn->nat_info = nc->nat_info;
        ctx->conn = nc = nconn;
        hmap_insert(&ct->buckets[bucket].connections, &nconn->node, ctx->hash);
        atomic_count_inc(&ct->n_conn);
    }

    return nc;

    /* This would be a user error or a DOS attack.
     * A user error is prevented by allocating enough
     * combinations of NAT addresses when combined with
     * ephemeral ports.  A DOS attack should be protected
     * against with firewall rules or a separate firewall.
     * Also using zone partitioning can limit DoS impact. */
nat_res_exhaustion:
    free(nc->alg);
    free(nc->nat_info);
    static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
    VLOG_WARN_RL(&rl, "Unable to NAT due to tuple space exhaustion - "
                 "if DoS attack, use firewalling and/or zone partitioning.");
    return NULL;
}