From fec1916a58fdbb2f2e93786106bb28ce7f1699f3 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Sun, 11 Oct 2026 02:33:56 +0200
Subject: [PATCH] Observe stream cache arrivals and exact pending block age

---
 src/stream/stream_buffer.c | 39 ++++++++++++++++++++++++++++++++++++++
 src/stream/stream_buffer.h |  5 +++++
 src/stream/stream_priv.h   |  2 ++
 3 files changed, 46 insertions(+)

diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c
index 3e88a740..f6efae3a 100644
--- a/src/stream/stream_buffer.c
+++ b/src/stream/stream_buffer.c
@@ -2,6 +2,7 @@
 
 #include "../../lib/debug_config.h"
 #include "../../lib/mem.h"
+#include "../../lib/u_async.h"
 #include "stream_buffer.h"
 
 #define STREAM_SLOT_NONE UINT32_MAX
@@ -12,6 +13,8 @@ struct stream_cached_slot {
     uint32_t next;
     uint8_t subblock;
     struct stream_cached_path *path; /* NULL — исходный путь {source}, без отдельной аллокации. */
+    uint32_t sub_next, sub_previous;
+    uint64_t seq, received_ms;
 };
 
 struct stream_cached_batch {
@@ -40,6 +43,9 @@ struct stream_buffer {
     uint32_t hash_size;
     uint32_t free_slot;
     struct stream_buffer_stats stats;
+    uint32_t sub_head[RS_MAX_BLOCKS], sub_tail[RS_MAX_BLOCKS];
+    stream_buffer_event_fn observer;
+    void *observer_arg;
 };
 
 /* Индекс пачки не зависит от порядка поступления; множество дырок не требует линейного поиска. */
@@ -105,6 +111,10 @@ static void buffer_evict_oldest(struct stream_buffer *b) {
     for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE;) {
         uint32_t next = b->slots[slot].next;
         unsigned sub = b->slots[slot].subblock;
+        uint32_t before = b->slots[slot].sub_previous, after = b->slots[slot].sub_next;
+        if (before == STREAM_SLOT_NONE) b->sub_head[sub] = after; else b->slots[before].sub_next = after;
+        if (after == STREAM_SLOT_NONE) b->sub_tail[sub] = before; else b->slots[after].sub_previous = before;
+        if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, b->block_bytes, 0);
         u_free(b->slots[slot].path); b->slots[slot].path = NULL;
         if (b->latest_valid[sub] && b->latest[sub] == batch->seq) b->latest_valid[sub] = 0;
         b->slots[slot].next = b->free_slot;
@@ -154,6 +164,7 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip
     }
     for (uint32_t i = 0; i < b->capacity; i++) b->slots[i].next = i + 1;
     b->slots[b->capacity - 1].next = STREAM_SLOT_NONE;
+    for (unsigned i = 0; i < RS_MAX_BLOCKS; i++) b->sub_head[i] = b->sub_tail[i] = STREAM_SLOT_NONE;
     if (rs_init(&b->rs, d->n, d->r, d->block_size) != RS_OK) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode RS initialization failed n=%u r=%u", d->n, d->r);
         stream_buffer_destroy(b);
@@ -254,6 +265,11 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t
     b->free_slot = b->slots[slot].next;
     b->slots[slot].subblock = view.subblock;
     b->slots[slot].path = stored_path;
+    b->slots[slot].seq = view.seq; b->slots[slot].received_ms = get_time_tb() / 10;
+    b->slots[slot].sub_next = STREAM_SLOT_NONE; b->slots[slot].sub_previous = b->sub_tail[view.subblock];
+    if (b->sub_tail[view.subblock] == STREAM_SLOT_NONE) b->sub_head[view.subblock] = slot;
+    else b->slots[b->sub_tail[view.subblock]].sub_next = slot;
+    b->sub_tail[view.subblock] = slot;
     b->slots[slot].next = batch->first_slot;
     batch->first_slot = slot;
     batch->count++;
@@ -267,9 +283,32 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t
     }
     DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer stored seq=%llu sub=%u present=%u/%u bytes=%zu",
                 (unsigned long long)view.seq, view.subblock, batch->count, b->descriptor.n, b->stats.stored_bytes);
+    if (b->observer) b->observer(b->observer_arg, b->data + slot * b->block_bytes, length, 1);
     return STREAM_OK;
 }
 
+void stream_buffer_set_observer(struct stream_buffer *b, stream_buffer_event_fn observer, void *arg) {
+    if (!b) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer observer null buffer"); return; }
+    b->observer = observer; b->observer_arg = arg;
+}
+
+/* Список субпотока упорядочен по монотонному времени вставки, включая поступившие вне порядка индексы. */
+int stream_buffer_pending(struct stream_buffer *b, uint64_t seq, uint8_t sub, uint64_t *count, uint64_t *oldest) {
+    if (!b || sub >= b->descriptor.n || (!count && !oldest)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer pending invalid arguments sub=%u", sub); return STREAM_ERR_ARGUMENT;
+    }
+    uint64_t found = 0, first_ms = 0;
+    for (uint32_t slot = b->sub_head[sub]; slot != STREAM_SLOT_NONE; slot = b->slots[slot].sub_next) {
+        if (b->slots[slot].seq < seq) continue;
+        if (!found) first_ms = b->slots[slot].received_ms;
+        found++;
+        if (!count) break;
+    }
+    if (count) *count = found;
+    if (oldest) *oldest = first_ms;
+    return found ? STREAM_OK : STREAM_WAIT;
+}
+
 /* Копия пути не держит указатели в переиспользуемом кольце. */
 int stream_buffer_path(struct stream_buffer *b, uint64_t seq, uint8_t subblock, struct stream_path *path) {
     if (!b || !path || subblock >= b->descriptor.n || seq == UINT64_MAX) {
diff --git a/src/stream/stream_buffer.h b/src/stream/stream_buffer.h
index 80206c4c..1247f0ad 100644
--- a/src/stream/stream_buffer.h
+++ b/src/stream/stream_buffer.h
@@ -14,6 +14,11 @@ extern "C" {
 #endif
 
 struct stream_buffer;
+/* Observer не выполняет I/O и не меняет кольцо; wire действует лишь внутри callback. Destroy observer не вызывает. */
+typedef void (*stream_buffer_event_fn)(void *arg, const uint8_t *wire, size_t length, int added);
+void stream_buffer_set_observer(struct stream_buffer *buffer, stream_buffer_event_fn observer, void *arg);
+/* Точные ожидающие блоки >= seq и время старейшего принятия. count=NULL ищет только старейший, без полного обхода. */
+int stream_buffer_pending(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, uint64_t *count, uint64_t *oldest_ms);
 
 struct stream_buffer_stats {
     size_t capacity_bytes;
diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h
index da1eeb33..1ee0131f 100644
--- a/src/stream/stream_priv.h
+++ b/src/stream/stream_priv.h
@@ -22,6 +22,8 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l
 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_local_metrics(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint8_t sub, uint64_t now_ms,
+                         struct stream_metric_snapshot *snapshot);
 
 int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length);