Resynchronize unilateral peer generations through nonce-bound INFO without epoch ping-pong

src/stream/stream_network.c

@@ -155,7 +155,7 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control allocation failed command=%u", m.command); return STREAM_ERR_MEMORY;
     }
     uint64_t queued_ms = get_time_tb() / 10; memcpy(e->data, &queued_ms, sizeof(queued_ms));
-    m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
+    m.sender_epoch = p->own_epoch; m.receiver_epoch = m.command == STREAM_CMD_INFO_REQ ? 0 : p->other_epoch;
     stream_wire_header(e->dgram, &m);
     if (length) memcpy(e->dgram + STREAM_WIRE_HEADER, payload, length);
     e->len = STREAM_WIRE_HEADER + length;
@@ -235,17 +235,26 @@ static void peer_sweep(struct stream_peer *p) {
 }
 
 /* Invalidation сохраняет транспорт чужих владельцев; старые DATA больше не имеют живого handle. */
-static void peer_reset(struct stream_peer *p, int reason) {
+static void peer_reset(struct stream_peer *p, int reason, int rotate, struct STREAM_RECEIVE *keep) {
     peer_cancel_wait(p);
     struct ll_entry *e;
     while ((e = queue_data_get(p->control))) { queue_dgram_free(e); queue_entry_free(e); }
-    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) flow_close(f, reason, 0);
+    for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
+        if (f == keep) continue;
+        if (!rotate && f->outgoing && !f->info.handle && f->info.state < STREAM_RECEIVE_REQUESTED) {
+            f->info_sent = f->info_ready = f->info.info_ready = 0; f->probe = f->probe_sent_ms = 0;
+            f->info.local_epoch = f->info.peer_epoch = f->info.info_ms = 0;
+            f->info.supplier_metrics_valid = f->info.rtt_valid = f->info.path_rtt_valid = 0;
+            f->info.state = STREAM_RECEIVE_CONNECTING;
+        } else flow_close(f, reason, 0);
+    }
     p->other_epoch = p->highwater = 0; p->next_handle = 1; p->last_sent = NULL;
-    p->own_epoch = new_epoch(); peer_sweep(p);
+    if (rotate) p->own_epoch = new_epoch();
+    peer_sweep(p);
     memset(&p->pressure, 0, sizeof(p->pressure)); memset(&p->requested, 0, sizeof(p->requested));
     p->pressure_id = 0; p->cooling = 0; stream_load_init(&p->load, get_time_tb() / 10);
-    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx reason=%d",
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx rotate=%d reason=%d",
-               (unsigned long long)p->node_id, (unsigned long long)p->own_epoch, reason);
+               (unsigned long long)p->node_id, (unsigned long long)p->own_epoch, rotate, reason);
 }
 
 /* NCD подтверждает UP собственного владельца, даже если ETCP уже использовался BGP/другим стримом. */
@@ -262,7 +271,7 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v
                     .subblock = f->subblock, .command = STREAM_CMD_CLOSED};
                 route_control(p, m, &reason, 1);
             }
-        peer_reset(p, STREAM_REASON_TRANSPORT);
+        peer_reset(p, STREAM_REASON_TRANSPORT, 1, NULL);
     } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id);
     DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream NCD event peer=%llx event=%d up=%d", (unsigned long long)p->node_id, event, p->up);
 }
@@ -591,7 +600,7 @@ static void peer_pump(void *arg) {
         if (etcp_send(conn, e) != 0) {
             queue_dgram_free(e); queue_entry_free(e);
             DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control send failed peer=%llx", (unsigned long long)p->node_id);
-            peer_reset(p, STREAM_REASON_TRANSPORT);
+            peer_reset(p, STREAM_REASON_TRANSPORT, 1, NULL);
         } else sent = 1;
         queue_resume_callback(p->control);
     } else {
@@ -712,8 +721,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
         if (!p) p = peer_open(net, sender);
         if (!p || (m.receiver_epoch && m.receiver_epoch != p->own_epoch)) return;
         if (p->other_epoch && p->other_epoch != m.sender_epoch) {
-            peer_reset(p, STREAM_REASON_TRANSPORT);
+            peer_reset(p, STREAM_REASON_TRANSPORT, 0, NULL);
-            /* Remote INFO начинает новое поколение; его следующий ответ подтвердит свежий local epoch. */
+            /* Изменившийся sender уже отличает пару поколений; own не вращаем в ответ, чтобы не получить ping-pong. */
         }
         p->other_epoch = m.sender_epoch;
         uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0};
@@ -729,7 +738,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
         return;
     }
     if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch
-        || (p->other_epoch && m.sender_epoch != p->other_epoch)) {
+        || (p->other_epoch && m.sender_epoch != p->other_epoch && m.command != STREAM_CMD_INFO_RESP)) {
         DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session peer=%llx command=%u handle=%llu sender=%llx/%llx receiver=%llx/%llx",
                     (unsigned long long)sender, m.command, (unsigned long long)m.handle,
                     (unsigned long long)m.sender_epoch, (unsigned long long)(p ? p->other_epoch : 0),
@@ -744,7 +753,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
             DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale INFO probe peer=%llx", (unsigned long long)sender);
         } else if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &f->info, &start) == STREAM_OK
                    && stream_metric_read(m.payload + STREAM_INFO_METRIC_OFF, &f->info.supplier_metrics, &f->info.supplier_metrics_valid) == STREAM_OK) {
-            p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1; peer_schedule(p);
+            if (p->other_epoch && p->other_epoch != m.sender_epoch) peer_reset(p, STREAM_REASON_TRANSPORT, 0, f);
+            p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1;
+            f->info.local_epoch = p->own_epoch; f->info.peer_epoch = p->other_epoch; peer_schedule(p);
             uint64_t rtt = get_time_tb() / 10 - f->probe_sent_ms;
             f->info.rtt_valid = f->probe_sent_ms != 0; f->info.rtt_ms = rtt > UINT32_MAX ? UINT32_MAX : (uint32_t)rtt;
             for (struct STREAM_RECEIVE *active = p->flows; active; active = active->next) {
@@ -847,7 +858,7 @@ static void transport_changed(struct ETCP_CONN *conn, int status, void *arg) {
     struct stream_network *net = arg;
     if (status != ETCP_CONN_STATUS_REINIT || net->stopping) return;
     struct stream_peer *p = peer_find(net, conn->peer_node_id);
-    if (p) peer_reset(p, STREAM_REASON_TRANSPORT);
+    if (p) peer_reset(p, STREAM_REASON_TRANSPORT, 1, NULL);
 }
 
 /* FIFO ограничена и по времени. Отмена неподтверждённого SUBSCRIBE/ACCEPT не допускает DATA без живого согласования. */

src/stream/stream_network.h

@@ -19,7 +19,7 @@ struct STREAM_RECEIVE;
 enum stream_receive_state { STREAM_RECEIVE_CONNECTING, STREAM_RECEIVE_INFO, STREAM_RECEIVE_REQUESTED,
                             STREAM_RECEIVE_ACTIVE, STREAM_RECEIVE_ENDED, STREAM_RECEIVE_CLOSED };
 struct stream_receive_info {
-    uint64_t handle, peer, start, next, blocks, gaps;
+    uint64_t handle, peer, start, next, blocks, gaps, local_epoch, peer_epoch;
     enum stream_receive_state state;
     int reason;
     struct stream_metric_snapshot metrics, supplier_metrics;