Browse Source
- block_availability: status column (0=processing, 1=completed) - md_ba_complete_block(): DELETE old + check exists + INSERT new id - BLOCK_PROCESSING subcmd (0x10) — INSERT with new id, no REPLACE/IGNORE - BLOCK_RELAY_FULL subcmd (0x11) — redirect when source/relay at capacity - relay_block_ctx: downstream forwarding from chunk file - md_file_load: per-file source download limit (max_downloads_per_file) - RELAY_FULL from source node when at capacity - relay_full handler: retry existing peers, not just new ones - sync.md: full documentationtopo_upd
evgeny
2 months ago
8 changed files with 974 additions and 617 deletions
@ -1,597 +0,0 @@
|
||||
# Архитектура media_delivery — План реализации |
||||
|
||||
## Транспортная модель |
||||
|
||||
- **Все сообщения** (управление + данные блоков) — через `etcp_router` (`etcp_route_send`) |
||||
- Перед передачей блоков устанавливается прямое соединение через `conn_mgr` (минимизация задержек) |
||||
- После установки `etcp_router` автоматически маршрутизирует через прямое соединение |
||||
|
||||
## Что уже есть в коде (используем, не дублируем) |
||||
|
||||
| Компонент | Для чего | |
||||
|---|---| |
||||
| `peers_*.node_type=4` + `adm_tags="supernode=yes"` | Определение суперузла (задаётся админом в чатгуи) | |
||||
| `member_sync` | Синхронизация мемберов (включая adm_tags) | |
||||
| `conn_mgr` | Прямые подключения между узлами (direct/reverse/indirect) | |
||||
| `etcp_router` (ETCP_RT_ID_MEDIA_DELIVERY=0x07) | Надёжная доставка сервисных сообщений (уже забинжено) | |
||||
| `media_index` | Регистрация блоков в SQLite (уже реализовано) | |
||||
| `etcp_add_conn_status_cbk` | Мониторинг статуса ETCP-соединений | |
||||
|
||||
## Новые модули |
||||
|
||||
``` |
||||
src/media_delivery/ |
||||
media_delivery.h/c — расширен: контекст, протокол, init/destroy |
||||
media_delivery_proto.h — packed-структуры протокольных пакетов |
||||
media_download.h/c — логика скачивания блоков |
||||
``` |
||||
|
||||
## Этап 1: BGP-коллбэк в `topo_group` |
||||
|
||||
Уведомление о появлении/обновлении/удалении узлов. Нужен для обнаружения новых суперузлов. |
||||
|
||||
**Файлы:** `src/routing_layer/topo_group.h`, `src/routing_layer/topo_group.c` |
||||
|
||||
Добавить цепочку подписчиков (аналогично `etcp_status_cbk_entry`): |
||||
|
||||
```c |
||||
typedef void (*topo_node_event_fn)(struct TOPO_GROUP* group, uint64_t node_id, |
||||
int event, void* arg); |
||||
// event: 0=NEW, 1=UPDATE, 2=REMOVE |
||||
|
||||
void topo_group_add_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg); |
||||
void topo_group_remove_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg); |
||||
``` |
||||
|
||||
Вызывать в `topo_group_process_nodeinfo` при создании/обновлении `TOPO_NODEQ`, и в `topo_group_process_withdraw` при удалении. |
||||
|
||||
--- |
||||
|
||||
## Этап 2: SQLite-таблицы доступности блоков |
||||
|
||||
Таблицы в `topo_sqlite_db` (уже открыт в `utun_instance`). **Создаются всегда** при `media_delivery_init`, независимо от `is_supernode`. У обычного узла просто пустые. |
||||
|
||||
### `block_availability` — какие узлы имеют блоки |
||||
|
||||
```sql |
||||
CREATE TABLE IF NOT EXISTS block_availability ( |
||||
id INTEGER PRIMARY KEY AUTOINCREMENT, |
||||
block_uuid BLOB NOT NULL, -- 16 байт |
||||
group_id INTEGER NOT NULL, -- идентификатор группы |
||||
node_id INTEGER NOT NULL, -- узел-обладатель блока |
||||
chunk INTEGER NOT NULL, -- номер чанка в файле |
||||
timestamp INTEGER NOT NULL -- когда добавлено/обновлено (unix epoch) |
||||
); |
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_ba_uuid_node |
||||
ON block_availability(block_uuid, node_id); |
||||
CREATE INDEX IF NOT EXISTS idx_ba_id ON block_availability(id); |
||||
CREATE INDEX IF NOT EXISTS idx_ba_node_id ON block_availability(node_id); |
||||
``` |
||||
|
||||
### `super_sync` — последний полученный ID от пира (хранит только принимающая сторона) |
||||
|
||||
```sql |
||||
CREATE TABLE IF NOT EXISTS super_sync ( |
||||
peer_node_id INTEGER PRIMARY KEY, -- другой суперузел |
||||
last_recv_id INTEGER NOT NULL DEFAULT 0 -- последний ID из БАЗЫ ПИРА, полученный мной |
||||
); |
||||
|
||||
-- last_recv_id — это НЕ локальный ID, а id в таблице block_availability пира. |
||||
-- Используется чтобы сказать пиру при SUPER_HELLO: «я принял твои записи до ID X». |
||||
-- Пиру это говорит с какого id начинать следующую отправку. |
||||
``` |
||||
|
||||
Отправитель не хранит состояние — он stateless. При подключении обмениваются `SUPER_HELLO`: |
||||
A говорит «я получил от тебя до ID X», B говорит «я получил от тебя до ID Y». |
||||
Каждая сторона начинает отправку с `(чужой last_recv_id) + 1`. |
||||
|
||||
### Операции над `block_availability` (прямые SQL-запросы) |
||||
|
||||
| Операция | SQL | |
||||
|---|---| |
||||
| Добавить/обновить блок | `INSERT OR REPLACE INTO block_availability(block_uuid,group_id,node_id,chunk,timestamp) VALUES(?,?,?,?,?)` | |
||||
| Удалить все блоки узла | `DELETE FROM block_availability WHERE node_id=?` | |
||||
| Удалить чужие блоки (выключение суперузла) | `DELETE FROM block_availability WHERE node_id != ?` | |
||||
| Удалить блоки группы | `DELETE FROM block_availability WHERE group_id=?` | |
||||
| Найти узлы с блоком | `SELECT node_id, timestamp FROM block_availability WHERE block_uuid=? ORDER BY timestamp DESC` | |
||||
| Получить записи с ID > N | `SELECT * FROM block_availability WHERE id > ? ORDER BY id ASC LIMIT ?` | |
||||
| sync_ptr пира | `SELECT last_recv_id FROM super_sync WHERE peer_node_id=?` — возвращает ID из **чужой базы** | |
||||
| обновить sync_ptr | `INSERT OR REPLACE INTO super_sync(peer_node_id,last_recv_id) VALUES(?,?)` — сохраняет ID из **чужой базы** | |
||||
|
||||
**Плюсы:** не теряем данные при рестарте, не нужен отдельный модуль, проще отладка через `sqlite3`. |
||||
**Ограничение:** все SQL-запросы только из uasync-потока (уже так). |
||||
|
||||
--- |
||||
|
||||
## Этап 2.1: Коллбэк изменения свойств узла в `member_sync` |
||||
|
||||
Многоподписочный коллбэк для отслеживания изменений `adm_tags` любого узла (включая себя). |
||||
Нужен `media_delivery` для динамического включения/выключения режима суперузла. |
||||
|
||||
**Файлы:** `src/chat/member_sync.h`, `src/chat/member_sync.c` |
||||
|
||||
```c |
||||
typedef void (*node_props_changed_fn)(uint64_t node_id, const char* adm_tags, void* arg); |
||||
|
||||
void member_sync_add_props_cbk(node_props_changed_fn fn, void* arg); |
||||
void member_sync_remove_props_cbk(node_props_changed_fn fn, void* arg); |
||||
``` |
||||
|
||||
Вызывается при получении обновлённых `adm_tags` через `MSG_ITEM_UPDATE` (в `member_sync.c` в функции обработки `adm_tags` для любого узла). |
||||
|
||||
`media_delivery` подписывается и в коллбэке: |
||||
- Если `node_id == self` и изменился факт `supernode=yes` → переключить `is_supernode` (см. Init) |
||||
- Если `node_id != self`: проверить стал/перестал быть суперузлом → запустить/остановить репликацию с ним |
||||
|
||||
--- |
||||
|
||||
## Этап 3: `media_delivery_proto.h` |
||||
|
||||
Packed-структуры подкоманд протокола (первый байт данных после etcp_router): |
||||
|
||||
```c |
||||
enum { |
||||
MEDIA_SUBCMD_SERVE_REG = 0x01, // req-узел→суперузел: обслуживай меня |
||||
MEDIA_SUBCMD_SERVE_ACK = 0x02, // суперузел→req-узел: подтверждение |
||||
MEDIA_SUBCMD_SERVE_LEAVE = 0x03, // req-узел→суперузел: отключение |
||||
MEDIA_SUBCMD_QUERY = 0x04, // req-узел→суперузел: кто имеет блоки? |
||||
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_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, // суперузел↔суперузел: рукопожатие при подключении |
||||
}; |
||||
``` |
||||
|
||||
Packed-структуры пакетов: |
||||
- `MEDIA_SERVE_REG`: `subcmd:1, group_id:8` |
||||
- `MEDIA_SERVE_ACK`: `subcmd:1, group_id:8, status:1` |
||||
- `MEDIA_SERVE_LEAVE`: `subcmd:1, group_id:8` |
||||
- `MEDIA_QUERY`: `subcmd:1, group_id:8, media_id:16, num_blocks:2, block_ids[num*16]` |
||||
<!-- src_node_id не передаётся — суперузел ищет по block_uuid независимо от автора. |
||||
src_node_id известен из сообщения чата (media_index_result), req-узел хранит его в media_download. --> |
||||
- `MEDIA_QUERY_RESP`: `subcmd:1, num_entries:2, entries[num × {node_id:8, rtt:2, block_id:16, chunk:4}]` |
||||
<!-- rtt — cumulative_rtt из topo_node суперузла до узла-обладателя (0.1ms). |
||||
req-узел может использовать для сортировки или перемерить сам через route_ping. --> |
||||
- `MEDIA_HAVE_BLOCK`: `subcmd:1, group_id:8, media_id:16, block_id:16, chunk:4, timestamp:8, node_sign:64` |
||||
- `MEDIA_HAVE_BLOCK_ACK`: `subcmd:1, media_id:16, block_id:16, status:1` (0=OK, 1=dup/already) |
||||
- `MEDIA_SUPER_REPL`: `subcmd:1, seq:4, num_entries:2, entries[num × repl_entry]` |
||||
- `MEDIA_SUPER_ACK`: `subcmd:1, ack_seq:4` |
||||
- `MEDIA_BLOCK_REQ`: `subcmd:1, media_id:16, block_id:16, chunk:4, offset:8` |
||||
- `MEDIA_BLOCK_CHUNK`: `subcmd:1, media_id:16, block_id:16, chunk:4, offset:4, data_len:2, data[data_len]` |
||||
<!-- data_len — реальный размер (1..1024 байт). offset — смещение в блоке. --> |
||||
- `MEDIA_BLOCK_DONE`: `subcmd:1, media_id:16, block_id:16, chunk:4, total_size:4, block_sig:64` |
||||
- `MEDIA_CANCEL`: `subcmd:1, media_id:16, block_id:16, chunk:4` |
||||
- `MEDIA_SUPER_HELLO`: `subcmd:1, last_recv_id:8` |
||||
<!-- last_recv_id — ID в МОЕЙ базе (отправителя HELLO) до которого пир подтвердил приём. |
||||
Получатель HELLO использует это значение чтобы знать с какого id начинать отправку. --> |
||||
|
||||
**Файл:** `src/media_delivery/media_delivery_proto.h` |
||||
|
||||
--- |
||||
|
||||
## Этап 4: Расширение `media_delivery_ctx` и init/destroy |
||||
|
||||
```c |
||||
#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_MAX_DOWNLOADS 10 |
||||
#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) |
||||
|
||||
struct media_super_peer { |
||||
struct ll_entry ll; // индекс по peer_node_id (8 байт) |
||||
uint64_t peer_node_id; |
||||
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; // таймер ретрансмита |
||||
void* connect_timer; // таймер переподключения (1 час при ошибке) |
||||
}; |
||||
|
||||
struct media_served_node { |
||||
struct ll_entry ll; // индекс по node_id (8 байт) |
||||
uint64_t node_id; |
||||
uint64_t group_id; |
||||
int64_t joined_at; // timestamp регистрации |
||||
}; |
||||
|
||||
struct media_download_peer { |
||||
uint64_t node_id; |
||||
uint8_t connected; // 1 = conn_mgr подключён |
||||
int num_blocks; // сколько блоков назначено этому узлу |
||||
struct { |
||||
uint8_t block_id[16]; |
||||
uint8_t started; // 1 = BLOCK_REQ отправлен, стрим активен |
||||
uint8_t received; // 1 = BLOCK_DONE получен |
||||
uint8_t validated; // 1 = подпись проверена |
||||
} blocks[64]; |
||||
}; |
||||
|
||||
struct media_download { |
||||
struct ll_entry ll; // индекс по media_id (16 байт) |
||||
uint8_t media_id[16]; |
||||
uint64_t group_id; |
||||
char dest_path[1024]; // путь для сохранения файла (как в media_index) |
||||
char media_base[512]; // базовый путь media-файлов |
||||
int num_blocks; |
||||
uint8_t* block_ids; // num_blocks * 16 |
||||
uint8_t* block_sigs; // num_blocks * 64 |
||||
uint8_t content_hash[32]; |
||||
int64_t file_size; |
||||
int64_t block_size; |
||||
int blocks_received; |
||||
int blocks_validated; |
||||
int num_peers; |
||||
struct media_download_peer peers[10]; |
||||
uint8_t active; // 1 = загрузка в процессе |
||||
uint8_t assembled; // 1 = файл собран |
||||
int err; // 0=OK, <0=ошибка |
||||
void* timeout_timer; // общий таймаут загрузки |
||||
void (*done_cb)(void* arg, int err); |
||||
void* done_arg; |
||||
}; |
||||
|
||||
struct media_delivery_ctx { |
||||
uint64_t self_node_id; |
||||
int initialized; |
||||
uint8_t is_supernode; // из adm_tags в peers_* |
||||
sqlite3* db; // = inst->topo_sqlite_db (кешируем для удобства) |
||||
struct ll_queue* served_nodes; // media_served_node — только суперузел |
||||
struct ll_queue* super_peers; // media_super_peer — только суперузел |
||||
struct ll_queue* downloads; // media_download — активные загрузки |
||||
struct UTUN_INSTANCE* inst; |
||||
void* bgp_cbk_handle; // handle для topo_group_remove_node_cbk |
||||
}; |
||||
``` |
||||
|
||||
**Init:** |
||||
1. Создать таблицы `block_availability` + `super_sync` (CREATE TABLE IF NOT EXISTS) — **всегда**, независимо от `is_supernode` |
||||
2. Прочитать `adm_tags` из `peers_*` → если `supernode=yes` → `is_supernode=1` |
||||
3. Инициализировать `downloads` |
||||
4. Подписаться на `topo_group_add_node_cbk` для всех групп (BGP-события) |
||||
5. Подписаться на `member_sync_add_props_cbk` для отслеживания изменений `adm_tags` |
||||
6. Подписаться на `etcp_add_conn_status_cbk` |
||||
|
||||
**Динамическое переключение суперузла** (через коллбэк `node_props_changed`, см. Этап 2.1): |
||||
|
||||
- **Включение** (`is_supernode` стал `1`): |
||||
- Найти других суперузлов (как при старте) |
||||
- Для каждого: `conn_mgr_connect_node` → `SUPER_HELLO` → репликация (см. 6.1, `last_recv_id` из `super_sync`) |
||||
- **Выключение** (`is_supernode` стал `0`): |
||||
- Очистить чужие записи: `DELETE FROM block_availability WHERE node_id != self` |
||||
- Остановить репликацию со всеми суперузлами, очистить `super_peers` |
||||
- `super_sync` не трогаем — пригодится если снова станем суперузлом |
||||
- `served_nodes` очищается автоматически по дисконнекту req-узлов |
||||
|
||||
**Файлы:** `src/media_delivery/media_delivery.h`, `src/media_delivery/media_delivery.c` |
||||
|
||||
--- |
||||
|
||||
## Этап 5: Обработчик `media_delivery_etcp_recv_cb` |
||||
|
||||
Диспатч по подкомандам: |
||||
|
||||
| Подкоманда | Действие суперузла | Действие req-узла | |
||||
|---|---|---| |
||||
| `SERVE_REG` | Добавить в served_nodes, ответить SERVE_ACK | — | |
||||
| `SERVE_ACK` | — | Отметить суперузел как обслуживающий | |
||||
| `SERVE_LEAVE` | Удалить из served_nodes | — | |
||||
| `QUERY` | SQL: `SELECT node_id, timestamp FROM block_availability WHERE block_uuid=?` → QUERY_RESP | — | |
||||
| `QUERY_RESP` | — | Получить список узлов, начать подключение | |
||||
| `HAVE_BLOCK` | SQL: `INSERT OR REPLACE` → ответ `HAVE_BLOCK_ACK` → репликация + уведомление автора | — | |
||||
| `HAVE_BLOCK_ACK` | — | Подтверждение получено, сброс таймера. Если dup/already — повторно не шлём | |
||||
| `CANCEL` | — | Остановить активный стрим для media_id/block_id, закрыть fd | |
||||
| `SUPER_REPL` | SQL: `INSERT OR REPLACE` для каждой записи, обновить `last_recv_id` в `super_sync`, ответить `SUPER_ACK(max_id)` | — | |
||||
| `SUPER_ACK` | Сдвинуть `peer_last_recv_id = ack_seq`, сбросить таймаут, отправить следующую пачку | — | |
||||
| `SUPER_HELLO` | Записать `hello.last_recv_id` в `peer->peer_last_recv_id` (это ID в нашей базе). Отправить ответный `SUPER_HELLO(мой_last_recv_id для этого пира)`. Начать репликацию с `peer_last_recv_id + 1` | — | |
||||
| `BLOCK_REQ` | — | Открыть файл на чтение, начать стрим чанков BLOCK_CHUNK | |
||||
| `BLOCK_CHUNK` | — | Записать чанк во временный файл блока | |
||||
| `BLOCK_DONE` | — | Закрыть файл, проверить подпись, отметить полученным | |
||||
|
||||
--- |
||||
|
||||
## Этап 6: Логика суперузла — обслуживание и репликация |
||||
|
||||
### 6.1 Старт суперузла |
||||
|
||||
При `is_supernode=1` (и при старте, и при динамическом включении): |
||||
- Таблицы `block_availability` + `super_sync` уже созданы в `media_delivery_init` |
||||
- Для каждой группы где мы член: найти других суперузлов |
||||
- `SELECT node_id FROM peers_<ch> WHERE node_type=4 AND node_id != self` |
||||
- Для каждого другого суперузла: `conn_mgr_connect_node` |
||||
- При успешном подключении обменяться `SUPER_HELLO` (обе стороны): |
||||
- A → B: `SUPER_HELLO(A.last_recv_id_for_B)` |
||||
- B → A: `SUPER_HELLO(B.last_recv_id_for_A)` |
||||
- `last_recv_id` для пира берётся из `super_sync`: `SELECT last_recv_id FROM super_sync WHERE peer_node_id=?` |
||||
- После обмена каждая сторона начинает отправку с `peer_last_recv_id + 1` |
||||
|
||||
При динамическом **включении** (был обычный узел, стал суперузел): |
||||
- `last_recv_id` для каждого пира — из `super_sync` (мог остаться от прошлого включения). |
||||
Если записи нет — 0. Не сбрасываем принудительно. |
||||
- Отправляем `SUPER_HELLO(last_recv_id)`, начинаем репликацию с `last_recv_id + 1` |
||||
|
||||
При динамическом **выключении** (был суперузел, стал обычный): |
||||
- `DELETE FROM block_availability WHERE node_id != self` — чистим чужие записи |
||||
- Останавливаем репликацию со всеми, очищаем `super_peers` |
||||
- `super_sync` сохраняем (last_recv_id остаётся) |
||||
- `served_nodes` очищается автоматически по дисконнекту req-узлов |
||||
|
||||
### 6.2 Обнаружение через BGP-коллбэк и `node_props_changed` |
||||
|
||||
Два источника — разные назначения, без дублирования: |
||||
|
||||
- **BGP-коллбэк** (`TOPO_NODE_EVENT_NEW/UPDATE`): узел появился в BGP-таблице группы (можем до него маршрутизировать). |
||||
Проверяем `node_type=4` (суперузел) → `conn_mgr_connect_node` → `SUPER_HELLO` → репликация. |
||||
BGP не даёт `adm_tags` напрямую, только `node_type`. |
||||
|
||||
- **`node_props_changed`**: изменились `adm_tags` любого узла (включая себя). Это основной источник для отслеживания переключения роли. |
||||
Проверяем изменение `supernode=yes` в `adm_tags` (strstr). |
||||
- `node_id == self` → dynamic switch `is_supernode` |
||||
- `node_id != self`: стал суперузлом → `conn_mgr_connect_node`. Перестал → disconnect. |
||||
|
||||
- **Устранение дублей**: если узел уже в `super_peers`, второй вызов `conn_mgr_connect_node` вернёт `CONN_MGR_ERR_ALREADY_CONNECTED` — игнорируем. |
||||
|
||||
При `TOPO_NODE_EVENT_REMOVE`: |
||||
- Удалить из `super_peers`, очистить контекст |
||||
- `DELETE FROM block_availability WHERE node_id=?` (записи ушедшего узла) |
||||
|
||||
### 6.3 Репликация (per-peer media_super_peer) |
||||
|
||||
Отправитель stateless, приёмник хранит `last_recv_id` в `super_sync`. |
||||
|
||||
**Рукопожатие при подключении:** |
||||
``` |
||||
A → B: SUPER_HELLO(last_recv_id_A_from_B) // «я принял твои (B) записи вплоть до ID X в ТВОЕЙ базе» |
||||
B → A: SUPER_HELLO(last_recv_id_B_from_A) // «я принял твои (A) записи вплоть до ID Y в ТВОЕЙ базе» |
||||
Обе стороны: peer_last_recv_id = значение из HELLO (это ID в СВОЕЙ базе, с которого начинать отправку) |
||||
A начинает отправку с Y+1 (записи своей базы с id > Y), B начинает с X+1 |
||||
``` |
||||
|
||||
**Отправка репликации (sender):** |
||||
``` |
||||
media_super_repl_send(peer): |
||||
rows = SQL("SELECT id, block_uuid, group_id, node_id, chunk, timestamp |
||||
FROM block_availability WHERE id > ? ORDER BY id LIMIT ?", |
||||
peer->peer_last_recv_id, 4) |
||||
if rows пуст: return |
||||
отправить MEDIA_SUPER_REPL с seq = max(id) из rows |
||||
peer->inflight_count++ |
||||
если inflight_count < 4 и есть ещё: отправить ещё |
||||
запустить таймер ретрансмита (peer->timeout_tb) |
||||
``` |
||||
|
||||
**Приём репликации (receiver):** |
||||
``` |
||||
на MEDIA_SUPER_REPL: |
||||
для каждой записи: INSERT OR REPLACE в block_availability |
||||
обновить last_recv_id = max(id из записей пира) в super_sync |
||||
// last_recv_id — это ID в ЧУЖОЙ базе (пира) |
||||
ответить SUPER_ACK(max_id_пира) |
||||
// подтверждаем ID пира (не локальный) |
||||
``` |
||||
|
||||
**Обработка SUPER_ACK (sender):** |
||||
``` |
||||
на MEDIA_SUPER_ACK(ack_seq): |
||||
// ack_seq — ID нашей записи в НАШЕЙ базе, подтверждённый пиром |
||||
peer->peer_last_recv_id = ack_seq |
||||
peer->inflight_count-- |
||||
peer->timeout_tb = MEDIA_REPL_TIMEOUT_TB // сброс таймаута |
||||
если есть ещё записи (id > ack_seq): отправить следующую пачку |
||||
иначе: остановить таймер |
||||
|
||||
таймер ретрансмита (сработал): |
||||
повторно отправить все неподтверждённые записи |
||||
peer->timeout_tb = min(timeout_tb * 2, MEDIA_REPL_TIMEOUT_MAX_TB) |
||||
|
||||
таймаут SUPER_HELLO (2 сек без ответа): |
||||
закрыть соединение, запустить таймер переподключения MEDIA_RECONNECT_COOLDOWN_TB (1 час) |
||||
|
||||
при дисконнекте: |
||||
inflight_count = 0 |
||||
запустить таймер переподключения MEDIA_RECONNECT_COOLDOWN_TB (1 час) |
||||
``` |
||||
|
||||
### 6.4 При новом HAVE_BLOCK |
||||
|
||||
Если есть подключённые суперузлы с `inflight_count < 4`: |
||||
- Сразу отправить новую запись через `SUPER_REPL` |
||||
|
||||
### 6.5 Обслуживание req-узлов |
||||
|
||||
- `served_nodes` — ll_queue с индексом по node_id |
||||
- При `etcp_conn_status_cbk` с DOWN/DELETE: удалить из served_nodes |
||||
- При `SERVE_REG`: добавить, ответить `SERVE_ACK` |
||||
- При `SERVE_LEAVE`: удалить |
||||
|
||||
**Когда req-узел отправляет `SERVE_REG`:** сразу после успешного `conn_mgr_connect_node` к суперузлу (перед отправкой `MEDIA_QUERY`). Но `MEDIA_HAVE_BLOCK` принимается суперузлом и без предварительного `SERVE_REG` — регистрация опциональна, нужна для мониторинга и упрощённого реконнекта. |
||||
|
||||
**Subcmd не по роли:** если не-суперузел получает `QUERY`, `HAVE_BLOCK`, `SUPER_REPL` или `SUPER_HELLO` — ответить `MEDIA_ERR_NOT_SUPERNODE` (status=0xFE в следующем байте после subcmd). Если суперузел получает `QUERY_RESP` или `HAVE_BLOCK_ACK` — игнорировать. |
||||
|
||||
--- |
||||
|
||||
## Этап 7: `media_download.h/c` — скачивание блоков |
||||
|
||||
### 7.1 Запуск загрузки |
||||
|
||||
```c |
||||
int media_download_start(struct UTUN_INSTANCE* inst, uint64_t group_id, |
||||
const struct media_index_result* result, |
||||
const char* dest_path, const char* media_base, |
||||
void (*done_cb)(void* arg, int err), void* done_arg); |
||||
``` |
||||
|
||||
1. Создать `media_download`, заполнить из `result` |
||||
2. Найти суперузлов: `SELECT node_id FROM peers_<ch> WHERE node_type=4` для всех групп |
||||
3. Для каждого суперузла получить RTT (из topo_node) |
||||
4. Выбрать суперузел с минимальным RTT |
||||
5. Отправить `MEDIA_QUERY` через `etcp_router` |
||||
6. Поставить таймаут `MEDIA_QUERY_TIMEOUT_TB` (2 сек) |
||||
7. При таймауте → следующий суперузел |
||||
|
||||
### 7.2 Обработка QUERY_RESP |
||||
|
||||
1. Получить массив `{node_id, rtt, block_id, chunk}` из ответа |
||||
2. Сгруппировать по node_id |
||||
3. Сортировать по RTT, взять до 10 лучших |
||||
4. Для каждого: `conn_mgr_connect_node(mgr, node_id, ...)` |
||||
5. При успешном подключении → `media_download_send_block_requests(dl, peer_index)` |
||||
|
||||
### 7.3 Стриминг блока чанками по 1KB с backpressure |
||||
|
||||
Передача блока (10-25MB) идёт потоком чанков по ~1KB каждый. |
||||
Congestion control через встроенный backpressure `etcp_router_on_send_ready`: |
||||
отправитель шлёт чанки пока `inflight` не заполнится, затем ждёт коллбэк «можно отправлять» |
||||
и продолжает стрим. |
||||
|
||||
Состояние одного стрима (блок-холдер → req-узел): |
||||
```c |
||||
struct media_block_stream { |
||||
uint8_t media_id[16]; |
||||
uint8_t block_id[16]; |
||||
uint32_t chunk; |
||||
uint64_t offset; // текущее смещение в блоке |
||||
uint32_t bytes_sent; // всего отправлено байт |
||||
uint32_t total_size; // ожидаемый размер блока |
||||
uint8_t done; // 1 = стрим завершён |
||||
void* waiter; // handle etcp_router_on_send_ready |
||||
FILE* file; // fd исходного файла |
||||
uint64_t dst_node_id; |
||||
uint64_t group_id; |
||||
}; |
||||
``` |
||||
|
||||
**Отправка (холдер):** |
||||
1. При `MEDIA_BLOCK_REQ` — создать `media_block_stream`, открыть файл на чтение |
||||
2. Вызвать `etcp_router_on_send_ready` для регистрации backpressure waiter |
||||
3. В коллбэке `on_send_ready`: читать чанк из файла (до 1024 байт), отправить `MEDIA_BLOCK_CHUNK` |
||||
4. Если `etcp_route_send` вернул быстрый успех — отправить следующий чанк сразу |
||||
5. Если очередь заполнена — waiter сработает когда можно отправлять дальше |
||||
6. Когда весь блок отправлен — отправить `MEDIA_BLOCK_DONE` с `block_sig` и `total_size` |
||||
7. Закрыть файл, освободить стрим |
||||
|
||||
**Приём (req-узел):** |
||||
1. При `MEDIA_BLOCK_REQ` — создать writer для блока: открыть временный файл по `dest_path + offset` |
||||
2. При `MEDIA_BLOCK_CHUNK` — записать данные файл, сдвинуть offset |
||||
3. При `MEDIA_BLOCK_DONE` — проверить что `total_size` совпадает с записанным, закрыть файл |
||||
4. Верифицировать `block_sig` (Ed25519 от подписи в `block_sigs`) |
||||
5. Если подпись верна: отметить блок полученным, `blocks_received++` |
||||
6. Если подпись не совпала: удалить файл, запросить блок с другого узла |
||||
7. Проверить `blocks_received == num_blocks` → сборка файла (см. 7.5) |
||||
|
||||
### 7.4 Распределение блоков по узлам |
||||
|
||||
- Для каждого подключённого узла: распределить неполученные блоки равномерно |
||||
- Отправить `MEDIA_BLOCK_REQ` для блоков назначенных этому узлу |
||||
- Один узел может параллельно стримить несколько блоков (отдельный стрим на каждый) |
||||
- Если узел дисконнектится — его неначатые блоки перераспределяются на других |
||||
|
||||
### 7.5 Сборка файла и обновление БД |
||||
|
||||
При получении каждого блока: |
||||
1. Проверить: `blocks_received == num_blocks`? |
||||
2. Если да — склеить блоки в порядке chunk в целевой файл: |
||||
``` |
||||
for (n = 0; n < num_blocks; n++) { |
||||
прочитать temp_файл блока n |
||||
записать в dest_path на offset = n * block_size |
||||
удалить temp_файл |
||||
} |
||||
``` |
||||
3. Проверить `content_hash` собранного файла (SHA256) |
||||
4. Если хеш совпал: |
||||
- Удалить записи отдельных блоков из `media_files` |
||||
- Вставить одну запись с `chunk=0`, `chunk_size=file_size` (цельный файл) |
||||
- Для каждого полученного и проверенного блока — отправить `MEDIA_HAVE_BLOCK` суперузлу |
||||
(с подтверждением и перезапросом — см. 7.6) |
||||
- Вызвать `done_cb(0)` |
||||
5. Если хеш не совпал — удалить собранный файл, перезапросить проблемные блоки |
||||
|
||||
### 7.6 HAVE_BLOCK с подтверждением и перезапросом |
||||
|
||||
Отправка `MEDIA_HAVE_BLOCK` — с обязательным подтверждением от суперузла: |
||||
|
||||
``` |
||||
req-узел → суперузел: MEDIA_HAVE_BLOCK |
||||
запустить таймер HAVE_BLOCK_TIMEOUT_TB (2 сек) |
||||
|
||||
суперузел → req-узел: MEDIA_HAVE_BLOCK_ACK(status=0 OK, 1 dup) |
||||
если OK: INSERT в block_availability → репликация → уведомление автора |
||||
если dup: запись уже есть, ничего не делаем |
||||
|
||||
req-узел получил ACK: |
||||
сброс таймера, блок считается доставленным суперузлу |
||||
|
||||
req-узел таймаут / ошибка: |
||||
выбрать другой суперузел (следующий по RTT из списка) |
||||
отправить MEDIA_HAVE_BLOCK повторно |
||||
если все суперузлы исчерпаны → done_cb(ERR) |
||||
``` |
||||
|
||||
Константа: `HAVE_BLOCK_TIMEOUT_TB = 20000` (2 сек). |
||||
|
||||
### 7.7 Отмена передачи (MEDIA_SUBCMD_CANCEL) |
||||
|
||||
Из GUI или при ошибке req-узел может отменить активную передачу блока: |
||||
|
||||
``` |
||||
req-узел → блок-холдер: MEDIA_CANCEL(media_id, block_id, chunk) |
||||
|
||||
блок-холдер при получении CANCEL: |
||||
остановить стрим для указанного блока |
||||
закрыть fd |
||||
отменить waiter (etcp_router_cancel_send_ready) |
||||
освободить media_block_stream |
||||
``` |
||||
|
||||
Также при дисконнекте блок-холдера или req-узла — все активные стримы отменяются автоматически (через `etcp_conn_status_cbk`). |
||||
|
||||
### 7.8 Обработка ошибок |
||||
|
||||
- Таймаут BLOCK_REQ (нет начала стрима за 10 сек) → перезапросить с другого узла |
||||
- Стрим stalled (нет чанков > 30 сек) → отправить CANCEL, запросить блок с другого узла |
||||
- BLOCK_DONE с неверной подписью → удалить файл, запросить с другого узла |
||||
- Все узлы для блока исчерпаны → `done_cb(ERR)` |
||||
- Дисконнект узла → перераспределить его блоки на других, отменить активные стримы с него |
||||
- Общий таймаут загрузки → `done_cb(ERR_TIMEOUT)` |
||||
- HAVE_BLOCK без ACK (2 сек) → переключиться на следующий суперузел |
||||
|
||||
--- |
||||
|
||||
## Этап 8: Интеграция |
||||
|
||||
**Файлы для изменения:** |
||||
- `src/Makefile.am` — добавить `media_delivery/media_download.c` |
||||
- `src/media_delivery/media_delivery.c` — полностью переписать обработчик |
||||
- `src/media_delivery/media_delivery.h` — расширить структуру |
||||
- `src/routing_layer/topo_group.h` + `topo_group.c` — добавить BGP-коллбэк |
||||
- `src/chat/member_sync.h` + `member_sync.c` — добавить `node_props_changed` коллбэк |
||||
|
||||
**Уже готово:** |
||||
- `media_delivery_init` вызывается в `utun_instance.c:650` |
||||
- `media_delivery_destroy` вызывается в `utun_instance.c:514` |
||||
- `ETCP_RT_ID_MEDIA_DELIVERY` забинжен в `media_delivery_init` |
||||
|
||||
--- |
||||
|
||||
## Решённые вопросы |
||||
|
||||
1. **Формат передачи блоков:** стриминг чанками по ~1KB с congestion control через `etcp_router_on_send_ready` (вариант А). Отправитель шлёт чанки пока inflight позволяет, затем ждёт backpressure-коллбэк. Завершается `MEDIA_BLOCK_DONE` с `block_sig` (Ed25519) и `total_size`. |
||||
|
||||
2. **Валидация при приёме:** каждый блок подписан Ed25519. При `BLOCK_DONE` проверяем `block_sig` против подписи из `media_index_result.block_sigs`. Если не совпала — удаляем файл, запрашиваем блок с другого узла. |
||||
|
||||
3. **Уведомление автора контента:** один раз, при успешной вставке `HAVE_BLOCK` в `block_availability` (после ответа `HAVE_BLOCK_ACK`). Без переповторов. Если что-то потерялось — req-узел перезапросит у другого суперузла. HAVE_BLOCK всегда с подтверждением: таймаут 2 сек → переключение на следующий суперузел. |
||||
|
||||
4. **Файловая система:** путь назначения (`dest_path`) и `media_base` передаются в `media_download_start` (точно как в `media_index_register_async`). Временные файлы блоков сохраняются рядом с dest_path с суффиксом `.chunk_N`. |
||||
|
||||
5. **Сборка файла:** в момент получения очередного блока проверяем `blocks_validated == num_blocks`. Если все на месте — склеиваем чанки по порядку в `dest_path`, проверяем `content_hash` (SHA256). При успехе удаляем записи отдельных блоков из `media_files`, вставляем одну запись с `chunk_size=file_size` (цельный файл). |
||||
@ -0,0 +1,251 @@
|
||||
# Репликация 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_<channel>`. |
||||
|
||||
Механизмы обнаружения: |
||||
|
||||
- **При старте** (`md_super_start`, стр. 730): суперузел ищет в БД все узлы с `node_type=4` в своей chat-группе и инициирует прямое подключение к каждому через `conn_mgr_open_invite`. |
||||
- **При появлении нового узла в BGP** (`md_on_bgp_node`, стр. 790): проверяет `node_type=4` в `peers_<channel>` — если да, вызывает `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_<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-записи для одного блока. |
||||
|
||||
## 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) | |
||||
Loading…
Reference in new issue