From da0a9ad31650c49c5c631585a9f105fd6b2b1cb5 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 17:55:19 +0200 Subject: [PATCH] Wake stream supplier discovery on Merkle and BGP callbacks --- src/stream/stream.c | 9 +++++++-- src/stream/stream_optimizer.c | 16 +++++++++++++++- src/stream/stream_optimizer.h | 2 ++ tests/test_stream_service.c | 26 +++++++++++++++++++++++--- 4 files changed, 47 insertions(+), 6 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index 43981bec..e4d389e0 100644 --- a/src/stream/stream.c +++ b/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"); } diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index a10e6c80..65f7a959 100644 --- a/src/stream/stream_optimizer.c +++ b/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; diff --git a/src/stream/stream_optimizer.h b/src/stream/stream_optimizer.h index 526a9962..a3566049 100644 --- a/src/stream/stream_optimizer.h +++ b/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); diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index f6da521e..109e2e64 100644 --- a/tests/test_stream_service.c +++ b/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, ¶meters, &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);