Keep automatic stream recovery active until signed END is drained

src/stream/stream_optimizer.c

@@ -41,7 +41,7 @@ struct stream_optimizer {
     struct stream_selection_info info;
     struct stream_probe_schedule schedule;
     uint64_t roster_ms, progress_ms, next_seq;
-    int progress_valid;
+    int progress_valid, draining;
 };
 
 unsigned stream_probe_schedule_tick(struct stream_probe_schedule *s, uint64_t now, unsigned pending, unsigned active) {
@@ -294,8 +294,16 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
     struct stream_local local;
     if (stream_local_get(o->inst, &o->descriptor, &local) != STREAM_OK) return;
     if (local.finished) {
-        for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; }
-        o->info.probes_active = 0; return;
+        /* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */
+        if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) {
+            for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; }
+            if (o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer END drained source=%llx next=%llu last=%llu",
+                (unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)local.last);
+            o->draining = 0; o->info.probes_active = 0; return;
+        }
+        if (!o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer draining END source=%llx next=%llu last=%llu",
+            (unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)local.last);
+        o->draining = 1;
     }
     if (now >= o->roster_ms) roster_update(o, h, now);
     probes_poll(o, h, now); probes_schedule(o, h, now);