# Репликация block_availability между суперузлами ## 1. Таблицы в SQLite ### `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-время добавления/обновления ``` Индексы: `block_uuid+node_id` (unique), `id`, `node_id`. ### `super_sync` Хранит указатель, до какого `id` синхронизировано с каждым другим суперузлом: ``` peer_node_id INTEGER PRIMARY KEY — id пира-суперузла last_recv_id INTEGER — максимальный id, подтверждённый пиром (что он принял от нас) ``` Обе таблицы создаются в `media_delivery_create_tables()` (media_delivery.c:940-971). На всех узлах есть `block_availability`. Суперузлы сохраняют в неё весь контент групп, в которых состоят. Обычные узлы — только свой контент. Если нет суперузлов или они недоступны, роль суперузла выполняет автор контента. ## 2. Обнаружение других суперузлов Суперузлы определяются через свойство `supernode=yes` в `adm_tags` таблицы `peers_`. Механизмы обнаружения: - **При старте** (`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` — удаляет пира. Как только суперузел видит появление другого суперузла в BGP-таблице — пробует установить прямую связь. Прямое подключение — независимый автономный процесс (без посредников и без обратного подключения — это сделает другой суперузел сам). Если не получается 10 сек — отключаемся и следующая попытка через час. ## 3. Протокол репликации ### Handshake: SUPER_HELLO (subcmd 0x0E) После установки прямого подключения инициатор отправляет `SUPER_HELLO` (`md_super_conn_cb`, стр. 484) с полем `last_recv_id` — до какого `id` он уже получил записи от этого пира. При получении `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` ### Репликация: SUPER_REPL (subcmd 0x08) `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) При получении `SUPER_ACK` (`md_handle_super_ack`, стр. 436): - Обновляется `peer_last_recv_id = ack_seq` - Декрементится `inflight_count` - Таймаут сбрасывается на начальное значение (`MEDIA_REPL_TIMEOUT_TB = 2с`) - Запускается следующая порция репликации (если есть записи) ### Приём репликации (`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` ## 4. Таймауты и retransmit - **Начальный таймаут:** 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` ## 5. Источники записей в block_availability Записи попадают в таблицу из трёх источников: 1. **HAVE_BLOCK** — когда req-узел принял полный блок, он уведомляет свой суперузел (`md_handle_have_block`, стр. 371) 2. **SUPER_REPL** — репликация от другого суперузла 3. **Локально** — при регистрации своего медиа через `media_index` При получении HAVE_BLOCK суперузел немедленно реплицирует новые записи всем подключённым super-пирам (`md_handle_have_block`, стр. 398-400). ## 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-записи для одного блока. ## 12. Relay-стриминг: передача блока в процессе загрузки ### Принцип Если узел C хочет скачать блок, а узел B уже качает этот блок от узла A: 1. C видит B в QUERY_RESP (благодаря BLOCK_PROCESSING, status=0) 2. C отправляет BLOCK_REQ к B 3. B принимает запрос, добавляет C в `relay_downstream` для этого блока 4. B получает BLOCK_CHUNK от A → пишет в файл → форвардит C 5. B получает BLOCK_DONE от A → форвардит C → удаляет relay-контекст ### Flow control - Relay-узел всегда отстаёт от апстрима: он не может слать быстрее, чем получает - Каждый downstream хранит `sent_offset` — сколько байт уже отправлено - При получении нового чанка: читается файл, отправляется отставание (до 1KB за раз) - При ошибке отправки — чанк пропускается, ретрай на следующем чанке или в BLOCK_DONE - Бэкпрессур через `etcp_router_on_send_ready` для асинхронной досылки ### Переполнение relay (RELAY_FULL) - Максимум 5 downstream-узлов на блок (`MD_MAX_RELAY_DOWNSTREAM`) - При переполнении B отправляет C пакет `RELAY_FULL` (subcmd 0x11) со списком node_id текущих downstream-узлов - C извлекает список и пробует запросить блок у одного из этих узлов ### Структуры данных **relay_block_ctx** — контекст relay для одного блока (индекс по block_id в `md->relay_blocks`): - `block_id[16]`, `media_id[16]`, `chunk` — идентификаторы блока - `chunk_file[2048]` — путь к `.chunk_N` файлу (источник данных для форвардинга) - `file_offset` — всего получено байт от апстрима (размер файла) - `downstream[5]` — массив relay_downstream **relay_downstream** — один пир, запросивший relay: - `node_id`, `group_id` — кому слать - `sent_offset` — сколько байт уже отправлено этому пиру - `waiter` — handle для бэкпрессур-досылки ### Диаграмма relay-потока ``` [Узел A: источник] [Узел B: relay] [Узел C: req] | | | | |<── BLOCK_REQ ────────────| | |── relay downstream add | | |── catchup send (если есть)| | | | |── BLOCK_CHUNK ──────────>| | | |── fwrite (в файл) | | |── fread → BLOCK_CHUNK ──>| | | | |── BLOCK_DONE ───────────>| | | |── fwd BLOCK_DONE ───────>| | |── md_relay_remove | ``` ### Жизненный цикл relay_block_ctx 1. **Создание** — в `md_dl_start_block` при отправке первого BLOCK_REQ 2. **Обновление file_offset** — в `media_download_handle_chunk` при каждом чанке 3. **Форвардинг downstream** — в `media_download_handle_chunk` после записи 4. **Завершение** — в `media_download_handle_done` после валидации: форвард BLOCK_DONE + удаление 5. **Отмена** — при `media_download_cancel`: контекст удаляется вместе с `md->relay_blocks` ### Константы relay | Константа | Значение | Описание | |-----------|----------|----------| | `MD_MAX_RELAY_DOWNSTREAM` | 5 | Макс число downstream-узлов на блок | | `MEDIA_RELAY_FULL_HDR_SIZE` | 39 | Размер заголовка RELAY_FULL (1+16+16+4+2) |