From f2c5187ef184e6c6dd8bf0cf1f1b1c5d79dc1589 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 18:06:58 +0200 Subject: [PATCH] Retry pending supplier buffers without resetting error backoff --- src/stream/stream_optimizer.c | 19 ++++++++++++++++++- tests/test_stream_service.c | 10 +++++++++- 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 65f7a959..687a9461 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -14,6 +14,8 @@ #define STREAM_OFFER_TTL_MS 60000 #define STREAM_MIGRATION_TIMEOUT_MS 5000 #define STREAM_BOOTSTRAP_RELAY_MS 100 +#define STREAM_AVAILABILITY_RETRY_MS 100 +#define STREAM_AVAILABILITY_WAIT_MS 5000 #define STREAM_PROBLEM_PENALTY_MS 1000 #define STREAM_HOP_PENALTY_MS 3 @@ -26,7 +28,7 @@ struct stream_offer { }; struct stream_supplier { struct stream_supplier *next; - uint64_t node, cycle, retry_ms; + uint64_t node, cycle, retry_ms, availability_retry_ms, availability_until_ms; uint64_t local_epoch, peer_epoch; struct STREAM_RECEIVE *probe; struct stream_offer *offers; @@ -137,6 +139,15 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin if (!info.info_ready && info.state != STREAM_RECEIVE_CLOSED) continue; if (info.state == STREAM_RECEIVE_CLOSED) { s->offers[s->probe_sub].info.info_ms = 0; + if (info.reason == STREAM_REASON_POSITION) { + if (!s->availability_until_ms) s->availability_until_ms = now + STREAM_AVAILABILITY_WAIT_MS; + if (now < s->availability_until_ms && !s->availability_retry_ms) { + s->availability_retry_ms = now + STREAM_AVAILABILITY_RETRY_MS; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream supplier buffer pending source=%llx peer=%llx sub=%u retry_ms=%u", + (unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, + s->probe_sub, STREAM_AVAILABILITY_RETRY_MS); + } + } if (info.reason == STREAM_REASON_TRANSPORT || info.reason == STREAM_REASON_MEMBER || info.reason == STREAM_REASON_PROTOCOL) { s->failures++; s->retry_ms = now + backoff(s->failures); DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer probe closed source=%llx peer=%llx sub=%u reason=%d" @@ -174,6 +185,12 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin static void probes_schedule(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) { unsigned pending = 0, active = 0; for (struct stream_supplier *s = o->suppliers; s; s = s->next) { + if (!s->probe && s->availability_retry_ms && now >= s->availability_retry_ms) { + s->availability_retry_ms = 0; + if (s->compatible && now < s->availability_until_ms && now >= s->retry_ms) { + s->cycle = 0; o->schedule.next_ms = 0; + } + } if (s->probe) active++; if (now < s->retry_ms) s->cycle = o->schedule.cycle; if (s->compatible && s->cycle != o->schedule.cycle) pending++; diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 109e2e64..ef1e7a1a 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -1191,6 +1191,14 @@ static void test_supernode_relay(void) { 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"); + test_clock_run_for(ua, 2000); + struct stream_selection_info stable_before, stable_after; + CHECK(stream_selection_get_info(viewers[2], &stable_before) == STREAM_OK); + for (unsigned version = 101; version <= 103; version++) relay_role(1, 1, version); + test_clock_run_for(ua, 2000); + CHECK(stream_selection_get_info(viewers[2], &stable_after) == STREAM_OK); + CHECK(stable_after.probes_started == stable_before.probes_started); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] repeated provider property callbacks preserve probes and their cycle"); stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 40000); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1; @@ -1296,7 +1304,7 @@ static void test_supernode_relay(void) { test_clock_run_for(ua, 50000); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); stream_close(viewers[2]); stream_close(publisher); - relay_drop_source = 0; relay_role(1, 0, 101); + relay_drop_source = 0; relay_role(1, 0, 104); 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);