Retry pending supplier buffers without resetting error backoff

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++;

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);