Restore automatic suppliers after proven resynchronization and test unilateral generation changes

src/stream/stream_optimizer.c

@@ -21,6 +21,7 @@ struct stream_offer {
 struct stream_supplier {
     struct stream_supplier *next;
     uint64_t node, cycle, retry_ms;
+    uint64_t local_epoch, peer_epoch;
     struct STREAM_RECEIVE *probe;
     struct stream_offer *offers;
     unsigned probe_sub, failures;
@@ -116,6 +117,7 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin
             }
         } else {
             s->offers[s->probe_sub].info = info; s->failures = 0; s->retry_ms = 0;
+            s->local_epoch = info.local_epoch; s->peer_epoch = info.peer_epoch;
             DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer offer peer=%llx sub=%u range=%llu..%llu hops=%u"
                 " rtt=%u/%d chain=%u/%d slots=%u accepting=%d lag=%llu stopped=%d age=%llu/%d queue=%llu",
                 (unsigned long long)s->node, s->probe_sub, (unsigned long long)info.first_available,
@@ -228,7 +230,11 @@ static void receive_failed(struct stream_optimizer *o, struct STREAM_RECEIVE **h
     struct stream_receive_info info; stream_receive_get_info(*handle, &info);
     struct stream_supplier *s = supplier_find(o, info.peer);
     if (s && s->offers) {
-        struct stream_offer *offer = &s->offers[sub]; offer->failures++; offer->retry_ms = now + backoff(offer->failures);
+        struct stream_offer *offer = &s->offers[sub];
+        int resynchronized = info.reason == STREAM_REASON_TRANSPORT && info.local_epoch && info.peer_epoch
+            && s->local_epoch && s->peer_epoch && (info.local_epoch != s->local_epoch || info.peer_epoch != s->peer_epoch);
+        if (resynchronized) { offer->failures = 0; offer->retry_ms = 0; }
+        else { offer->failures++; offer->retry_ms = now + backoff(offer->failures); }
         if (info.reason == STREAM_REASON_OVERLOAD) s->retry_ms = offer->retry_ms;
     }
     o->info.failures++;

tests/test_stream_service.c

@@ -314,6 +314,13 @@ static void test_network(void) {
         }
         CHECK(info.blocks == 1 && info.start == 1 && info.next == 2 && info.handle);
     }
+    /* Меняется только локальная stream generation; физический ETCP и поколение источника сохраняются. */
+    struct stream_receive_info previous_generation; CHECK(stream_receive_get_info(up[0], &previous_generation) == STREAM_OK);
+    struct ETCP_CONN *stable = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(stable && stable->links_up);
+    int notified = 0;
+    for (struct etcp_status_cbk_entry *cb = nodes[1]->conn_status_cbks; cb; cb = cb->next)
+        if (cb->arg == stream_network_get(nodes[1])) { cb->fn(stable, ETCP_CONN_STATUS_REINIT, cb->arg); notified = 1; break; }
+    CHECK(notified && stable->links_up);
     /* Новый bootstrap может получить уже сохранённый блок: DUPLICATE обязан установить общий старт. */
     CHECK(stream_receive_restart(subscribers[1]) == STREAM_OK);
     network_data[1] = 0; network_next[1] = 1; first_request[0] = 0;
@@ -323,6 +330,9 @@ static void test_network(void) {
         CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, i, UINT64_MAX, &up[i]) == STREAM_OK);
     }
     run_until(&network_data[1], 1);
+    struct stream_receive_info current_generation; CHECK(stream_receive_get_info(up[0], &current_generation) == STREAM_OK);
+    CHECK(current_generation.local_epoch != previous_generation.local_epoch && current_generation.peer_epoch == previous_generation.peer_epoch);
+    CHECK(stable == instance_find_conn(nodes[1], nodes[3]->node_id) && stable->links_up);
     /* Node2 видит подписку relay через обычный Merkle-обмен, а не копирование из его SQLite. */
     deadline = get_time_tb() + 150000;
     int ready = 0;