Track signed timestamp progress in each local stream subflow

src/stream/stream.c

@@ -19,6 +19,7 @@ struct stream_session {
     struct stream_source *source;
     struct stream_buffer *buffer;
     uint8_t *output;
+    struct stream_metrics *metrics;
     void *delivery;
     uint64_t next_seq;
     uint64_t start_seq, last_seq, passed_before;
@@ -50,6 +51,19 @@ struct stream_service {
 static void session_deliver(void *arg);
 static void service_reconcile(void *arg);
 
+/* Observer выполняет только учёт; пользовательские callbacks и отправка остаются отложенными. */
+static void cache_event(void *arg, const uint8_t *wire, size_t length, int added) {
+    struct stream_session *s = arg;
+    if (s->closing || s->invalidated) return;
+    uint8_t sub = wire[STREAM_BLOCK_SUB_OFF]; uint64_t seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
+    if (added) {
+        stream_metrics_observe(&s->metrics[sub], seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
+                               wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
+        if (s->end_valid && seq == s->last_seq) stream_metrics_end(&s->metrics[sub]);
+    }
+    stream_network_cache_event(s->service->network, &s->descriptor, wire, length, added);
+}
+
 /* Активность CHAT проверяется до чтения профиля, выделения 32 МБ и создания handles. */
 static struct TOPO_GROUP *stream_group(struct UTUN_INSTANCE *inst, uint64_t id) {
     struct TOPO_GROUP *group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, id) : NULL;
@@ -105,9 +119,11 @@ static void session_free(struct stream_session *s) {
     while (*link != s) link = &(*link)->next;
     *link = s->next;
     if (s->delivery) uasync_call_soon_cancel(s->service->inst->ua, s->delivery);
+    if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL);
     if (s->source) stream_source_destroy(s->source);
     else stream_buffer_destroy(s->buffer);
     u_free(s->output);
+    u_free(s->metrics);
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream released group=%llx source=%llx epoch=%llx reason=%d",
                (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
                (unsigned long long)s->descriptor.source_epoch, s->reason);
@@ -303,6 +319,8 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) {
     for (struct stream_session *s = service->sessions; s; s = s->next)
         if (s->holds) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream stop during active operation rejected"); return; }
     service->stopping = 1;
+    for (struct stream_session *s = service->sessions; s; s = s->next)
+        if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL);
     stream_network_destroy(service->network); service->network = NULL;
     if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service);
     if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile);
@@ -351,7 +369,8 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
         if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session allocation failed"); u_free(h); return STREAM_ERR_MEMORY; }
         s->service = service; s->descriptor = *d; memcpy(s->public_key, key, 32);
         s->output = u_malloc((uint32_t)stream_input_size(d));
-        int result = s->output ? STREAM_OK : STREAM_ERR_MEMORY;
+        s->metrics = u_calloc(d->n, sizeof(*s->metrics));
+        int result = s->output && s->metrics ? STREAM_OK : STREAM_ERR_MEMORY;
         if (result == STREAM_OK && publisher) {
             result = stream_source_create(&s->source, inst->ua, d, inst->my_ed25519_privkey, source_ready, s);
             if (result == STREAM_OK) s->buffer = stream_source_buffer(s->source);
@@ -361,6 +380,8 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
             DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session initialization failed result=%d", result);
             session_free(s); u_free(h); return result;
         }
+        for (unsigned i = 0; i < d->n; i++) stream_metrics_init(&s->metrics[i]);
+        stream_buffer_set_observer(s->buffer, cache_event, s);
     }
     h->service = service; h->session = s; h->publisher = publisher;
     h->next = service->handles; service->handles = h; s->owners++;
@@ -496,6 +517,7 @@ int stream_publish_finish(struct STREAM_HANDLE *h) {
         s->last_seq = stats.blocks ? stats.highest_seq : UINT64_MAX;
         result = stream_end_sign(&s->descriptor, s->last_seq, s->service->inst->my_ed25519_privkey, s->end);
         s->end_valid = result == STREAM_OK;
+        if (s->end_valid) for (unsigned i = 0; i < s->descriptor.n; i++) stream_metrics_end(&s->metrics[i]);
     }
     if (result != STREAM_OK) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
     else { session_schedule(s); stream_network_changed(s->service->network); }
@@ -572,9 +594,20 @@ int stream_ingest_end(struct UTUN_INSTANCE *inst, const struct stream_descriptor
         return STREAM_ERR_CONFLICT;
     }
     memcpy(s->end, wire, STREAM_END_SIZE); s->end_valid = s->finished = 1; s->last_seq = last;
+    for (unsigned i = 0; i < s->descriptor.n; i++)
+        if (last == UINT64_MAX || (s->metrics[i].initialized && s->metrics[i].last_seq == last)) stream_metrics_end(&s->metrics[i]);
     session_schedule(s); stream_network_changed(inst->stream->network); return STREAM_OK;
 }
 
+/* Общая метрика прихода субпотока в экземпляр, независимо от поставщика конкретного блока. */
+int stream_local_metrics(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint8_t sub, uint64_t now,
+                         struct stream_metric_snapshot *snapshot) {
+    struct stream_local local;
+    if (stream_local_get(inst, key, &local) != STREAM_OK || sub >= local.descriptor.n || !snapshot) return STREAM_WAIT;
+    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
+    return stream_metrics_poll(&s->metrics[sub], now, snapshot);
+}
+
 /* Все выбранные upstream прошли эту границу; дальше старую неполную пачку они уже не выдадут. */
 void stream_local_progress(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t before) {
     struct stream_local local;