Retain stable NAT results and compact the address database

7 days ago · db8a5ee2df
2 changed files with 80 additions and 42 deletions
--- a/src/routing_layer/nat_detection.c
+++ b/src/routing_layer/nat_detection.c
@ -13,7 +13,6 @@
 #include "route_ping.h"
 #include "topo_node.h"
 #include "topo_group.h"
-#include "ntp_time.h"
 #include "stcp_client.h"
 #include "stun_nat.h"
 #include "../lib/debug_config.h"
@ -27,7 +26,7 @@
 #define NAT_MAX_CONTACTS 256
 #define NAT_TICK_TB 1000
 #define NAT_ROUND_TB 300000
-#define NAT_REFRESH_TB 1800000
+#define NAT_REFRESH_TB 18000000
 #define NAT_RETRY_TB 300000
 #define NAT_PROBE_COUNT 3
 #define NAT_QUERY_TAG_SIZE (ROUTE_OBSERVE_TAG_SIZE + 9)
@ -52,6 +51,7 @@ struct nat_socket {
    struct ETCP_SOCKET *socket, *auxiliary;
    struct sockaddr_storage local, observer_address, helper_address, alternate_address;
    struct sockaddr_storage baseline, helper_seen, alternate_seen;
+    struct sockaddr_storage measured_address;
    struct address_profile measured_profile;
    uint64_t select_due;
    uint8_t fallback_allowed;
@ -361,6 +361,7 @@ static void clear_profile(struct nat_socket *state, uint8_t reason) {
    memset(&state->socket->nat_addr, 0, sizeof(state->socket->nat_addr));
    state->completed_tb = 0;
    memset(&state->measured_profile, 0, sizeof(state->measured_profile));
+    memset(&state->measured_address, 0, sizeof(state->measured_address));
    if (changed) {
        DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT profile invalidated socket=%s generation=%016llx reason=%s",
                   state->socket->name, (unsigned long long)state->socket->generation, nat_check_reason_name(reason));
@ -369,39 +370,47 @@ static void clear_profile(struct nat_socket *state, uint8_t reason) {
 }

 /* Единственный выбор публикуемого профиля. STUN переносит только тип IP/netif,
- * никогда не mapped endpoint тестового порта. ETCP сохраняет приоритет до TTL. */
+ * никогда не mapped endpoint тестового порта. Успешный ETCP действует до явного сброса. */
 static void select_profile(struct nat_socket *state) {
-    uint64_t now = get_time_tb();
-    struct address_profile selected = {0};
-    if (state->completed_tb && now >= state->completed_tb &&
-        now - state->completed_tb <= ADDRESS_PROFILE_TTL_US / 100) selected = state->measured_profile;
+    const struct address_profile *current = &state->socket->address_profile;
+    struct address_profile selected = *current;
+    struct sockaddr_storage selected_address = {0};
+    if (state->measured_profile.source == ADDRESS_SOURCE_ETCP) {
+        selected = state->measured_profile;
+        if (selected.mapping == ADDRESS_MAP_EIM) selected_address = state->measured_address;
+    }
    else if (state->fallback_allowed && !state->socket->is_tcp) {
        struct stun_nat_result result;
        if (!stun_nat_get_socket_result(state->detector->instance, state->socket, &result) &&
-            result.generation && result.evidence_id && result.measured_at_us && result.measured_at_tb &&
-            now >= result.measured_at_tb && now - result.measured_at_tb < ADDRESS_PROFILE_TTL_US / 100 &&
+            result.generation &&
            (result.status == STUN_NAT_COMPLETE || result.status == STUN_NAT_PARTIAL)) {
-            if (result.mapping == STUN_MAP_EIM) selected.mapping = result.translated == 0 ? ADDRESS_MAP_UNCHANGED : ADDRESS_MAP_EIM;
+            struct address_profile fallback = {0};
+            if (result.mapping == STUN_MAP_EIM) fallback.mapping = result.translated == 0 ? ADDRESS_MAP_UNCHANGED : ADDRESS_MAP_EIM;
            else if (result.mapping == STUN_MAP_DEPENDENT || result.mapping == STUN_MAP_ADM || result.mapping == STUN_MAP_APDM)
-                selected.mapping = ADDRESS_MAP_DEPENDENT;
-            if (result.filtering == STUN_FILTER_EIF) selected.inbound = ADDRESS_IN_OPEN;
-            else if (result.filtering == STUN_FILTER_ADF || result.filtering == STUN_FILTER_APDF) selected.inbound = ADDRESS_IN_CONTACT;
-            if (selected.mapping || selected.inbound) {
-                selected.source = ADDRESS_SOURCE_STUN;
-                selected.evidence_id = result.evidence_id; selected.checked_at = result.measured_at_us;
+                fallback.mapping = ADDRESS_MAP_DEPENDENT;
+            if (result.filtering == STUN_FILTER_EIF) fallback.inbound = ADDRESS_IN_OPEN;
+            else if (result.filtering == STUN_FILTER_ADF || result.filtering == STUN_FILTER_APDF) fallback.inbound = ADDRESS_IN_CONTACT;
+            if (fallback.mapping || fallback.inbound) {
+                fallback.source = ADDRESS_SOURCE_STUN;
+                if (current->source == ADDRESS_SOURCE_STUN) {
+                    if (!fallback.mapping) fallback.mapping = current->mapping;
+                    if (fallback.mapping == current->mapping && !fallback.inbound) fallback.inbound = current->inbound;
+                }
+                selected = fallback;
            }
        }
    }
-    const struct address_profile *current = &state->socket->address_profile;
-    if (selected.mapping == current->mapping && selected.inbound == current->inbound && selected.source == current->source &&
-        selected.evidence_id == current->evidence_id && selected.checked_at == current->checked_at) return;
+    int address_changed = selected_address.ss_family != state->socket->nat_addr.ss_family ||
+        (selected_address.ss_family && !route_ping_addr_equal(&selected_address, &state->socket->nat_addr, 1));
+    if (selected.mapping == current->mapping && selected.inbound == current->inbound &&
+        selected.source == current->source && !address_changed) return;
    DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT selected socket=%s generation=%016llx source=%s previous=%s mapping=%s inbound=%s"
-               " evidence=%016llx reason=%s",
+               " address_changed=%d reason=%s",
               state->socket->name, (unsigned long long)state->socket->generation, address_profile_source_name(selected.source),
               address_profile_source_name(current->source), address_mapping_name(selected.mapping), address_inbound_name(selected.inbound),
-               (unsigned long long)selected.evidence_id, nat_check_reason_name(state->reason));
+               address_changed, nat_check_reason_name(state->reason));
    state->socket->address_profile = selected;
-    if (selected.source != ADDRESS_SOURCE_ETCP) memset(&state->socket->nat_addr, 0, sizeof(state->socket->nat_addr));
+    state->socket->nat_addr = selected_address;
    publish(state->detector);
 }

@ -410,7 +419,7 @@ static void owner_finish(struct nat_socket *state, uint8_t reason) {
    state->active_ping = 0; state->phase = NAT_IDLE; state->reason = reason;
    if (reason == NAT_REASON_NO_OBSERVER || reason == NAT_REASON_NO_HELPER) state->fallback_allowed = 1;
    state->preferred_observer = 0;
-    state->due = get_time_tb() + (reason ? NAT_RETRY_TB : NAT_REFRESH_TB);
+    state->due = get_time_tb() + (state->socket->address_profile.source || !reason ? NAT_REFRESH_TB : NAT_RETRY_TB);
    if (reason) {
        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT round incomplete socket=%s generation=%016llx round=%016llx reason=%s",
                   state->socket->name, (unsigned long long)state->socket->generation,
@ -420,20 +429,23 @@ static void owner_finish(struct nat_socket *state, uint8_t reason) {
 }

 static void owner_apply(struct nat_socket *state, uint8_t mapping, uint8_t inbound) {
-    int64_t now = ntp_time_get_us(state->detector->instance);
-    if (now <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT profile time unavailable"); owner_finish(state, NAT_REASON_INVALID_OBSERVATION); return; }
-    struct address_profile profile = {.mapping = mapping, .inbound = inbound, .source = ADDRESS_SOURCE_ETCP,
-        .evidence_id = state->round, .checked_at = now};
+    struct address_profile profile = {.mapping = mapping, .inbound = inbound, .source = ADDRESS_SOURCE_ETCP};
    uint8_t partial_reason = inbound != ADDRESS_IN_UNKNOWN ? NAT_REASON_NONE : !state->filter_clean ? NAT_REASON_CONTACT_HISTORY :
        mapping == ADDRESS_MAP_DEPENDENT ? NAT_REASON_DEPENDENT_MAPPING : NAT_REASON_TIMEOUT;
-    state->measured_profile = profile; state->fallback_allowed = 0;
-    memset(&state->socket->nat_addr, 0, sizeof(state->socket->nat_addr));
-    if (mapping == ADDRESS_MAP_EIM) state->socket->nat_addr = state->baseline;
+    if (inbound == ADDRESS_IN_UNKNOWN && state->measured_profile.source == ADDRESS_SOURCE_ETCP &&
+        mapping == state->measured_profile.mapping && route_ping_addr_equal(&state->measured_address, &state->baseline, 1)) {
+        profile.inbound = state->measured_profile.inbound;
+        DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT incomplete ingress preserves previous result socket=%s round=%016llx inbound=%s reason=%s",
+                    state->socket->name, (unsigned long long)state->round, address_inbound_name(profile.inbound),
+                    nat_check_reason_name(partial_reason));
+    }
+    state->measured_profile = profile; state->measured_address = state->baseline; state->fallback_allowed = 0;
    state->completed_tb = get_time_tb();
-    DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT profile socket=%s id=%u generation=%016llx evidence=%016llx mapping=%s inbound=%s"
+    DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT profile socket=%s id=%u generation=%016llx round=%016llx mapping=%s inbound=%s effective_inbound=%s"
               " local=%s external=%s observer=%016llx helper=%016llx clean=%u partial_reason=%s",
               state->socket->name, state->socket->sock_id, (unsigned long long)state->socket->generation,
               (unsigned long long)state->round, address_mapping_name(mapping), address_inbound_name(inbound),
+               address_inbound_name(profile.inbound),
               sockaddr_storage_to_str(&state->local).str, sockaddr_storage_to_str(&state->baseline).str,
               (unsigned long long)state->observer, (unsigned long long)state->helper, state->filter_clean,
               nat_check_reason_name(partial_reason));
@ -896,8 +908,6 @@ static void detector_tick(void *arg) {
    for (struct ETCP_SOCKET *socket = detector->instance->etcp_sockets; socket; socket = socket->next)
        if (!socket->probe_only) socket_state(detector, socket, 1);
    for (struct nat_socket *state = detector->sockets; state; state = state->next) {
-        if (state->completed_tb && now - state->completed_tb > ADDRESS_PROFILE_TTL_US / 100)
-            clear_profile(state, NAT_REASON_TIMEOUT);
        if (now >= state->select_due) { select_profile(state); state->select_due = now + 10000; }
        if (state->phase && now >= state->deadline) { owner_finish(state, NAT_REASON_TIMEOUT); continue; }
        if (!state->phase && now >= state->due && (detector->automatic || state->forced)) {
--- a/src/routing_layer/topo_node_sqlite.c
+++ b/src/routing_layer/topo_node_sqlite.c
@ -121,9 +121,7 @@ static int address_row_put(sqlite3_stmt *statement, struct TOPO_NODE *node, uint
    if (has_reality) sqlite3_bind_blob(statement, 10, &reality, sizeof(reality), SQLITE_STATIC);
    else sqlite3_bind_null(statement, 10);
    sqlite3_bind_int(statement, 11, kind); sqlite3_bind_int(statement, 12, profile->inbound);
-    sqlite3_bind_int64(statement, 13, (sqlite3_int64)profile->evidence_id);
-    sqlite3_bind_int64(statement, 14, (sqlite3_int64)profile->checked_at);
-    sqlite3_bind_int(statement, 15, profile->source);
+    sqlite3_bind_int(statement, 13, profile->source);
    int result = sqlite3_step(statement);
    sqlite3_reset(statement);
    if (result != SQLITE_DONE) {
@ -147,8 +145,8 @@ int topo_node_sqlite_addrs_put(sqlite3 *db, uint64_t node_id, struct TOPO_NODE *
    if (sqlite3_step(statement) != SQLITE_DONE) goto fail;
    sqlite3_finalize(statement); statement = NULL;
    if (sqlite3_prepare_v2(db, "INSERT INTO node_addresses(node_id,family,protocol,address,port,addr_type,config_type,"
-        "nat_mapping,socket_id,options,address_kind,nat_inbound,nat_evidence,nat_checked,nat_source)"
-        " VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", -1, &statement, NULL) != SQLITE_OK) goto fail;
+        "nat_mapping,socket_id,options,address_kind,nat_inbound,nat_source)"
+        " VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)", -1, &statement, NULL) != SQLITE_OK) goto fail;
    int written = 0;
    for (const struct TOPO_ADDR4 *address = node->v4_addrs; address; address = address->next) {
        uint8_t config;
@ -244,6 +242,7 @@ int topo_node_sqlite_init(sqlite3* db) {

    /* Проекция адресов производная. Старый формат пересоздаём, не интерпретируя NAT_TYPE_* как новый профиль. */
    sqlite3_stmt *schema = NULL;
+    int legacy_profile = 0;
    if (sqlite3_prepare_v2(db, "SELECT 1 FROM sqlite_master WHERE type='table' AND name='node_addresses'",
                          -1, &schema, NULL) != SQLITE_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "address schema inspection failed: %s", sqlite3_errmsg(db)); return -1;
@ -252,7 +251,7 @@ int topo_node_sqlite_init(sqlite3* db) {
    sqlite3_finalize(schema); schema = NULL;
    if (exists) {
        int compatible = sqlite3_prepare_v2(db,
-            "SELECT nat_mapping,nat_inbound,nat_evidence,nat_checked,nat_source,address_kind FROM node_addresses LIMIT 0",
+            "SELECT nat_mapping,nat_inbound,nat_source,address_kind FROM node_addresses LIMIT 0",
            -1, &schema, NULL);
        sqlite3_finalize(schema);
        if (compatible == SQLITE_ERROR) {
@ -262,6 +261,17 @@ int topo_node_sqlite_init(sqlite3* db) {
            DEBUG_INFO(DEBUG_CATEGORY_BGP, "old address projection reset; signed NODEINFO will rebuild endpoint profiles");
        } else if (compatible != SQLITE_OK) {
            DEBUG_ERROR(DEBUG_CATEGORY_BGP, "address schema read failed: %s", sqlite3_errmsg(db)); return -1;
+        } else {
+            if (sqlite3_prepare_v2(db, "SELECT count(*) FROM pragma_table_info('node_addresses')"
+                " WHERE name IN ('nat_evidence','nat_checked')", -1, &schema, NULL) != SQLITE_OK ||
+                sqlite3_step(schema) != SQLITE_ROW) {
+                DEBUG_ERROR(DEBUG_CATEGORY_BGP, "address profile schema inspection failed: %s", sqlite3_errmsg(db));
+                sqlite3_finalize(schema); return -1;
+            }
+            int fields = sqlite3_column_int(schema, 0);
+            sqlite3_finalize(schema); schema = NULL;
+            if (fields == 1) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "incomplete legacy NAT profile schema"); return -1; }
+            legacy_profile = fields == 2;
        }
    }

@ -292,8 +302,6 @@ int topo_node_sqlite_init(sqlite3* db) {
        "  config_type INTEGER DEFAULT 0,"
        "  nat_mapping INTEGER NOT NULL DEFAULT 0,"
        "  nat_inbound INTEGER NOT NULL DEFAULT 0,"
-        "  nat_evidence INTEGER NOT NULL DEFAULT 0,"
-        "  nat_checked INTEGER NOT NULL DEFAULT 0,"
        "  nat_source INTEGER NOT NULL DEFAULT 0,"
        "  address_kind INTEGER NOT NULL DEFAULT 0,"
        "  socket_id   INTEGER NOT NULL DEFAULT 0,"
@ -331,6 +339,24 @@ int topo_node_sqlite_init(sqlite3* db) {
        return -1;
    }

+    /* Старые подписи непригодны; адреса и максимальные версии остаются для bootstrap и новой публикации. */
+    if (legacy_profile) {
+        const char *migration = "SAVEPOINT compact_nat_profile;"
+            "ALTER TABLE node_addresses DROP COLUMN nat_evidence;"
+            "ALTER TABLE node_addresses DROP COLUMN nat_checked;"
+            "UPDATE node_addresses SET nat_mapping=0,nat_inbound=0,nat_source=0,"
+            "addr_type=CASE WHEN addr_type=0 OR config_type IN (3,4) THEN 0 ELSE 4 END;"
+            "UPDATE node_snapshots SET record=X'';"
+            "RELEASE compact_nat_profile;";
+        if (sqlite3_exec(db, migration, NULL, NULL, NULL) != SQLITE_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_BGP, "compact NAT profile migration failed: %s", sqlite3_errmsg(db));
+            if (sqlite3_exec(db, "ROLLBACK TO compact_nat_profile; RELEASE compact_nat_profile", NULL, NULL, NULL) != SQLITE_OK)
+                DEBUG_ERROR(DEBUG_CATEGORY_BGP, "compact NAT profile migration rollback failed: %s", sqlite3_errmsg(db));
+            return -1;
+        }
+        DEBUG_INFO(DEBUG_CATEGORY_BGP, "NAT profile compacted to mapping/inbound/source; bootstrap addresses and node versions preserved");
+    }
+
    /* migration: добавить колонку options в уже существующую node_addresses (idempotent) */
    {
        sqlite3_stmt* cs = NULL;
@ -1033,7 +1059,9 @@ struct TOPO_NODE* topo_node_sqlite_snapshot_load(sqlite3* db, struct TOPO_GROUPS
        struct TOPO_GROUP group = { .instance = groups->instance };
        const uint8_t* data = sqlite3_column_blob(stmt, 0);
        int len = sqlite3_column_bytes(stmt, 0);
-        if (topo_node_deserialize(&group, data, (size_t)len, &ni, NULL, NULL, NULL, NULL) < 0 ||
+        if (!len) {
+            DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "snapshot invalidated by format change node=%016llx", (unsigned long long)node_id);
+        } else if (topo_node_deserialize(&group, data, (size_t)len, &ni, NULL, NULL, NULL, NULL) < 0 ||
            !ni || ni->node_id != node_id || topo_node_verify(ni) < 0) {
            DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid persisted snapshot node=%016llx", (unsigned long long)node_id);
            topo_node_destroy(groups, ni); ni = NULL;