From b31aa9f71d08ccfe6a7c33c556ca1b8f926e1c92 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 00:36:29 +0200 Subject: [PATCH] Retain stream block paths and locate gaps in ring delivery --- src/Makefile.am | 2 ++ src/stream/stream_buffer.c | 48 +++++++++++++++++++++++++++++++++++++- src/stream/stream_buffer.h | 6 +++++ 3 files changed, 55 insertions(+), 1 deletion(-) diff --git a/src/Makefile.am b/src/Makefile.am index 3daa5d61..0a8cab17 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -7,6 +7,8 @@ stream_core_files = \ stream/stream.h \ stream/stream.c \ stream/stream_priv.h \ + stream/stream_wire.h \ + stream/stream_wire.c \ stream/stream_member.h \ stream/stream_member.c \ stream/stream_proto.h \ diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c index 95698aaa..0f1c738b 100644 --- a/src/stream/stream_buffer.c +++ b/src/stream/stream_buffer.c @@ -9,6 +9,7 @@ struct stream_cached_slot { uint32_t next; uint8_t subblock; + struct stream_path *path; /* NULL — исходный путь {source}, без отдельной аллокации. */ }; struct stream_cached_batch { @@ -102,6 +103,7 @@ 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; + 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; b->free_slot = slot; @@ -169,6 +171,7 @@ void stream_buffer_destroy(struct stream_buffer *b) { (unsigned long long)b->descriptor.source_node_id, b->stats.blocks, b->stats.batches, b->stats.stored_bytes, (unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates); for (uint32_t i = 0; i < b->stats.batches; i++) u_free(b->heap[i]); + if (b->slots) for (uint32_t i = 0; i < b->capacity; i++) u_free(b->slots[i].path); rs_free(b->rs); u_free(b->decode_output); u_free(b->data); @@ -180,7 +183,17 @@ void stream_buffer_destroy(struct stream_buffer *b) { /* Вставить только подписанный блок; OOM/конфликт проверяются до вытеснения ранее сохранённых данных. */ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t length) { + struct stream_path path = {.count = 1}; + if (b) path.nodes[0] = b->descriptor.source_node_id; + return stream_buffer_put_path(b, wire, length, &path); +} + +/* Путь и память готовятся до вытеснения: отказ не повреждает окно кольца. */ +int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t length, const struct stream_path *path) { if (!b || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer put invalid arguments length=%zu", length); return STREAM_ERR_ARGUMENT; } + if (!path || !path->count || path->count > STREAM_PATH_MAX + || stream_path_validate(path, b->descriptor.source_node_id, path->nodes[path->count - 1], 0) != STREAM_OK) + return STREAM_ERR_FORMAT; struct stream_cached_batch *batch = NULL; if (length == b->block_bytes) { batch = buffer_batch(b, stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8)); @@ -214,12 +227,18 @@ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t lengt (unsigned long long)view.seq, view.subblock, (unsigned long long)b->heap[0]->seq); return STREAM_EXPIRED; } + struct stream_path *stored_path = path->count > 1 ? u_malloc(sizeof(*stored_path)) : NULL; + if (path->count > 1 && !stored_path) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path allocation failed seq=%llu", (unsigned long long)view.seq); + return STREAM_ERR_MEMORY; + } + if (stored_path) *stored_path = *path; int new_batch = batch == NULL; if (new_batch) { batch = u_calloc(1, sizeof(*batch)); if (!batch) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)view.seq); - return STREAM_ERR_MEMORY; + u_free(stored_path); return STREAM_ERR_MEMORY; } batch->seq = view.seq; batch->first_slot = STREAM_SLOT_NONE; @@ -232,6 +251,7 @@ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t lengt uint32_t slot = b->free_slot; b->free_slot = b->slots[slot].next; b->slots[slot].subblock = view.subblock; + b->slots[slot].path = stored_path; b->slots[slot].next = batch->first_slot; batch->first_slot = slot; batch->count++; @@ -248,6 +268,32 @@ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t lengt return STREAM_OK; } +/* Копия пути не держит указатели в переиспользуемом кольце. */ +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) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path invalid arguments"); return STREAM_ERR_ARGUMENT; + } + struct stream_cached_batch *batch = buffer_batch(b, seq); + uint32_t slot = batch ? buffer_slot(b, batch, subblock) : STREAM_SLOT_NONE; + if (slot == STREAM_SLOT_NONE) return seq < b->stats.expired_before ? STREAM_EXPIRED : STREAM_GAP; + *path = b->slots[slot].path ? *b->slots[slot].path : (struct stream_path){.nodes = {b->descriptor.source_node_id}, .count = 1}; + return STREAM_OK; +} + +/* Поиск используется только на дыре/вытеснении, обычная выдача читает точный индекс через hash. */ +int stream_buffer_next_seq(struct stream_buffer *b, uint64_t seq, uint8_t subblock, uint64_t *next) { + if (!b || !next || subblock >= b->descriptor.n) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer next 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 && buffer_slot(b, batch, subblock) != STREAM_SLOT_NONE) found = batch->seq; + } + if (found == UINT64_MAX) return STREAM_WAIT; + *next = found; return STREAM_OK; +} + /* Найти точную позицию либо последний блок субпотока, не меняя его историю/срок хранения. */ int stream_buffer_read(struct stream_buffer *b, uint64_t seq, uint8_t subblock, const uint8_t **wire, size_t *length, uint64_t *actual_seq) { diff --git a/src/stream/stream_buffer.h b/src/stream/stream_buffer.h index 47445208..fcdf8808 100644 --- a/src/stream/stream_buffer.h +++ b/src/stream/stream_buffer.h @@ -7,6 +7,7 @@ #define STREAM_BUFFER_H #include "stream_block.h" +#include "stream_wire.h" #ifdef __cplusplus extern "C" { @@ -34,6 +35,11 @@ void stream_buffer_destroy(struct stream_buffer *buffer); /* Новый блок проверяется до изменения кольца. Повтор неизменного блока: STREAM_DUPLICATE. * STREAM_EXPIRED не меняет более свежие данные; неверная подпись/конфликт возвращает отрицательный код. */ int stream_buffer_put(struct stream_buffer *buffer, const uint8_t *wire, size_t length); +/* Сохранённая история содержит текущего владельца кольца; первый путь повтора остаётся неизменным. */ +int stream_buffer_put_path(struct stream_buffer *buffer, const uint8_t *wire, size_t length, const struct stream_path *path); +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); /* seq=UINT64_MAX читает последний имеющийся блок именно запрошенного субпотока. * WAIT — блок впереди головы/пустое кольцо; GAP — дыра; EXPIRED — вытеснен.