Race initial stream suppliers with relay delay of 100ms

src/stream/stream_optimizer.c

@@ -2,6 +2,7 @@
 
 #include "../../lib/debug_config.h"
 #include "../../lib/mem.h"
+#include "../../lib/u_async.h"
 #include "../utun_instance.h"
 #include "../transport_layer/node_conn_direct.h"
 #include "stream_optimizer.h"
@@ -12,6 +13,7 @@
 #define STREAM_PROBE_CYCLE_MS 30000
 #define STREAM_OFFER_TTL_MS 60000
 #define STREAM_MIGRATION_TIMEOUT_MS 5000
+#define STREAM_BOOTSTRAP_RELAY_MS 100
 #define STREAM_PROBLEM_PENALTY_MS 1000
 #define STREAM_HOP_PENALTY_MS 3
 
@@ -44,10 +46,18 @@ struct stream_optimizer {
     struct stream_selection_settings settings;
     struct stream_selection_info info;
     struct stream_probe_schedule schedule;
+    void *startup_timer;
+    uint64_t started_ms, startup_peer;
     uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms, deficient_ms, relay_poll_ms;
-    int progress_valid, draining;
+    int progress_valid, draining, startup_complete;
 };
 
+/* Точный срок гонки не зависит от фазы общего 100-мс таймера; selector остаётся отложенным. */
+static void startup_timeout(void *arg) {
+    struct stream_optimizer *o = arg; o->startup_timer = NULL;
+    stream_network_wake(stream_network_get(o->inst));
+}
+
 unsigned stream_probe_schedule_tick(struct stream_probe_schedule *s, uint64_t now, unsigned pending, unsigned active) {
     if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream probe schedule missing"); return 0; }
     unsigned budget = 1;
@@ -221,12 +231,17 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig
         for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
             if (!s->online || !s->compatible || s->node == exclude || !s->offers || now < s->retry_ms
                 || (s->probe && s->probe_sub == sub)) continue;
+            if (!o->startup_complete && o->startup_peer && s->node != o->startup_peer) continue;
             struct stream_offer *offer = &s->offers[sub]; struct stream_receive_info *info = &offer->info;
             if (now < offer->retry_ms || !info->info_ms || now - info->info_ms > STREAM_OFFER_TTL_MS || !info->accepting || !info->free_slots
                 || (seq != UINT64_MAX && info->first_available != UINT64_MAX && seq < info->first_available)) continue;
             if (!pass && o->descriptor.r && usage(o, s->node, sub) >= o->descriptor.r) continue;
             if (!best) { best = s; continue; }
             struct stream_receive_info *previous = &best->offers[sub].info;
+            if (!o->startup_complete && seq == UINT64_MAX) {
+                if (info->info_order < previous->info_order) best = s;
+                continue;
+            }
             uint64_t score = stream_supplier_score(info, &info->supplier_metrics, info->supplier_metrics_valid);
             uint64_t old_score = stream_supplier_score(previous, &previous->supplier_metrics, previous->supplier_metrics_valid);
             int better = score < old_score;
@@ -240,6 +255,18 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig
     return best;
 }
 
