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)
{
    unsigned bucket = hash_to_bucket(ctx->hash);
    struct conn *nc = NULL;

    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;
        }

        nc = new_conn(&ct->buckets[bucket], pkt, &ctx->key, now);
        ctx->conn = nc;
        nc->rev_key = nc->key;
        conn_key_reverse(&nc->rev_key);

        if (helper) {
            nc->alg = 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->passive_mode) {
                    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);

                if (!nat_res) {
                    goto nat_res_exhaustion;
                }

                /* Update nc with nat adjustments made to
                 * conn_for_un_nat_copy by nat_select_range_tuple(). */
                *nc = *conn_for_un_nat_copy;
                ct_rwlock_unlock(&ct->resources_lock);
            }
            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);
        }
        hmap_insert(&ct->buckets[bucket].connections, &nc->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:
    ovs_list_remove(&nc->exp_node);
    delete_conn(nc);
    /* conn_for_un_nat_copy is a local variable in process_one; this
     * memset() serves to document that conn_for_un_nat_copy is from
     * this point on unused. */
    memset(conn_for_un_nat_copy, 0, sizeof *conn_for_un_nat_copy);
    ct_rwlock_unlock(&ct->resources_lock);
    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;
}