jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
2793 Commits
15 Branches
2 Tags
160 MiB
 
 
 
 
 
 
Tree: 5def12351f
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '5def12351f'
${ noResults }
utun2/src/stream/stream.c

934 lines
55 KiB
Raw Blame History

#include <stdio.h>
#include <string.h>
#include <limits.h>
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/platform_compat.h"
#include "../../lib/u_async.h"
#include "../../lib/strbuf.h"
#include "../../lib/sqlite3.h"
#include "../utun_instance.h"
#include "../chat/chat_core.h"
#include "../chat/chat_event.h"
#include "stream_member.h"
#include "stream_source.h"
#include "stream_priv.h"
#include "stream_optimizer.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;
struct stream_metrics *metrics;
struct stream_optimizer *optimizer;
void *delivery;
uint64_t next_seq;
uint64_t start_seq, last_seq, passed_before;
uint64_t restart_from, restart_last;
uint8_t end[STREAM_END_SIZE];
int end_valid;
unsigned owners, relay_owners, holds;
int anchored, closing, invalidated, finished, end_delivered, reason;
int restarting, restart_gap, notify_close;
};
struct STREAM_HANDLE {
struct STREAM_HANDLE *next;
struct stream_service *service;
struct stream_session *session;
int publisher;
int relay;
int selection_owned;
int reason;
};
struct stream_service {
struct UTUN_INSTANCE *inst;
struct stream_session *sessions;
struct STREAM_HANDLE *handles;
struct stream_network *network;
stream_event_fn callback;
void *arg;
void *reconcile;
void *sources_pending;
uint64_t sources_group;
char *sources_snapshot;
int sources_valid;
int stopping, dispatching, registered, initializing;
};
static void session_deliver(void *arg);
static void service_reconcile(void *arg);
/* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */
int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) {
if (!inst || !out || *out || !group_id) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list invalid arguments"); return STREAM_ERR_ARGUMENT;
}
struct TOPO_GROUP *group = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (!group || group->stopping || group->group_type != TOPO_GROUP_TYPE_CHAT) return STREAM_WAIT;
for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source namespace invalid group=%llx", (unsigned long long)group_id);
return STREAM_ERR_STATE;
}
sqlite3 *db = chat_core_get_db(inst);
if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list missing DB"); return STREAM_ERR_STATE; }
char query[320];
snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') "
"FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id ORDER BY p.node_id", group->channel_id);
struct strbuf json = strbuf_new();
sqlite3_stmt *statement = NULL;
int result = STREAM_OK, count = 0;
if (strbuf_addf(&json, "[") < 0) result = STREAM_ERR_MEMORY;
if (result == STREAM_OK && sqlite3_prepare_v2(db, query, -1, &statement, NULL) != SQLITE_OK) result = STREAM_ERR_STATE;
int step = SQLITE_DONE;
while (result == STREAM_OK && (step = sqlite3_step(statement)) == SQLITE_ROW) {
uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0);
if (!stream_member_online(inst, group_id, node)) continue;
struct stream_member_state state;
int loaded = stream_member_load(inst, group_id, node, &state, NULL);
if (loaded == STREAM_WAIT || loaded == STREAM_ERR_NOT_MEMBER) continue;
if (loaded < 0) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source excluded group=%llx node=%llx profile_result=%d",
(unsigned long long)group_id, (unsigned long long)node, loaded); continue;
}
if (!state.published) continue;
if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "",
(unsigned long long)node, (unsigned long long)state.source.source_epoch, node == inst->node_id) < 0
|| strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0
|| strbuf_addf(&json, "}") < 0) result = STREAM_ERR_MEMORY;
count++;
}
if (result == STREAM_OK && step != SQLITE_DONE) result = STREAM_ERR_STATE;
sqlite3_finalize(statement);
if (result == STREAM_OK && strbuf_addf(&json, "]") < 0) result = STREAM_ERR_MEMORY;
if (result == STREAM_OK) *out = strbuf_detach(&json);
strbuf_free(&json);
if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source list failed group=%llx result=%d DB=%s",
(unsigned long long)group_id, result, sqlite3_errmsg(db));
else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source list group=%llx count=%d", (unsigned long long)group_id, count);
return result;
}
/* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */
static void sources_deliver(void *arg) {
struct stream_service *service = arg;
service->sources_pending = NULL;
if (service->stopping || !service->sources_group) return;
char *json = NULL;
int result = stream_sources_json(service->inst, service->sources_group, &json);
int valid = result == STREAM_OK || result == STREAM_WAIT;
const char *list = json ? json : "[]";
if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) {
u_free(json); return;
}
size_t length = strlen(list);
uint8_t *event = length <= INT_MAX - 9 ? u_malloc((uint32_t)(9 + length)) : NULL;
if (!event) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source event allocation failed bytes=%zu", length); u_free(json); return;
}
stream_write_be(event, service->sources_group, 8); event[8] = (uint8_t)valid;
memcpy(event + 9, list, length);
char *snapshot = json ? json : u_strdup(list);
if (!snapshot) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source snapshot allocation failed");
u_free(service->sources_snapshot); service->sources_snapshot = snapshot; service->sources_valid = valid;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu",
(unsigned long long)service->sources_group, valid, length);
chat_event_post(service->inst, CHAT_EVT_STREAM_SOURCES, event, (int)(length + 9));
u_free(event);
}
/* Наблюдение принадлежит сервису и снимается до освобождения ядра. */
int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id) {
struct stream_service *service = inst ? inst->stream : NULL;
if (!service || service->stopping) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id);
return STREAM_ERR_STATE;
}
service->sources_group = group_id;
u_free(service->sources_snapshot); service->sources_snapshot = NULL;
if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
service->sources_pending = group_id ? uasync_call_soon(inst->ua, service, sources_deliver) : NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx", (unsigned long long)group_id);
if (group_id && !service->sources_pending) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY;
}
return STREAM_OK;
}
/* Observer выполняет только учёт; пользовательские callbacks и отправка остаются отложенными. */
static void cache_event(void *arg, const uint8_t *wire, size_t length, int added) {
struct stream_session *s = arg;
if (s->closing || s->invalidated) return;
uint8_t sub = wire[STREAM_BLOCK_SUB_OFF]; uint64_t seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
if (added) {
if (!s->metrics[sub].ended) stream_metrics_observe(&s->metrics[sub], seq,
(uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
if (s->end_valid && seq == s->last_seq) stream_metrics_end(&s->metrics[sub]);
}
stream_network_cache_event(s->service->network, &s->descriptor, wire, length, added);
}
/* Активность 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,
!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
&& 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);
stream_optimizer_destroy(s->optimizer);
if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL);
if (s->source) stream_source_destroy(s->source);
else stream_buffer_destroy(s->buffer);
u_free(s->output);
u_free(s->metrics);
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->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);
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 (!s->service->stopping) {
s->holds++;
if (utun_set_client_activity(s->service->inst, s->service->inst->client_activity_requested) < 0)
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream close failed to restore requested core activity");
s->holds--;
}
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 = 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);
} else if (s->restart_gap) {
event.type = STREAM_EVENT_GAP; event.first_seq = s->restart_from; event.last_seq = s->restart_last;
s->restart_gap = 0; emit = again = 1;
} 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 == STREAM_WAIT || result == STREAM_GAP) && s->next_seq < s->passed_before) {
uint64_t next = s->passed_before, decoded;
if (stream_buffer_next_decodable(s->buffer, s->next_seq + 1, &decoded) == STREAM_OK && decoded < next) next = decoded;
event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = next - 1;
s->next_seq = next; emit = 1; again = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream consumer GAP source=%llx first=%llu last=%llu next=%llu",
(unsigned long long)s->descriptor.source_node_id, (unsigned long long)event.first_seq,
(unsigned long long)event.last_seq, (unsigned long long)next);
} else if (result < 0) {
session_close(s, STREAM_CLOSE_FAILED, 1, 1);
} else if (s->finished && s->end_valid && s->next_seq > s->last_seq && !s->end_delivered) {
event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = s->last_seq;
s->end_delivered = 1; emit = 1;
}
} else if (s->finished && s->end_valid && s->last_seq == UINT64_MAX && !s->end_delivered) {
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);
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->start_seq = s->next_seq = seq; }
session_schedule(s);
stream_network_changed(s->service->network);
}
/* Отложенная сверка не разрушает экземпляры из 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) return;
struct TOPO_GROUP *watched = service->sources_group ? stream_group(service->inst, service->sources_group) : NULL;
if (watched && channel && !strcmp(watched->channel_id, channel) && !service->sources_pending) {
service->sources_pending = uasync_call_soon(service->inst->ua, service, sources_deliver);
if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed");
}
if (!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;
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;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream membership invalidated group=%llx member=%llx own=%d result=%d",
(unsigned long long)group->group_id, (unsigned long long)node_id, own, result);
} else if (own && result == STREAM_OK && memcmp(key, service->inst->my_ed25519_pubkey, 32)) {
s->invalidated = 1; s->reason = STREAM_CLOSE_MEMBER;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream own member key changed group=%llx", (unsigned long long)group->group_id);
} else if (!own && 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;
if (stream_network_init(inst, &service->network) != STREAM_OK) { stream_service_destroy(inst); return STREAM_ERR_STATE; }
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->stopping) 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;
for (struct stream_session *s = service->sessions; s; s = s->next)
if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL);
for (struct stream_session *s = service->sessions; s; s = s->next) {
stream_optimizer_destroy(s->optimizer); s->optimizer = NULL;
}
stream_network_destroy(service->network); service->network = NULL;
if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service);
if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile);
if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending);
u_free(service->sources_snapshot);
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; }
if (utun_set_client_activity(inst, inst->client_activity_requested) < 0)
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream stop failed to restore requested core activity");
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;
if (service->sources_group == group_id && !service->sources_pending) {
service->sources_pending = uasync_call_soon(inst->ua, service, sources_deliver);
if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed");
}
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, 1);
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, 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 || (!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;
}
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));
s->metrics = u_calloc(d->n, sizeof(*s->metrics));
int result = s->output && s->metrics ? 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);
if (result == STREAM_OK && automatic && !publisher) result = stream_optimizer_create(inst, d, &s->optimizer);
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;
}
for (unsigned i = 0; i < d->n; i++) stream_metrics_init(&s->metrics[i]);
stream_buffer_set_observer(s->buffer, cache_event, s);
}
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 || 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;
}
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--; s->relay_owners -= relay;
if (created) session_close(s, STREAM_CLOSE_FAILED, 0, 0);
}
u_free(h); s->holds--;
if (s->closing && !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);
if (created && s->optimizer) stream_network_wake(service->network);
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, 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, 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, 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) {
return subscribe_open(inst, group_id, source_id, 1, out);
}
int stream_subscribe_manual_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) {
return subscribe_open(inst, group_id, source_id, 0, out);
}
int stream_selection_configure(struct STREAM_HANDLE *h, const struct stream_selection_settings *settings) {
if (!stream_handle_active(h) || !h->session->optimizer) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream selection configuration requires automatic receiver"); return STREAM_ERR_STATE;
}
return stream_optimizer_configure(h->session->optimizer, settings);
}
int stream_selection_get_info(const struct STREAM_HANDLE *h, struct stream_selection_info *info) {
if (!stream_handle_active(h) || !info) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream selection info invalid handle/output"); return STREAM_ERR_STATE; }
memset(info, 0, sizeof(*info));
if (h->session->optimizer) stream_optimizer_info(h->session->optimizer, info);
return STREAM_OK;
}
/* 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) {
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);
}
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;
}
/* Блочная публикация использует те же holds и обработку отказов, что и источник непрерывных байтов. */
int stream_publish_block(struct STREAM_HANDLE *h, const uint8_t *data, size_t length) {
if (!stream_handle_active(h) || !h->publisher) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream block write not an active publisher"); return STREAM_ERR_STATE;
}
struct stream_session *s = h->session; s->holds++;
int result = stream_source_write_block(s->source, data, length);
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) {
struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
s->last_seq = stats.blocks ? stats.highest_seq : UINT64_MAX;
result = stream_end_sign(&s->descriptor, s->last_seq, s->service->inst->my_ed25519_privkey, s->end);
s->end_valid = result == STREAM_OK;
if (s->end_valid) for (unsigned i = 0; i < s->descriptor.n; i++) stream_metrics_end(&s->metrics[i]);
}
if (result != STREAM_OK) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
else { session_schedule(s); stream_network_changed(s->service->network); }
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) {
struct stream_path path = {.count = 1};
if (wire && length >= STREAM_BLOCK_RS_OFF) path.nodes[0] = stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8);
return stream_ingest_path(inst, wire, length, &path);
}
/* Кольцо хранит уже дополненный путь: источник ... отправитель ... этот узел. */
int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length, const struct stream_path *path) {
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; }
if (!path || !path->count || path->count >= STREAM_PATH_MAX
|| stream_path_validate(path, source, path->nodes[path->count - 1], inst->node_id) != STREAM_OK)
return STREAM_ERR_FORMAT;
if (s->end_valid && (s->last_seq == UINT64_MAX || stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8) > s->last_seq)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA beyond signed END source=%llx", (unsigned long long)source);
return STREAM_ERR_CONFLICT;
}
struct stream_path retained = *path; retained.nodes[retained.count++] = inst->node_id;
int result = stream_buffer_put_path(s->buffer, wire, length, &retained);
if (result == STREAM_OK || (result == STREAM_DUPLICATE && !s->anchored)) {
if (!s->anchored) {
uint64_t received = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8);
uint64_t start = s->restarting && received < s->restart_from ? s->restart_from : received;
s->anchored = 1; s->start_seq = s->next_seq = start;
if (s->restarting && s->next_seq > s->restart_from) { s->restart_gap = 1; s->restart_last = s->next_seq - 1; }
if (s->restarting) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream restart anchored source=%llx received=%llu floor=%llu start=%llu",
(unsigned long long)source, (unsigned long long)received, (unsigned long long)s->restart_from, (unsigned long long)start);
s->restarting = 0;
}
session_schedule(s);
stream_network_changed(s->service->network);
}
return result;
}
struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst) { return inst && inst->stream ? inst->stream->network : NULL; }
int stream_service_needs_activity(const struct UTUN_INSTANCE *inst) {
if (!inst || !inst->stream || inst->stream->stopping) return 0;
for (struct stream_session *s = inst->stream->sessions; s; s = s->next)
if (!s->closing && !s->invalidated && s->owners) return 1;
return 0;
}
/* Доступ network только к существующей локальной подписке точной эпохи. */
int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct stream_local *local) {
if (!key || !local || !inst || !inst->stream || inst->stream->stopping) return STREAM_WAIT;
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,
.relay_only = !s->source && s->owners == s->relay_owners, .end = s->end_valid ? s->end : NULL};
return STREAM_OK;
}
/* Выданный локальный handle проверяется до создания сетевых ресурсов. */
int stream_handle_context(struct STREAM_HANDLE *h, struct UTUN_INSTANCE **inst, struct stream_descriptor *d) {
if (!inst || !d || !stream_handle_active(h) || h->session->source || (h->session->optimizer && !h->selection_owned)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "network receive requires active manual reader or internal selector handle"); return STREAM_ERR_STATE;
}
*inst = h->service->inst; *d = h->session->descriptor; return STREAM_OK;
}
/* END ретранслируется без изменений и не может завершить поток по заявлению поставщика без подписи источника. */
int stream_ingest_end(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, const uint8_t *wire, size_t length) {
struct stream_local local;
if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) return STREAM_WAIT;
struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
uint64_t last; int result = stream_end_verify(&s->descriptor, s->public_key, wire, length, &last);
if (result != STREAM_OK) return result;
struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
if ((s->end_valid && s->last_seq != last) || (stats.blocks && (last == UINT64_MAX || stats.highest_seq > last))) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "signed END conflicts with retained stream source=%llx", (unsigned long long)key->source_node_id);
return STREAM_ERR_CONFLICT;
}
memcpy(s->end, wire, STREAM_END_SIZE); s->end_valid = s->finished = 1; s->last_seq = last;
for (unsigned i = 0; i < s->descriptor.n; i++)
if (last == UINT64_MAX || (s->metrics[i].initialized && s->metrics[i].last_seq == last)) stream_metrics_end(&s->metrics[i]);
session_schedule(s); stream_network_changed(inst->stream->network); return STREAM_OK;
}
/* Общая метрика прихода субпотока в экземпляр, независимо от поставщика конкретного блока. */
int stream_local_metrics(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint8_t sub, uint64_t now,
struct stream_metric_snapshot *snapshot) {
struct stream_local local;
if (stream_local_get(inst, key, &local) != STREAM_OK || sub >= local.descriptor.n || !snapshot) return STREAM_WAIT;
struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
return stream_metrics_poll(&s->metrics[sub], now, snapshot);
}
/* Все выбранные upstream прошли эту границу; дальше старую неполную пачку они уже не выдадут. */
void stream_local_progress(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t before) {
struct stream_local local;
if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) return;
struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
if (before > s->passed_before) { s->passed_before = before; session_schedule(s); }
}
/* Явный новый старт не смешивается с DATA старых upstream handles; кольцо остаётся пригодным для relay. */
int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key) {
struct stream_local local;
if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream restart requires local receiver"); return STREAM_ERR_STATE;
}
struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
if (s->delivery) { uasync_call_soon_cancel(inst->ua, s->delivery); s->delivery = NULL; }
if (s->anchored) {
if (!s->restart_gap) s->restart_from = s->next_seq;
s->restarting = 1;
}
s->restart_gap = 0;
s->anchored = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u floor=%llu known=%d",
(unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
(unsigned long long)key->source_epoch, local.buffer != NULL, (unsigned long long)s->restart_from, s->restarting); return STREAM_OK;
}
/* Заимствованный локальный handle нужен только на время tick; владельцем остаётся сама подписка. */
void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms) {
struct stream_service *service = inst ? inst->stream : NULL;
if (!service || service->stopping) return;
for (struct stream_session *s = service->sessions; s; s = s->next) {
if (!s->optimizer || s->closing || s->invalidated) continue;
struct STREAM_HANDLE subscription = {.service = service, .session = s, .selection_owned = 1};
stream_optimizer_tick(s->optimizer, &subscription, now_ms);
}
}
/* Routed READY будит selector, не создавая пользовательского события или второго экземпляра. */
void stream_relay_ready(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer) {
struct stream_session *s = inst && inst->stream ? session_find(inst->stream, key->group_id, key->source_node_id) : NULL;
if (s && s->descriptor.source_epoch == key->source_epoch && s->optimizer) stream_optimizer_refresh(s->optimizer, peer);
}
/* Выборка идёт из 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;
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
&& 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) 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 };
}
if (!found) { u_free(candidates); candidates = NULL; }
*out = candidates; *count = found;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidates source=%llx eligible=%zu BGP=%zu",
(unsigned long long)s->descriptor.source_node_id, found, capacity);
return STREAM_OK;
}