+/* Первый пригодный INFO среди всех субпотоков выигрывает гонку; порядок приёма не зависит от порядка списка пиров. */
+static struct stream_supplier *startup_choose(struct stream_optimizer *o, uint64_t now, unsigned *subblock) {
+    struct stream_supplier *first = NULL; uint64_t order = UINT64_MAX;
+    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
+        struct stream_supplier *s = supplier_choose(o, sub, UINT64_MAX, 0, now);
+        if (s && s->offers[sub].info.info_order < order) {
+            first = s; order = s->offers[sub].info.info_order; *subblock = sub;
+        }
+    }
+    return first;
+}
+
 static void receive_failed(struct stream_optimizer *o, struct STREAM_RECEIVE **handle, unsigned sub, uint64_t now) {
     struct stream_receive_info info; stream_receive_get_info(*handle, &info);
     struct stream_supplier *s = supplier_find(o, info.peer);
@@ -273,11 +300,18 @@ int stream_optimizer_create(struct UTUN_INSTANCE *inst, const struct stream_desc
     if (o) o->selected = u_calloc(d->n, sizeof(*o->selected));
     if (!o || !o->selected) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer allocation failed"); if (o) u_free(o); return STREAM_ERR_MEMORY; }
     o->inst = inst; o->descriptor = *d; o->settings = (struct stream_selection_settings){2000, 5000, 20};
+    o->started_ms = get_time_tb() / 10;
+    o->startup_timer = uasync_set_timeout(inst->ua, STREAM_BOOTSTRAP_RELAY_MS * 10, o, startup_timeout, "stream_bootstrap");
+    if (!o->startup_timer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream bootstrap timer allocation failed");
+        u_free(o->selected); u_free(o); return STREAM_ERR_MEMORY;
+    }
     o->info.automatic = 1; *out = o; return STREAM_OK;
 }
 
 void stream_optimizer_destroy(struct stream_optimizer *o) {
     if (!o) return;
+    if (o->startup_timer) uasync_cancel_timeout(o->inst->ua, o->startup_timer);
     for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
         stream_receive_close(o->selected[sub].current); stream_receive_close(o->selected[sub].replacement);
     }
@@ -316,6 +350,13 @@ void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) {
 void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
     struct stream_local local;
     if (stream_local_get(o->inst, &o->descriptor, &local) != STREAM_OK) return;
+    if (local.decoded && !o->startup_complete) {
+        o->startup_complete = 1;
+        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap complete group=%llx source=%llx epoch=%llx peer=%llx elapsed_ms=%llu start=%llu",
+            (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
+            (unsigned long long)o->descriptor.source_epoch, (unsigned long long)o->startup_peer,
+            (unsigned long long)(now - o->started_ms), (unsigned long long)local.start);
+    }
     if (local.finished) {
         /* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */
         if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) {
@@ -385,6 +426,21 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
     }
     int migrating = 0;
     for (unsigned sub = 0; sub < o->descriptor.n; sub++) if (o->selected[sub].replacement) migrating = 1;
+    if (!local.anchored && !busy && !o->startup_complete) {
+        o->startup_peer = 0;
+        unsigned sub = 0; struct stream_supplier *s = startup_choose(o, now, &sub);
+        if (s && stream_receive_open(h, s->node, (uint8_t)sub, UINT64_MAX, &o->selected[sub].current) == STREAM_OK) {
+            o->startup_peer = s->node; o->selected[sub].opened = 1; o->selected[sub].opened_ms = now; busy++;
+            if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; }
+            for (struct stream_supplier *other = o->suppliers; other; other = other->next)
+                if (other != s && other->probe) { stream_receive_close(other->probe); other->probe = NULL; }
+            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap winner group=%llx source=%llx epoch=%llx peer=%llx sub=%u"
+                " order=%llu elapsed_ms=%llu relay=%d",
+                (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
+                (unsigned long long)o->descriptor.source_epoch, (unsigned long long)s->node, sub,
+                (unsigned long long)s->offers[sub].info.info_order, (unsigned long long)(now - o->started_ms), s->relay);
+        }
+    }
     for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
         struct stream_selection *pick = &o->selected[sub]; struct stream_receive_info info = {0};
         if (pick->replacement || (!local.anchored && busy)) continue;
@@ -476,9 +532,11 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
     int deficient = active < required || (viable < required && newer);
     if (!deficient) o->deficient_ms = 0;
     else if (!o->deficient_ms) o->deficient_ms = now;
