From d83e85b8cab09ef645139590fe2cd8075b72c8ac Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 00:44:27 +0200 Subject: [PATCH] Report stream ring bounds and handle expired and final cursors --- src/stream/stream.c | 3 ++- src/stream/stream_network.c | 47 +++++++++++++++++++++++-------------- src/stream/stream_wire.h | 2 +- 3 files changed, 32 insertions(+), 20 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index 76f8c5d8..4d569155 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -513,7 +513,8 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l if (!s) { DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "unsubscribed stream ignored group=%llx source=%llx", (unsigned long long)group, (unsigned long long)source); return STREAM_WAIT; } if (s->source) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "incoming DATA to own publisher rejected"); return STREAM_ERR_STATE; } - if (!path || path->count >= STREAM_PATH_MAX || stream_path_validate(path, source, path->nodes[path->count - 1], inst->node_id) != STREAM_OK) + if (!path || !path->count || path->count >= STREAM_PATH_MAX + || stream_path_validate(path, source, path->nodes[path->count - 1], inst->node_id) != STREAM_OK) return STREAM_ERR_FORMAT; if (s->end_valid && (s->last_seq == UINT64_MAX || stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8) > s->last_seq)) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA beyond signed END source=%llx", (unsigned long long)source); diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 10f9dd56..2454feee 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -267,25 +267,27 @@ static int authorized(struct stream_network *net, uint64_t peer, const struct st /* INFO и ACCEPT описывают реальный последний/начальный блок с его историей, а не BGP-метрику. */ static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, uint64_t peer, uint64_t self, uint8_t *payload) { const uint8_t *wire; size_t length; uint64_t actual, first, last; struct stream_path path; - if (stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual) != STREAM_OK + int result = stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual); + if (result == STREAM_EXPIRED) return STREAM_REASON_EXPIRED; + if (result != STREAM_OK || stream_buffer_range(local->buffer, sub, &first, &last) != STREAM_OK || stream_buffer_path(local->buffer, actual, sub, &path) != STREAM_OK) return STREAM_REASON_POSITION; if (stream_path_validate(&path, local->descriptor.source_node_id, self, peer) != STREAM_OK || path.count == STREAM_PATH_MAX) return STREAM_REASON_LOOP; - stream_write_be(payload, actual, 8); stream_write_be(payload + 8, last, 8); - payload[16] = local->descriptor.n; payload[17] = local->descriptor.r; - stream_write_be(payload + 18, local->descriptor.block_size, 2); - /* byte 20 — число узлов, следует непосредственно за параметрами. */ - stream_path_write(payload + 20, &path); return 0; + stream_write_be(payload, actual, 8); stream_write_be(payload + 8, first, 8); stream_write_be(payload + 16, last, 8); + payload[24] = local->descriptor.n; payload[25] = local->descriptor.r; + stream_write_be(payload + 26, local->descriptor.block_size, 2); + stream_path_write(payload + 28, &path); return 0; } static int parse_info(const struct stream_message *m, const struct stream_descriptor *d, uint64_t peer, uint64_t self, uint64_t *start) { struct stream_path path; size_t used; - if (m->length < STREAM_INFO_SIZE || m->payload[16] != d->n || m->payload[17] != d->r - || stream_read_be(m->payload + 18, 2) != d->block_size - || stream_path_parse(m->payload + 20, m->length - 20, &path, &used) != STREAM_OK || m->length != 20 + used + if (m->length < STREAM_INFO_SIZE || m->payload[24] != d->n || m->payload[25] != d->r + || stream_read_be(m->payload + 26, 2) != d->block_size + || stream_path_parse(m->payload + 28, m->length - 28, &path, &used) != STREAM_OK || m->length != 28 + used || path.count == STREAM_PATH_MAX || stream_path_validate(&path, d->source_node_id, peer, self) != STREAM_OK - || stream_read_be(m->payload, 8) == UINT64_MAX || stream_read_be(m->payload, 8) > stream_read_be(m->payload + 8, 8)) { + || stream_read_be(m->payload, 8) == UINT64_MAX || stream_read_be(m->payload, 8) > stream_read_be(m->payload + 16, 8) + || stream_read_be(m->payload + 8, 8) > stream_read_be(m->payload + 16, 8)) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO/ACCEPT invalid command=%u bytes=%zu", m->command, m->length); return STREAM_ERR_FORMAT; } @@ -316,7 +318,8 @@ static int downstream_send(struct STREAM_RECEIVE *f) { if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 1); return 0; } struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock}; const uint8_t *wire; size_t length; uint64_t actual; - int result = stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual); + int result = f->info.next == UINT64_MAX ? STREAM_WAIT + : stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual); if (result == STREAM_OK) { struct stream_path path; if (stream_buffer_path(local.buffer, actual, f->subblock, &path) != STREAM_OK @@ -332,7 +335,11 @@ static int downstream_send(struct STREAM_RECEIVE *f) { } if (result == STREAM_EXPIRED || result == STREAM_GAP) { uint64_t next; - if (stream_buffer_next_seq(local.buffer, f->info.next, f->subblock, &next) == STREAM_OK && next > f->info.next) { + int available = stream_buffer_next_seq(local.buffer, f->info.next, f->subblock, &next) == STREAM_OK; + if (!available && local.finished && local.last != UINT64_MAX && f->info.next <= local.last) { + next = local.last + 1; available = 1; + } + if (available && next > f->info.next) { uint8_t payload[16]; stream_write_be(payload, f->info.next, 8); stream_write_be(payload + 8, next, 8); m.command = STREAM_CMD_GAP; if (raw_send(p, m, payload, sizeof(payload)) == STREAM_OK) { @@ -442,19 +449,22 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st } else if (f && !reason) { reason = info_payload(local, m.subblock, f->info.start, p->node_id, p->network->inst->node_id, payload); /* Идемпотентный ACCEPT возвращает первоначальный индекс даже если тот уже вытеснен. */ - if (reason == STREAM_REASON_POSITION) { + if (reason == STREAM_REASON_POSITION || reason == STREAM_REASON_EXPIRED) { reason = info_payload(local, m.subblock, UINT64_MAX, p->node_id, p->network->inst->node_id, payload); if (!reason) stream_write_be(payload, f->info.start, 8); } } if (reason) { - uint8_t value = (uint8_t)reason; m.command = STREAM_CMD_REJECT; control_put(p, m, &value, 1); + uint8_t rejection[17] = {(uint8_t)reason}; uint64_t first = UINT64_MAX, last = UINT64_MAX; + if (m.subblock < local->descriptor.n) stream_buffer_range(local->buffer, m.subblock, &first, &last); + stream_write_be(rejection + 1, first, 8); stream_write_be(rejection + 9, last, 8); + m.command = STREAM_CMD_REJECT; control_put(p, m, rejection, sizeof(rejection)); DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u reason=%d watermark=%llu", (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, reason, (unsigned long long)p->highwater); return; } f->deadline = get_time_tb() + STREAM_LEASE_TB; - m.command = STREAM_CMD_ACCEPT; control_put(p, m, payload, STREAM_INFO_SIZE + payload[20] * 8); + m.command = STREAM_CMD_ACCEPT; control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream ACCEPT queued peer=%llx handle=%llu sub=%u start=%llu lease=6000ms", (unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)f->info.start); } @@ -490,10 +500,10 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t int reason = info_payload(&local, m.subblock, UINT64_MAX, sender, net->inst->node_id, payload); m.command = reason ? STREAM_CMD_REJECT : STREAM_CMD_INFO_RESP; if (reason) { uint8_t value = (uint8_t)reason; control_put(p, m, &value, 1); } - else control_put(p, m, payload, STREAM_INFO_SIZE + payload[20] * 8); + else control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8); return; } - if (!p || !p->up || m.receiver_epoch != p->own_epoch + if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch || (p->other_epoch && m.sender_epoch != p->other_epoch)) { DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session packet peer=%llx command=%u handle=%llu", (unsigned long long)sender, m.command, (unsigned long long)m.handle); return; @@ -546,7 +556,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t else flow_close(f, STREAM_REASON_PROTOCOL, 1); } else if (m.command == STREAM_CMD_RENEW && f && !m.length) f->deadline = get_time_tb() + STREAM_LEASE_TB; else if (m.command == STREAM_CMD_UNSUBSCRIBE && f && m.length == 1) flow_close(f, m.payload[0], 1); - else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && m.length == 1) flow_close(f, m.payload[0], 0); + else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && (m.length == 1 || m.length == 17)) + flow_close(f, m.payload[0], 0); else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream unknown/inactive issuance peer=%llx command=%u handle=%llu sub=%u bytes=%zu", (unsigned long long)sender, m.command, (unsigned long long)m.handle, m.subblock, m.length); p->busy--; peer_sweep(p); diff --git a/src/stream/stream_wire.h b/src/stream/stream_wire.h index f4dad5ea..bbfdf391 100644 --- a/src/stream/stream_wire.h +++ b/src/stream/stream_wire.h @@ -9,7 +9,7 @@ #define STREAM_WIRE_HEADER 52 #define STREAM_PATH_MAX 32 #define STREAM_END_SIZE 106 -#define STREAM_INFO_SIZE 21 +#define STREAM_INFO_SIZE 29 /* actual:8, first:8, last:8, n:1, r:1, block_size:2, path_count:1 */ enum stream_command { STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT,