Define bounded receiver-driven media replication protocol
This commit is contained in:
evgeny committed
1 parent
04dc53672a
commit
b47a82e3ac
2 files changed
+69
-60
No files matched your search
@@ -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);
|
||||
|
||||
// ── Внутренний учёт выдач ──
|
||||
|
||||
|
||||
@@ -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];
|
||||
|
||||
Reference in new issue
Block a user