From 0e9478d03349f3c3763ce58b83f141647ab0d803 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 20:08:07 +0200 Subject: [PATCH] Account variable stream parts and continue after rejected batches --- src/stream/stream.c | 6 ++++++ src/stream/stream_network.c | 17 +++++++++++------ src/stream/stream_wire.h | 2 +- 3 files changed, 18 insertions(+), 7 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index e4d389e0..2180426e 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -306,6 +306,12 @@ static void session_deliver(void *arg) { } event.type = STREAM_EVENT_DATA; event.first_seq = event.last_seq = s->next_seq++; event.data = s->output; event.length = length; emit = 1; again = 1; + } else if (result == STREAM_REJECTED) { + event.type = STREAM_EVENT_GAP; event.first_seq = event.last_seq = s->next_seq++; + emit = again = 1; + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream rejected batch GAP source=%llx epoch=%llx seq=%llu; receiving continues", + (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch, + (unsigned long long)event.first_seq); } else if (result == STREAM_EXPIRED) { event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = stats.expired_before - 1; s->next_seq = stats.expired_before; emit = 1; again = 1; diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index d35ba9a9..7a030462 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -131,7 +131,7 @@ static uint64_t flow_age(struct STREAM_RECEIVE *f, uint64_t now) { if (f->outgoing || !f->load.queued.packets) return 0; struct stream_local local; uint64_t oldest = 0; if (stream_local_get(f->peer->network->inst, &f->descriptor, &local) != STREAM_OK) return 0; - if (stream_buffer_pending(local.buffer, f->info.next, f->subblock, NULL, &oldest) != STREAM_OK) return 0; + if (stream_buffer_pending(local.buffer, f->info.next, f->subblock, NULL, NULL, &oldest) != STREAM_OK) return 0; if (oldest < f->admitted_ms) oldest = f->admitted_ms; return now > oldest ? now - oldest : 0; } @@ -736,9 +736,9 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st f->info.handle = m.handle; f->requested = requested; f->info.start = f->info.next = stream_read_be(payload + 8, 8); if (f->info.start == UINT64_MAX) { f->info.start = f->info.next = 0; stream_write_be(payload + 8, 0, 8); } f->info.state = STREAM_RECEIVE_ACTIVE; - uint64_t count = 0; - stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL); - load_account(f, count, count * stream_block_size(&f->descriptor), 1, 0); + uint64_t count = 0, bytes = 0; + stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, &bytes, NULL); + load_account(f, count, bytes, 1, 0); f->pending_highest = stream_read_be(payload + 24, 8); if (f->pending_highest == UINT64_MAX) f->pending_highest = 0; const uint8_t *wire; size_t length; uint64_t seq; @@ -886,9 +886,14 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t } } else if (m.command == STREAM_CMD_DATA && f && f->info.state == STREAM_RECEIVE_ACTIVE) { struct stream_path path; size_t used; - if (stream_path_parse(m.payload, m.length, &path, &used) != STREAM_OK || m.length - used != stream_block_size(&f->descriptor) + if (stream_path_parse(m.payload, m.length, &path, &used) != STREAM_OK || stream_path_validate(&path, f->descriptor.source_node_id, sender, net->inst->node_id) != STREAM_OK) { flow_close(f, STREAM_REASON_LOOP, 1); + } else if (m.length - used <= STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE + || m.length - used > stream_block_capacity(&f->descriptor)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA part size invalid peer=%llx bytes=%zu max=%zu", + (unsigned long long)sender, m.length - used, stream_block_capacity(&f->descriptor)); + flow_close(f, STREAM_REASON_PROTOCOL, 1); } else { const uint8_t *block = m.payload + used; uint64_t seq = stream_read_be(block + STREAM_BLOCK_SEQ_OFF, 8); if (block[STREAM_BLOCK_SUB_OFF] != f->subblock || seq != f->info.next @@ -899,7 +904,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t int result = stream_ingest_path(net->inst, block, m.length - used, &path); if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) { stream_metrics_observe(&f->metrics, seq, (uint16_t)stream_read_be(block + STREAM_BLOCK_TIMESTAMP_OFF, 2), - block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)stream_block_size(&f->descriptor), get_time_tb() / 10); + block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)(m.length - used), get_time_tb() / 10); if (f->info.path.count != path.count || memcmp(f->info.path.nodes, path.nodes, path.count * sizeof(uint64_t))) f->info.path_rtt_valid = 0; f->info.next = seq + 1; f->info.blocks++; f->info.path = path; f->deadline = get_time_tb() + STREAM_LEASE_TB; diff --git a/src/stream/stream_wire.h b/src/stream/stream_wire.h index 1c3a550b..347362b1 100644 --- a/src/stream/stream_wire.h +++ b/src/stream/stream_wire.h @@ -10,7 +10,7 @@ extern "C" { #endif -#define STREAM_WIRE_VERSION 4 +#define STREAM_WIRE_VERSION 5 #define STREAM_WIRE_HEADER 52 #define STREAM_PATH_MAX 32 #define STREAM_END_SIZE 106