diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c index 155fb4f6..3e88a740 100644 --- a/src/stream/stream_buffer.c +++ b/src/stream/stream_buffer.c @@ -6,10 +6,12 @@ #define STREAM_SLOT_NONE UINT32_MAX +struct stream_cached_path { uint8_t count; uint64_t nodes[]; }; + struct stream_cached_slot { uint32_t next; uint8_t subblock; - struct stream_path *path; /* NULL — исходный путь {source}, без отдельной аллокации. */ + struct stream_cached_path *path; /* NULL — исходный путь {source}, без отдельной аллокации. */ }; struct stream_cached_batch { @@ -227,12 +229,12 @@ int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t (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; + struct stream_cached_path *stored_path = path->count > 1 ? u_malloc(sizeof(*stored_path) + path->count * 8u) : 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; + if (stored_path) { stored_path->count = path->count; memcpy(stored_path->nodes, path->nodes, path->count * 8u); } int new_batch = batch == NULL; if (new_batch) { batch = u_calloc(1, sizeof(*batch)); @@ -276,7 +278,11 @@ int stream_buffer_path(struct stream_buffer *b, uint64_t seq, uint8_t subblock, 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}; + *path = (struct stream_path){.nodes = {b->descriptor.source_node_id}, .count = 1}; + if (b->slots[slot].path) { + path->count = b->slots[slot].path->count; + memcpy(path->nodes, b->slots[slot].path->nodes, path->count * 8u); + } return STREAM_OK; } diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 576190d2..e962033d 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -143,6 +143,10 @@ static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) { else if (!f->outgoing) route_control(p, m, &payload, 1); } f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason; + if (reason != STREAM_REASON_STOPPED) + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream issuance failure peer=%llx handle=%llu sub=%u reason=%d next=%llu", + (unsigned long long)f->info.peer, (unsigned long long)f->info.handle, f->subblock, + reason, (unsigned long long)f->info.next); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream issuance closed peer=%llx group=%llx source=%llx handle=%llu sub=%u out=%d reason=%d blocks=%llu", (unsigned long long)f->info.peer, (unsigned long long)f->descriptor.group_id, (unsigned long long)f->descriptor.source_node_id, (unsigned long long)f->info.handle, f->subblock, @@ -492,7 +496,11 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t } struct stream_peer *p = peer_find(net, sender); if (routed && m.command == STREAM_CMD_REVERSE_REQ) { - if (m.length || m.handle || m.receiver_epoch || !authorized(net, sender, &m, &local)) return; + if (m.length || m.handle || m.receiver_epoch) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream REVERSE_REQ malformed peer=%llx bytes=%zu handle=%llu receiver_epoch=%llx", + (unsigned long long)sender, m.length, (unsigned long long)m.handle, (unsigned long long)m.receiver_epoch); return; + } + if (!authorized(net, sender, &m, &local)) return; if (!p) p = peer_open(net, sender); if (p) { p->idle = get_time_tb(); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream REVERSE_REQ peer=%llx attempt=%llx NCD=%p", (unsigned long long)sender, (unsigned long long)m.sender_epoch, (void *)p->ncd); } @@ -502,7 +510,11 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed DATA/control forbidden command=%u", m.command); return; } if (m.command == STREAM_CMD_INFO_REQ && !routed) { - if (m.handle || m.length || m.subblock >= local.descriptor.n || !authorized(net, sender, &m, &local)) return; + if (m.handle || m.length || m.subblock >= local.descriptor.n) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO_REQ malformed peer=%llx bytes=%zu handle=%llu sub=%u n=%u", + (unsigned long long)sender, m.length, (unsigned long long)m.handle, m.subblock, local.descriptor.n); return; + } + if (!authorized(net, sender, &m, &local)) return; if (!p) p = peer_open(net, sender); if (!p || (m.receiver_epoch && m.receiver_epoch != p->own_epoch)) return; if (p->other_epoch && p->other_epoch != m.sender_epoch) { diff --git a/tests/test_stream_core.c b/tests/test_stream_core.c index aa085049..fb548983 100644 --- a/tests/test_stream_core.c +++ b/tests/test_stream_core.c @@ -78,6 +78,34 @@ static void test_signed_blocks(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] signed blocks: canonical bytes, all field classes, wrong key and malformed sizes"); } +/* Wire сохраняет все 64 бита и не меняет выход при усечении; путь не допускает повторного посещения. */ +static void test_wire(void) { + uint8_t wire[STREAM_WIRE_HEADER + 8]; + struct stream_message input = {.descriptor = descriptor, .sender_epoch = UINT64_MAX, .receiver_epoch = UINT64_MAX - 1, + .handle = UINT64_MAX - 2, .command = STREAM_CMD_SUBSCRIBE, .subblock = 7}; + stream_wire_header(wire, &input); stream_write_be(wire + STREAM_WIRE_HEADER, UINT64_MAX - 3, 8); + struct stream_message parsed; + CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_OK); + CHECK(parsed.sender_epoch == input.sender_epoch && parsed.receiver_epoch == input.receiver_epoch && parsed.handle == input.handle); + CHECK(parsed.descriptor.group_id == descriptor.group_id && parsed.descriptor.source_epoch == descriptor.source_epoch); + CHECK(parsed.subblock == 7 && parsed.length == 8 && stream_read_be(parsed.payload, 8) == UINT64_MAX - 3); + for (size_t i = 0; i < STREAM_WIRE_HEADER; i++) { + memset(&parsed, 0xa5, sizeof(parsed)); uint8_t before[sizeof(parsed)]; memcpy(before, &parsed, sizeof(parsed)); + CHECK(stream_wire_parse(wire, i, &parsed) == STREAM_ERR_FORMAT && !memcmp(before, &parsed, sizeof(parsed))); + } + wire[1]++; CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_ERR_FORMAT); wire[1]--; + wire[2] = 255; CHECK(stream_wire_parse(wire, sizeof(wire), &parsed) == STREAM_ERR_FORMAT); + struct stream_path path = {.nodes = {descriptor.source_node_id, 1, 2}, .count = 3}, read; + uint8_t encoded[1 + STREAM_PATH_MAX * 8]; size_t used = 0, length = stream_path_write(encoded, &path); + CHECK(stream_path_parse(encoded, length, &read, &used) == STREAM_OK && used == length && read.count == 3 && read.nodes[2] == 2); + CHECK(stream_path_validate(&read, descriptor.source_node_id, 2, 3) == STREAM_OK); + CHECK(stream_path_validate(&read, descriptor.source_node_id, 2, 1) == STREAM_ERR_FORMAT); + read.nodes[2] = 1; CHECK(stream_path_validate(&read, descriptor.source_node_id, 1, 0) == STREAM_ERR_FORMAT); + for (size_t i = 0; i < length; i++) CHECK(stream_path_parse(encoded, i, &read, &used) == STREAM_ERR_FORMAT); + encoded[0] = STREAM_PATH_MAX + 1; CHECK(stream_path_parse(encoded, sizeof(encoded), &read, &used) == STREAM_ERR_FORMAT); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream wire: 64-bit fields, truncation, version, path repeats/receiver loop and size"); +} + /* Любая пара потерь при 6+2 восстанавливается из подписанных блоков, отсутствующие блоки не регенерируются. */ static void test_fec_and_padding(void) { struct stream_encoder *encoder = NULL; @@ -317,6 +345,7 @@ int main(void) { for (unsigned i = 0; i < sizeof(private_key); i++) private_key[i] = (uint8_t)(i * 37 + 11); size_t allocations = u_get_allocated_count(); test_signed_blocks(); CHECK(u_get_allocated_count() == allocations); + test_wire(); CHECK(u_get_allocated_count() == allocations); test_fec_and_padding(); CHECK(u_get_allocated_count() == allocations); test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations); test_relative_metrics(); CHECK(u_get_allocated_count() == allocations);