Wake stream supplier discovery on Merkle and BGP callbacks

src/stream/stream.c

@@ -378,6 +378,7 @@ static void service_reconcile(void *arg) {
         if (s->closing && !s->holds && !s->delivery) session_free(s);
         s = next;
     }
+    stream_network_wake(service->network);
 }
 
 /* Общий props сигнал относится и к Merkle commit, и к BGP. Payload adm_tags здесь не используется. */
@@ -392,11 +393,15 @@ static void member_changed(uint64_t node_id, const char *tags, const char *chann
     }
     if (!service->sessions) return;
     stream_network_membership_changed(service->network, node_id, channel);
+    int changed = 0;
     for (struct stream_session *s = service->sessions; s; s = s->next) {
         int own = node_id == service->inst->node_id;
-        if (s->closing || (!own && (s->source || s->descriptor.source_node_id != node_id))) continue;
+        if (s->closing) continue;
         struct TOPO_GROUP *group = stream_group(service->inst, s->descriptor.group_id);
         if (!group || !channel || strcmp(group->channel_id, channel)) continue;
+        changed = 1;
+        if (s->optimizer) stream_optimizer_roster_changed(s->optimizer, node_id);
+        if (!own && (s->source || s->descriptor.source_node_id != node_id)) continue;
         struct stream_member_state state; uint8_t key[32];
         int result = stream_member_load(service->inst, group->group_id, node_id, &state, key);
         if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT) {
@@ -413,7 +418,7 @@ static void member_changed(uint64_t node_id, const char *tags, const char *chann
             s->invalidated = 1; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
         }
     }
-    if (service->reconcile) return;
+    if (!changed || service->reconcile) return;
     service->reconcile = uasync_call_soon(service->inst->ua, service, service_reconcile);
     if (!service->reconcile) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream member reconciliation scheduling failed");
 }

src/stream/stream_optimizer.c

@@ -93,7 +93,14 @@ static void roster_update(struct stream_optimizer *o, struct STREAM_HANDLE *h, u
         }
         s->online = 1; s->source = candidates[i].source;
         s->relay = stream_relay_confirmed(o->inst, &o->descriptor, s->node);
-        s->compatible = node_conn_direct_compatible(o->inst, s->node);
+        int compatible = node_conn_direct_compatible(o->inst, s->node);
+        if (compatible && !s->compatible && s->cycle != o->schedule.cycle && now >= s->retry_ms) {
+            o->schedule.next_ms = 0;
+            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer new supplier wakes probe group=%llx source=%llx peer=%llx",
+                        (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
+                        (unsigned long long)s->node);
+        }
+        s->compatible = compatible;
     }
     u_free(candidates);
     struct stream_supplier **link = &o->suppliers;
