/* Кольцо проверенных блоков одного запуска: общий бюджет данных для всех субпотоков. * Пустые субпотоки не резервируют блоки: 32 МБ занимают реально сохранённые кодограммы. * Вытеснение целой старейшей пачки не закрепляет данные для медленного получателя. * Все вызовы — один uasync-поток. read возвращает память до следующего put/destroy. * public_key должен происходить из проверенной записи источника в группе. */ #ifndef STREAM_BUFFER_H #define STREAM_BUFFER_H #include "stream_block.h" #include "stream_wire.h" #ifdef __cplusplus 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; size_t stored_bytes; uint32_t blocks; uint32_t batches; uint64_t expired_before; uint64_t highest_seq; uint64_t evicted_batches; uint64_t duplicates; }; /* out заранее NULL; bytes=0 выбирает 32 МБ, меньший бюджет допускается для стенда. * Должна помещаться хотя бы одна полная пачка n блоков; верхний предел — STREAM_BUFFER_BYTES. */ int stream_buffer_create(struct stream_buffer **out, const struct stream_descriptor *descriptor, const uint8_t public_key[SC_PUBKEY_SIZE], size_t bytes); 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); int stream_buffer_next_decodable(struct stream_buffer *buffer, uint64_t seq, uint64_t *next); /* seq=UINT64_MAX читает последний имеющийся блок именно запрошенного субпотока. * WAIT — блок впереди головы/пустое кольцо; GAP — дыра; EXPIRED — вытеснен. * При отсутствии блока выходные указатели/номер не меняются. */ int stream_buffer_read(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, const uint8_t **wire, size_t *length, uint64_t *actual_seq); int stream_buffer_range(struct stream_buffer *buffer, uint8_t subblock, uint64_t *first, uint64_t *last); /* Восстановить пачку из проверенных блоков; достаточно capacity >= действительной длины. * Выход не меняется при WAIT/GAP/EXPIRED/ошибке; восстановленные отсутствующие блоки в кольцо не добавляются. */ int stream_buffer_decode(struct stream_buffer *buffer, uint64_t seq, uint8_t *output, size_t capacity, size_t *length); int stream_buffer_get_stats(const struct stream_buffer *buffer, struct stream_buffer_stats *stats); #ifdef __cplusplus } #endif #endif