-    if (now >= o->relay_poll_ms) {
+    if (!o->startup_complete || now >= o->relay_poll_ms) {
         o->relay_poll_ms = now + STREAM_PROBE_PERIOD_MS;
         int needed = deficient && now - o->deficient_ms >= STREAM_MIGRATION_TIMEOUT_MS;
+        if (!o->startup_complete && !o->startup_peer && now - o->started_ms >= STREAM_BOOTSTRAP_RELAY_MS) needed = 1;
+        if (!o->startup_complete && o->startup_peer && stream_relay_confirmed(o->inst, &o->descriptor, o->startup_peer)) needed = 1;
         stream_relay_demand(o->inst, &o->descriptor, needed, now);
     }
     /* Пока нет первого блока, раз в секунду показываем причину ожидания каждого поставщика. */

tests/test_stream_service.c

@@ -54,10 +54,17 @@ static int automatic_drop_relay;
 static int automatic_drop_tail;
 static etcp_recv_fn automatic_handlers[4];
 static uint8_t automatic_request[STREAM_WIRE_HEADER + 1];
-static etcp_recv_fn relay_raw_handlers[3], relay_route_handler;
+static etcp_recv_fn relay_raw_handlers[4], relay_route_handler;
 static unsigned relay_data[3], relay_ends[3], relay_requests_seen;
 static int relay_drop_source;
 static uint8_t relay_request_wire[STREAM_WIRE_HEADER];
+static struct {
+    int hold_info, hold_request, check_starts;
+    uint8_t info[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
+    size_t info_length;
+    unsigned source_subscribes, helper_subscribes, accepted;
+    uint64_t request_tb, start;
+} relay_race;
 static struct {
     unsigned data, reinits, responses;
     int hold_info;
@@ -1009,7 +1016,27 @@ static void relay_role(unsigned member, int enabled, unsigned version) {
 
 /* Прямой stream-вход источника блокируем только у зрителей; Merkle/BGP и получение суперузла остаются настоящими. */
 static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
-    unsigned index = 0; while (index < 3 && conn->instance != nodes[index]) index++; CHECK(index < 3);
+    unsigned index = 0; while (index < 4 && conn->instance != nodes[index]) index++; CHECK(index < 4);
+    if (entry->len >= STREAM_WIRE_HEADER) {
+        unsigned command = entry->dgram[2];
+        if (command == STREAM_CMD_SUBSCRIBE && conn->peer_node_id == nodes[2]->node_id) {
+            if (index == 3) relay_race.source_subscribes++;
+            if (index == 1) relay_race.helper_subscribes++;
+        }
+        if (index == 2 && conn->peer_node_id == nodes[3]->node_id && command == STREAM_CMD_INFO_RESP && relay_race.hold_info) {
+            if (!relay_race.info_length) {
+                CHECK(entry->len <= sizeof(relay_race.info)); relay_race.info_length = entry->len;
+                memcpy(relay_race.info, entry->dgram, entry->len);
+            }
+            queue_dgram_free(entry); queue_entry_free(entry); return;
+        }
+        if (relay_race.check_starts && index == 2 && conn->peer_node_id == nodes[1]->node_id && command == STREAM_CMD_ACCEPT) {
+            struct stream_message m; CHECK(stream_wire_parse(entry->dgram, entry->len, &m) == STREAM_OK && m.length >= 16);
+            uint64_t start = stream_read_be(m.payload + 8, 8);
+            if (!relay_race.accepted) relay_race.start = start;
+            CHECK(start == relay_race.start && m.subblock < 8); relay_race.accepted |= 1u << m.subblock;
+        }
+    }
     if (relay_drop_source && index != 1 && conn->peer_node_id == nodes[3]->node_id) {
         queue_dgram_free(entry); queue_entry_free(entry); return;
     }
@@ -1021,6 +1048,8 @@ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
     if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
         && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_REQUEST) {
         memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++;
+        if (!relay_race.request_tb) relay_race.request_tb = get_time_tb();
+        if (relay_race.hold_request) { queue_dgram_free(entry); queue_entry_free(entry); return; }
     }
     relay_route_handler(conn, entry);
 }
@@ -1041,6 +1070,15 @@ static void relay_test_control(const struct stream_message *message) {
     CHECK(etcp_route_send(nodes[2], GROUP2, nodes[1]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0);
 }
 
+/* Запоздалый настоящий INFO подаётся тому же raw dispatcher; старый probe nonce не должен менять победителя. */
+static void relay_release_info(void) {
+    CHECK(relay_race.info_length);
+    struct ETCP_CONN *conn = instance_find_conn(nodes[2], nodes[3]->node_id); CHECK(conn && conn->links_up);
+    struct ll_entry *entry = ll_alloc_lldgram((uint16_t)relay_race.info_length); CHECK(entry);
+    memcpy(entry->dgram, relay_race.info, relay_race.info_length); entry->len = relay_race.info_length;
+    relay_raw_handlers[2](conn, entry); relay_race.info_length = 0; relay_race.hold_info = 0;
+}
+
 /* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */
 static void test_supernode_relay(void) {
     int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE];
@@ -1055,14 +1093,14 @@ static void test_supernode_relay(void) {
         CHECK(member_sync_start(nodes[i], nodes[3]->node_id, "710002", NULL, NULL) == 0);
     }
     relay_role(1, 1, 100);
-    for (unsigned i = 0; i < 3; i++) {
+    for (unsigned i = 0; i < 4; i++) {
         relay_raw_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM];
         CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_test_raw) == 0);
-        stream_set_event_callback(nodes[i], relay_test_event, NULL);
+        if (i < 3) stream_set_event_callback(nodes[i], relay_test_event, NULL);
     }
     relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
     CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0);
