diff --git a/src/stream/stream.c b/src/stream/stream.c index 4d569155..f31d590b 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -229,6 +229,7 @@ static void member_changed(uint64_t node_id, const char *tags, const char *chann (void)tags; struct stream_service *service = arg; if (service->stopping || !service->sessions) return; + stream_network_membership_changed(service->network, node_id, channel); for (struct stream_session *s = service->sessions; s; s = s->next) { int own = node_id == service->inst->node_id; if (s->closing || (!own && (s->source || s->descriptor.source_node_id != node_id))) continue; diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 2454feee..0a533aa4 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -392,7 +392,7 @@ static void peer_pump(void *arg) { struct stream_peer *p = arg; p->pump = NULL; if (!p->up || !p->own_epoch || p->network->stopping) return; struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd); - if (!conn || !conn->send_input_q || conn->state == 2 || conn->close_requested) return; + if (!conn || !conn->send_input_q || conn->state == 2 || conn->close_requested || conn->reinit_pending || !conn->initialized) return; if (conn->send_input_q->count) { p->wait_q = conn->send_input_q; if (queue_waiter_wait(p->wait_q, &p->waiter, send_ready, p) < 0) @@ -554,7 +554,11 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t } else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) { if (stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED; 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_RENEW && f && !m.length) { + f->deadline = get_time_tb() + STREAM_LEASE_TB; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream lease renewed peer=%llx handle=%llu deadline=%llu", + (unsigned long long)sender, (unsigned long long)m.handle, (unsigned long long)f->deadline); + } 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 || m.length == 17)) flow_close(f, m.payload[0], 0); @@ -674,6 +678,25 @@ void stream_network_cancel(struct stream_network *net, const struct stream_descr } } +/* Исключение поставщика/получателя или self закрывает выдачи, сохраняя локальное кольцо остальных трансляций. */ +void stream_network_membership_changed(struct stream_network *net, uint64_t node, const char *channel) { + if (!net || net->stopping || !channel) return; + for (struct stream_peer *p = net->peers; p; p = p->next) { + if (node != p->node_id && node != net->inst->node_id) continue; + p->busy++; + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) { + if (f->info.state == STREAM_RECEIVE_CLOSED) continue; + struct TOPO_GROUP *group = topo_groups_find(net->inst->topo_groups, f->descriptor.group_id); + if (!group || strcmp(group->channel_id, channel)) continue; + struct stream_member_state state; + int result = stream_member_load(net->inst, group->group_id, node, &state, NULL); + if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT) + flow_close(f, STREAM_REASON_MEMBER, 1); + } + p->busy--; peer_sweep(p); + } +} + int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, struct STREAM_RECEIVE **out) { struct UTUN_INSTANCE *inst; struct stream_descriptor d; if (!out || *out || stream_handle_context(subscription, &inst, &d) != STREAM_OK || sub >= d.n || !node || node == inst->node_id) { diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index 2155d54b..2d7b4a5f 100644 --- a/src/stream/stream_network.h +++ b/src/stream/stream_network.h @@ -27,4 +27,5 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) void stream_network_destroy(struct stream_network *network); void stream_network_changed(struct stream_network *network); void stream_network_cancel(struct stream_network *network, const struct stream_descriptor *descriptor); +void stream_network_membership_changed(struct stream_network *network, uint64_t node_id, const char *channel); #endif diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 46dd597c..624c3f34 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -259,13 +259,17 @@ static void test_network(void) { /* REINIT не обязан разрушать ETCP_CONN, но старые stream handles становятся недействительны. */ conn = instance_find_conn(nodes[2], nodes[1]->node_id); CHECK(conn != NULL); etcp_conn_reinit(conn, "stream network test"); + deadline = get_time_tb() + 100000; + do { test_clock_poll(ua, 100); CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK); } + while (info.state != STREAM_RECEIVE_CLOSED && get_time_tb() < deadline); CHECK(stream_receive_get_info(down[0], &info) == STREAM_OK && info.state == STREAM_RECEIVE_CLOSED && info.reason == STREAM_REASON_TRANSPORT); for (unsigned i = 0; i < 8; i++) { stream_receive_close(up[i]); stream_receive_close(down[i]); } for (unsigned i = 1; i < 4; i++) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, saved_raw[i]) == 0); CHECK(etcp_router_bind(nodes[3], ETCP_RT_ID_STREAM, saved_route) == 0); - stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL); for (unsigned i = 1; i < 3; i++) stream_close(subscribers[i]); stream_close(publisher); + test_clock_run_for(ua, 1000); + stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL); for (unsigned i = 0; i < 2; i++) topo_group_peer_close(connections[i]); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live raw ETCP chain: common index, FEC, retained paths/LOOP, duplicate/reverse, END, renew/lease, REINIT"); }