jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
abe3d4bc89abc3e0cf700ce51f29781a961f5b37
utun2/src/stream/stream.c
T

1192 lines
70 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#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 "../radio/radio.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;
struct stream_media_store *media;
void *media_work;
uint64_t media_next, media_seek;
int media_seeking, media_failed;
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, decoded, closing, invalidated, finished, end_delivered, reason;
int restarting, restart_gap, notify_close;
int audio_enabled, radio_muted;
};
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_media_cursor *media_cursor;
};
struct source_watch {
struct source_watch *next;
struct stream_service *service;
uint64_t owner, group;
void *pending;
char *snapshot;
int valid;
};
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;
struct source_watch *watches;
int stopping, dispatching, registered, initializing;
};
static void session_deliver(void *arg);
static void service_reconcile(void *arg);
static void session_media_schedule(struct stream_session *s);
/* Общий каталог не выделяет 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 source_watch *watch = arg;
struct stream_service *service = watch->service;
watch->pending = NULL;
if (service->stopping) return;
char *json = NULL;
int result = stream_sources_json(service->inst, watch->group, &json);
int valid = result == STREAM_OK;
const char *list = json ? json : "[]";
if (watch->snapshot && valid == watch->valid && !strcmp(watch->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, watch->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(watch->snapshot); watch->snapshot = snapshot; watch->valid = valid;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu",
(unsigned long long)watch->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, uint64_t owner) {
struct stream_service *service = inst ? inst->stream : NULL;
if (!service || (service->stopping && group_id)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id);
return STREAM_ERR_STATE;
}
struct source_watch **link = &service->watches;
while (*link && (*link)->owner != owner) link = &(*link)->next;
struct source_watch *watch = *link;
if (!group_id) {
if (watch) {
if (watch->pending) uasync_call_soon_cancel(inst->ua, watch->pending);
*link = watch->next; u_free(watch->snapshot); u_free(watch);
}
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch removed owner=%llu", (unsigned long long)owner);
return STREAM_OK;
}
if (!watch) {
watch = u_calloc(1, sizeof(*watch));
if (!watch) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch allocation failed"); return STREAM_ERR_MEMORY; }
watch->service = service; watch->owner = owner; *link = watch;
}
watch->group = group_id;
u_free(watch->snapshot); watch->snapshot = NULL;
if (!watch->pending) watch->pending = uasync_call_soon(inst->ua, watch, sources_deliver);
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx owner=%llu",
(unsigned long long)group_id, (unsigned long long)owner);
if (!watch->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;
}
int stream_radio_muted(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id) {
struct stream_session *s = inst && inst->stream ? session_find(inst->stream, group_id, source_id) : NULL;
return s && s->radio_muted;
}
/* Курсор означает просмотр. Raw readers, publisher и relay не отключают рацию. */
static void session_radio_update(struct stream_session *s) {
int muted = 0;
if (!s->closing && !s->invalidated && !s->finished && !s->source && s->audio_enabled)
for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next)
if (h->session == s && h->media_cursor && !h->relay && !h->publisher) { muted = 1; break; }
if (muted == s->radio_muted) return;
s->radio_muted = muted;
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "stream viewer mute group=%llx source=%llx epoch=%llx muted=%d audio=%d",
(unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
(unsigned long long)s->descriptor.source_epoch, muted, s->audio_enabled);
radio_refresh_member_mute(s->service->inst, s->descriptor.group_id, s->descriptor.source_node_id);
}
/* Две опции заменяются одной подписанной записью; список отражает экземпляры, не отдельные 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; state.audio_enabled = s->audio_enabled; }
}
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->media_work) uasync_call_soon_cancel(s->service->inst->ua, s->media_work);
stream_media_store_destroy(s->media);
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;
session_radio_update(s);
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; }
if (s->media_work) { uasync_call_soon_cancel(s->service->inst->ua, s->media_work); s->media_work = 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);
}
/* Поздний reader ищет ближайший сохранённый KEY с ограниченным бюджетом, затем индексирует вперёд. */
static void session_media_work(void *arg) {
struct stream_session *s = arg; s->media_work = NULL; s->holds++;
int again = 0;
struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
if (!s->closing && !s->media_failed) {
for (unsigned work = 0; work < 8 && s->media_seeking; work++) {
size_t length = 0;
int result = stream_buffer_decode(s->buffer, s->media_seek, s->output, stream_input_size(&s->descriptor), &length);
if (result == STREAM_OK) {
struct stream_media_block b;
if (stream_media_parse(s->output, length, &b) != STREAM_OK) { s->media_failed = 1; break; }
for (unsigned i = 0; i < b.video_count; i++) {
if (!b.video[i].offset && (b.video[i].flags & STREAM_MEDIA_KEY)) {
s->media_next = s->media_seek; s->media_seeking = 0; break;
}
}
}
if (!s->media_seeking) break;
if (s->media_seek <= stats.expired_before) { s->media_seeking = 0; s->media_next = s->next_seq; break; }
s->media_seek--;
}
if (s->media_seeking) again = 1;
for (unsigned work = 0; work < 8 && !s->media_seeking && !s->media_failed && s->media_next < s->next_seq; work++) {
size_t length = 0;
int result = stream_buffer_decode(s->buffer, s->media_next, s->output, stream_input_size(&s->descriptor), &length);
if (result == STREAM_OK) {
result = stream_media_store_feed(s->media, s->media_next++, s->output, length);
if (result < 0) s->media_failed = 1;
} else if (result == STREAM_EXPIRED || result == STREAM_REJECTED || result == STREAM_WAIT || result == STREAM_GAP) {
uint64_t next = s->next_seq, available;
if (stream_buffer_next_decodable(s->buffer, s->media_next + 1, &available) == STREAM_OK && available < next) next = available;
stream_media_store_gap(s->media, s->media_next, next - 1); s->media_next = next;
} else {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index decode failed source=%llx seq=%llu result=%d",
(unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->media_next, result);
s->media_failed = 1;
}
}
if (!s->media_seeking && s->end_valid && (s->last_seq == UINT64_MAX || s->media_next > s->last_seq))
stream_media_store_finish(s->media);
if (!s->media_failed && s->media_next < s->next_seq) again = 1;
}
if (!s->closing && s->service->callback && !s->service->stopping && s->owners > s->relay_owners) {
struct stream_event event = {.descriptor = s->descriptor, .type = STREAM_EVENT_AVAILABLE, .first_seq = s->media_next};
s->service->dispatching++;
s->service->callback(s->service->inst, &event, s->service->arg);
s->service->dispatching--;
}
s->holds--;
if (s->closing) { if (!s->holds && !s->delivery) session_free(s); }
else if (again) session_media_schedule(s);
}
/* Индексация не зависит от позиции/паузы плеера и не создаёт отдельную очередь медиаданных. */
static void session_media_schedule(struct stream_session *s) {
if (!s->media || s->media_work || s->closing || s->service->stopping) return;
s->media_work = uasync_call_soon(s->service->inst->ua, s, session_media_work);
if (!s->media_work) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index scheduling failed source=%llx",
(unsigned long long)s->descriptor.source_node_id);
}
/* Уведомление использует собственный 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) {
if (!s->decoded) {
s->decoded = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u signature=verified",
(unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
(unsigned long long)s->descriptor.source_epoch, (unsigned long long)s->next_seq,
(unsigned)(s->descriptor.n - s->descriptor.r));
stream_network_wake(service->network);
}
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_REJECTED) {
event.type = STREAM_EVENT_GAP; event.first_seq = event.last_seq = s->next_seq++;
emit = again = 1;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream rejected batch GAP source=%llx epoch=%llx seq=%llu; receiving continues",
(unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch,
(unsigned long long)event.first_seq);
} 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 (!s->closing && s->media) {
if (event.type == STREAM_EVENT_DATA && emit && !s->media_seeking && s->media_next == event.first_seq) {
int result = stream_media_store_feed(s->media, s->media_next++, event.data, event.length);
if (result < 0) s->media_failed = 1;
}
session_media_schedule(s);
}
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;
}
stream_network_wake(service->network);
}
/* Общий 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;
for (struct source_watch *watch = service->watches; watch; watch = watch->next) {
struct TOPO_GROUP *watched = stream_group(service->inst, watch->group);
if (watched && channel && !strcmp(watched->channel_id, channel) && !watch->pending) {
watch->pending = uasync_call_soon(service->inst->ua, watch, sources_deliver);
if (!watch->pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed");
}
}
if (!service->sessions) return;
stream_network_membership_changed(service->network, node_id, channel);
int changed = 0;
for (struct stream_session *s = service->sessions; s; s = s->next) {
int own = node_id == service->inst->node_id;
if (s->closing) continue;
struct TOPO_GROUP *group = stream_group(service->inst, s->descriptor.group_id);
if (!group || !channel || strcmp(group->channel_id, channel)) continue;
changed = 1;
if (s->optimizer) stream_optimizer_roster_changed(s->optimizer, node_id);
if (!own && (s->source || s->descriptor.source_node_id != node_id)) 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 (!own && result == STREAM_OK) s->audio_enabled = state.published && state.audio_enabled;
session_radio_update(s);
}
if (!changed || 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);
while (service->watches) stream_watch_sources(inst, 0, service->watches->owner);
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;
for (struct source_watch *watch = service->watches; watch; watch = watch->next) {
if (watch->group == group_id && !watch->pending) {
watch->pending = uasync_call_soon(inst->ua, watch, sources_deliver);
if (!watch->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, int audio_enabled,
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->audio_enabled = audio_enabled;
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 : STREAM_DEFAULT_N, .r = parameters ? parameters->r : STREAM_DEFAULT_R,
.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;
struct stream_session *previous = session_find(inst->stream, group_id, inst->node_id);
if (previous && previous->source && previous->end_valid)
session_close(previous, STREAM_CLOSE_SOURCE_CHANGED, 0, 1);
return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, 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, s->audio_enabled, 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, state.audio_enabled, 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, s->audio_enabled, 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, state.audio_enabled, 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;
stream_media_cursor_destroy(h->media_cursor); h->media_cursor = NULL;
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) {
session_radio_update(s);
int was_relay = s->owners == s->relay_owners;
if ((h->publisher && !s->end_valid) || s->owners == 1) session_close(s, STREAM_CLOSE_STOPPED, 1, !h->relay);
else {
s->owners--; s->relay_owners -= h->relay;
if (h->publisher) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "finished publisher released; retained media readers source=%llx owners=%u",
(unsigned long long)s->descriptor.source_node_id, s->owners);
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;
}
/* Первый frontend создаёт индекс точной эпохи, остальные владельцы используют его без копии RS-кольца. */
int stream_media_enable(struct STREAM_HANDLE *h, unsigned tracks) {
if (!stream_handle_active(h)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media enable: detached handle"); return STREAM_ERR_STATE; }
struct stream_session *s = h->session;
if (!s->media) {
int result = stream_media_store_create(&s->media, s->buffer, &s->descriptor);
if (result != STREAM_OK) return result;
struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
s->media_next = stats.expired_before;
if (s->anchored && s->next_seq > stats.expired_before) {
s->media_seek = s->next_seq - 1; s->media_seeking = 1;
}
}
if (tracks && !h->media_cursor) {
int result = stream_media_cursor_create(&h->media_cursor, s->media, tracks);
if (result != STREAM_OK) return result;
session_radio_update(s);
}
session_media_schedule(s);
return STREAM_OK;
}
/* Выбранная эпоха проверяется до открытия и по реально используемой session; ошибка снимает только новый owner. */
int stream_media_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, uint64_t epoch,
unsigned tracks, struct STREAM_HANDLE **out) {
if (!out || *out || !epoch || !source_id || !(tracks & STREAM_MEDIA_TRACK_VIDEO) || (tracks & ~3u)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media open invalid source=%llx epoch=%llx tracks=%u",
(unsigned long long)source_id, (unsigned long long)epoch, tracks); return STREAM_ERR_ARGUMENT;
}
struct stream_member_state member;
int result = stream_member_load(inst, group_id, source_id, &member, NULL);
if (result != STREAM_OK) return result;
struct stream_session *s = inst->stream ? session_find(inst->stream, group_id, source_id) : NULL;
if (!member.published || member.source.source_epoch != epoch || (s && s->descriptor.source_epoch != epoch)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media selected source changed group=%llx source=%llx expected=%llx published=%llx local=%llx",
(unsigned long long)group_id, (unsigned long long)source_id, (unsigned long long)epoch,
(unsigned long long)(member.published ? member.source.source_epoch : 0),
(unsigned long long)(s ? s->descriptor.source_epoch : 0)); return STREAM_ERR_CONFLICT;
}
struct STREAM_HANDLE *h = NULL;
result = stream_subscribe_open(inst, group_id, source_id, &h);
if (result == STREAM_OK) result = stream_media_enable(h, tracks);
if (result != STREAM_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media reader open failed group=%llx source=%llx epoch=%llx result=%d",
(unsigned long long)group_id, (unsigned long long)source_id, (unsigned long long)epoch, result);
stream_close(h); return result;
}
*out = h;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media reader opened group=%llx source=%llx epoch=%llx tracks=%u",
(unsigned long long)group_id, (unsigned long long)source_id, (unsigned long long)epoch, tracks);
return STREAM_OK;
}
/* Чтение изменяет только локальный cursor; live ingest/READY/выдачи продолжают двигаться независимо. */
int stream_media_read(struct STREAM_HANDLE *h, int64_t position, int want_video, int64_t audio_ahead,
struct stream_media_packet *packet) {
if (!stream_handle_active(h) || !h->media_cursor) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media read: detached/unconfigured reader"); return STREAM_ERR_STATE;
}
if (h->session->media_failed) return STREAM_ERR_FORMAT;
if (h->session->media_seeking || h->session->media_next < h->session->next_seq) return STREAM_WAIT;
return stream_media_cursor_read(h->media_cursor, position, want_video, audio_ahead, packet);
}
int stream_media_get_status(struct STREAM_HANDLE *h, struct stream_media_status *status) {
if (!stream_handle_active(h) || !h->media_cursor || !status) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media status: detached/unconfigured reader"); return STREAM_ERR_STATE;
}
if (h->session->media_failed) return STREAM_ERR_FORMAT;
return stream_media_cursor_status(h->media_cursor, status);
}
int stream_media_resync(struct STREAM_HANDLE *h) {
if (!stream_handle_active(h) || !h->media_cursor) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media resync: detached/unconfigured reader"); return STREAM_ERR_STATE;
}
stream_media_cursor_reset(h->media_cursor);
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_set_audio_enabled(struct STREAM_HANDLE *h, int enabled) {
if (!stream_handle_active(h) || !h->publisher || h->session->finished || (enabled != 0 && enabled != 1)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream audio change invalid publisher/state enabled=%d", enabled); return STREAM_ERR_STATE;
}
struct stream_session *s = h->session;
if (s->audio_enabled == enabled) return STREAM_OK;
s->holds++;
int previous = s->audio_enabled;
s->audio_enabled = enabled;
int result = service_publish_options(s->service, stream_group(s->service->inst, s->descriptor.group_id));
if (result != STREAM_OK || s->closing) {
s->audio_enabled = previous;
if (result == STREAM_OK) result = STREAM_ERR_STATE;
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream audio state publication failed result=%d", result);
} else DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream audio group=%llx epoch=%llx enabled=%d",
(unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_epoch, enabled);
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, .decoded = s->decoded, .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;
session_radio_update(s);
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->decoded = 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;
}
Reference in new issue View Git Blame Copy Permalink