-    struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE};
+    struct stream_parameters parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE};
     struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0};
     CHECK(stream_publish_open(nodes[3], GROUP2, &parameters, &publisher) == STREAM_OK);
     struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK);
@@ -1074,11 +1112,45 @@ static void test_supernode_relay(void) {
     CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && !relay_requests_seen);
     stream_close(viewers[2]); viewers[2] = NULL;
     test_clock_run_for(ua, 10000);
+    /* Помощь начинается через 100 мс, но прямой INFO выигрывает до ответа помощника и отменяет его запрос. */
+    relay_race.hold_info = relay_race.hold_request = 1;
+    uint64_t started = get_time_tb();
+    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
+    test_clock_run_for(ua, 900); CHECK(!relay_requests_seen);
+    deadline = get_time_tb() + 10000;
+    while (!relay_requests_seen && get_time_tb() < deadline) test_clock_poll(ua, 100);
+    CHECK(relay_requests_seen && relay_race.info_length);
+    CHECK(relay_race.request_tb - started >= 990 && relay_race.request_tb - started < 1500);
+    relay_release_info(); run_until(&relay_data[2], 2); test_clock_run_for(ua, 2000);
+    CHECK(relay_race.source_subscribes && !relay_race.helper_subscribes);
+    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
+    struct stream_message cancelled; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &cancelled) == STREAM_OK);
+    relay_race.hold_request = 0; relay_test_control(&cancelled); test_clock_run_for(ua, 2000);
+    CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap direct wins after helper request: request_delay_ms=%llu, late REQUEST cancelled",
+               (unsigned long long)((relay_race.request_tb - started) / 10));
+    stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 1000);
+    /* Закрытие reader до дедлайна отменяет и таймер, и объединённую работу сети. */
+    unsigned requests_before = relay_requests_seen;
+    relay_race.hold_info = 1;
+    CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
+    test_clock_run_for(ua, 500); stream_close(viewers[2]); viewers[2] = NULL;
+    test_clock_run_for(ua, 2000); CHECK(relay_requests_seen == requests_before);
+    memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1;
     relay_drop_source = 1;
+    started = get_time_tb();
     CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
     deadline = get_time_tb() + 300000;
-    while (relay_data[2] < 2 && get_time_tb() < deadline) test_clock_poll(ua, 100);
+    while (relay_data[2] < 3 && get_time_tb() < deadline) test_clock_poll(ua, 100);
-    CHECK(relay_data[2] == 2 && relay_requests_seen && !relay_data[1]);
+    CHECK(relay_data[2] == 3 && relay_requests_seen && !relay_data[1]);
+    CHECK(get_time_tb() - started < 10000 && relay_race.request_tb - started >= 990);
+    test_clock_run_for(ua, 1000); CHECK(relay_race.accepted == 255);
+    CHECK(relay_race.helper_subscribes && !relay_race.source_subscribes);
+    relay_release_info(); test_clock_run_for(ua, 1000);
+    CHECK(!relay_race.source_subscribes && relay_data[2] == 3);
+    relay_race.check_starts = 0;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap helper wins: elapsed_ms=%llu, n=8 r=2, common_start=%llu, stale direct INFO ignored",
+               (unsigned long long)((get_time_tb() - started) / 10), (unsigned long long)relay_race.start);
     CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only);
     /* Поздний обычный подписчик также удерживает раздачу, даже без собственного запроса помощи. */
     CHECK(stream_subscribe_open(nodes[0], GROUP2, nodes[3]->node_id, &viewers[0]) == STREAM_OK);
@@ -1163,9 +1235,9 @@ static void test_supernode_relay(void) {
     stream_close(viewers[2]); stream_close(publisher);
     relay_drop_source = 0; relay_role(1, 0, 101);
     CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0);
-    for (unsigned i = 0; i < 3; i++) {
+    for (unsigned i = 0; i < 4; i++) {
         CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0);
-        stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL);
+        if (i < 3) stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL);
     }
     for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]);
     debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, route_level);