From 921340ae60e1d1f0628f06f996b9f9d23940ee19 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 02:34:55 +0200 Subject: [PATCH] Track signed timestamp progress in each local stream subflow --- src/stream/stream.c | 35 ++++++++++++++++++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index 48402b0e..95187388 100644 --- a/src/stream/stream.c +++ b/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;