Publish actual part lengths and expose rejected batch status

src/stream/stream_buffer.h

@@ -1,5 +1,5 @@
-/* Кольцо проверенных блоков одного запуска: общий бюджет данных для всех субпотоков.
+/* Кольцо структурно проверенных частей одного запуска; подпись проверяется после восстановления.
- * Пустые субпотоки не резервируют блоки: 32 МБ занимают реально сохранённые кодограммы.
+ * Слоты резервируются по максимальному размеру, stored_bytes учитывает фактические длины.
  * Вытеснение целой старейшей пачки не закрепляет данные для медленного получателя.
  * Все вызовы — один uasync-поток. read возвращает память до следующего put/destroy.
  * public_key должен происходить из проверенной записи источника в группе. */
@@ -17,8 +17,10 @@ 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 ищет только старейший, без полного обхода. */
+/* Точное число/сумма байтов частей >= seq и время старейшего принятия.
-int stream_buffer_pending(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, uint64_t *count, uint64_t *oldest_ms);
+ * count=bytes=NULL ищет только старейший, без полного обхода. */
+int stream_buffer_pending(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock,
+                          uint64_t *count, uint64_t *bytes, uint64_t *oldest_ms);
 
 struct stream_buffer_stats {
     size_t capacity_bytes;
@@ -29,6 +31,7 @@ struct stream_buffer_stats {
     uint64_t highest_seq;
     uint64_t evicted_batches;
     uint64_t duplicates;
+    uint64_t rejected_batches;
 };
 
 /* out заранее NULL; bytes=0 выбирает 32 МБ, меньший бюджет допускается для стенда.
@@ -37,8 +40,9 @@ int stream_buffer_create(struct stream_buffer **out, const struct stream_descrip
                          const uint8_t public_key[SC_PUBKEY_SIZE], size_t bytes);
 void stream_buffer_destroy(struct stream_buffer *buffer);
 
-/* Новый блок проверяется до изменения кольца. Повтор неизменного блока: STREAM_DUPLICATE.
+/* Формат новой части проверяется до изменения кольца. Повтор неизменной части: STREAM_DUPLICATE.
- * STREAM_EXPIRED не меняет более свежие данные; неверная подпись/конфликт возвращает отрицательный код. */
+ * STREAM_EXPIRED не меняет более свежие данные; конфликт возвращает отрицательный код.
+ * Части ещё не подтверждены подписью источника и могут пересылаться до локального decode. */
 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);
@@ -54,8 +58,9 @@ int stream_buffer_read(struct stream_buffer *buffer, uint64_t seq, uint8_t subbl
                        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 >= действительной длины.
+/* Восстановить пачку и проверить одну подпись; достаточно capacity >= действительной длины.
- * Выход не меняется при WAIT/GAP/EXPIRED/ошибке; восстановленные отсутствующие блоки в кольцо не добавляются. */
+ * STREAM_REJECTED означает неверную подпись/подписанные метаданные; такая пачка повторно не декодируется.
+ * Выход не меняется при WAIT/GAP/EXPIRED/REJECTED/ошибке; отсутствующие части в кольцо не добавляются. */
 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);
 

src/stream/stream_source.c

@@ -48,10 +48,12 @@ static int source_flush(struct stream_source *source) {
     }
     uint8_t flags = source->has_batch && now_ms - source->last_batch_ms >= STREAM_TIMESTAMP_PERIOD_MS ? STREAM_BLOCK_TIME_RESET : 0;
     const uint8_t *blocks[RS_MAX_BLOCKS];
-    int result = stream_encoder_encode(source->encoder, source->next_seq, (uint16_t)now_ms, flags, source->input, source->used, blocks);
+    size_t block_length = 0;
+    int result = stream_encoder_encode(source->encoder, source->next_seq, (uint16_t)now_ms, flags,
+                                       source->input, source->used, blocks, &block_length);
     if (result != STREAM_OK) return source_fail(source, result);
     for (unsigned i = 0; i < source->descriptor.n; i++) {
-        result = stream_buffer_put(source->buffer, blocks[i], stream_block_size(&source->descriptor));
+        result = stream_buffer_put(source->buffer, blocks[i], block_length);
         if (result != STREAM_OK) return source_fail(source, result < 0 ? result : STREAM_ERR_STATE);
     }
     uint64_t seq = source->next_seq++;
@@ -60,9 +62,10 @@ static int source_flush(struct stream_source *source) {
     source->has_batch = 1;
     source->last_batch_ms = now_ms;
     if (!source->report_ms || now_ms - source->report_ms >= 1000) {
-        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source progress source=%llx seq=%llu bytes=%llu timestamp=%u flags=%u",
+        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source progress source=%llx seq=%llu bytes=%llu timestamp=%u flags=%u part=%zu max=%zu",
                     (unsigned long long)source->descriptor.source_node_id, (unsigned long long)seq,
-                    (unsigned long long)source->emitted_bytes, (uint16_t)now_ms, flags);
+                    (unsigned long long)source->emitted_bytes, (uint16_t)now_ms, flags,
+                    block_length, stream_block_capacity(&source->descriptor));
         source->report_ms = now_ms;
     }
     if (source->ready) source->ready(source, seq, source->arg);