From bc149497f3fa3672da125d598dcd66b3a0733a4e Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 21:15:55 +0200 Subject: [PATCH] Document media replication protocol and resource ownership --- doc/service_lifecycle.md | 12 ++ src/media_delivery/sync.md | 247 ++++++++++++++----------------------- 2 files changed, 107 insertions(+), 152 deletions(-) diff --git a/doc/service_lifecycle.md b/doc/service_lifecycle.md index c1a4e4fd..6823c6c3 100644 --- a/doc/service_lifecycle.md +++ b/doc/service_lifecycle.md @@ -126,6 +126,18 @@ completion. Отказ запуска/неверные аргументы даю закрытия temp; вызывающий отвечает за очистку своего результата при отмене. Подробные файловые контракты, 64-битные размеры и ошибки RNG описаны в `media_async.h`. +`media_delivery_init()` вызывается после подготовки chat core и media_index, пока +общая БД и группы живы; подписка на роли обязательна. Репликация хранит отдельный +контекст и CM handle для каждой пары «CHAT-группа–пир». DOWN отменяет текущий RPC, +сохраняя handle для восстановления CM; TIMEOUT закрывает его перед повторным открытием. +Отдельный ETCP DOWN не означает потерю всех путей CM. До free контекста отменяются +его waiter, pump, watchdog и собственный handle; destroy отменяет также сверку ролей. +Обычный stop сохраняет каталог, его эпоху и RX-курсоры. Потеря локальной роли удаляет +чужие объявления вместе с RX-курсорами только этой группы одной транзакцией; +удаление канала удаляет также собственные объявления. Страница репликации и её +RX-курсор фиксируются одним самостоятельным commit; вложенная внешняя транзакция +не считается сохранённым прогрессом. Wire-контракт находится в `media_delivery_proto.h`. + Перед файловой подготовкой назначение записывается в `media_preparations`. Успешный `media_index_commit()` регистрирует весь файл и снимает эту запись в одном savepoint; после ошибки/отмены путь остаётся известен, даже если worker уже опубликовал копию. diff --git a/src/media_delivery/sync.md b/src/media_delivery/sync.md index 2feb5b21..a9e34997 100644 --- a/src/media_delivery/sync.md +++ b/src/media_delivery/sync.md @@ -1,185 +1,128 @@ # Репликация block_availability между суперузлами -## 1. Таблицы в SQLite +## Каталог и курсоры -### `block_availability` -Хранит информацию о том, какие узлы имеют какие блоки: +Нормативные wire-структуры находятся в `media_delivery_proto.h`, состояние и владение +ресурсами — в `media_delivery.h`. Все обработчики выполняются в uasync-потоке ядра. -``` -id INTEGER PRIMARY KEY AUTOINCREMENT -block_uuid BLOB — UUID блока (16 байт) -group_id INTEGER — id группы -node_id INTEGER — id узла-владельца блока -chunk INTEGER — номер чанка -timestamp INTEGER — unix-время добавления/обновления -``` +`block_availability` хранит положительные объявления PROCESSING/COMPLETE: +`id AUTOINCREMENT, block_uuid[16], group_id, node_id, chunk, timestamp, status`. +Уникальный ключ — `(group_id, block_uuid, node_id)`, индекс страниц — `(group_id,id)`. +Это каталог кандидатов для загрузки, не подтверждённый снимок удалённой файловой системы. +Протокола удалённых tombstone здесь нет; загрузчик проверяет полученные блоки и меняет +держателя при недоступности. Пересоздание источника не удаляет ранее полученные объявления. -Индексы: `block_uuid+node_id` (unique), `id`, `node_id`. +`super_sync` хранит **нашу позицию приёма от источника**: +`(group_id, peer_node_id) PRIMARY KEY, source_epoch[16], last_recv_id`. +Позиции отправки в БД и ACK нет. ID входящей строки не становится локальным ID. -### `super_sync` -Хранит указатель, до какого `id` синхронизировано с каждым другим суперузлом: +`media_replication_meta` хранит версию схемы и случайную эпоху локального каталога. +Обычный stop/start сохраняет эпоху, каталог и RX-курсоры. При переходе со старой схемы +или отсутствии одной из таблиц вспомогательный каталог пересоздаётся одной транзакцией +с новой эпохой. `media_files` и данные чата при этом сохраняются. -``` -peer_node_id INTEGER PRIMARY KEY — id пира-суперузла -last_recv_id INTEGER — максимальный id, подтверждённый пиром (что он принял от нас) -``` +## Роли и соединения -Обе таблицы создаются в `media_delivery_create_tables()` (media_delivery.c:940-971). +Пир индексируется по паре `(node_id,group_id)`. Обе стороны репликации должны быть +актуальными суперузлами одной CHAT-группы: `adm_tags.supernode=yes`, без `deleted=yes`. +Производный `node_type` и общий диагностический флаг `md.is_supernode` не дают полномочий. +Группа и отправитель берутся только из router; группа не дублируется в SYNC payload. -На всех узлах есть `block_availability`. Суперузлы сохраняют в неё весь контент групп, в которых состоят. Обычные узлы — только свой контент. Если нет суперузлов или они недоступны, роль суперузла выполняет автор контента. +Перечитывание ролей выполняется при init, событиях member_sync и каждые 30 секунд. +Так обнаруживаются и группы, созданные после init. Chat core и media_index должны +быть готовы до media_delivery_init; отказ подписки на роли завершает init ошибкой. -## 2. Обнаружение других суперузлов +Каждый контекст владеет отдельным CM handle. CM может предоставить прямой или +непрямой путь; handle сохраняется при DOWN, восстановлением занимается CM. +После TIMEOUT свой handle закрывается, повтор открытия выполняется с задержкой. +Raw ETCP DOWN и потеря BGP-маршрута не удаляют объявления или RX-курсоры. -Суперузлы определяются через свойство `supernode=yes` в `adm_tags` таблицы `peers_`. +При потере локальной роли закрываются только peers этой группы; чужие объявления +и все её RX-курсоры удаляются атомарно, собственные объявления сохраняются. +Удаление канала отменяет его работу и удаляет весь его каталог вместе с курсорами. +Обычный destroy сохраняет БД; до free отменяются CM callbacks, waiter, pump и таймеры. -Механизмы обнаружения: +## Запросы страниц -- **При старте** (`md_super_start`, стр. 730): суперузел ищет в БД все узлы с `node_type=4` в своей chat-группе и инициирует прямое подключение к каждому через `conn_mgr_open`. -- **При появлении нового узла в BGP** (`md_on_bgp_node`, стр. 790): проверяет `node_type=4` в `peers_` — если да, вызывает `md_super_connect`. -- **При изменении свойств** (`md_on_props_changed`, стр. 825): если `adm_tags` изменились на `supernode=yes` — подключается; если на `no` — удаляет пира. +| Команда | Подкоманда | Размер | +|---------|------------|--------| +| SYNC_REQ | 0x08 | 41 байт | +| SYNC_RESP | 0x09 | 53 + N × 45 байт, N ≤ 32 | +| CHANGED | 0x0E | 1 байт | -Как только суперузел видит появление другого суперузла в BGP-таблице — пробует установить прямую связь. Прямое подключение — независимый автономный процесс (без посредников и без обратного подключения — это сделает другой суперузел сам). Если не получается 10 сек — отключаемся и следующая попытка через час. +Все числа SYNC кодируются little endian. UUID запроса и эпоха — непрозрачные 16 байт. +ID строк/курсоры ограничены INT64_MAX. Максимум ответа — 1493 байта. -## 3. Протокол репликации +После CM UP получатель создаёт запрос с новым UUID и сохранёнными эпохой/курcором +источника. На пару «группа–пир» приходится один ожидающий запрос. +При таймауте повторяются его прежние UUID и параметры, с задержками 2, 4, 8, 16, 30 секунд. +Успешный commit сбрасывает задержку до 2 секунд. DOWN отменяет RPC; UP создаёт новый UUID. -### Handshake: SUPER_HELLO (subcmd 0x0E) +Источник строит страницу внутри read transaction: -После установки прямого подключения инициатор отправляет `SUPER_HELLO` (`md_super_conn_cb`, стр. 484) с полем `last_recv_id` — до какого `id` он уже получил записи от этого пира. +1. При совпадении эпох читает после запрошенного курсора, иначе после нуля. +2. Фиксирует верхнюю границу H из `sqlite_sequence`, которая не уменьшается при DELETE. +3. Выбирает только строки router-группы с `after < id ≤ H`, ORDER BY id, LIMIT 33. +4. Передаёт первые 32. Если есть 33-я, `more=1`, `next_id` равен ID последней переданной. +5. В последней странице `more=0`, `next_id=H`, даже при отсутствии строк. -При получении `SUPER_HELLO` (`md_handle_super_hello`, стр. 454): -- Создаёт/обновляет `media_super_peer` в памяти -- Устанавливает `connected=1`, `hello_done=1` -- Отправляет ответный `SUPER_HELLO` со своим `last_recv_id` -- Обнуляет `inflight_count` и запускает репликацию через `md_super_repl_send` +Так курсор проходит удалённые ID и промежутки, принадлежащие другим группам. +Повторный запрос может получить более свежий согласованный снимок: принимается только +первый корректный ответ актуального UUID, следующие игнорируются. -### Репликация: SUPER_REPL (subcmd 0x08) +Ошибка чтения/завершения snapshot даёт DB_ERROR без записей; получатель сохраняет RPC +и повторяет его по исходному дедлайну. Если при совпавшей эпохе курсор выше H, +CURSOR_INVALID сбрасывает только эту RX-пару, оставляя объявления; новый запрос с нуля +будет создан после задержки 30 секунд. Эпоха источника от чужого курсора не меняется. -`md_super_repl_send` (стр. 228): -- Запрашивает из `block_availability` записи с `id > peer->peer_last_recv_id` -- Порциями по 4 записи (константа `MEDIA_MAX_REPL_INFLIGHT = 4`) -- Каждая запись: 52 байта (id:8, block_uuid:16, group_id:8, node_id:8, chunk:4, timestamp:8) -- Пакет: `subcmd` + `seq` (макс id в пачке) + `num_entries` + сырые строки -- Инкрементит `inflight_count` +## Атомарный приём и сходимость -### Подтверждение: SUPER_ACK (subcmd 0x09) +До SQL проверяются весь пакет и его точная длина, группа/пир/UUID, эпоха, диапазон +from/next, количество, порядок ID, статусы, дубли ключей и согласованность chunk. +Неподходящий ответ не меняет запрос, курсор или дедлайн. -При получении `SUPER_ACK` (`md_handle_super_ack`, стр. 436): -- Обновляется `peer_last_recv_id = ack_seq` -- Декрементится `inflight_count` -- Таймаут сбрасывается на начальное значение (`MEDIA_REPL_TIMEOUT_TB = 2с`) -- Запускается следующая порция репликации (если есть записи) +Все объявления и RX-эпоха/курсор применяются одной собственной транзакцией. +Внешняя незавершённая транзакция приводит к отказу этой попытки: вложенный savepoint +не гарантировал бы окончательный commit. При ошибке строки, курсора или commit +прогресс не публикуется; учитывается и автоматический rollback SQLite. -### Приём репликации (`md_handle_super_repl`, стр. 404): -- Разбирает каждую запись, вызывает `md_ba_insert` (INSERT OR REPLACE в `block_availability`) -- Сохраняет `max_id` в `super_sync` через `md_ss_set(db, from_node, max_id)` -- Отправляет `SUPER_ACK` с подтверждённым `max_id` +Объединение выбирает максимум `(status,timestamp)`: COMPLETE выше PROCESSING, +при равном статусе выигрывает более свежий timestamp. Chunk одного блока в группе +неизменяем. Только реальное изменение выполняет REPLACE с новым локальным ID; +повтор/старое объявление не меняет строку и sqlite_sequence. Это останавливает эхо +при распространении одной записи через несколько суперузлов. -## 4. Таймауты и retransmit +HAVE_BLOCK и BLOCK_PROCESSING проходят тот же merge; несовпадение router-группы +с полем объявления отклоняется. ACK HAVE: 0 — изменение, 1 — no-op, 2 — ошибка БД. +Собственные зарегистрированные блоки добавляются непосредственно из media_files, +в том числе на единственном суперузле; повторный анонс использует стабильное время. -- **Начальный таймаут:** 2 секунды (`MEDIA_REPL_TIMEOUT_TB`) -- **Exponential backoff:** при отсутствии ACK таймаут удваивается (до 10 минут — `MEDIA_REPL_TIMEOUT_MAX_TB`) -- **При успешном ACK:** таймаут сбрасывается на 2 секунды -- **При неудачном подключении:** повтор через 1 час (`MEDIA_RECONNECT_COOLDOWN_TB`) -- Механика: `md_repl_schedule_retrans` → `md_repl_retrans_cb` → `md_super_repl_send` +После commit изменений отправляется CHANGED остальным peers этой группы. +CHANGED во время RPC ставит dirty, и после commit выполняется ещё один запрос. +`more=1` также сразу запускает следующую страницу. Потеря CHANGED восстанавливается +периодическим запросом каждые 30 секунд; дубликаты не создают новые локальные ID. -## 5. Источники записей в block_availability +## Очереди, диагностика и тесты -Записи попадают в таблицу из трёх источников: +Отправка репликации использует force=0 и общий router backpressure. +Контекст хранит один отложенный запрос ответа, один свой RPC и признак CHANGED; +очередь ответов не растёт. Более поздний входящий запрос заменяет отложенный. +Если это старый RPC, его ответ будет проигнорирован, а актуальный запрос повторится. -1. **HAVE_BLOCK** — когда req-узел принял полный блок, он уведомляет свой суперузел (`md_handle_have_block`, стр. 371) -2. **SUPER_REPL** — репликация от другого суперузла -3. **Локально** — при регистрации своего медиа через `media_index` +Waiter только планирует отменяемый pump; один вызов отправляет максимум три пакета. +Watchdog каждые 200ms проверяет дедлайны, повторяет неудачные отправки и восстанавливает +пробуждение после reset router-очереди. Репликация не перезапускает общий MEDIA-канал, +который также используется загрузками и выдачами блоков. -При получении HAVE_BLOCK суперузел немедленно реплицирует новые записи всем подключённым super-пирам (`md_handle_have_block`, стр. 398-400). +Категория media на DEBUG показывает group/peer, UUID/epoch, from/next, count/more, +результат merge и commit. Ошибки SQL показывают фазу и коды, таймауты — UUID и задержку. -## 6. Кто есть суперузел - -- `node_type=4` в `peers_` + `adm_tags` содержит `supernode=yes` -- `md->is_supernode` устанавливается при инициализации и может меняться через `media_delivery_set_supernode` / `md_on_props_changed` -- Если узел перестаёт быть суперузлом — вызывается `md_super_stop`: удаляются все чужие блоки из `block_availability`, очищается очередь свер-пиров - -## 7. Обслуживаемые узлы (served_nodes) - -- Обычный узел при старте загрузки отправляет суперузлу `SERVE_REG` (subcmd 0x01) -- Суперузел сохраняет его в `served_nodes` (ll_queue с индексом по node_id) -- Если соединение рвётся — узел автоматически выпадает из списка (`md_on_conn_status`) -- Также автор контента получает от суперузла обновления о доступности его блоков - -## 8. Очистка - -- **При удалении узла из BGP** (`md_on_bgp_node`, TOPO_NODE_EVENT_REMOVE): удаляются все его записи из `block_availability`, свер-пир удаляется -- **При разрыве соединения** (`md_on_conn_status`, ETCP_CONN_STATUS_DOWN/DELETE): свер-пир помечается как отключённый (`connected=0, hello_done=0, inflight_count=0`), таймеры отменяются; served_node удаляется -- **При отключении режима суперузла** (`md_super_stop`): удаляются все чужие блоки (`WHERE node_id != self`), очищается очередь свер-пиров - -## 9. Диаграмма потоков - -``` -[Узел A: обычный] [Суперузел S1] [Суперузел S2] - | | | - |── HAVE_BLOCK ─────────────────>| | - | |── md_ba_insert (локально) | - | |── SUPER_REPL ────────────>| - | | |── md_ba_insert (локально) - | |<─────────── SUPER_ACK ────| - | |── md_ss_set(S2, max_id) | - | | | - |── QUERY ──────────────────────>| | - |<───── QUERY_RESP (node list) ──| | - | | | - |── BLOCK_REQ ─────────────────────────────────────────────>| - |<───── BLOCK_CHUNK (stream) ───────────────────────────────| - |<───── BLOCK_DONE ─────────────────────────────────────────| - | | | - |── HAVE_BLOCK ─────────────────>| | - | |── SUPER_REPL ────────────>| -``` - -## 10. Константы - -| Константа | Значение | Описание | -|-----------|----------|----------| -| `MEDIA_MAX_REPL_INFLIGHT` | 4 | Макс пакетов в полёте на одного пира | -| `MEDIA_REPL_TIMEOUT_TB` | 20000 | Начальный таймаут 2с (0.1ms units) | -| `MEDIA_REPL_TIMEOUT_MAX_TB` | 6000000 | Макс таймаут 10 мин | -| `MEDIA_RECONNECT_COOLDOWN_TB` | 360000000 | Повтор подключения через 1 час | -| `MEDIA_HELLO_TIMEOUT_TB` | 20000 | Таймаут SUPER_HELLO 2с | - -## 11. BLOCK_PROCESSING: регистрация блока в процессе загрузки - -Когда узел начинает скачивать блок, он отправляет суперузлу `BLOCK_PROCESSING` (subcmd 0x10). -Суперузел вставляет запись в `block_availability` со `status=0` (processing) через `INSERT OR IGNORE` -— не перезаписывает существующие completed-записи (status=1). - -Когда блок полностью скачан, узел отправляет `HAVE_BLOCK` (subcmd 0x06). -Суперузел вызывает `INSERT OR REPLACE` со `status=1` — processing-запись заменяется на completed. - -Записи с `status=0` реплицируются между суперузлами так же, как и completed (поле `status` включено -в сериализацию `SUPER_REPL`, размер строки: 56 байт вместо 52). - -### BLOCK_PROCESSING vs HAVE_BLOCK - -| Операция | subcmd | SQL | Статус | id | -|----------|--------|-----|--------|----| -| Начал качать | 0x10 | INSERT (status=0) | processing | новый X | -| Скачал | 0x06 | DELETE (status=0) + проверка + INSERT (status=1) | completed | новый Y>X | - -**Гарантия:** каждый переход создаёт новую запись с новым id. Старая удаляется — id не переиспользуется. -Если completed уже существует — HAVE_BLOCK игнорируется (no-op). -Если processing пытается вставиться поверх completed — `SQLITE_CONSTRAINT`, логируется WARN. - -### Схема block_availability (обновлённая) - -``` -id INTEGER PRIMARY KEY AUTOINCREMENT — монотонный, не переиспользуется -block_uuid BLOB — UUID блока (16 байт) -group_id INTEGER — id группы -node_id INTEGER — id узла-владельца блока -chunk INTEGER — номер чанка -timestamp INTEGER — unix-время добавления/обновления -status INTEGER — 0=processing (качается), 1=completed (скачан) -``` - -Уникальный индекс: `(block_uuid, node_id)` — предотвращает две completed-записи для одного блока. +`test_media_replication` использует настоящие handlers/SQLite/router send_q с +управляемыми CM и доставкой: страницы до 10000 записей, ID выше 2^32, усечения, +дубли/чужие ответы, SQL/commit faults, retry, смена эпохи, две группы, смена роли, +сходимость трёх узлов и backpressure/reset. `test_media_delivery_sql` проверяет +настоящую схему, restart/reset и сохранность media_files. Full/chat/integration +проверяют реальные транспорт, CHAT-членство и подписанные роли. ## 12. Попытки передачи и relay