@@ -339,6 +346,13 @@ int stream_optimizer_configure(struct stream_optimizer *o, const struct stream_s
 
 void stream_optimizer_info(struct stream_optimizer *o, struct stream_selection_info *info) { *info = o->info; }
 
+/* Несколько изменений до одного прохода объединяются; только новый доступный поставщик ускоряет расписание пробы. */
+void stream_optimizer_roster_changed(struct stream_optimizer *o, uint64_t peer) {
+    if (o->roster_ms) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer roster invalidated group=%llx source=%llx peer=%llx",
+        (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id, (unsigned long long)peer);
+    o->roster_ms = 0;
+}
+
 /* READY отменяет ожидание старого цикла/ошибки только для подтверждённого поставщика. */
 void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) {
     o->roster_ms = o->schedule.next_ms = 0;

src/stream/stream_optimizer.h

@@ -18,6 +18,8 @@ void stream_optimizer_destroy(struct stream_optimizer *optimizer);
 void stream_optimizer_tick(struct stream_optimizer *optimizer, struct STREAM_HANDLE *subscription, uint64_t now_ms);
 int stream_optimizer_configure(struct stream_optimizer *optimizer, const struct stream_selection_settings *settings);
 void stream_optimizer_info(struct stream_optimizer *optimizer, struct stream_selection_info *info);
+/* Merkle/BGP помечает список поставщиков устаревшим; пробы/цикл/backoff сохраняются до отложенной сверки. */
+void stream_optimizer_roster_changed(struct stream_optimizer *optimizer, uint64_t peer);
 /* Подтверждённая помощь немедленно обновляет кандидатов и снимает старый backoff конкретного пира. */
 void stream_optimizer_refresh(struct stream_optimizer *optimizer, uint64_t peer);
 

tests/test_stream_service.c

@@ -59,11 +59,11 @@ static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuse
 static int relay_drop_source;
 static uint8_t relay_request_wire[STREAM_WIRE_HEADER];
 static struct {
-    int hold_info, hold_request, hold_ready, check_starts;
+    int hold_info, hold_request, hold_ready, check_starts, watch_provider;
     uint8_t info[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
     size_t info_length;
     unsigned source_subscribes, helper_subscribes, accepted, ready_held;
-    uint64_t request_tb, start;
+    uint64_t request_tb, start, epoch, provider_tb, provider_info_tb;
 } relay_race;
 static struct {
     unsigned data, reinits, responses;
@@ -1019,6 +1019,8 @@ static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
     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 (relay_race.watch_provider && index == 2 && conn->peer_node_id == nodes[1]->node_id
+            && command == STREAM_CMD_INFO_RESP && !relay_race.provider_info_tb) relay_race.provider_info_tb = get_time_tb();
         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++;
@@ -1071,6 +1073,17 @@ static void relay_test_status(struct ETCP_CONN *conn, struct ll_entry *entry) {
     relay_status_handler(conn, entry);
 }
 
+/* Фиксируем принятую Merkle/BGP-опцию перед stream callback; PROFILE без нужной подписки не является поставщиком. */
+static void relay_test_provider(uint64_t node, const char *tags, const char *channel, void *arg) {
+    (void)tags; (void)arg;
+    if (!relay_race.watch_provider || relay_race.provider_tb || node != nodes[1]->node_id
+        || !channel || strcmp(channel, "710002")) return;
+    struct stream_member_state member;
+    if (stream_member_load(nodes[2], GROUP2, node, &member, NULL) == STREAM_OK
+        && stream_member_online(nodes[2], GROUP2, node)
+        && stream_member_has_subscription(&member, nodes[3]->node_id, relay_race.epoch)) relay_race.provider_tb = get_time_tb();
+}
+
 /* Внутренний приём не вызывает GUI callback; публичный viewer на том же экземпляре вызывает его. */
 static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
     (void)arg; unsigned index = 0; while (index < 3 && inst != nodes[index]) index++; CHECK(index < 3);
@@ -1119,6 +1132,7 @@ static void test_supernode_relay(void) {
     CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0);
     relay_status_handler = nodes[2]->router_bindings.callbacks[ETCP_RT_ID_STREAM];
     CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_test_status) == 0);
+    CHECK(member_sync_add_props_cbk(nodes[2], relay_test_provider, NULL) == 0);
     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);
@@ -1161,7 +1175,8 @@ static void test_supernode_relay(void) {
     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);
     /* Merkle объявляет буфер помощника раньше STATUS READY: выбранная выдача должна удержать STARTING-запрос. */
-    memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_ready = 1;
+    memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_ready = relay_race.watch_provider = 1;
+    relay_race.epoch = origin.descriptor.source_epoch;
     relay_drop_source = 1;
     unsigned cancellations_before = relay_cancels_seen;
     unsigned expected_data = relay_data[2] + 1;
@@ -1171,6 +1186,10 @@ static void test_supernode_relay(void) {
         test_clock_poll(ua, 100);
     CHECK(relay_race.ready_held && relay_cancels_seen == cancellations_before);
     CHECK(relay_data[2] == expected_data && relay_race.helper_subscribes);
+    CHECK(relay_race.provider_tb && relay_race.provider_info_tb >= relay_race.provider_tb);
+    CHECK(relay_race.provider_info_tb - relay_race.provider_tb < 2000);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] provider callback starts INFO without 1s roster wait: elapsed_ms=%llu",
+               (unsigned long long)((relay_race.provider_info_tb - relay_race.provider_tb) / 10));
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] helper INFO before delayed READY keeps selected pending issuance; no premature CANCEL");
     stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 40000);
     CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
@@ -1280,6 +1299,7 @@ static void test_supernode_relay(void) {
     relay_drop_source = 0; relay_role(1, 0, 101);
     CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0);
     CHECK(etcp_router_bind(nodes[2], ETCP_RT_ID_STREAM, relay_status_handler) == 0);
+    member_sync_remove_props_cbk(nodes[2], relay_test_provider, NULL);
     for (unsigned i = 0; i < 4; i++) {
         CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0);
         if (i < 3) stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL);