From ccc9210232768c12a4043856d774041af2b391a5 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 17:20:36 +0200 Subject: [PATCH] Retain selected relay issuance before delayed READY and ACCEPT --- src/stream/stream_network.c | 15 +++++++------- tests/test_stream_service.c | 40 +++++++++++++++++++++++++++++-------- 2 files changed, 40 insertions(+), 15 deletions(-) diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index c457a537..d35ba9a9 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -1241,13 +1241,14 @@ static struct stream_relay_request *relay_request_new(struct stream_network *net return r; } -/* Наличие транспортных handles/INFO/ENDED не является действующей раздачей. */ -static unsigned relay_active(struct stream_network *net, const struct stream_descriptor *key, uint64_t peer, int outgoing) { +/* Downstream удерживает приём только в ACTIVE; выбранный upstream удерживает запрос также до ACCEPT, но INFO-пробы — нет. */ +static unsigned relay_flows(struct stream_network *net, const struct stream_descriptor *key, uint64_t peer, int outgoing, int pending) { unsigned count = 0; for (struct stream_peer *p = net->peers; p; p = p->next) { if (peer && p->node_id != peer) continue; for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) - if (f->outgoing == outgoing && !f->info.probe_only && f->info.state == STREAM_RECEIVE_ACTIVE + if (f->outgoing == outgoing && !f->info.probe_only + && (f->info.state == STREAM_RECEIVE_ACTIVE || (pending && f->info.state < STREAM_RECEIVE_ACTIVE)) && same_stream(&f->descriptor, key)) count++; } return count; @@ -1307,11 +1308,11 @@ void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descrip if (!net || net->stopping || stream_local_get(inst, key, &local) != STREAM_OK) return; for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) { if (!r->outgoing || !same_stream(&r->descriptor, key) || r->status >= STREAM_RELAY_REJECTED) continue; - unsigned active = relay_active(net, key, r->peer, 1); + unsigned in_use = relay_flows(net, key, r->peer, 1, 1); int drained = local.finished && (local.last == UINT64_MAX || (local.anchored && local.next > local.last)); if (needed && r->status == STREAM_RELAY_BOUND && now - r->progress_ms >= STREAM_RELAY_START_MS) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_TRANSPORT, now); - else if (local.relay_only || drained || (!needed && !active)) + else if (local.relay_only || drained || (!needed && !in_use)) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); else return; } @@ -1442,7 +1443,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) { && stream_member_supernode(net->inst, r->descriptor.group_id, r->outgoing ? r->peer : net->inst->node_id) == 1; if (!valid) relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_MEMBER, now); } - unsigned active = relay_active(net, &r->descriptor, r->peer, r->outgoing); + unsigned active = relay_flows(net, &r->descriptor, r->peer, r->outgoing, 0); if (r->status < STREAM_RELAY_REJECTED && active) { if (r->status != STREAM_RELAY_BOUND) { r->status = STREAM_RELAY_BOUND; r->dirty = r->urgent = 1; r->progress_ms = now; @@ -1505,7 +1506,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) { unsigned pending = 0; for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) if (!r->outgoing && same_stream(&r->descriptor, &relay->descriptor) && r->status <= STREAM_RELAY_READY) pending++; - unsigned active = relay_active(net, &relay->descriptor, 0, 0); + unsigned active = relay_flows(net, &relay->descriptor, 0, 0, 0); if (now >= relay->checked_ms) { relay->checked_ms = now + 1000; relay->allowed = stream_member_supernode(net->inst, relay->descriptor.group_id, net->inst->node_id) == 1; diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 206bdc92..f6da521e 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -55,14 +55,14 @@ static int automatic_drop_tail; static etcp_recv_fn automatic_handlers[4]; static uint8_t automatic_request[STREAM_WIRE_HEADER + 1]; static etcp_recv_fn relay_raw_handlers[4], relay_route_handler, relay_status_handler; -static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuses_seen; +static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuses_seen, relay_cancels_seen; static int relay_drop_source; static uint8_t relay_request_wire[STREAM_WIRE_HEADER]; static struct { - int hold_info, hold_request, check_starts; + int hold_info, hold_request, hold_ready, check_starts; uint8_t info[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8]; size_t info_length; - unsigned source_subscribes, helper_subscribes, accepted; + unsigned source_subscribes, helper_subscribes, accepted, ready_held; uint64_t request_tb, start; } relay_race; static struct { @@ -1045,6 +1045,8 @@ static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { /* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { + if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER + && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_CANCEL) relay_cancels_seen++; if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_REQUEST) { memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++; @@ -1057,7 +1059,15 @@ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { /* Счётчик STATUS выявляет ответы на дубликаты, не меняя подписанное управление и его порядок. */ static void relay_test_status(struct ETCP_CONN *conn, struct ll_entry *entry) { if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER - && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_STATUS) relay_statuses_seen++; + && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_STATUS) { + relay_statuses_seen++; + size_t payload = ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER; + if (relay_race.hold_ready && entry->len > payload && entry->dgram[payload] == STREAM_RELAY_READY) { + relay_race.ready_held++; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "bootstrap test holds READY ordinal=%u", relay_race.ready_held); + queue_dgram_free(entry); queue_entry_free(entry); return; + } + } relay_status_handler(conn, entry); } @@ -1150,19 +1160,33 @@ static void test_supernode_relay(void) { CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); test_clock_run_for(ua, 500); stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 2000); CHECK(relay_requests_seen == requests_before); - memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1; + /* Merkle объявляет буфер помощника раньше STATUS READY: выбранная выдача должна удержать STARTING-запрос. */ + memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_ready = 1; relay_drop_source = 1; + unsigned cancellations_before = relay_cancels_seen; + unsigned expected_data = relay_data[2] + 1; + CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); + deadline = get_time_tb() + 40000; + while (relay_data[2] < expected_data && relay_cancels_seen == cancellations_before && get_time_tb() < deadline) + test_clock_poll(ua, 100); + CHECK(relay_race.ready_held && relay_cancels_seen == cancellations_before); + CHECK(relay_data[2] == expected_data && relay_race.helper_subscribes); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] helper INFO before delayed READY keeps selected pending issuance; no premature CANCEL"); + stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 40000); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1; started = get_time_tb(); + expected_data = relay_data[2] + 1; CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); deadline = get_time_tb() + 300000; - while (relay_data[2] < 3 && get_time_tb() < deadline) test_clock_poll(ua, 100); - CHECK(relay_data[2] == 3 && relay_requests_seen && !relay_data[1]); + while (relay_data[2] < expected_data && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(relay_data[2] == expected_data && relay_requests_seen && !relay_data[1]); CHECK(get_time_tb() - started < 10000 && relay_race.request_tb - started >= 990); uint64_t connected_ms = (get_time_tb() - started) / 10; test_clock_run_for(ua, 1000); CHECK(relay_race.accepted == 255); CHECK(relay_race.helper_subscribes && !relay_race.source_subscribes); relay_release_info(); test_clock_run_for(ua, 1000); - CHECK(!relay_race.source_subscribes && relay_data[2] == 3); + CHECK(!relay_race.source_subscribes && relay_data[2] == expected_data); struct stream_message duplicate; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &duplicate) == STREAM_OK); CHECK(duplicate.receiver_epoch); unsigned statuses_before = relay_statuses_seen;