From cd5e658fb90d0ed10ed490687270f2438b400b85 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 01:13:52 +0200 Subject: [PATCH] Add internal ownership for exact-epoch stream relays --- src/stream/stream.c | 81 +++++++++++++++++++++++++++++++--------- src/stream/stream_priv.h | 8 +++- 2 files changed, 71 insertions(+), 18 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index e57f7c30..7f8016e3 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -32,9 +32,9 @@ struct stream_session { uint64_t restart_from, restart_last; uint8_t end[STREAM_END_SIZE]; int end_valid; - unsigned owners, holds; + unsigned owners, relay_owners, holds; int anchored, closing, invalidated, finished, end_delivered, reason; - int restarting, restart_gap; + int restarting, restart_gap, notify_close; }; struct STREAM_HANDLE { @@ -42,6 +42,7 @@ struct STREAM_HANDLE { struct stream_service *service; struct stream_session *session; int publisher; + int relay; int selection_owned; int reason; }; @@ -196,7 +197,8 @@ static int service_publish_options(struct stream_service *service, struct TOPO_G DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "local subscription list exceeds member limit group=%llx", (unsigned long long)group->group_id); return STREAM_ERR_ARGUMENT; } - struct stream_subscription_key entry = {s->descriptor.source_node_id, s->descriptor.source_epoch}; + struct stream_subscription_key entry = {s->descriptor.source_node_id, s->descriptor.source_epoch, + !s->source && s->owners == s->relay_owners}; size_t at = state.count++; while (at && (state.subscriptions[at - 1].source_node_id > entry.source_node_id || (state.subscriptions[at - 1].source_node_id == entry.source_node_id @@ -251,6 +253,7 @@ static void session_schedule(struct stream_session *s) { /* Снять локальные ссылки и публикацию; CLOSED также отложен и отменяется при stop всего сервиса. */ static void session_close(struct stream_session *s, int reason, int update_profile, int notify) { if (s->closing) return; + s->notify_close = notify && s->owners > s->relay_owners; s->closing = 1; s->reason = reason; s->owners = 0; stream_optimizer_destroy(s->optimizer); s->optimizer = NULL; stream_network_cancel(s->service->network, &s->descriptor); @@ -279,7 +282,7 @@ static void session_deliver(void *arg) { s->delivery = NULL; s->holds++; struct stream_event event = { .descriptor = s->descriptor }; int emit = 0, again = 0; - if (s->closing) { event.type = STREAM_EVENT_CLOSED; event.reason = s->reason; emit = 1; } + if (s->closing) { event.type = STREAM_EVENT_CLOSED; event.reason = s->reason; emit = s->notify_close; } else if (s->invalidated) { DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "invalidated stream awaits deferred close source=%llx", (unsigned long long)s->descriptor.source_node_id); @@ -316,6 +319,7 @@ static void session_deliver(void *arg) { event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = UINT64_MAX; s->end_delivered = 1; emit = 1; } + if (!s->closing && s->owners == s->relay_owners) emit = 0; if (emit && service->callback && !service->stopping) { service->dispatching++; service->callback(service->inst, &event, service->arg); @@ -493,11 +497,12 @@ void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callb /* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, int automatic, + int relay, struct STREAM_HANDLE **out) { struct stream_service *service = inst->stream; struct stream_session *s = session_find(service, d->group_id, d->source_node_id); int created = s == NULL; - if (s && (publisher || s->descriptor.source_epoch != d->source_epoch || (!s->source && !!s->optimizer != automatic))) { + if (s && (publisher || s->descriptor.source_epoch != d->source_epoch || (!relay && !s->source && !!s->optimizer != automatic))) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream instance conflicts with publication/epoch/selection mode automatic=%d existing_auto=%d", automatic, s->optimizer != NULL); return STREAM_ERR_STATE; } @@ -523,10 +528,12 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript for (unsigned i = 0; i < d->n; i++) stream_metrics_init(&s->metrics[i]); stream_buffer_set_observer(s->buffer, cache_event, s); } - h->service = service; h->session = s; h->publisher = publisher; - h->next = service->handles; service->handles = h; s->owners++; + int was_relay = s->owners && s->owners == s->relay_owners; + h->service = service; h->session = s; h->publisher = publisher; h->relay = relay; + h->next = service->handles; service->handles = h; s->owners++; s->relay_owners += relay; s->holds++; - int result = created ? service_publish_options(service, stream_group(inst, d->group_id)) : STREAM_OK; + int result = created || was_relay != (s->owners == s->relay_owners) + ? service_publish_options(service, stream_group(inst, d->group_id)) : STREAM_OK; if (result == STREAM_OK && created && utun_set_client_activity(inst, inst->client_activity_requested) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream open failed to activate core"); result = STREAM_ERR_STATE; } @@ -537,7 +544,10 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript struct STREAM_HANDLE **link = &service->handles; while (*link != h) link = &(*link)->next; *link = h->next; - if (!s->closing) { s->owners--; session_close(s, STREAM_CLOSE_FAILED, 0, 0); } + if (!s->closing) { + s->owners--; s->relay_owners -= relay; + if (created) session_close(s, STREAM_CLOSE_FAILED, 0, 0); + } u_free(h); s->holds--; if (!s->holds && !s->delivery) session_free(s); return result; @@ -577,14 +587,14 @@ int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const str } } while (!d.source_epoch); if (stream_descriptor_validate(&d) != STREAM_OK) return STREAM_ERR_ARGUMENT; - return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, out); + return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, 0, out); } static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, int automatic, struct STREAM_HANDLE **out) { int result = stream_open_check(inst, group_id, out); if (result != STREAM_OK) return result; struct stream_session *s = session_find(inst->stream, group_id, source_id); - if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, out); + if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, 0, out); struct stream_member_state state; uint8_t key[32]; result = stream_member_load(inst, group_id, source_id, &state, key); if (result != STREAM_OK) return result; @@ -592,7 +602,25 @@ static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_ DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id); return STREAM_WAIT; } - return session_open(inst, &state.source, key, 0, automatic, out); + return session_open(inst, &state.source, key, 0, automatic, 0, out); +} + +/* Запрос старой эпохи не может незаметно подписать суперузел на новый запуск. */ +int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct STREAM_HANDLE **out) { + int result = key ? stream_open_check(inst, key->group_id, out) : STREAM_ERR_ARGUMENT; + if (result != STREAM_OK) return result; + struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id); + if (s && s->descriptor.source_epoch == key->source_epoch) + return session_open(inst, &s->descriptor, s->public_key, 0, 1, 1, out); + struct stream_member_state state; uint8_t public_key[32]; + result = stream_member_load(inst, key->group_id, key->source_node_id, &state, public_key); + if (result != STREAM_OK) return result; + if (!state.published || state.source.source_epoch != key->source_epoch) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay source epoch unavailable source=%llx requested=%llx published=%d actual=%llx", + (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, state.published, + (unsigned long long)state.source.source_epoch); return STREAM_ERR_STATE; + } + return session_open(inst, &state.source, public_key, 0, 1, 1, out); } int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) { @@ -627,7 +655,18 @@ void stream_close(struct STREAM_HANDLE *h) { *link = h->next; } struct stream_session *s = h->session; - if (s && (h->publisher || --s->owners == 0)) session_close(s, STREAM_CLOSE_STOPPED, 1, 1); + if (s) { + int was_relay = s->owners == s->relay_owners; + if (h->publisher || s->owners == 1) session_close(s, STREAM_CLOSE_STOPPED, 1, !h->relay); + else { + s->owners--; s->relay_owners -= h->relay; + if (was_relay != (s->owners == s->relay_owners) + && service_publish_options(service, stream_group(service->inst, s->descriptor.group_id)) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream owner mode publication failed; closing inconsistent instance"); + session_close(s, STREAM_CLOSE_FAILED, 0, 1); + } + } + } u_free(h); } @@ -762,7 +801,8 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id); if (!s || s->descriptor.source_epoch != key->source_epoch) return STREAM_WAIT; *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq, .next = s->next_seq, - .anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid, .end = s->end_valid ? s->end : NULL}; + .anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid, + .relay_only = !s->source && s->owners == s->relay_owners, .end = s->end_valid ? s->end : NULL}; return STREAM_OK; } @@ -862,13 +902,20 @@ int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate int source = node->node_id == s->descriptor.source_node_id; DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate peer=%llx type=%u source=%d online=1", (unsigned long long)node->node_id, identity ? identity->client_type : UINT8_MAX, source); - if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source) { + if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source + && stream_member_supernode(inst, group->group_id, node->node_id) != 1) { DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate excluded peer=%llx reason=mobile_relay", (unsigned long long)node->node_id); continue; } struct stream_member_state state; - if (stream_member_load(inst, group->group_id, node->node_id, &state, NULL) != STREAM_OK - || !stream_member_has_subscription(&state, s->descriptor.source_node_id, s->descriptor.source_epoch)) continue; + if (stream_member_load(inst, group->group_id, node->node_id, &state, NULL) != STREAM_OK) continue; + int confirmed = stream_relay_confirmed(inst, &s->descriptor, node->node_id); + if (!stream_member_has_subscription(&state, s->descriptor.source_node_id, s->descriptor.source_epoch) && !confirmed) continue; + int relay_only = confirmed; + for (size_t i = 0; i < state.count; i++) + if (state.subscriptions[i].source_node_id == s->descriptor.source_node_id + && state.subscriptions[i].source_epoch == s->descriptor.source_epoch) relay_only = state.subscriptions[i].relay_only; + if (!source && s->owners == s->relay_owners && relay_only) continue; candidates[found++] = (struct stream_candidate){ .node_id = node->node_id, .source = (uint8_t)source, .client_type = identity ? identity->client_type : UINT8_MAX }; } diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h index e3368495..2bb8ce7d 100644 --- a/src/stream/stream_priv.h +++ b/src/stream/stream_priv.h @@ -11,7 +11,7 @@ struct stream_local { struct stream_descriptor descriptor; struct stream_buffer *buffer; uint64_t start, last, next; - int anchored, publisher, finished; + int anchored, publisher, finished, relay_only; const uint8_t *end; }; /* Заимствованные буфер/END действуют только до следующего callback/мутации сервиса. */ @@ -25,6 +25,12 @@ void stream_local_progress(struct UTUN_INSTANCE *inst, const struct stream_descr int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key); int stream_local_metrics(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint8_t sub, uint64_t now_ms, struct stream_metric_snapshot *snapshot); +/* Внутреннее владение точной эпохой; close обязателен и после detach. Режим существующего экземпляра сохраняется. */ +int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct STREAM_HANDLE **out); +/* Подтверждение поставщика действует только в живой попытке помощи, до прихода Merkle-опции. */ +int stream_relay_confirmed(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer); +/* Вызывается selector один раз на экземпляр; relay-only экземпляры новых запросов помощи не создают. */ +void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, int needed, uint64_t now_ms); int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length);