From b47a82e3ac14c0e00c241b66ca16ca1dd0004b21 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 20:45:14 +0200 Subject: [PATCH] Define bounded receiver-driven media replication protocol --- src/media_delivery/media_delivery.h | 68 +++++++++-------------- src/media_delivery/media_delivery_proto.h | 61 +++++++++++++------- 2 files changed, 69 insertions(+), 60 deletions(-) diff --git a/src/media_delivery/media_delivery.h b/src/media_delivery/media_delivery.h index c58e9d3a..26237849 100644 --- a/src/media_delivery/media_delivery.h +++ b/src/media_delivery/media_delivery.h @@ -16,33 +16,38 @@ extern "C" { #include "../lib/ll_queue.h" #include "../lib/sqlite3.h" #include "../lib/file_io.h" +#include "media_delivery_proto.h" struct UTUN_INSTANCE; // ── константы ── -#define MEDIA_MAX_REPL_INFLIGHT 4 #define MEDIA_REPL_TIMEOUT_TB 20000 // начальный таймаут 2s (0.1ms) -#define MEDIA_REPL_TIMEOUT_MAX_TB 6000000 // макс 10 min +#define MEDIA_REPL_TIMEOUT_MAX_TB 300000 // предел повторов 30s +#define MEDIA_REPL_POLL_TB 300000 // периодическая сверка 30s, включая потерю CHANGED +#define MEDIA_REPL_WATCHDOG_TB 2000 // контроль дедлайна/очереди/CM каждые 200ms #define MEDIA_QUERY_TIMEOUT_TB 20000 // таймаут QUERY 2s #define MEDIA_HAVE_BLOCK_TIMEOUT_TB 20000 // таймаут HAVE_BLOCK_ACK 2s #define MEDIA_RECONNECT_COOLDOWN_TB 360000000 // 1 час (0.1ms) -#define MEDIA_HELLO_TIMEOUT_TB 20000 // таймаут SUPER_HELLO 2s #define MEDIA_DL_STALL_TIMEOUT_TB 20000 // таймаут простоя загрузки 2s (0.1ms) #define MEDIA_DL_MAX_ATTEMPTS 3 // попыток на блок до ошибки // ── структуры ── struct media_super_peer { - struct ll_entry ll; // индекс по peer_node_id (8 байт) - uint64_t peer_node_id; - uint64_t group_id; // CHAT-группа, в которой установлена связь с пиром - uint64_t peer_last_recv_id; // что пир говорит он получил от нас (из SUPER_HELLO / SUPER_ACK) - uint32_t timeout_tb; // текущий таймаут (растёт при ошибках) - uint8_t inflight_count; // пакетов в полёте (макс 4) - uint8_t connected; // 1 = прямое подключение установлено - uint8_t hello_done; // 1 = SUPER_HELLO обмен завершён, можно реплицировать - void* timeout_timer; // таймер ретрансмита + uint64_t peer_node_id; // Эти два поля подряд — 16-байтный ключ очереди. + uint64_t group_id; + struct media_pkt_sync_req request; // Неизменяемый запрос до успешного commit/отмены. + struct media_pkt_sync_req response; // Один ожидающий выдачи запрос пира; ответы вычисляются из БД. + uint64_t deadline_tb; // Дедлайн запроса, idle-сверки или повторного открытия CM. + uint32_t timeout_tb; // Задержка повторов 2..30s, сбрасывается после commit. + uint8_t connected; // Рабочий путь по событию CM, включая непрямой. + uint8_t request_pending, request_send, response_pending; + uint8_t dirty; // CHANGED во время запроса: после commit проверить ещё раз. + uint8_t changed; // Наш CHANGED ожидает места в router send_q. + void* timeout_timer; // Watchdog восстанавливает и потерянный waiter при reset router. + void* pump; // Отменяемая отложенная отправка; waiter не вызывает её рекурсивно. + struct queue_waiter_handle waiter; struct CONN_MGR_HANDLE* connection; // собственное CM-владение; восстановлением пути управляет CM struct media_delivery_ctx* md; // обратная ссылка }; @@ -56,32 +61,10 @@ struct md_file_load { uint64_t downloader_ids[10]; // кто сейчас качает (для редиректа) }; -/* - * block_availability: жизненный цикл записей и гарантия уникальности id - * - * Каждое событие создаёт новую запись с новым id (AUTOINCREMENT). - * Старая запись удаляется — id НЕ переиспользуется. - * Это критично для репликации: курсор super_sync.last_recv_id - * всегда монотонно растёт и не пропускает события. - * - * BLOCK_PROCESSING (subcmd 0x10): - * md_ba_insert(... , status=0) — чистый INSERT, новый id=X. - * - * HAVE_BLOCK (subcmd 0x06): - * md_ba_complete_block(): - * 1. DELETE FROM block_availability WHERE block_uuid=? AND node_id=? AND status=0 - * (удаляет processing-запись id=X, если была) - * 2. Если completed уже есть → no-op (возврат 0) - * 3. INSERT ... status=1 — новый id=Y. - * - * SUPER_REPL (subcmd 0x08): - * Принимающая сторона: - * - status=0: INSERT OR IGNORE (дубликаты при репликации — норма) - * - status=1: md_ba_complete_block() — та же логика что HAVE_BLOCK - * - * Итог: processing id=X удаляется, completed вставляется с новым id=Y>X. - * Репликация через md_ba_get_since(id > last_recv_id) всегда видит Y. - */ +/* Каталог уникален по (group,block,holder). Объединение выбирает максимум (status,timestamp). + * Только реальное изменение получает новый AUTOINCREMENT ID; повторы не меняют sqlite_sequence. + * ID источника не становится локальным ID. Страница и RX-курсор сохраняются атомарно. + * Верхняя граница выдачи — sqlite_sequence, включая удалённые ID и промежутки других групп. */ struct media_served_node { struct ll_entry ll; // индекс по node_id (8 байт) @@ -117,7 +100,9 @@ struct media_delivery_ctx { struct media_destination* destinations; // Активные писатели нормализованных путей, входящие и исходящие. uint64_t self_node_id; int initialized; - uint8_t is_supernode; // из adm_tags в peers_* + uint8_t is_supernode; // Диагностическое «есть хотя бы одна группа с нашей ролью supernode». + uint8_t source_epoch[16]; // Сохранённое поколение каталога; перезапуск сервиса его не меняет. + void* reconcile_timer; // Периодическое перечитывание ролей и восстановление CM после ошибок. uint8_t active_streams; // Выдачи индексированных файлов; растущий relay ограничен отдельно на блок. uint8_t max_downloads_per_file; // лимит параллельных загрузок с одного файла (source node), default 3 uint8_t stream_completed; // 1 = хотя бы один стрим завершился (для тестов) @@ -145,8 +130,9 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst); // Вызывает init целиком; использовать вместо init, а не как следующий этап. 0/-1. int media_delivery_bind(struct UTUN_INSTANCE* inst); -// Включить/выключить локальную работу суперузла и обмен доступностью блоков. -void media_delivery_set_supernode(struct UTUN_INSTANCE* inst, int is_supernode); +// Перечитать актуальные роли из peers_* и согласовать владение CM по каждой CHAT-группе; только uasync. +// Обычно вызывается автоматически по member_sync и периодически; не изменяет подписанные роли в БД. +void media_delivery_refresh_roles(struct UTUN_INSTANCE* inst); // ── Внутренний учёт выдач ── diff --git a/src/media_delivery/media_delivery_proto.h b/src/media_delivery/media_delivery_proto.h index 03d95cef..53fce8d6 100644 --- a/src/media_delivery/media_delivery_proto.h +++ b/src/media_delivery/media_delivery_proto.h @@ -20,13 +20,13 @@ enum { MEDIA_SUBCMD_QUERY_RESP = 0x05, // суперузел→req-узел: список узлов MEDIA_SUBCMD_HAVE_BLOCK = 0x06, // узел→суперузел: у меня есть блок MEDIA_SUBCMD_HAVE_BLOCK_ACK = 0x07, // суперузел→узел: подтверждение приёма - MEDIA_SUBCMD_SUPER_REPL = 0x08, // суперузел↔суперузел: репликация записей - MEDIA_SUBCMD_SUPER_ACK = 0x09, // подтверждение репликации + MEDIA_SUBCMD_SYNC_REQ = 0x08, // получатель→суперузел: страница после курсора + MEDIA_SUBCMD_SYNC_RESP = 0x09, // суперузел→получатель: страница каталога MEDIA_SUBCMD_BLOCK_REQ = 0x0A, // req-узел→блок-холдер: дай блок MEDIA_SUBCMD_BLOCK_CHUNK = 0x0B, // блок-холдер→req-узел: чанк данных (1KB) MEDIA_SUBCMD_BLOCK_DONE = 0x0C, // блок-холдер→req-узел: блок передан полностью MEDIA_SUBCMD_CANCEL = 0x0D, // req-узел→блок-холдер: отмена передачи блока - MEDIA_SUBCMD_SUPER_HELLO = 0x0E, // суперузел↔суперузел: рукопожатие при подключении + MEDIA_SUBCMD_CHANGED = 0x0E, // суперузел→суперузел: каталог мог измениться (один байт) MEDIA_SUBCMD_BLOCK_OVERLOADED = 0x0F, // автор→req-узел: перегружен, попробуй позже MEDIA_SUBCMD_BLOCK_PROCESSING = 0x10, // узел→суперузел: начал качать блок (status=processing) MEDIA_SUBCMD_BLOCK_RELAY_FULL = 0x11, // relay-узел→req-узел: перегружен, вот список кому транслирую @@ -51,6 +51,18 @@ enum { #define MEDIA_BLOCK_OK 0 #define MEDIA_BLOCK_UNAVAILABLE 1 +/* Репликация положительных объявлений, не снимок удалённой файловой системы. + * Группа/источник — только router; обе стороны должны быть суперузлами этой CHAT-группы. + * Все числа SYNC — little endian. ID ограничены INT64_MAX; epoch/UUID — непрозрачные 16 байт. + * Новый запрос имеет новый UUID; timeout повторяет тот же запрос. Ответ принимается только + * для текущего UUID. Эпоха сохраняется в БД источника и меняется при пересоздании каталога. + * При несовпадении ожидаемой эпохи источник читает с нуля. OK применяется вместе с RX-курсором + * одной транзакцией. CHANGED — подсказка; периодический SYNC восстанавливает потерянные подсказки. */ +#define MEDIA_SYNC_MAX_ENTRIES 32 +#define MEDIA_SYNC_OK 0 +#define MEDIA_SYNC_DB_ERROR 1 +#define MEDIA_SYNC_CURSOR_INVALID 2 + /* ── packed-структуры пакетов ── */ #pragma pack(push, 1) @@ -117,20 +129,37 @@ struct media_pkt_have_block_ack { uint8_t status; // 0=OK, 1=dup/already }; -struct media_pkt_super_repl { - uint8_t subcmd; // MEDIA_SUBCMD_SUPER_REPL - uint32_t seq; // макс id в пачке - uint16_t num_entries; - // далее: записи raw block_availability +struct media_pkt_sync_req { + uint8_t subcmd; // MEDIA_SUBCMD_SYNC_REQ + uint8_t request_id[16]; // UUID текущего запроса страницы. + uint8_t source_epoch[16]; // Ожидаемая эпоха источника; нули для первого запроса. + uint64_t after_id; // Уже атомарно сохранённый RX-курсор. }; -#define MEDIA_SUPER_REPL_HDR_SIZE (sizeof(struct media_pkt_super_repl)) // 1+4+2=7 - -struct media_pkt_super_ack { - uint8_t subcmd; // MEDIA_SUBCMD_SUPER_ACK - uint32_t ack_seq; // подтверждённый max id +struct media_pkt_sync_entry { + uint64_t id; // ID строки у источника, строго возрастают в странице. + uint8_t block_id[16]; + uint64_t node_id; // Ненулевой держатель; уникален с block_id внутри группы. + uint32_t chunk; // Неизменяемый номер блока. + int64_t timestamp; // Unix-время объявления, неотрицательное. + uint8_t status; // 0=PROCESSING, 1=COMPLETE; COMPLETE имеет приоритет. }; +struct media_pkt_sync_resp { + uint8_t subcmd; // MEDIA_SUBCMD_SYNC_RESP + uint8_t request_id[16]; // Точное эхо UUID запроса. + uint8_t status; // MEDIA_SYNC_*; ошибка не продвигает курсор, записей нет. + uint8_t source_epoch[16]; + uint64_t from_id; // after_id запроса при совпадении эпох, иначе 0. + uint64_t next_id; // При more — последний ID страницы; иначе верхняя граница снимка. + uint8_t more; // 0/1; 1 требует полной страницы из 32 записей. + uint16_t num_entries; // 0..32; далее media_pkt_sync_entry[num_entries]. +}; + +#define MEDIA_SYNC_REQ_SIZE sizeof(struct media_pkt_sync_req) // 41 +#define MEDIA_SYNC_RESP_HDR_SIZE sizeof(struct media_pkt_sync_resp) // 53 +#define MEDIA_SYNC_PACKET_MAX (MEDIA_SYNC_RESP_HDR_SIZE + MEDIA_SYNC_MAX_ENTRIES * sizeof(struct media_pkt_sync_entry)) // 1493 + struct media_pkt_block_req { uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_REQ uint8_t media_id[16]; @@ -174,12 +203,6 @@ struct media_pkt_cancel { uint32_t chunk; }; -struct media_pkt_super_hello { - uint8_t subcmd; // MEDIA_SUBCMD_SUPER_HELLO - uint64_t last_recv_id; // ID в МОЕЙ базе до которого пир подтвердил приём - uint64_t group_id; // CHAT-группа, в которой установлена связь -}; - struct media_pkt_block_overloaded { uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_OVERLOADED uint8_t media_id[16];