jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Document media replication protocol and resource ownership

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 21:15:55 +02:00
1 parent 0bcdb9930a
commit bc149497f3
2 files changed
+107 -152
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/service_lifecycle.md
+12
View File
@@ -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 уже опубликовал копию.
src/media_delivery/sync.md
+95 -152
View File
@@ -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_<channel>`.
При потере локальной роли закрываются только 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_<channel>` — если да, вызывает `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_<channel>` + `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