diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 28ceb0ba..3634180f 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -32,7 +32,8 @@ struct stream_relay_request { struct stream_relay_request *next; struct stream_descriptor descriptor; uint64_t peer, id, epoch, server_epoch, deadline_ms, retain_ms, sent_ms, checked_ms, progress_ms, progress_seq; - int outgoing, status, reason, dirty, granted; + int outgoing, status, reason, dirty, granted, blocked; + struct queue_waiter_handle waiter; uint8_t end[STREAM_END_SIZE]; }; struct stream_relay { @@ -165,6 +166,12 @@ static void send_ready(struct ll_queue *queue, void *arg) { (void)queue; struct stream_peer *p = arg; p->wait_q = NULL; peer_schedule(p); } +/* Waiter только будит отправку актуального состояния запроса; DATA и старые STATUS в нём не копируются. */ +static void relay_send_ready(struct ll_queue *queue, void *arg) { + (void)queue; struct stream_relay_request *r = arg; + r->blocked = 0; r->dirty = 1; r->sent_ms = 0; +} + /* Сохранённый указатель очереди используется только пока waiter действительно зарегистрирован. */ static void peer_cancel_wait(struct stream_peer *p) { if (p->waiter.internal || p->waiter.call_soon_id) @@ -198,7 +205,8 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin } /* Routed-управление обязательно подписано end-to-end, DATA этим путём не передаётся. */ -static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length) { +static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length, + struct stream_relay_request *request) { struct UTUN_INSTANCE *inst = net->inst; int route = etcp_router_has_route(inst, m.descriptor.group_id, peer); if (!route) { @@ -211,7 +219,13 @@ static int route_control(struct stream_network *net, uint64_t peer, struct strea } if (!etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) { DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for queue room peer=%llx command=%u", - (unsigned long long)peer, m.command); return STREAM_WAIT; + (unsigned long long)peer, m.command); + if (request) { + request->blocked = 1; + etcp_router_on_send_ready(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM, &request->waiter, + relay_send_ready, request); + } + return STREAM_WAIT; } struct ll_entry *e = ll_alloc_lldgram((uint16_t)(1 + STREAM_WIRE_HEADER + length)); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream routed control allocation failed"); return STREAM_ERR_MEMORY; } @@ -251,7 +265,7 @@ static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) { .command = f->outgoing ? STREAM_CMD_UNSUBSCRIBE : STREAM_CMD_CLOSED, .sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch}; if (p->up) control_put(p, m, &payload, 1); - else if (!f->outgoing) route_control(p->network, p->node_id, m, &payload, 1); + else if (!f->outgoing) route_control(p->network, p->node_id, m, &payload, 1, NULL); } f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason; if (reason != STREAM_REASON_STOPPED) @@ -329,7 +343,7 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock, .command = STREAM_CMD_CLOSED, .sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch}; - route_control(p->network, p->node_id, m, &reason, 1); + route_control(p->network, p->node_id, m, &reason, 1, NULL); } peer_reset(p, STREAM_REASON_TRANSPORT, STREAM_RESET_ROTATE, NULL); } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id); @@ -985,7 +999,7 @@ static void network_tick(void *arg) { if (!p->up && pending && (!p->reverse_at || now - p->reverse_at >= 10000)) { struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ, .sender_epoch = p->own_epoch}; - if (route_control(net, p->node_id, m, NULL, 0) == STREAM_OK) p->reverse_at = now; + if (route_control(net, p->node_id, m, NULL, 0, NULL) == STREAM_OK) p->reverse_at = now; } p->busy--; peer_sweep(p); if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p); @@ -1036,7 +1050,8 @@ void stream_network_destroy(struct stream_network *net) { stream_close(relay->owner); u_free(relay); } while (net->relay_requests) { - struct stream_relay_request *r = net->relay_requests; net->relay_requests = r->next; u_free(r); + struct stream_relay_request *r = net->relay_requests; net->relay_requests = r->next; + etcp_router_cancel_send_ready(net->inst, r->descriptor.group_id, r->peer, ETCP_RT_ID_STREAM, &r->waiter); u_free(r); } etcp_remove_conn_status_cbk(net->inst, transport_changed, net); etcp_unbind(net->inst, ETCP_ID_STREAM); etcp_router_unbind(net->inst, ETCP_RT_ID_STREAM); @@ -1214,6 +1229,8 @@ static unsigned relay_active(struct stream_network *net, const struct stream_des /* Неуспешная помощь закрывает только выдачи своего пира и трансляции, не общий ETCP. */ static void relay_finish(struct stream_network *net, struct stream_relay_request *r, int status, int reason, uint64_t now) { if (r->status >= STREAM_RELAY_REJECTED) return; + etcp_router_cancel_send_ready(net->inst, r->descriptor.group_id, r->peer, ETCP_RT_ID_STREAM, &r->waiter); + memset(&r->waiter, 0, sizeof(r->waiter)); r->blocked = 0; r->status = status; r->reason = reason; r->retain_ms = now + STREAM_RELAY_RETAIN_MS; r->dirty = 1; if (r->outgoing) { struct stream_peer *p = peer_find(net, r->peer); @@ -1265,8 +1282,9 @@ void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descrip if (!r->outgoing || !same_stream(&r->descriptor, key) || r->status >= STREAM_RELAY_REJECTED) continue; unsigned active = relay_active(net, key, r->peer, 1); int drained = local.finished && (local.last == UINT64_MAX || (local.anchored && local.next > local.last)); - if (local.relay_only || drained || (!needed && !active) - || (needed && r->status == STREAM_RELAY_BOUND && now - r->progress_ms >= STREAM_RELAY_START_MS)) + 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)) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); else return; } @@ -1280,7 +1298,8 @@ void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descrip if (!peers[i].supernode || peer == inst->node_id || peer == key->source_node_id || !stream_member_online(inst, key->group_id, peer) || !node_conn_direct_compatible(inst, peer)) continue; struct stream_relay_request *previous = relay_request_find(net, key, peer, 0, 1); - if (previous) continue; + if (previous && (previous->status < STREAM_RELAY_REJECTED || previous->status == STREAM_RELAY_REJECTED + || previous->reason != STREAM_REASON_STOPPED)) continue; struct stream_member_state member; if (stream_member_load(inst, key->group_id, peer, &member, NULL) != STREAM_OK) continue; chosen = peer; break; @@ -1338,7 +1357,7 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_PROTOCOL, now); } else relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); } else if (status >= STREAM_RELAY_REJECTED) { - relay_finish(net, r, STREAM_RELAY_CANCELLED, m->payload[1], now); + relay_finish(net, r, STREAM_RELAY_REJECTED, m->payload[1], now); } return; } @@ -1428,7 +1447,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) { } } } - if (!r->dirty && !(r->outgoing && r->status <= STREAM_RELAY_READY && now - r->sent_ms >= 1000)) continue; + if (r->blocked || (!r->dirty && !(r->outgoing && r->status <= STREAM_RELAY_READY && now - r->sent_ms >= 1000))) continue; if (r->sent_ms && now - r->sent_ms < 1000) continue; struct stream_message m = {.descriptor = r->descriptor, .handle = r->id, .sender_epoch = r->outgoing ? r->epoch : net->relay_epoch, @@ -1442,7 +1461,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) { if (r->status == STREAM_RELAY_ENDED) { memcpy(payload + 6, r->end, STREAM_END_SIZE); length += STREAM_END_SIZE; } } r->sent_ms = now; - if (route_control(net, r->peer, m, payload, length) == STREAM_OK) r->dirty = 0; + if (route_control(net, r->peer, m, payload, length, r) == STREAM_OK) r->dirty = 0; } struct stream_relay **link = &net->relays; while (*link) { @@ -1469,7 +1488,8 @@ static void relay_tick(struct stream_network *net, uint64_t now) { while (*request_link) { struct stream_relay_request *r = *request_link; if (r->status < STREAM_RELAY_REJECTED || now < r->retain_ms) { request_link = &r->next; continue; } - *request_link = r->next; net->relay_request_count--; u_free(r); + *request_link = r->next; net->relay_request_count--; + etcp_router_cancel_send_ready(net->inst, r->descriptor.group_id, r->peer, ETCP_RT_ID_STREAM, &r->waiter); u_free(r); } } diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index e6cd28bb..ece5ae71 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -1087,6 +1087,12 @@ static void test_supernode_relay(void) { test_clock_run_for(ua, 80000); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK); /* Локальный viewer меняет :r на обычную подписку и делит существующий буфер. */ + CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, "CREATE TRIGGER reject_relay_viewer BEFORE UPDATE OF userinfo ON peers_710002 " + "BEGIN SELECT RAISE(ABORT,'injected relay viewer failure'); END", NULL, NULL, NULL) == SQLITE_OK); + struct STREAM_HANDLE *failed = NULL; + CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &failed) < 0 && !failed); + CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, "DROP TRIGGER reject_relay_viewer", NULL, NULL, NULL) == SQLITE_OK); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only); CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &viewers[1]) == STREAM_OK); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && !local.relay_only); unsigned before = relay_data[1];