From d6f935d7796228edf4694ecb6ec95b03657ffbe2 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Sun, 11 Oct 2026 01:03:39 +0200
Subject: [PATCH] Locate recoverable stream batches when skipping delivery gaps

---
 src/stream/stream_buffer.c | 12 ++++++++++++
 src/stream/stream_buffer.h |  1 +
 src/stream/stream_priv.h   |  2 ++
 3 files changed, 15 insertions(+)

diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c
index 0f1c738b..155fb4f6 100644
--- a/src/stream/stream_buffer.c
+++ b/src/stream/stream_buffer.c
@@ -341,6 +341,18 @@ int stream_buffer_range(struct stream_buffer *b, uint8_t subblock, uint64_t *fir
     return STREAM_OK;
 }
 
+/* При GAP потребитель сохраняет все ещё декодируемые пачки, даже если выбранные upstream уже далеко впереди. */
+int stream_buffer_next_decodable(struct stream_buffer *b, uint64_t seq, uint64_t *next) {
+    if (!b || !next) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer next decodable invalid arguments"); return STREAM_ERR_ARGUMENT; }
+    uint64_t found = UINT64_MAX;
+    for (uint32_t i = 0; i < b->stats.batches; i++) {
+        struct stream_cached_batch *batch = b->heap[i];
+        if (batch->seq >= seq && batch->seq < found && batch->count >= b->descriptor.n - b->descriptor.r) found = batch->seq;
+    }
+    if (found == UINT64_MAX) return STREAM_WAIT;
+    *next = found; return STREAM_OK;
+}
+
 /* RS пишет в собственный scratch; ошибка или недостающие блоки не повреждают выход потребителя. */
 int stream_buffer_decode(struct stream_buffer *b, uint64_t seq, uint8_t *output, size_t capacity, size_t *length) {
     if (!b || !output || !length || seq == UINT64_MAX) {
diff --git a/src/stream/stream_buffer.h b/src/stream/stream_buffer.h
index fcdf8808..80206c4c 100644
--- a/src/stream/stream_buffer.h
+++ b/src/stream/stream_buffer.h
@@ -40,6 +40,7 @@ int stream_buffer_put_path(struct stream_buffer *buffer, const uint8_t *wire, si
 int stream_buffer_path(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, struct stream_path *path);
 /* Найти ближайший имеющийся блок >= seq для скачка через GAP; WAIT при отсутствии. */
 int stream_buffer_next_seq(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, uint64_t *next);
+int stream_buffer_next_decodable(struct stream_buffer *buffer, uint64_t seq, uint64_t *next);
 
 /* seq=UINT64_MAX читает последний имеющийся блок именно запрошенного субпотока.
  * WAIT — блок впереди головы/пустое кольцо; GAP — дыра; EXPIRED — вытеснен.
diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h
index 27787fb6..da1eeb33 100644
--- a/src/stream/stream_priv.h
+++ b/src/stream/stream_priv.h
@@ -20,6 +20,8 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor
 int stream_handle_context(struct STREAM_HANDLE *handle, struct UTUN_INSTANCE **inst, struct stream_descriptor *descriptor);
 int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length, const struct stream_path *path);
 int stream_ingest_end(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, const uint8_t *wire, size_t length);
+void stream_local_progress(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t before);
+int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key);
 
 int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length);