From a877ba8a732dd7caec3ceaca27b7f82132223ac2 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 17:33:29 +0200 Subject: [PATCH] Keep automatic stream recovery active until signed END is drained --- src/stream/stream_optimizer.c | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 70b029c7..c6ecd865 100644 --- a/src/stream/stream_optimizer.c +++ b/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);