From 30106c8f93716f5ec3f35b000ebb0bcfe03a0f36 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 00:00:36 +0200 Subject: [PATCH] Attach stream subscriptions to CHAT lifecycle and member profiles --- src/Makefile.am | 5 + src/chat/member_sync.c | 7 +- src/chat/member_sync.h | 3 +- src/routing_layer/topo_group.c | 2 + src/stream/stream.c | 524 +++++++++++++++++++++++++++++++++ src/stream/stream.h | 81 +++++ src/stream/stream_priv.h | 11 + src/utun_instance.c | 4 +- src/utun_instance.h | 2 + tests/Makefile.am | 5 + tests/test_stream_service.c | 315 ++++++++++++++++++++ 11 files changed, 954 insertions(+), 5 deletions(-) create mode 100644 src/stream/stream.c create mode 100644 src/stream/stream.h create mode 100644 src/stream/stream_priv.h create mode 100644 tests/test_stream_service.c diff --git a/src/Makefile.am b/src/Makefile.am index 37779348..3daa5d61 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -4,6 +4,11 @@ noinst_LIBRARIES = libutun.a EXTRA_DIST = lwip_tcp/COPYING dnsmasq/COPYING dnsmasq/COPYING-v3 stream_core_files = \ + stream/stream.h \ + stream/stream.c \ + stream/stream_priv.h \ + stream/stream_member.h \ + stream/stream_member.c \ stream/stream_proto.h \ stream/stream_block.h \ stream/stream_block.c \ diff --git a/src/chat/member_sync.c b/src/chat/member_sync.c index df71dbc0..60951c82 100644 --- a/src/chat/member_sync.c +++ b/src/chat/member_sync.c @@ -317,13 +317,14 @@ static void _fire_props_changed(struct UTUN_INSTANCE* inst, uint64_t node_id, co while (pc) { pc->fn(node_id, adm_tags, channel_id, pc->arg); pc = pc->next; } } -void member_sync_add_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg) { - if (!inst || !fn || !CC(inst)) return; +int member_sync_add_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg) { + if (!inst || !fn || !CC(inst)) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "props callback invalid arguments"); return -1; } struct ms_props_cbk* e = u_malloc(sizeof(*e)); - if (!e) return; + if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "props callback allocation failed"); return -1; } e->fn = fn; e->arg = arg; e->next = CC(inst)->props_cbks; CC(inst)->props_cbks = e; + return 0; } void member_sync_remove_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg) { diff --git a/src/chat/member_sync.h b/src/chat/member_sync.h index cab7f8d9..565b3403 100644 --- a/src/chat/member_sync.h +++ b/src/chat/member_sync.h @@ -202,7 +202,8 @@ int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id); * Многоподписочный — можно добавить несколько подписчиков. */ typedef void (*node_props_changed_fn)(uint64_t node_id, const char* adm_tags, const char* channel_id, void* arg); -void member_sync_add_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg); +/* 0 — зарегистрирован, -1 — неверные аргументы/память; потребитель может откатить свой init. */ +int member_sync_add_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg); void member_sync_remove_props_cbk(struct UTUN_INSTANCE* inst, node_props_changed_fn fn, void* arg); #ifdef __cplusplus diff --git a/src/routing_layer/topo_group.c b/src/routing_layer/topo_group.c index f0634841..6ebff032 100644 --- a/src/routing_layer/topo_group.c +++ b/src/routing_layer/topo_group.c @@ -29,6 +29,7 @@ #include "topo_recovery.h" #include "broadcast.h" #include "../radio/radio.h" +#include "../stream/stream.h" #include "../chat/chat_setting.h" #define TOPO_GROUP_UTUN 0x8000000000000000ULL @@ -624,6 +625,7 @@ static struct TOPO_GROUP* topo_group_create(struct UTUN_INSTANCE* instance, uint static void topo_group_destroy(struct TOPO_GROUP* group) { if (!group) return; group->stopping = 1; + stream_cancel_group(group->instance, group->group_id); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 0 enter grp=%p id=0x%016llx rec_list=%p connect=%p", group, (unsigned long long)group->group_id, group->recovery, group->connect); diff --git a/src/stream/stream.c b/src/stream/stream.c new file mode 100644 index 00000000..2084a8bd --- /dev/null +++ b/src/stream/stream.c @@ -0,0 +1,524 @@ +#include +#include + +#include "../../lib/debug_config.h" +#include "../../lib/mem.h" +#include "../../lib/platform_compat.h" +#include "../../lib/u_async.h" +#include "../utun_instance.h" +#include "../chat/chat_core.h" +#include "stream_member.h" +#include "stream_source.h" +#include "stream_priv.h" + +struct stream_session { + struct stream_session *next; + struct stream_service *service; + struct stream_descriptor descriptor; + uint8_t public_key[SC_PUBKEY_SIZE]; + struct stream_source *source; + struct stream_buffer *buffer; + uint8_t *output; + void *delivery; + uint64_t next_seq; + unsigned owners, holds; + int anchored, closing, invalidated, finished, end_delivered, reason; +}; + +struct STREAM_HANDLE { + struct STREAM_HANDLE *next; + struct stream_service *service; + struct stream_session *session; + int publisher; + int reason; +}; + +struct stream_service { + struct UTUN_INSTANCE *inst; + struct stream_session *sessions; + struct STREAM_HANDLE *handles; + stream_event_fn callback; + void *arg; + void *reconcile; + int stopping, dispatching, registered, initializing; +}; + +static void session_deliver(void *arg); +static void service_reconcile(void *arg); + +/* Активность CHAT проверяется до чтения профиля, выделения 32 МБ и создания handles. */ +static struct TOPO_GROUP *stream_group(struct UTUN_INSTANCE *inst, uint64_t id) { + struct TOPO_GROUP *group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, id) : NULL; + return group && group->group_type == TOPO_GROUP_TYPE_CHAT && !group->stopping ? group : NULL; +} + +/* Один локальный экземпляр на источник группы; закрывающийся старый запуск не мешает новому. */ +static struct stream_session *session_find(struct stream_service *service, uint64_t group_id, uint64_t source_id) { + for (struct stream_session *s = service->sessions; s; s = s->next) + if (!s->closing && !s->invalidated && s->descriptor.group_id == group_id && s->descriptor.source_node_id == source_id) return s; + return NULL; +} + +/* Две опции заменяются одной подписанной записью; список отражает экземпляры, не отдельные handles. */ +static int service_publish_options(struct stream_service *service, struct TOPO_GROUP *group) { + if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream options group missing"); return STREAM_ERR_STATE; } + struct stream_member_state state = {0}; + for (struct stream_session *s = service->sessions; s; s = s->next) { + if (s->closing || s->invalidated || s->descriptor.group_id != group->group_id) continue; + if (state.count == STREAM_MEMBER_MAX_SUBSCRIPTIONS) { + 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}; + 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 + && state.subscriptions[at - 1].source_epoch > entry.source_epoch))) { + state.subscriptions[at] = state.subscriptions[at - 1]; at--; + } + state.subscriptions[at] = entry; + if (s->source && !s->finished) { state.source = s->descriptor; state.published = 1; } + } + char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1]; + int result = stream_member_format(&state, source, subscriptions); + if (result != STREAM_OK) return result; + struct chat_profile_option options[] = { + {STREAM_MEMBER_SOURCE, source[0] ? source : NULL}, + {STREAM_MEMBER_SUBSCRIPTIONS, subscriptions[0] ? subscriptions : NULL} + }; + if (chat_core_set_my_profile_options(service->inst, group->channel_id, options, 2) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "local member stream options update failed group=%llx owners=%zu", + (unsigned long long)group->group_id, state.count); return STREAM_ERR_STATE; + } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream options group=%llx published=%d subscriptions=%zu", + (unsigned long long)group->group_id, state.published, state.count); + return STREAM_OK; +} + +/* Ресурсы живут до возврата пользовательского callback/write, даже если он закрыл последний handle. */ +static void session_free(struct stream_session *s) { + struct stream_session **link = &s->service->sessions; + while (*link != s) link = &(*link)->next; + *link = s->next; + if (s->delivery) uasync_call_soon_cancel(s->service->inst->ua, s->delivery); + if (s->source) stream_source_destroy(s->source); + else stream_buffer_destroy(s->buffer); + u_free(s->output); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream released group=%llx source=%llx epoch=%llx reason=%d", + (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id, + (unsigned long long)s->descriptor.source_epoch, s->reason); + u_free(s); +} + +/* Delivery выполняется отдельным callback, а не из source_ready при busy источнике. */ +static void session_schedule(struct stream_session *s) { + if (s->service->stopping || s->delivery) return; + s->delivery = uasync_call_soon(s->service->inst->ua, s, session_deliver); + if (!s->delivery) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream delivery scheduling failed source=%llx", + (unsigned long long)s->descriptor.source_node_id); +} + +/* Снять локальные ссылки и публикацию; CLOSED также отложен и отменяется при stop всего сервиса. */ +static void session_close(struct stream_session *s, int reason, int update_profile, int notify) { + if (s->closing) return; + s->closing = 1; s->reason = reason; s->owners = 0; + for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next) + if (h->session == s) { h->session = NULL; h->reason = reason; } + if (s->delivery) { uasync_call_soon_cancel(s->service->inst->ua, s->delivery); s->delivery = NULL; } + struct TOPO_GROUP *group = s->service->inst->topo_groups + ? topo_groups_find(s->service->inst->topo_groups, s->descriptor.group_id) : NULL; + if (update_profile && group && service_publish_options(s->service, group) != STREAM_OK) + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream closed with stale own member options group=%llx", + (unsigned long long)s->descriptor.group_id); + if (notify && s->service->callback && !s->service->stopping) session_schedule(s); + else if (!s->holds) session_free(s); +} + +/* Уведомление использует собственный scratch экземпляра; close внутри callback не освобождает его преждевременно. */ +static void session_deliver(void *arg) { + struct stream_session *s = arg; + struct stream_service *service = s->service; + 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; } + else if (s->invalidated) { + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "invalidated stream awaits deferred close source=%llx", + (unsigned long long)s->descriptor.source_node_id); + } else if (s->anchored) { + struct stream_buffer_stats stats; + stream_buffer_get_stats(s->buffer, &stats); + size_t length = 0; + int result = s->next_seq == UINT64_MAX ? STREAM_WAIT + : stream_buffer_decode(s->buffer, s->next_seq, s->output, stream_input_size(&s->descriptor), &length); + if (result == STREAM_OK) { + event.type = STREAM_EVENT_DATA; event.first_seq = event.last_seq = s->next_seq++; + event.data = s->output; event.length = length; emit = 1; again = 1; + } else if (result == STREAM_EXPIRED) { + event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = stats.expired_before - 1; + s->next_seq = stats.expired_before; emit = 1; again = 1; + } else if (result < 0) { + session_close(s, STREAM_CLOSE_FAILED, 1, 1); + } else if (s->finished && s->next_seq > stats.highest_seq && !s->end_delivered) { + event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = stats.highest_seq; + s->end_delivered = 1; emit = 1; + } + } else if (s->finished && !s->end_delivered) { + event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = UINT64_MAX; + s->end_delivered = 1; emit = 1; + } + if (emit && service->callback && !service->stopping) { + service->dispatching++; + service->callback(service->inst, &event, service->arg); + service->dispatching--; + } + s->holds--; + if (s->closing) { if (!s->holds && !s->delivery) session_free(s); } + else if (again) session_schedule(s); +} + +/* Готовая source-пачка уже полностью в кольце. Пользовательский callback здесь не вызывается. */ +static void source_ready(struct stream_source *source, uint64_t seq, void *arg) { + (void)source; + struct stream_session *s = arg; + if (s->closing) return; + if (!s->anchored) { s->anchored = 1; s->next_seq = seq; } + session_schedule(s); +} + +/* Отложенная сверка не разрушает экземпляры из BGP/member callbacks. */ +static void service_reconcile(void *arg) { + struct stream_service *service = arg; + service->reconcile = NULL; + struct stream_session *snapshot = service->sessions; + for (struct stream_session *s = snapshot; s; s = s->next) s->holds++; + struct stream_session *s = snapshot; + while (s) { + struct stream_session *next = s->next; + if (!s->closing && s->invalidated) session_close(s, s->reason, 1, 1); + else if (!s->closing && !s->source) { + struct stream_member_state state; uint8_t key[32]; + int result = stream_member_load(service->inst, s->descriptor.group_id, s->descriptor.source_node_id, &state, key); + if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT) + session_close(s, STREAM_CLOSE_MEMBER, 1, 1); + else if (result == STREAM_OK && (memcmp(key, s->public_key, 32) + || (state.published && (state.source.source_epoch != s->descriptor.source_epoch + || state.source.n != s->descriptor.n || state.source.r != s->descriptor.r + || state.source.block_size != s->descriptor.block_size)))) + session_close(s, STREAM_CLOSE_SOURCE_CHANGED, 1, 1); + } + s = next; + } + s = snapshot; + while (s) { + struct stream_session *next = s->next; + s->holds--; + if (s->closing && !s->holds && !s->delivery) session_free(s); + s = next; + } +} + +/* Общий props сигнал относится и к Merkle commit, и к BGP. Payload adm_tags здесь не используется. */ +static void member_changed(uint64_t node_id, const char *tags, const char *channel, void *arg) { + (void)tags; + struct stream_service *service = arg; + if (service->stopping || !service->sessions) return; + for (struct stream_session *s = service->sessions; s; s = s->next) { + if (s->closing || s->source || s->descriptor.source_node_id != node_id) continue; + struct TOPO_GROUP *group = stream_group(service->inst, s->descriptor.group_id); + if (!group || !channel || strcmp(group->channel_id, channel)) continue; + struct stream_member_state state; uint8_t key[32]; + int result = stream_member_load(service->inst, group->group_id, node_id, &state, key); + if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT) { + s->invalidated = 1; s->reason = STREAM_CLOSE_MEMBER; + } else if (result == STREAM_OK && (memcmp(key, s->public_key, 32) + || (state.published && (state.source.source_epoch != s->descriptor.source_epoch + || state.source.n != s->descriptor.n || state.source.r != s->descriptor.r + || state.source.block_size != s->descriptor.block_size)))) { + s->invalidated = 1; s->reason = STREAM_CLOSE_SOURCE_CHANGED; + } + } + if (service->reconcile) return; + service->reconcile = uasync_call_soon(service->inst->ua, service, service_reconcile); + if (!service->reconcile) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream member reconciliation scheduling failed"); +} + +/* Снять старые собственные опции после аварии; неполный/отсутствующий профиль не является активной подпиской. */ +int stream_service_init(struct UTUN_INSTANCE *inst) { + if (!inst || !inst->core_started || !chat_core_get_db(inst) || !inst->topo_groups) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream init requires core and chat DB"); return STREAM_ERR_STATE; + } + if (inst->stream) return STREAM_OK; + struct stream_service *service = u_calloc(1, sizeof(*service)); + if (!service) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream service allocation failed"); return STREAM_ERR_MEMORY; } + service->inst = inst; service->initializing = 1; inst->stream = service; + for (struct ll_entry *e = inst->topo_groups->group_list->head; e; e = e->next) { + struct TOPO_GROUP *group = (struct TOPO_GROUP *)e; + if (group->group_type != TOPO_GROUP_TYPE_CHAT) continue; + struct stream_member_state state; + int result = stream_member_load(inst, group->group_id, inst->node_id, &state, NULL); + if (result == STREAM_OK && !state.published && !state.count) continue; + if (result == STREAM_WAIT || result == STREAM_ERR_NOT_MEMBER) continue; + if (service_publish_options(service, group) != STREAM_OK) { + service->initializing = 0; stream_service_destroy(inst); return STREAM_ERR_STATE; + } + } + if (member_sync_add_props_cbk(inst, member_changed, service) < 0) { + service->initializing = 0; stream_service_destroy(inst); return STREAM_ERR_MEMORY; + } + service->registered = 1; service->initializing = 0; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream service initialized node=%llx", (unsigned long long)inst->node_id); + return STREAM_OK; +} + +/* Detached handles принадлежат callers и остаются безопасными для close после остановки чата/ядра. */ +void stream_service_destroy(struct UTUN_INSTANCE *inst) { + struct stream_service *service = inst ? inst->stream : NULL; + if (!service) return; + if (service->dispatching || service->initializing) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream service destroy inside callback/init rejected"); return; + } + for (struct stream_session *s = service->sessions; s; s = s->next) + if (s->holds) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream stop during active operation rejected"); return; } + service->stopping = 1; + if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service); + if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile); + while (service->sessions) { + struct stream_session *s = service->sessions; + if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0); + else session_free(s); + } + for (struct STREAM_HANDLE *h = service->handles; h; h = h->next) { h->service = NULL; h->session = NULL; h->reason = STREAM_CLOSE_SERVICE; } + inst->stream = NULL; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream service stopped node=%llx", (unsigned long long)inst->node_id); + u_free(service); +} + +void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) { + struct stream_service *service = inst ? inst->stream : NULL; + if (!service) return; + struct stream_session *s = service->sessions; + while (s) { + struct stream_session *next = s->next; + if (!s->closing && s->descriptor.group_id == group_id) session_close(s, STREAM_CLOSE_GROUP, 1, 0); + s = next; + } +} + +void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg) { + if (!inst || !inst->stream || inst->stream->stopping) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream callback before init"); return; } + inst->stream->callback = callback; inst->stream->arg = arg; + if (callback) for (struct stream_session *s = inst->stream->sessions; s; s = s->next) + if (!s->closing) session_schedule(s); +} + +/* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */ +static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, + 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)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source already owned with another publication/epoch"); return STREAM_ERR_STATE; + } + struct STREAM_HANDLE *h = u_calloc(1, sizeof(*h)); + if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream handle allocation failed"); return STREAM_ERR_MEMORY; } + if (created) { + s = u_calloc(1, sizeof(*s)); + if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session allocation failed"); u_free(h); return STREAM_ERR_MEMORY; } + s->service = service; s->descriptor = *d; memcpy(s->public_key, key, 32); + s->output = u_malloc((uint32_t)stream_input_size(d)); + int result = s->output ? STREAM_OK : STREAM_ERR_MEMORY; + if (result == STREAM_OK && publisher) { + result = stream_source_create(&s->source, inst->ua, d, inst->my_ed25519_privkey, source_ready, s); + if (result == STREAM_OK) s->buffer = stream_source_buffer(s->source); + } else if (result == STREAM_OK) result = stream_buffer_create(&s->buffer, d, key, STREAM_BUFFER_BYTES); + s->next = service->sessions; service->sessions = s; + if (result != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session initialization failed result=%d", result); + session_free(s); u_free(h); return result; + } + } + h->service = service; h->session = s; h->publisher = publisher; + h->next = service->handles; service->handles = h; s->owners++; + s->holds++; + int result = created ? service_publish_options(service, stream_group(inst, d->group_id)) : STREAM_OK; + if (s->closing && result == STREAM_OK) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream open cancelled during member callbacks"); result = STREAM_ERR_STATE; + } + if (result != STREAM_OK) { + 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); } + u_free(h); s->holds--; + if (!s->holds && !s->delivery) session_free(s); + return result; + } + s->holds--; + *out = h; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream handle opened group=%llx source=%llx epoch=%llx publisher=%d owners=%u", + (unsigned long long)d->group_id, (unsigned long long)d->source_node_id, + (unsigned long long)d->source_epoch, publisher, s->owners); + return STREAM_OK; +} + +/* Контекст не создаётся для неизвестной группы, неавторизованного self или неподтверждённого ключа. */ +static int stream_open_check(struct UTUN_INSTANCE *inst, uint64_t group_id, struct STREAM_HANDLE **out) { + if (!out || *out || !inst || !inst->chat_started || !inst->stream || inst->stream->stopping || !stream_group(inst, group_id)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream open invalid state/group=%llx", (unsigned long long)group_id); return STREAM_ERR_STATE; + } + struct stream_member_state own; uint8_t key[32]; + int result = stream_member_load(inst, group_id, inst->node_id, &own, key); + if (result != STREAM_OK) return result; + if (memcmp(key, inst->my_ed25519_pubkey, 32)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream own member key differs from core identity"); return STREAM_ERR_SIGNATURE; + } + return STREAM_OK; +} + +int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const struct stream_parameters *parameters, + struct STREAM_HANDLE **out) { + int result = stream_open_check(inst, group_id, out); + if (result != STREAM_OK) return result; + struct stream_descriptor d = {.group_id = group_id, .source_node_id = inst->node_id, + .n = parameters ? parameters->n : 8, .r = parameters ? parameters->r : 2, + .block_size = parameters ? parameters->block_size : STREAM_RS_DEFAULT_SIZE}; + do { + if (random_bytes((uint8_t *)&d.source_epoch, sizeof(d.source_epoch)) != 0) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source epoch generation failed"); return STREAM_ERR_CRYPTO; + } + } 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, out); +} + +int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, 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, 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; + if (!state.published) { 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, out); +} + +/* Publisher авторитетно останавливает трансляцию; reader снимает только собственное владение. */ +void stream_close(struct STREAM_HANDLE *h) { + if (!h) return; + struct stream_service *service = h->service; + if (service) { + struct STREAM_HANDLE **link = &service->handles; + while (*link != h) link = &(*link)->next; + *link = h->next; + } + struct stream_session *s = h->session; + if (s && (h->publisher || --s->owners == 0)) session_close(s, STREAM_CLOSE_STOPPED, 1, 1); + u_free(h); +} + +int stream_handle_active(const struct STREAM_HANDLE *h) { return h && h->session && !h->session->closing && !h->session->invalidated; } +int stream_handle_reason(const struct STREAM_HANDLE *h) { + return h && h->session && h->session->invalidated ? h->session->reason : h ? h->reason : STREAM_CLOSE_STOPPED; +} + +int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) { + if (!info || !stream_handle_active(h)) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream info detached/invalid handle"); return STREAM_ERR_STATE; } + struct stream_session *s = h->session; + struct stream_info result = {.descriptor = s->descriptor, .owners = s->owners, .publisher = s->source != NULL, .finished = s->finished}; + stream_buffer_get_stats(s->buffer, &result.buffer); + *info = result; + return STREAM_OK; +} + +size_t stream_publish_capacity(const struct STREAM_HANDLE *h) { + if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream capacity not a publisher"); return 0; } + return stream_source_input_capacity(h->session->source); +} + +int stream_publish_write(struct STREAM_HANDLE *h, const uint8_t *data, size_t length, size_t *accepted) { + if (accepted) *accepted = 0; + if (!stream_handle_active(h) || !h->publisher || !accepted) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream write not an active publisher"); return STREAM_ERR_STATE; + } + struct stream_session *s = h->session; s->holds++; + int result = stream_source_write(s->source, data, length, accepted); + if (result != STREAM_OK && result != STREAM_ERR_ARGUMENT) session_close(s, STREAM_CLOSE_FAILED, 1, 1); + s->holds--; + if (s->closing && !s->holds && !s->delivery) session_free(s); + return result; +} + +int stream_publish_finish(struct STREAM_HANDLE *h) { + if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream finish not a publisher"); return STREAM_ERR_STATE; } + struct stream_session *s = h->session; + if (s->finished) return STREAM_OK; + s->holds++; + s->finished = 1; + int result = service_publish_options(s->service, stream_group(s->service->inst, s->descriptor.group_id)); + if (result != STREAM_OK || s->closing) { + if (!s->closing) s->finished = 0; + s->holds--; + if (s->closing && !s->holds && !s->delivery) session_free(s); + return result != STREAM_OK ? result : STREAM_ERR_STATE; + } + result = stream_source_finish(s->source); + if (result != STREAM_OK) session_close(s, STREAM_CLOSE_FAILED, 1, 1); + else session_schedule(s); + s->holds--; + if (s->closing && !s->holds && !s->delivery) session_free(s); + return result; +} + +/* Сначала только ключ/наличие подписки, затем crypto/FEC. Парсер пути и peer authorization — выше этого входа. */ +int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length) { + if (!inst || !wire || length < STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream ingest invalid length=%zu", length); return STREAM_ERR_ARGUMENT; + } + if (!inst->stream || inst->stream->stopping) return STREAM_WAIT; + uint64_t group = stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8), source = stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8); + struct stream_session *s = session_find(inst->stream, group, source); + if (!s) { DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "unsubscribed stream ignored group=%llx source=%llx", + (unsigned long long)group, (unsigned long long)source); return STREAM_WAIT; } + if (s->source) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "incoming DATA to own publisher rejected"); return STREAM_ERR_STATE; } + int result = stream_buffer_put(s->buffer, wire, length); + if (result == STREAM_OK) { + if (!s->anchored) { s->anchored = 1; s->next_seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); } + session_schedule(s); + } + return result; +} + +/* Выборка идёт из live BGP-путей группы, не из nodes.online или глобальных соединений. */ +int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate **out, size_t *count) { + if (!out || *out || !count || !stream_handle_active(h)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates invalid handle/output"); return STREAM_ERR_ARGUMENT; + } + struct stream_session *s = h->session; + struct UTUN_INSTANCE *inst = s->service->inst; + struct TOPO_GROUP *group = stream_group(inst, s->descriptor.group_id); + if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates group missing"); return STREAM_ERR_STATE; } + size_t capacity = (size_t)group->nodes->count; + struct stream_candidate *candidates = capacity ? u_calloc((uint32_t)capacity, sizeof(*candidates)) : NULL; + if (capacity && !candidates) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates allocation failed count=%zu", capacity); return STREAM_ERR_MEMORY; } + size_t found = 0; + for (struct ll_entry *e = group->nodes->head; e; e = e->next) { + struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e; + if (node->node_id == inst->node_id || !stream_member_online(inst, group->group_id, node->node_id)) continue; + struct TOPO_NODE *identity = topo_node_registry_find(inst->topo_groups, node->node_id); + int source = node->node_id == s->descriptor.source_node_id; + if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source) 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; + candidates[found++] = (struct stream_candidate){ .node_id = node->node_id, .source = (uint8_t)source, + .client_type = identity ? identity->client_type : UINT8_MAX }; + } + if (!found) { u_free(candidates); candidates = NULL; } + *out = candidates; *count = found; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidates source=%llx eligible=%zu BGP=%zu", + (unsigned long long)s->descriptor.source_node_id, found, capacity); + return STREAM_OK; +} diff --git a/src/stream/stream.h b/src/stream/stream.h new file mode 100644 index 00000000..d9df809c --- /dev/null +++ b/src/stream/stream.h @@ -0,0 +1,81 @@ +/* Стриминг CHAT-сервиса: локальные handles владеют общим экземпляром трансляции. + * Все API — uasync-поток. Последний reader close освобождает экземпляр; publisher close + * останавливает свою трансляцию и отсоединяет её readers. Отсоединённый handle всё равно закрывает вызывающий. + * DATA/GAP/END/CLOSED callback отложен, data/описание заимствованы до его возврата. + * Close handles из callback допустим; остановка сервиса/группы и destroy ядра — вне callbacks. + * Источник и локальные подписки публикуются атомарно через опции собственного мембера. */ +#ifndef STREAM_H +#define STREAM_H + +#include "stream_buffer.h" + +#ifdef __cplusplus +extern "C" { +#endif + +struct UTUN_INSTANCE; +struct STREAM_HANDLE; +struct stream_service; + +struct stream_parameters { uint8_t n, r; uint16_t block_size; }; + +enum stream_event_type { STREAM_EVENT_DATA, STREAM_EVENT_GAP, STREAM_EVENT_END, STREAM_EVENT_CLOSED }; +enum stream_close_reason { + STREAM_CLOSE_STOPPED, STREAM_CLOSE_GROUP, STREAM_CLOSE_SERVICE, STREAM_CLOSE_SOURCE_CHANGED, + STREAM_CLOSE_MEMBER, STREAM_CLOSE_FAILED +}; + +struct stream_event { + struct stream_descriptor descriptor; + enum stream_event_type type; + uint64_t first_seq, last_seq; + const uint8_t *data; + size_t length; + int reason; +}; +typedef void (*stream_event_fn)(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg); + +struct stream_info { + struct stream_descriptor descriptor; + struct stream_buffer_stats buffer; + unsigned owners; + int publisher; + int finished; +}; + +struct stream_candidate { + uint64_t node_id; + uint8_t client_type; + uint8_t source; +}; + +int stream_service_init(struct UTUN_INSTANCE *inst); +void stream_service_destroy(struct UTUN_INSTANCE *inst); +void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id); +void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg); + +/* out заранее NULL. parameters=NULL — 6+2 и полный подписанный блок 1300 байт. + * Одна публикация этого узла в группе; новый запуск получает новый source_epoch. */ +int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const struct stream_parameters *parameters, + struct STREAM_HANDLE **out); +/* Параметры берутся из подписанного профиля источника; дубликаты локальных владельцев делят один буфер. + * WAIT — источник ещё не объявил/не подтвердил трансляцию. Offline не создаёт ложного онлайн-кандидата. */ +int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out); +void stream_close(struct STREAM_HANDLE *handle); +int stream_handle_active(const struct STREAM_HANDLE *handle); +int stream_handle_reason(const struct STREAM_HANDLE *handle); +int stream_get_info(const struct STREAM_HANDLE *handle, struct stream_info *info); + +size_t stream_publish_capacity(const struct STREAM_HANDLE *handle); +int stream_publish_write(struct STREAM_HANDLE *handle, const uint8_t *data, size_t length, size_t *accepted); +/* Сбросить остаток, снять статус публикации; DATA перед END. Буфер живёт до close. */ +int stream_publish_finish(struct STREAM_HANDLE *handle); + +/* Только подходящие online-подписчики этой группы/эпохи; мобильные relay исключены, источник допускается. + * out заранее NULL; массив освобождается u_free(). Наличие кандидата не гарантирует прямой транспорт. */ +int stream_get_candidates(const struct STREAM_HANDLE *handle, struct stream_candidate **out, size_t *count); + +#ifdef __cplusplus +} +#endif +#endif diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h new file mode 100644 index 00000000..a6a0b9e8 --- /dev/null +++ b/src/stream/stream_priv.h @@ -0,0 +1,11 @@ +/* Внутренний вход проверенного DATA-dispatcher: авторизация пира/handle/пути делается сетевым протоколом. + * Подпись источника проверяется кольцом. Отсутствие локальной подписки отсекается до crypto/FEC. + * Внешний frontend использует stream.h, а не прямую вставку сетевых блоков. */ +#ifndef STREAM_PRIV_H +#define STREAM_PRIV_H + +#include "stream.h" + +int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length); + +#endif diff --git a/src/utun_instance.c b/src/utun_instance.c index 4aa4d24d..57379bd1 100644 --- a/src/utun_instance.c +++ b/src/utun_instance.c @@ -1,5 +1,6 @@ // utun_instance.c - Root instance implementation #include "utun_instance.h" +#include "stream/stream.h" #include "etcp_router.h" #include "config_parser.h" #include "config_updater.h" @@ -777,7 +778,7 @@ int chat_service_start(struct UTUN_INSTANCE* instance) { snprintf(db_file, sizeof(db_file), "%s/chats.db", instance->config->global.db_path); if (db_sync_enable(instance) < 0) return -1; topo_group_invite_init(instance); - if (chat_core_init(instance, db_file) < 0 || chat_sync_init(instance) < 0) goto fail; + if (chat_core_init(instance, db_file) < 0 || chat_sync_init(instance) < 0 || stream_service_init(instance) < 0) goto fail; instance->media_async = media_async_create(); if (!instance->media_async || media_delivery_init(instance) < 0) goto fail; if (member_files_init(instance) < 0 || dm_mailbox_init(instance) < 0 || dm_core_init(instance) < 0 || @@ -839,6 +840,7 @@ void chat_service_stop(struct UTUN_INSTANCE* instance) { instance->chat_started = 0; chat_headless_control_destroy(instance); call_headless_destroy(instance); radio_headless_destroy(instance); + stream_service_destroy(instance); chat_media_startup_backfill_destroy(instance); dm_media_quiesce(instance); member_files_quiesce(instance); diff --git a/src/utun_instance.h b/src/utun_instance.h index a7cf2e3d..69b6a859 100644 --- a/src/utun_instance.h +++ b/src/utun_instance.h @@ -51,6 +51,7 @@ struct file_transfer; struct call_ctx; /* call/call.c */ struct radio_instance; /* radio/radio.c */ struct radio_headless; /* radio/radio_headless.c */ +struct stream_service; /* stream/stream.c — подписки и публикация данных */ struct uasync_s; struct ROUTE_TABLE; struct ETCP_CONN; @@ -239,6 +240,7 @@ struct UTUN_INSTANCE { void* call_audio; // call/call_headless.c — audio stream socket struct radio_instance* radio; // radio/radio.c — PTT walkie-talkie (frame cb + counters) struct radio_headless* radio_headless; // radio/radio_headless.c — audio stream socket + struct stream_service* stream; // живёт в чат-сервисе, независимо от UTUN struct chat_setting_state chat_settings; // текущие настройки чата chat_event_handler_fn chat_event_handler; // получатель событий чата void* headless; // контекст TCP-управления headless-чатом diff --git a/tests/Makefile.am b/tests/Makefile.am index c44c39d4..05a32cf8 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -94,6 +94,7 @@ endif # Короткие интеграционные проверки (до 1с). check_PROGRAMS += \ test_stream_core \ + test_stream_service \ test_u_async_comprehensive \ test_nat_detection \ test_audio_compressor \ @@ -449,6 +450,10 @@ test_stream_core_SOURCES = test_stream_core.c $(test_clock_sources) test_stream_core_LDADD = $(LIBUTUN) $(COMMON_LIBS) test_stream_core_LDFLAGS = $(test_clock_ldflags) +test_stream_service_SOURCES = test_stream_service.c $(test_clock_sources) +test_stream_service_LDADD = $(LIBUTUN) $(COMMON_LIBS) +test_stream_service_LDFLAGS = $(test_clock_ldflags) + test_intensive_memory_pool_SOURCES = test_intensive_memory_pool.c test_intensive_memory_pool_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_intensive_memory_pool_LDADD = $(COMMON_LIBS) diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c new file mode 100644 index 00000000..e582f030 --- /dev/null +++ b/tests/test_stream_service.c @@ -0,0 +1,315 @@ +/* Живые CHAT-ядра и Merkle: атомарные опции, локальные владельцы, BGP-кандидаты и teardown. */ +#include +#include +#include + +#include + +#include "../src/stream/stream.h" +#include "../src/stream/stream_member.h" +#include "../src/stream/stream_priv.h" +#include "../src/chat/chat_core.h" +#include "../src/chat/merkle_tree.h" +#include "../src/routing_layer/topo_node_sqlite.h" +#include "../src/utun_instance.h" +#include "../lib/debug_config.h" +#include "../lib/json_flat.h" +#include "../lib/mem.h" +#include "test_clock.h" +#include "test_utils.h" + +#define GROUP1 UINT64_C(710001) +#define GROUP2 UINT64_C(710002) +#define CHECK(expr) do { if (!(expr)) { \ + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_service failed line=%d: %s", __LINE__, #expr); exit(1); \ +} } while (0) + +static struct UTUN_INSTANCE *nodes[4]; +static struct UASYNC *ua; +static uint8_t group_private[32], group_public[32], group_x[32], expected[1000]; +static struct STREAM_HANDLE *reader1, *reader2; +static unsigned data_events[4], end_events[4], closed_events[4]; +static int close_readers_in_callback; + +static void hex(const uint8_t *input, char *out) { + for (unsigned i = 0; i < 32; i++) snprintf(out + i * 2, 3, "%02x", input[i]); +} + +/* Проверяем заимствованные байты и после закрытия последнего локального владельца внутри callback. */ +static void on_stream(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)arg; + unsigned node = 0; + while (node < 4 && nodes[node] != inst) node++; + CHECK(node < 4 && event->descriptor.group_id == GROUP1); + if (event->type == STREAM_EVENT_DATA) { + CHECK(event->length == sizeof(expected) && !memcmp(event->data, expected, sizeof(expected))); + data_events[node]++; + if (node == 1 && close_readers_in_callback) { + close_readers_in_callback = 0; + stream_close(reader1); reader1 = NULL; + stream_close(reader2); reader2 = NULL; + CHECK(!memcmp(event->data, expected, sizeof(expected))); + } + } else if (event->type == STREAM_EVENT_END) end_events[node]++; + else if (event->type == STREAM_EVENT_CLOSED) closed_events[node]++; + else CHECK(0); +} + +/* Реальные удостоверения участников: join/update/tree подписи, одинаковые на всех узлах. */ +static void put_member(unsigned destination, unsigned member, uint64_t group) { + char channel[32], userinfo[256], name[32]; + snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group); + snprintf(name, sizeof(name), "node%u", member); + snprintf(userinfo, sizeof(userinfo), "{\"name\":\"%s\",\"pad\":\"retained\",\"avatar_server\":\"0011223344556677\"," + "\"avatar_file\":\"retained-avatar\"}", name); + uint8_t join[64], tree[64], update[64], body[MS_USERINFO_MAX + 112]; + int size = member_sync_build_join_msg(group_x, group_public, nodes[member]->node_id, + nodes[member]->my_keys.public_key, 123, body, sizeof(body)); + CHECK(size > 0 && sc_ed25519_sign(nodes[member]->my_ed25519_privkey, body, (size_t)size, join) == SC_OK); + CHECK(member_sync_sign_pubkey(group_private, nodes[member]->my_keys.public_key, tree) == 0); + size = member_sync_build_update_msg(join, 124, userinfo, body, sizeof(body)); + CHECK(size > 0 && sc_ed25519_sign(nodes[member]->my_ed25519_privkey, body, (size_t)size, update) == SC_OK); + CHECK(member_sync_put(nodes[destination], channel, nodes[member]->node_id, nodes[member]->my_keys.public_key, + nodes[member]->my_ed25519_pubkey, join, 123, update, 124, userinfo, NULL, NULL, 0, 0, tree) >= 0); +} + +static void run_until(unsigned *value, unsigned expected_value) { + uint64_t deadline = get_time_tb() + 100000; + while (*value < expected_value && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(*value == expected_value); +} + +/* Дожидаемся именно Merkle-профиля, не копируем stream-опции вручную между ядрами. */ +static void wait_state(unsigned receiver, unsigned member, uint64_t group, uint64_t epoch, int subscribed) { + uint64_t deadline = get_time_tb() + 150000; + while (get_time_tb() < deadline) { + struct stream_member_state state; + int result = stream_member_load(nodes[receiver], group, nodes[member]->node_id, &state, NULL); + if (result == STREAM_OK && (subscribed ? stream_member_has_subscription(&state, nodes[0]->node_id, epoch) + : state.published && state.source.source_epoch == epoch)) return; + test_clock_poll(ua, 100); + } + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Merkle stream state timeout receiver=%u member=%u group=%llu epoch=%llx", + receiver, member, (unsigned long long)group, (unsigned long long)epoch); + CHECK(0); +} + +static void profile(unsigned node, uint64_t group, char json[MS_USERINFO_MAX + 1], uint8_t signature[64], uint64_t *version) { + char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group); + CHECK(chat_core_get_my_profile(nodes[node], channel, json, MS_USERINFO_MAX + 1, signature, version) == 0); +} + +static void root_hash(unsigned node, uint64_t group, uint8_t hash[32]) { + char channel[32]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)group); + CHECK(merkle_tree_root(nodes[node]->topo_sqlite_db, MT_DATA_MEMBERS, channel, hash) == 0); +} + +/* Ошибка размера/SQLite не меняет подпись, userinfo, версию и производное Merkle-дерево. */ +static void test_profile_atomicity(void) { + char before[MS_USERINFO_MAX + 1], after[MS_USERINFO_MAX + 1], huge[MS_USERINFO_MAX + 1]; + uint8_t before_signature[64], after_signature[64], before_hash[32], after_hash[32]; + uint64_t before_version, after_version; + profile(0, GROUP1, before, before_signature, &before_version); root_hash(0, GROUP1, before_hash); + memset(huge, 'p', MS_USERINFO_MAX); huge[MS_USERINFO_MAX] = 0; + struct chat_profile_option options[] = {{"stream_source", "0000000000000001:8:2:1189"}, {"stream_subscriptions", huge}}; + CHECK(chat_core_set_my_profile_options(nodes[0], "710001", options, 2) < 0); + profile(0, GROUP1, after, after_signature, &after_version); root_hash(0, GROUP1, after_hash); + CHECK(!strcmp(before, after) && before_version == after_version && !memcmp(before_signature, after_signature, 64)); + CHECK(!memcmp(before_hash, after_hash, 32)); + CHECK(sqlite3_exec(nodes[0]->topo_sqlite_db, "CREATE TRIGGER reject_stream BEFORE UPDATE OF userinfo ON peers_710001 " + "BEGIN SELECT RAISE(ABORT,'injected stream profile failure'); END", NULL, NULL, NULL) == SQLITE_OK); + struct STREAM_HANDLE *failed = NULL; + CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &failed) < 0 && !failed); + CHECK(sqlite3_exec(nodes[0]->topo_sqlite_db, "DROP TRIGGER reject_stream", NULL, NULL, NULL) == SQLITE_OK); + profile(0, GROUP1, after, after_signature, &after_version); root_hash(0, GROUP1, after_hash); + CHECK(!strcmp(before, after) && before_version == after_version && !memcmp(before_hash, after_hash, 32)); + CHECK(!memcmp(before_signature, after_signature, 64)); + options[0] = (struct chat_profile_option){"pad", "retained"}; options[1] = (struct chat_profile_option){"extra", "value"}; + CHECK(chat_core_set_my_profile_options(nodes[0], "710001", options, 2) == 0); + profile(0, GROUP1, after, after_signature, &after_version); + char value[64]; + CHECK(json_flat_get(after, "avatar_file", value, sizeof(value)) == 0 && !strcmp(value, "retained-avatar")); + CHECK(json_flat_get(after, "avatar_server", value, sizeof(value)) == 0 && !strcmp(value, "0011223344556677")); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] profile updates: retained avatar/name/options and atomic size/DB rollback"); +} + +/* Канонический список исключает неоднозначные IDs, дубли и обрезание значений. */ +static void test_member_format(void) { + struct stream_member_state state = {.source = {.group_id = GROUP1, .source_node_id = 1, .source_epoch = 2, + .n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}, .published = 1, .count = 2, + .subscriptions = {{1, 2}, {UINT64_MAX, UINT64_MAX}}}; + char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1], json[MS_USERINFO_MAX + 1]; + CHECK(stream_member_format(&state, source, subscriptions) == STREAM_OK); + snprintf(json, sizeof(json), "{\"stream_source\":\"%s\",\"stream_subscriptions\":\"%s\"}", source, subscriptions); + struct stream_member_state parsed; + CHECK(stream_member_parse(GROUP1, 1, json, &parsed) == STREAM_OK); + CHECK(parsed.published && parsed.source.source_epoch == 2 && parsed.count == 2); + CHECK(parsed.subscriptions[1].source_node_id == UINT64_MAX && parsed.subscriptions[1].source_epoch == UINT64_MAX); + const char *bad[] = { + "{\"stream_subscriptions\":\"1:2\"}", + "{\"stream_subscriptions\":\"0000000000000001:0000000000000002,\"}", + "{\"stream_subscriptions\":\"0000000000000001:0000000000000002,0000000000000001:0000000000000002\"}", + "{\"stream_source\":\"0000000000000002:256:2:1189\"}", + "{\"stream_source\":\"0000000000000002:8:2:1189\"}", + "{\"stream_source\":\"\",\"stream_source\":\"\"}" + }; + for (unsigned i = 0; i < sizeof(bad) / sizeof(bad[0]); i++) + CHECK(stream_member_parse(GROUP1, 1, bad[i], &parsed) == STREAM_ERR_FORMAT); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] member format: full 64-bit IDs, canonical lists, duplicate/overflow rejection"); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); + debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); + debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_DISABLED); + test_member_format(); + size_t allocations = u_get_allocated_count(); + test_clock_init(100); utun_instance_set_tun_init_enabled(0); + ua = uasync_create(); CHECK(ua != NULL); + struct SC_MYKEYS channel; CHECK(sc_generate_keypair(&channel) == SC_OK); + memcpy(group_x, channel.public_key, 32); + uint8_t seed[64]; SHA512(channel.private_key, 32, seed); memcpy(group_private, seed, 32); + CHECK(sc_derive_ed25519_pubkey(channel.private_key, group_public) == SC_OK); + char directory[512] = "/tmp/utun-stream-service-XXXXXX"; CHECK(test_mkdtemp(directory) == 0); + for (unsigned i = 0; i < 4; i++) { + struct SC_MYKEYS keys; CHECK(sc_generate_keypair(&keys) == SC_OK); + char priv[65], pub[65], config[1600]; hex(keys.private_key, priv); hex(keys.public_key, pub); + snprintf(config, sizeof(config), "[global]\nmy_private_key=%s\nmy_public_key=%s\ndb_path=%s/node%u\n" + "[server:udp]\naddr=127.0.0.1:0\ntype=public\n[allowed_keys]\nallow_all=1\n" + "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n", priv, pub, directory, i); + nodes[i] = utun_instance_create_from_str(ua, config); CHECK(nodes[i] != NULL); + nodes[i]->client_type = i == 0 || i == 2 ? CLIENT_TYPE_MOBILE : CLIENT_TYPE_DESKTOP; + snprintf(nodes[i]->name, sizeof(nodes[i]->name), "node%u", i); + CHECK(utun_core_start(nodes[i]) == 0 && chat_service_start(nodes[i]) == 0 && nodes[i]->stream); + for (unsigned g = 0; g < 2; g++) { + char id[32]; uint64_t group = g ? GROUP2 : GROUP1; snprintf(id, sizeof(id), "%llu", (unsigned long long)group); + uint8_t signature[64] = {0}; + CHECK(topo_node_sqlite_channel_put(nodes[i]->topo_sqlite_db, id, "stream", nodes[0]->node_id, + group_x, NULL, group_public, i ? NULL : group_private, signature) == 0); + chat_core_ensure_channel_ready(nodes[i], id); + } + } + for (unsigned destination = 0; destination < 4; destination++) { + for (unsigned member = 0; member < 4; member++) { + put_member(destination, member, GROUP1); put_member(destination, member, GROUP2); + struct TOPO_NODE *identity = topo_node_registry_find(nodes[member]->topo_groups, nodes[member]->node_id); + CHECK(identity && topo_node_sqlite_snapshot_put(nodes[destination]->topo_sqlite_db, identity) >= 0); + } + } + struct TOPO_PEER_REQUEST *connections[3] = {0}; + for (unsigned i = 1; i < 4; i++) + CHECK(topo_group_peer_open(topo_groups_find(nodes[0]->topo_groups, GROUP1), nodes[i]->node_id, &connections[i - 1]) == 0); + uint64_t deadline = get_time_tb() + 150000; + while (get_time_tb() < deadline) { + int ready = 1; + for (unsigned i = 0; i < 3; i++) if (topo_group_peer_phase(connections[i]) != TOPO_PEER_READY) ready = 0; + if (ready) break; + test_clock_poll(ua, 100); + } + for (unsigned i = 0; i < 3; i++) CHECK(topo_group_peer_phase(connections[i]) == TOPO_PEER_READY); + test_profile_atomicity(); + stream_set_event_callback(nodes[0], on_stream, NULL); stream_set_event_callback(nodes[1], on_stream, NULL); + struct STREAM_HANDLE *publisher = NULL, *publisher2 = NULL, *other_group = NULL, *mobile = NULL, *desktop = NULL; + CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK); + CHECK(stream_publish_open(nodes[0], GROUP2, NULL, &publisher2) == STREAM_OK); + struct stream_info info; CHECK(stream_get_info(publisher, &info) == STREAM_OK); + uint64_t epoch = info.descriptor.source_epoch; + for (unsigned i = 1; i < 4; i++) wait_state(i, 0, GROUP1, epoch, 0); + CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[0]->node_id, &reader1) == STREAM_OK); + CHECK(stream_subscribe_open(nodes[1], GROUP1, nodes[0]->node_id, &reader2) == STREAM_OK); + CHECK(stream_get_info(reader1, &info) == STREAM_OK && info.owners == 2 && info.buffer.blocks == 0); + CHECK(stream_subscribe_open(nodes[2], GROUP1, nodes[0]->node_id, &mobile) == STREAM_OK); + CHECK(stream_subscribe_open(nodes[3], GROUP1, nodes[0]->node_id, &desktop) == STREAM_OK); + struct stream_member_state state; + CHECK(stream_member_load(nodes[1], GROUP1, nodes[1]->node_id, &state, NULL) == STREAM_OK && state.count == 1); + for (unsigned i = 2; i < 4; i++) wait_state(1, i, GROUP1, epoch, 1); + struct stream_candidate *candidates = NULL; size_t count = 0; + CHECK(stream_get_candidates(reader1, &candidates, &count) == STREAM_OK); + int source_found = 0, desktop_found = 0; + for (size_t i = 0; i < count; i++) { + CHECK(candidates[i].node_id != nodes[2]->node_id); + if (candidates[i].node_id == nodes[0]->node_id) { CHECK(candidates[i].source); source_found = 1; } + if (candidates[i].node_id == nodes[3]->node_id) desktop_found = 1; + } + CHECK(source_found && desktop_found); u_free(candidates); candidates = NULL; + struct TOPO_GROUP *group = topo_groups_find(nodes[1]->topo_groups, GROUP1); + struct TOPO_GROUP_NODE *peer = topo_node_find_by_id(group, nodes[3]->node_id); CHECK(peer != NULL); + topo_nodeq_remove_node(group, peer); /* BGP withdraw не удаляет сохранённую опцию мембера. */ + CHECK(stream_member_load(nodes[1], GROUP1, nodes[3]->node_id, &state, NULL) == STREAM_OK); + CHECK(stream_member_has_subscription(&state, nodes[0]->node_id, epoch)); + CHECK(stream_get_candidates(reader1, &candidates, &count) == STREAM_OK); + for (size_t i = 0; i < count; i++) CHECK(candidates[i].node_id != nodes[3]->node_id); + u_free(candidates); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] Merkle subscriptions and BGP gating; mobile source allowed, mobile relay excluded"); + + /* GROUP2 не соединена, но подписанный профиль можно читать из общей таблицы. */ + struct stream_info second; CHECK(stream_get_info(publisher2, &second) == STREAM_OK); + char source_value[MS_USERINFO_MAX + 1], subs_value[MS_USERINFO_MAX + 1]; + CHECK(stream_member_load(nodes[0], GROUP2, nodes[0]->node_id, &state, NULL) == STREAM_OK); + CHECK(stream_member_format(&state, source_value, subs_value) == STREAM_OK); + char profile2[MS_USERINFO_MAX + 1]; uint8_t sig2[64], join2[64]; uint64_t version2, join_ts; + profile(0, GROUP2, profile2, sig2, &version2); + CHECK(topo_node_sqlite_member_get_join(nodes[0]->topo_sqlite_db, "710002", nodes[0]->node_id, join2, &join_ts) == 0); + uint8_t tree2[64]; uint64_t parent; + CHECK(topo_node_sqlite_member_get_sign(nodes[0]->topo_sqlite_db, "710002", nodes[0]->node_id, &parent, tree2) == 0); + CHECK(member_sync_put(nodes[1], "710002", nodes[0]->node_id, nodes[0]->my_keys.public_key, nodes[0]->my_ed25519_pubkey, + join2, join_ts, sig2, version2, profile2, NULL, NULL, 0, parent, tree2) >= 0); + CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[0]->node_id, &other_group) == STREAM_OK); + + for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3); + struct stream_encoder *encoder = NULL; + CHECK(stream_encoder_create(&encoder, &info.descriptor, nodes[0]->my_ed25519_privkey) == STREAM_OK); + const uint8_t *blocks[RS_MAX_BLOCKS]; + CHECK(stream_encoder_encode(encoder, 100, 65530, 0, expected, sizeof(expected), blocks) == STREAM_OK); + close_readers_in_callback = 1; + for (unsigned i = 0; i < 8; i++) if (i != 1 && i != 5) CHECK(stream_ingest_block(nodes[1], blocks[i], 1300) == STREAM_OK); + run_until(&data_events[1], 1); run_until(&closed_events[1], 1); + CHECK(!reader1 && !reader2 && stream_handle_active(other_group)); + uint8_t invalid[1300]; memcpy(invalid, blocks[0], sizeof(invalid)); invalid[1299] ^= 1; + CHECK(stream_ingest_block(nodes[1], invalid, sizeof(invalid)) == STREAM_WAIT); /* Нет подписки — нет crypto. */ + stream_encoder_destroy(encoder); + size_t accepted = 0; + CHECK(stream_publish_write(publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected)); + CHECK(stream_publish_finish(publisher) == STREAM_OK); + run_until(&data_events[0], 1); run_until(&end_events[0], 1); + CHECK(stream_member_load(nodes[0], GROUP1, nodes[0]->node_id, &state, NULL) == STREAM_OK && !state.published && state.count == 1); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] opaque data/FEC delivery, close inside callback, independent group and DATA before END"); + + stream_close(publisher); publisher = NULL; + CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK); + CHECK(stream_get_info(publisher, &info) == STREAM_OK && info.descriptor.source_epoch != epoch); + deadline = get_time_tb() + 150000; + while ((stream_handle_active(mobile) || stream_handle_active(desktop)) && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(!stream_handle_active(mobile) && !stream_handle_active(desktop)); + CHECK(stream_handle_reason(mobile) == STREAM_CLOSE_SOURCE_CHANGED && stream_handle_reason(desktop) == STREAM_CLOSE_SOURCE_CHANGED); + stream_close(mobile); stream_close(desktop); + chat_core_delete_channel(nodes[1], "710002"); + CHECK(!stream_handle_active(other_group) && stream_handle_reason(other_group) == STREAM_CLOSE_GROUP); + stream_close(other_group); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] source epoch invalidation and channel deletion detach local handles"); + + char stale[MS_USERINFO_MAX + 1]; uint8_t stale_signature[64]; uint64_t stale_version; + profile(0, GROUP1, stale, stale_signature, &stale_version); + CHECK(utun_service_start(nodes[0]) == 0); + sqlite3 *db = nodes[0]->topo_sqlite_db; struct ETCP_SOCKET *sockets = nodes[0]->etcp_sockets; + CHECK(stream_publish_write(publisher2, expected, 17, &accepted) == STREAM_OK); + chat_service_stop(nodes[0]); + CHECK(!nodes[0]->stream && nodes[0]->utun_started && nodes[0]->topo_sqlite_db == db && nodes[0]->etcp_sockets == sockets); + CHECK(!stream_handle_active(publisher) && !stream_handle_active(publisher2)); + stream_close(publisher); stream_close(publisher2); + sqlite3_stmt *restore = NULL; + CHECK(sqlite3_prepare_v2(db, "UPDATE peers_710001 SET userinfo=?,update_sig=?,update_ts=? WHERE node_id=?", -1, &restore, NULL) == SQLITE_OK); + sqlite3_bind_text(restore, 1, stale, -1, SQLITE_STATIC); sqlite3_bind_blob(restore, 2, stale_signature, 64, SQLITE_STATIC); + sqlite3_bind_int64(restore, 3, (sqlite3_int64)stale_version); sqlite3_bind_int64(restore, 4, (sqlite3_int64)nodes[0]->node_id); + CHECK(sqlite3_step(restore) == SQLITE_DONE); sqlite3_finalize(restore); + CHECK(chat_service_start(nodes[0]) == 0 && nodes[0]->stream); + CHECK(stream_member_load(nodes[0], GROUP1, nodes[0]->node_id, &state, NULL) == STREAM_OK && !state.published && !state.count); + CHECK(data_events[0] == 1); /* Pending 100ms batch не выполнилась после stop. */ + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] CHAT stop keeps UTUN/core; source timer cancelled; stale signed options cleared on restart"); + for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); + for (unsigned i = 0; i < 4; i++) utun_instance_destroy(nodes[i]); + uasync_destroy(ua, 1); test_clock_finish(); + CHECK(u_get_allocated_count() == allocations); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream service checks complete, no tracked allocations remain"); + return 0; +}