Browse Source
- SoundManager: add currentPcmCursor()/isCurrentPcmPlaying() via ma_sound_get_cursor_in_pcm_frames / ma_sound_is_playing - messagedelegate: replace wall-clock position tracking with real miniaudio cursor, remove 200ms hardcoded delay - messagedelegate: end-of-playback detection via isCurrentPcmPlaying() instead of wall-clock comparison (was killing audio ~200ms early) - messagedelegate: EMA smoothing of displayFrac/displayElapsed (0.15 factor, 16ms timer) - messagedelegate: timer continues after audio stops to animate EMA decay to 0 (not mouse-move dependent) - media_delivery: download progress callback, Ed25519 sig verify, author fallback, content hash check - conn_mgr: already-connected returns OK with callback instead of error - chatgui: voice download-pending auto-play, file download progress bar, media bar colorstopo_upd
evgeny
2 months ago
23 changed files with 1165 additions and 162 deletions
@ -0,0 +1,530 @@
|
||||
# db_sync — децентрализованная реплицируемая таблица с цепным хешированием |
||||
|
||||
## 1. Архитектура |
||||
|
||||
### 1.1 Обзор |
||||
|
||||
``` |
||||
UTUN_INSTANCE |
||||
│ |
||||
┌────▼────┐ |
||||
│ DB_SYNC │ (глобальный контекст, один на инстанс) |
||||
└────┬────┘ |
||||
│ |
||||
┌────────────────┼────────────────┐ |
||||
▼ ▼ ▼ |
||||
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ |
||||
│DB_SYNC │ │DB_SYNC │ │DB_SYNC │ |
||||
│INSTANCE │ │INSTANCE │ │INSTANCE ... │ |
||||
│msg_GENERAL │ │msg_SECRET │ │ │ |
||||
└──────┬───────┘ └──────┬───────┘ └──────────────┘ |
||||
│ │ |
||||
┌────────┼────────┐ │ |
||||
▼ ▼ ▼ ▼ |
||||
┌────────┐┌────────┐┌──────┐ |
||||
│SI_PEER ││SI_PEER ││SI_...│ (per-instance peer state) |
||||
│node_A ││node_B ││ │ |
||||
└────────┘└────────┘└──────┘ |
||||
``` |
||||
|
||||
**DB_SYNC** — глобальный модуль (struct, один на UTUN_INSTANCE): |
||||
- `inst` — указатель на UTUN_INSTANCE |
||||
- `db` — SQLite-соединение (своё или shared `inst->topo_sqlite_db`) |
||||
- `instances[]` — динамический массив DB_SYNC_INSTANCE |
||||
- `peer_check_timer` — периодическая проверка (5с), ищет несинхронизированных пиров |
||||
- `done_cbks` — связанный список коллбэков завершения синхронизации |
||||
|
||||
**DB_SYNC_INSTANCE** — одна таблица синхронизации: |
||||
- `table_name` — имя SQLite-таблицы (например `msg_GENERAL`) |
||||
- `hash` — instance_hash = первые 8 байт SHA256(name || id_be) для маршрутизации сообщений |
||||
- `next_id` — автоинкрементный ID записей |
||||
- `last_timestamp_ms` — последний использованный timestamp (монотонный) |
||||
- `peers[]` — динамический массив SI_PEER (состояние синхронизации с каждым пиром) |
||||
- `recalc` — отложенный пересчёт цепных хешей (асинхронно, батчами по 50) |
||||
- `ttl_timer` — периодическая TTL-очистка (каждый час) |
||||
- `on_insert` / `on_insert_arg` — коллбэк при вставке записи |
||||
|
||||
**SI_PEER** — состояние пира в контексте инстанса: |
||||
- `node_id` — идентификатор пира |
||||
- `synced_pos` — позиция (0-based) до которой данные синхронизированы |
||||
- `verified_pos` — последняя подтверждённая позиция (совпадение хешей) |
||||
- `last_peer_count` — последнее известное количество записей у пира |
||||
- `sync_state` — 0=idle, 1=syncing, 2=synced |
||||
- `sync_start_tb` — время начала синхронизации (для детекции таймаутов) |
||||
- `retry_count` — счётчик ретраев (зарезервировано, не используется) |
||||
|
||||
### 1.2 Структура данных (SQLite) |
||||
|
||||
Каждый инстанс — это SQLite-таблица: |
||||
|
||||
```sql |
||||
CREATE TABLE "msg_GENERAL" ( |
||||
timestamp INTEGER NOT NULL, -- мс, монотонный (NTP или local) |
||||
node_id INTEGER NOT NULL, -- author node_id (8 байт) |
||||
id INTEGER NOT NULL, -- автоинкремент внутри инстанса |
||||
chain_hash BLOB NOT NULL, -- SHA256(prev_hash || id || ts || author || sig) |
||||
flags INTEGER DEFAULT 0, -- DB_REC_FLAG_WAS_SENT (0x01) |
||||
data BLOB, -- JSON-данные записи |
||||
author_signature BLOB NOT NULL, -- Ed25519(ts || data), 64 байта |
||||
local_attrs TEXT DEFAULT '', -- локальные атрибуты (не реплицируются) |
||||
delivered_peers INTEGER DEFAULT 0, -- счётчик пиров, принявших PUSH |
||||
delivery_chain TEXT DEFAULT '', -- цепочка peer_id hex через запятую |
||||
PRIMARY KEY (timestamp, author_signature) |
||||
); |
||||
CREATE INDEX idx_msg_GENERAL_ttl ON msg_GENERAL (node_id, timestamp); |
||||
``` |
||||
|
||||
**Цепной хеш:** |
||||
``` |
||||
chain_hash[pos] = SHA256(chain_hash[pos-1] || id || timestamp || author || author_signature) |
||||
chain_hash[0] = SHA256(0x00..00 [32] || id_0 || ts_0 || author_0 || sig_0) |
||||
``` |
||||
|
||||
Упорядочение: `ORDER BY timestamp, author_signature`. |
||||
|
||||
### 1.3 Ключевые механизмы |
||||
|
||||
| Механизм | Описание | |
||||
|----------|----------| |
||||
| **Цепной хеш** | Каждая запись ссылается на хеш предыдущей — защита от вставки/удаления в середине цепочки | |
||||
| **Ed25519-подписи** | Каждая запись подписана автором. Проверяется при вставке (и локальной, и от пира) | |
||||
| **Async recalc** | Пересчёт `chain_hash` батчами по 50 записей (через `uasync_set_timeout`) — не блокирует event loop | |
||||
| **Master/Slave** | Узел с бОльшим `node_id` — мастер (шлёт `INIT_SYNC`). Узел с меньшим — слейв (шлёт `REQUEST_SYNC`) | |
||||
| **Периодическая проверка** | Каждые 5 секунд — поиск пиров с sync_state=0 и запуск синхронизации, детекция таймаутов (15с) | |
||||
| **TTL-очистка** | Каждый час — удаление неподтверждённых (`flags&1==0`) записей локального автора старше `db_sync_ttl` | |
||||
| **Lazy-register** | При получении сообщения с неизвестным instance_hash — поиск таблицы `msg_<ch_id>` с совпадающим хешем | |
||||
| **Push** | Новая локальная запись немедленно рассылается всем пирам в sync_state >= 1 | |
||||
| **Delivery tracking** | `delivered_peers` / `delivery_chain` — отслеживание доставки PUSH каждому пиру | |
||||
|
||||
### 1.4 Интеграция с ETCP |
||||
|
||||
- **Service ID:** `ETCP_RT_ID_DB_SYNC = 0x20` |
||||
- **Биндинг:** `etcp_bind(inst, 0x20, db_sync_recv_cb)` — приём сообщений |
||||
- **Connection callbacks:** `etcp_add_conn_status_cbk(inst, db_sync_on_conn_status, NULL)` — отслеживание UP/DOWN соединений |
||||
- **Отправка:** через `etcp_send()` в прямое P2P-соединение |
||||
- **Маршрутизация:** каждый пакет начинается с `[svc:1][hash_be:8]`, hash идентифицирует инстанс на приёмной стороне |
||||
|
||||
--- |
||||
|
||||
## 2. Диаграмма состояний синхронизации |
||||
|
||||
### 2.1 Состояния пира (sync_state) |
||||
|
||||
``` |
||||
┌─────────────────────────────────────────────┐ |
||||
│ │ |
||||
▼ │ |
||||
┌───────┐ INIT_SYNC/REQUEST_SYNC ┌─────────┐ │ |
||||
│ IDLE │ ──────────────────────────► │ SYNCING │ │ |
||||
│ (0) │ │ (1) │ │ |
||||
└───────┘ └────┬────┘ │ |
||||
▲ │ │ |
||||
│ ┌─────────┼──────┘ |
||||
│ │ │ |
||||
│ ┌───────────▼──┐ ┌───▼──────────┐ |
||||
│ │ SYNC_DONE │ │ TIMEOUT+ERROR │ |
||||
│ │ matched ✓ │ │ (15s) │ |
||||
│ │ (2) │ │ reset → IDLE │ |
||||
│ └──────────────┘ └───────────────┘ |
||||
│ │ |
||||
└─────────────────────────────────────────────┘ |
||||
CONN DOWN / ERROR (DB_ERR_NOT_FOUND, DB_ERR_DISABLED) |
||||
``` |
||||
|
||||
### 2.2 Диаграмма протокола синхронизации |
||||
|
||||
``` |
||||
MASTER (node_id > peer) SLAVE (node_id < peer) |
||||
─────────────────────── ────────────────────── |
||||
|
||||
conn UP → initiate sync |
||||
│ |
||||
├─ INIT_SYNC ──────────────────────────► |
||||
│ [my_count:4] │ flush recalc |
||||
│ │ определяет tp = min(peer_c, my_c) - 1 |
||||
│ │ вычисляет ch8_at_tp |
||||
│ │ строит sparse hashes (интервалы: 1,1,1,2,2,4×8) |
||||
│ │ |
||||
│ ◄────────────────────────── INIT_RESP │ |
||||
│ [tp:4][ch8:8][has_tail:1] │ |
||||
│ [sparse_cnt:1][(pos,ch8)*] │ |
||||
│ │ |
||||
│ flush recalc │ |
||||
│ ┌─ peer ch8 == 0? ──► send ALL data │ |
||||
│ ├─ ch8 match + no_tail? ──► DONE │ |
||||
│ ├─ ch8 match + has_tail? ──► send │ |
||||
│ │ from tp+1, want_from=tp+1 │ |
||||
│ └─ ch8 mismatch? ──► находим │ |
||||
│ первую разницу через sparse │ |
||||
│ hashes → fm (first mismatch) │ |
||||
│ vp = fm-1, send from fm │ |
||||
│ │ |
||||
│ SEND_DATA ──────────────────────────► │ |
||||
│ [from:4][count:2][vp:4] │ вставляет записи, проверяет подписи |
||||
│ [want_from:4][records...] │ проверяет want_from: |
||||
│ │ want≠NONE → отправляет встречный SEND_DATA |
||||
│ │ want=NONE + ss=1 → SYNC_DONE |
||||
│ │ |
||||
│ ◄────────────────────────── SEND_DATA │ |
||||
│ (встречные данные при want→NONE) │ |
||||
│ │ |
||||
│ ┌─ want=NONE + ss=1 → SYNC_DONE ────►│ |
||||
│ │ [mc:4][ch8_last:8] │ сверяет mc==pc и ch8==pch8 |
||||
│ │ │ match → ss=2, DONE |
||||
│ │ │ mc<pc → request tail |
||||
│ │ │ mc>pc → send tail |
||||
│ └─────────────────────────────────────│ |
||||
``` |
||||
|
||||
### 2.3 Временная диаграмма жизненного цикла |
||||
|
||||
``` |
||||
┌─ init ─────────────────────────────────────────────────────────────────────┐ |
||||
│ db_sync_init(inst) │ |
||||
│ ├─ открывает SQLite (shared или свой chats.db) │ |
||||
│ ├─ etcp_bind(inst, 0x20, recv_cb) │ |
||||
│ ├─ etcp_add_conn_status_cbk(conn_status) │ |
||||
│ └─ запускает peer_check_timer (5s) │ |
||||
│ │ |
||||
│ ... позже, при создании канала ... │ |
||||
│ │ |
||||
│ db_sync_instance_add(inst, "msg_GENERAL", hash, 1) │ |
||||
│ ├─ CREATE TABLE IF NOT EXISTS... │ |
||||
│ ├─ загружает next_id = MAX(id)+1 │ |
||||
│ ├─ db_verify_chain() — проверка целостности цепных хешей │ |
||||
│ ├─ запускает TTL-таймер (1 час) │ |
||||
│ └─ для всех активных соединений: si_peer_add() + initiate_sync() │ |
||||
├────────────────────────────────────────────────────────────────────────────┤ |
||||
│ │ |
||||
│ ┌─── DB_SYNC_INSTANCE активен ───────────────────────────────────────────┐│ |
||||
│ │ ││ |
||||
│ │ conn UP ──► on_conn_up() ││ |
||||
│ │ └─ для всех enabled инстансов: ││ |
||||
│ │ si_peer_add() → initiate_sync() ││ |
||||
│ │ ││ |
||||
│ │ conn DOWN ──► on_conn_down() ││ |
||||
│ │ └─ сброс sync_state для всех инстансов ││ |
||||
│ │ ││ |
||||
│ │ recv_cb() — приём сообщений ││ |
||||
│ │ ├─ извлекает instance_hash, type, payload ││ |
||||
│ │ ├─ lazy-register если инстанс не найден ││ |
||||
│ │ └─ диспетчеризация по type ││ |
||||
│ │ ││ |
||||
│ │ peer_check_timer (каждые 5с) ││ |
||||
│ │ ├─ находит пиров c sync_state=0, запускает sync ││ |
||||
│ │ └─ детектит таймауты (15s), сбрасывает sync_state ││ |
||||
│ │ ││ |
||||
│ │ ttl_timer (каждый час) ││ |
||||
│ │ └─ удаляет неподтверждённые локальные записи старше TTL ││ |
||||
│ │ ││ |
||||
│ │ recalc_timer (асинхронно, батчи по 50) ││ |
||||
│ │ └─ пересчитывает chain_hash для записей после вставки ││ |
||||
│ └─────────────────────────────────────────────────────────────────────────┘│ |
||||
│ │ |
||||
│ db_sync_instance_remove(si) │ |
||||
│ ├─ отменяет таймеры │ |
||||
│ ├─ освобождает peers[] │ |
||||
│ └─ сдвигает массив instances[] (таблица БД не удаляется) │ |
||||
│ │ |
||||
│ db_sync_destroy(inst) │ |
||||
│ ├─ etcp_unbind(0x20) │ |
||||
│ ├─ останавливает все таймеры │ |
||||
│ ├─ освобождает все инстансы + done_cbks │ |
||||
│ ├─ закрывает SQLite │ |
||||
│ └─ u_free(db) │ |
||||
└─────────────────────────────────────────────────────────────────────────────┘ |
||||
``` |
||||
|
||||
--- |
||||
|
||||
## 3. Протокол (форматы кодограмм) |
||||
|
||||
### 3.1 Общий заголовок ETCP |
||||
|
||||
Каждое сообщение db_sync инкапсулируется в ETCP-дейтаграмму: |
||||
|
||||
``` |
||||
┌─────────────────┬──────────────┬──────────────────────────────┐ |
||||
│ svc_id (1 byte) │ hash_be (8) │ payload (variable) │ |
||||
│ ETCP_RT_ID=0x20 │ instance_hash│ зависит от type │ |
||||
└─────────────────┴──────────────┴──────────────────────────────┘ |
||||
``` |
||||
|
||||
**instance_hash** = первые 8 байт `SHA256(table_name || id_be)` в big-endian. |
||||
Это позволяет приёмной стороне найти `DB_SYNC_INSTANCE` по хешу. |
||||
|
||||
### 3.2 Типы сообщений |
||||
|
||||
| #define | Value | Направление | Назначение | |
||||
|---------|-------|-------------|------------| |
||||
| `DB_MSG_INIT_SYNC` | 0x01 | Master→Slave | Инициирует синхронизацию | |
||||
| `DB_MSG_INIT_RESP` | 0x02 | Slave→Master | Ответ с хешами на точке расхождения | |
||||
| `DB_MSG_SEND_DATA` | 0x04 | ↔ | Передача батча записей | |
||||
| `DB_MSG_PUSH` | 0x05 | →Peers | Рассылка новой записи всем synced-пирам | |
||||
| `DB_MSG_ACK_PUSH` | 0x06 | →Author | Подтверждение получения PUSH | |
||||
| `DB_MSG_SYNC_DONE` | 0x07 | ↔ | Завершение синхронизации | |
||||
| `DB_MSG_ERROR` | 0x08 | ↔ | Ошибка (NOT_FOUND, DISABLED) | |
||||
| `DB_MSG_REQUEST_SYNC` | 0x09 | Slave→Master | Запрос на инициацию синхронизации | |
||||
|
||||
### 3.3 DB_MSG_INIT_SYNC (0x01) |
||||
|
||||
**Отправитель:** мастер (node_id > peer_id) |
||||
**Назначение:** сообщить пиру своё количество записей для поиска точки расхождения |
||||
|
||||
``` |
||||
┌──────┬──────────────┐ |
||||
│ 0x01 │ count (4 BE) │ |
||||
└──────┴──────────────┘ |
||||
total: 5 bytes |
||||
``` |
||||
|
||||
- `count` — количество записей в таблице мастера (uint32, big-endian) |
||||
|
||||
### 3.4 DB_MSG_INIT_RESP (0x02) |
||||
|
||||
**Отправитель:** слейв (node_id < peer_id) |
||||
**Назначение:** вернуть хеш на точке пересечения tp и разреженные хеши для бинарного поиска расхождения |
||||
|
||||
``` |
||||
┌──────┬──────────────┬──────────────┬─────────────┬─────────────┬──────────────────────────────────┐ |
||||
│ 0x02 │ tp (4 BE) │ ch8_at_tp(8) │ has_tail(1) │ sparse_cnt │ sparse_entries[sparse_cnt] │ |
||||
│ │ │ │ │ (1) │ { pos(4BE), ch8(8) } × sparse_cnt│ |
||||
└──────┴──────────────┴──────────────┴─────────────┴─────────────┴──────────────────────────────────┘ |
||||
total: 14 + 12×sparse_cnt bytes (max 206) |
||||
``` |
||||
|
||||
- `tp` — точка пересечения: `min(peer_count, my_count) - 1` (uint32, network byte order) |
||||
- `ch8_at_tp` — первые 8 байт chain_hash на позиции tp (uint64, host byte order) |
||||
- `has_tail` — 1 если у слейва есть записи после tp (uint8) |
||||
- `sparse_cnt` — количество разреженных точек (uint8) |
||||
- `sparse_entries` — массив `{pos:4, ch8:8}` от tp назад с интервалами: 1, 1, 1, 2, 2, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4 (макс. 16) |
||||
|
||||
**Пример sparse-интервалов при tp=10:** |
||||
``` |
||||
tp=10 → entry[0]: pos=9 (tp - 1) |
||||
entry[1]: pos=8 (tp - 2) |
||||
entry[2]: pos=7 (tp - 3) |
||||
entry[3]: pos=5 (tp - 5) |
||||
entry[4]: pos=3 (tp - 7) |
||||
entry[5]: pos=0 (tp - 11) — только если tp достаточно велико |
||||
``` |
||||
|
||||
### 3.5 DB_MSG_SEND_DATA (0x04) |
||||
|
||||
**Отправитель:** любой узел |
||||
**Назначение:** передать батч записей (до 32) и/или запросить встречные данные |
||||
|
||||
``` |
||||
┌──────┬───────────┬───────────┬───────────┬──────────────┬──────────────────────────┐ |
||||
│ 0x04 │ from(4BE) │ count(2) │ vp(4BE) │ want_from(4) │ records[count] │ |
||||
│ │ │ │ │ │ ≤32 штук │ |
||||
└──────┴───────────┴───────────┴───────────┴──────────────┴──────────────────────────┘ |
||||
header: 14 bytes + records |
||||
``` |
||||
|
||||
- `from` — начальная позиция записей (uint32, network byte order) |
||||
- `count` — количество записей в этом батче (uint16, network byte order), ≤ 32 |
||||
- `vp` — verified position (uint32, BE): последняя позиция, где хеши совпали. `0xFFFFFFFF` если нет |
||||
- `want_from` — запрос встречных данных от пира начиная с этой позиции. `0xFFFFFFFF` = `DB_WANT_FROM_NONE` (не запрашивать) |
||||
- `records` — массив записей |
||||
|
||||
#### Формат одной записи (record): |
||||
|
||||
``` |
||||
┌───────┬───────┬──────────┬──────────┬────────────┬──────────┬──────────┐ |
||||
│ id(8) │ ts(8) │ author(8)│ dlen(4BE)│ data(dlen) │ sig_len │ sig(64) │ |
||||
│ │ │ │ │ │ =64 (1) │ │ |
||||
└───────┴───────┴──────────┴──────────┴────────────┴──────────┴──────────┘ |
||||
минимальный размер: 28 + 1 + 64 = 93 bytes (при dlen=0) |
||||
``` |
||||
|
||||
- `id` — идентификатор записи (uint64, host byte order) |
||||
- `ts` — timestamp в мс (uint64, host byte order) |
||||
- `author` — node_id автора (uint64, host byte order) |
||||
- `dlen` — длина JSON-данных (uint32, network byte order) |
||||
- `data` — JSON-данные (dlen байт) |
||||
- `sig_len` — длина подписи, всегда 64 (uint8) |
||||
- `sig` — Ed25519-подпись: `Ed25519(ts[8] || data[dlen])`, 64 байта |
||||
|
||||
**Note:** поля `id`, `ts`, `author` в host byte order (LE на x86), а `dlen` в network byte order (BE). |
||||
|
||||
### 3.6 DB_MSG_PUSH (0x05) |
||||
|
||||
**Отправитель:** автор записи → все synced-пиры |
||||
**Назначение:** рассылка новой записи в реальном времени |
||||
|
||||
Формат идентичен одной record из SEND_DATA: |
||||
|
||||
``` |
||||
┌──────┬───────┬───────┬──────────┬──────────┬────────────┬──────────┬──────────┐ |
||||
│ 0x05 │ id(8) │ ts(8) │ author(8)│ dlen(4BE)│ data(dlen) │ sig_len │ sig(64) │ |
||||
│ │ │ │ │ │ │ =64 (1) │ │ |
||||
└──────┴───────┴───────┴──────────┴──────────┴────────────┴──────────┴──────────┘ |
||||
минимальный размер: 29 + 1 + 64 = 94 bytes (при dlen=0) |
||||
``` |
||||
|
||||
### 3.7 DB_MSG_ACK_PUSH (0x06) |
||||
|
||||
**Отправитель:** получатель PUSH → автору |
||||
**Назначение:** подтверждение получения PUSH-записи |
||||
|
||||
``` |
||||
┌──────┬──────────┬──────────────┐ |
||||
│ 0x06 │ ts (8) │ author (8) │ |
||||
└──────┴──────────┴──────────────┘ |
||||
total: 17 bytes |
||||
``` |
||||
|
||||
- `ts` — timestamp записи (uint64, host byte order) |
||||
- `author` — node_id автора (uint64, host byte order) |
||||
|
||||
При получении ACK_PUSH автор помечает запись флагом `DB_REC_FLAG_WAS_SENT` (0x01) и обновляет `delivery_chain`/`delivered_peers`. |
||||
|
||||
### 3.8 DB_MSG_SYNC_DONE (0x07) |
||||
|
||||
**Отправитель:** любой узел |
||||
**Назначение:** финальная сверка — сравнение количества записей и последнего chain_hash |
||||
|
||||
``` |
||||
┌──────┬─────────────┬──────────────────┐ |
||||
│ 0x07 │ count (4BE) │ chain_hash8 (8) │ |
||||
└──────┴─────────────┴──────────────────┘ |
||||
total: 13 bytes |
||||
``` |
||||
|
||||
- `count` — количество записей в таблице отправителя (uint32, network byte order) |
||||
- `chain_hash8` — первые 8 байт последнего chain_hash (uint64, host byte order) |
||||
|
||||
**Обработка:** |
||||
- `count == my_count && ch8 == my_ch8` → sync завершён, ss→2, fire `db_sync_done_cb` |
||||
- `my_count < peer_count` → запрашиваем хвост (SEND_DATA from=my_count, want=my_count) |
||||
- `my_count > peer_count` → отправляем хвост (SEND_DATA from=peer_count) |
||||
- `count == my_count && ch8 != my_ch8` → принимаем как сошедшееся (конвергенция) |
||||
|
||||
### 3.9 DB_MSG_ERROR (0x08) |
||||
|
||||
**Отправитель:** любой узел |
||||
**Назначение:** сигнализировать об ошибке (инстанс не найден или выключен) |
||||
|
||||
``` |
||||
┌──────┬──────────┐ |
||||
│ 0x08 │ code (1) │ |
||||
└──────┴──────────┘ |
||||
total: 2 bytes |
||||
``` |
||||
|
||||
Коды ошибок: |
||||
| #define | Value | Значение | |
||||
|---------|-------|----------| |
||||
| `DB_ERR_NOT_FOUND` | 0x01 | Инстанс с таким hash не найден | |
||||
| `DB_ERR_DISABLED` | 0x02 | Инстанс существует, но disabled | |
||||
|
||||
При получении ERROR сбрасывается `sync_state` пира в 0. |
||||
|
||||
### 3.10 DB_MSG_REQUEST_SYNC (0x09) |
||||
|
||||
**Отправитель:** слейв → мастер |
||||
**Назначение:** запросить инициацию синхронизации (слейв говорит мастеру «начни sync») |
||||
|
||||
``` |
||||
┌──────┐ |
||||
│ 0x09 │ |
||||
└──────┘ |
||||
total: 1 byte (без payload) |
||||
``` |
||||
|
||||
При получении REQUEST_SYNC мастер вызывает `initiate_sync()` → шлёт `INIT_SYNC`. |
||||
|
||||
--- |
||||
|
||||
## 4. Алгоритм синхронизации (по шагам) |
||||
|
||||
### 4.1 Запуск |
||||
|
||||
1. При conn UP для каждого enabled инстанса: `si_peer_add()` → `initiate_sync()` |
||||
2. `initiate_sync()`: |
||||
- Если `node_id > peer_id` (мастер): шлёт `INIT_SYNC` с `my_count` |
||||
- Если `node_id < peer_id` (слейв): шлёт `REQUEST_SYNC` |
||||
|
||||
### 4.2 Обработка INIT_SYNC (слейв) |
||||
|
||||
1. `db_sync_flush_recalc()` — синхронно завершает все отложенные пересчёты |
||||
2. Вычисляет `tp = min(peer_count, my_count) - 1` (точка пересечения) |
||||
3. Вычисляет `ch8_at_tp` — chain_hash8 на позиции tp |
||||
4. Определяет `has_tail = (my_count > tp + 1)` |
||||
5. Строит разреженные хеши: от tp назад с интервалами 1,1,1,2,2,4×8 |
||||
6. Отправляет `INIT_RESP` с `tp`, `ch8_at_tp`, `has_tail`, `sparse_hashes` |
||||
7. Устанавливает `sync_state = 1` для этого пира |
||||
|
||||
### 4.3 Обработка INIT_RESP (мастер) |
||||
|
||||
1. `db_sync_flush_recalc()` — синхронно завершает все отложенные пересчёты |
||||
2. Извлекает `tp`, `peer_ch8`, `has_tail`, `sc`, `sparse_hashes` |
||||
3. Вычисляет `my_ch8_at_tp` |
||||
|
||||
**Ветки:** |
||||
- **peer_ch8 == 0 && sc == 0** (пир пуст): шлёт **все** свои данные батчами по 32 записи. Последний батч с `want_from = my_count` (запрос встречных данных, которые будут пустыми → SYNC_DONE) |
||||
- **my_ch8 == peer_ch8** (совпадение на tp): |
||||
- `!has_tail && my_count <= tp+1`: обе стороны идентичны → сразу `SYNC_DONE` |
||||
- `has_tail`: отправляем свои записи после tp и запрашиваем записи пира (`want_from = tp+1`) |
||||
- **my_ch8 != peer_ch8** (расхождение): ищем первую различающуюся позицию `fm` через sparse-хеши. `vp = fm-1`. Отправляем данные с fm и запрашиваем данные пира с fm. |
||||
|
||||
### 4.4 Обработка SEND_DATA |
||||
|
||||
1. `db_sync_flush_recalc()` — синхронно завершает все отложенные пересчёты |
||||
2. Парсит `from`, `count`, `vp`, `want_from` |
||||
3. Для каждой записи (макс. 32): `si_parse_record()` → `db_record_insert()` |
||||
- Проверка Ed25519-подписи (если неверна — удаление записи) |
||||
- Проверка дубликата по `(timestamp, author_signature)` |
||||
- Вычисление chain_hash |
||||
- Вставка в SQLite, запуск async recalc |
||||
- Вызов `on_insert` коллбэка |
||||
4. Корректирует `synced_pos` (и соседних пиров, если запись вставлена не в конец) |
||||
5. Если `want_from != DB_WANT_FROM_NONE`: отправляет встречный SEND_DATA |
||||
6. Если `want_from == DB_WANT_FROM_NONE` и `sync_state == 1`: отправляет `SYNC_DONE` |
||||
|
||||
### 4.5 Обработка SYNC_DONE |
||||
|
||||
1. `db_sync_flush_recalc()` |
||||
2. Сравнивает `my_count`, `my_ch8_last` с `peer_count`, `peer_ch8_last` |
||||
3. **Совпало** → sync_state=2, fire `db_sync_done_cb` |
||||
4. **my < peer** → запрашиваем хвост SEND_DATA(from=my_count, want=my_count) |
||||
5. **my > peer** → отправляем хвост SEND_DATA(from=peer_count) |
||||
6. **count совпало но хеши разные** → принимаем как converged |
||||
|
||||
### 4.6 PUSH (реальное время) |
||||
|
||||
1. При локальной вставке (`db_sync_insert_signed`): |
||||
- Формирует PUSH-сообщение |
||||
- Рассылает всем пирам с `sync_state >= 1` |
||||
- Обновляет `delivery_chain` для каждого пира |
||||
2. При получении PUSH: |
||||
- Вставляет запись через `db_record_insert()` |
||||
- Шлёт `ACK_PUSH` обратно автору |
||||
- Если запись вставлена в середину (не в конец) — сбрасывает `synced_pos` всех пиров до позиции вставки-1 |
||||
3. При получении ACK_PUSH: |
||||
- Помечает запись флагом `WAS_SENT` |
||||
- Обновляет `delivery_chain` |
||||
|
||||
--- |
||||
|
||||
## 5. Безопасность |
||||
|
||||
1. **Ed25519-подписи:** каждая запись подписана `Ed25519(timestamp || json_data)`. Проверяется всегда — и при локальной вставке, и при приёме от пира. Невалидные записи отбрасываются и удаляются из БД. |
||||
2. **Цепной хеш:** SHA256-цепочка защищает от вставки/удаления записей в середине истории. При проверке (`db_sync_chain_verify`) или при обнаружении расхождения во время отправки — запускается пересчёт с позиции ошибки. |
||||
3. **Публичные ключи:** Ed25519 публичный ключ пира получается из трёх источников (по приоритету): |
||||
- Локальный `inst->my_ed25519_pubkey` (для своих записей) |
||||
- `topo_node_sqlite_get_ed25519_pubkey()` (из БД узлов) |
||||
- `conn->peer_ed25519_pubkey` (из ETCP handshake) |
||||
4. **Отсутствие удаления/редактирования:** записи не редактируются и не удаляются явно — только TTL-очистка неподтверждённых. |
||||
|
||||
--- |
||||
|
||||
## 6. Конфигурация |
||||
|
||||
```ini |
||||
[global] |
||||
db_sync_enabled = 1 # включить модуль (по умолчанию 0) |
||||
db_sync_ttl = 86400 # TTL неподтверждённых записей в секундах (по умолчанию 86400) |
||||
db_path = /var/lib/utun # путь к БД (по умолчанию /tmp/utun_db_sync) |
||||
``` |
||||
@ -0,0 +1,158 @@
|
||||
/**
|
||||
* @file test_conn_mgr_already_connected.c |
||||
* @brief conn_mgr: подключение к уже-соединённому узлу через instance_find_conn |
||||
* |
||||
* Проверяет, что conn_mgr_connect_node находит существующее ETCP-соединение |
||||
* и сразу вызывает callback с CONN_MGR_OK, а не уходит в фазы установки нового соединения. |
||||
*/ |
||||
#include <stdio.h> |
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
#include <stdarg.h> |
||||
#include "../lib/platform_compat.h" |
||||
#include "test_utils.h" |
||||
#ifndef _WIN32 |
||||
#include <unistd.h> |
||||
#endif |
||||
|
||||
#include "etcp.h" |
||||
#include "etcp_connections.h" |
||||
#include "../src/config_parser.h" |
||||
#include "../src/config_updater.h" |
||||
#include "../src/utun_instance.h" |
||||
#include "topo_group.h" |
||||
#include "conn_mgr.h" |
||||
#include "../lib/u_async.h" |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
|
||||
#define TOTAL_TIMEOUT_TB 10000 /* 1 sec */ |
||||
#define POLL_MS 5 |
||||
static struct UTUN_INSTANCE *g_a = NULL, *g_b = NULL; |
||||
static struct UASYNC *ua = NULL; |
||||
static volatile int result = 0, cb_fired = 0, cb_ok = 0; |
||||
static char tdir[] = "/tmp/utun_cm2_XXXXXX"; |
||||
static char ca[256], cb[256]; |
||||
static int pa = 0, pb = 0; |
||||
|
||||
static int wf(const char* p, const char* f, ...) { |
||||
va_list ap; FILE* fp = fopen(p, "w"); if (!fp) return -1; |
||||
va_start(ap, f); vfprintf(fp, f, ap); va_end(ap); fclose(fp); return 0; |
||||
} |
||||
static char* gv(const char* p, const char* k) { |
||||
struct utun_config* c = parse_config(p); if (!c) return NULL; |
||||
char* r = (strcmp(k, "pub") == 0) ? u_strdup(c->global.my_public_key_hex) : u_strdup(c->global.my_private_key_hex); |
||||
free_config(c); return r; |
||||
} |
||||
static void fail(const char* msg) { fprintf(stderr, "FAIL: %s\n", msg); fflush(stderr); result = 2; } |
||||
static void done(void) { if (result == 0) result = 1; } |
||||
static void test_mgr_connect(void* arg); |
||||
static void test_check(void* arg); |
||||
static void test3_preconnect(void* arg); |
||||
static void test3_check(void* arg); |
||||
|
||||
static void conn_cb(int r, uint64_t id, void* arg) { |
||||
(void)arg; |
||||
cb_fired = 1; cb_ok = (r == CONN_MGR_OK); |
||||
fprintf(stderr, "conn_cb: result=%d node=0x%llx ok=%d\n", r, (unsigned long long)id, cb_ok); fflush(stderr); |
||||
} |
||||
|
||||
static void test_find_conn(void* arg) { |
||||
(void)arg; |
||||
struct ETCP_CONN* c = instance_find_conn(g_a, g_b->node_id); |
||||
if (!c || !c->links) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test_find_conn, "t1w"); return; } |
||||
int has = 0; struct ETCP_LINK* l = c->links; |
||||
while (l) { if (l->link_state == 3 && l->initialized) { has = 1; break; } l = l->next; } |
||||
if (!has) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test_find_conn, "t1w2"); return; } |
||||
fprintf(stderr, "Test 1: instance_find_conn OK (link_state=%d)\n", l->link_state); fflush(stderr); |
||||
uasync_call_soon(ua, NULL, test_mgr_connect); |
||||
} |
||||
|
||||
static void test_mgr_connect(void* arg) { |
||||
(void)arg; |
||||
cb_fired = 0; cb_ok = 0; |
||||
struct CONN_MGR* mgr = g_a->conn_mgr; |
||||
struct TOPO_NODEQ* nq = topo_node_find_by_id(mgr->group, g_b->node_id); |
||||
if (nq) { fprintf(stderr, "node already in UTUN group\n"); fflush(stderr); } |
||||
fprintf(stderr, "Test 2: conn_mgr_connect_node(B)\n"); fflush(stderr); |
||||
conn_mgr_connect_node(mgr, g_b->node_id, 0, conn_cb, NULL); |
||||
uasync_call_soon(ua, NULL, test_check); |
||||
} |
||||
|
||||
static void test_check(void* arg) { |
||||
(void)arg; |
||||
if (!cb_fired) { fail("callback not fired"); done(); return; } |
||||
if (!cb_ok) { fail("callback result != CONN_MGR_OK"); done(); return; } |
||||
uint8_t st, ty; |
||||
conn_mgr_get_status(g_a->conn_mgr, g_b->node_id, &st, &ty); |
||||
if (st != CONN_MGR_STATE_CONNECTED) { fail("state != CONNECTED"); done(); return; } |
||||
fprintf(stderr, "Test 2: OK state=%d type=%d\n", st, ty); fflush(stderr); |
||||
fprintf(stderr, "=== ALL DONE ===\n"); fflush(stderr); |
||||
done(); |
||||
} |
||||
|
||||
static void test3_preconnect(void* arg) { |
||||
/* Test 3: conn_mgr_connect_node BEFORE ETCP link is up (no instance_find_conn match) */ |
||||
(void)arg; |
||||
struct CONN_MGR* mgr = g_a->conn_mgr; |
||||
cb_fired = 0; cb_ok = 0; |
||||
struct ETCP_CONN* c = instance_find_conn(g_a, g_b->node_id); |
||||
if (c) { fprintf(stderr, "Test3: SKIP — conn already established\n"); return; } |
||||
fprintf(stderr, "Test 3: conn_mgr_connect_node BEFORE link up\n"); fflush(stderr); |
||||
conn_mgr_connect_node(mgr, g_b->node_id, 0, conn_cb, NULL); |
||||
uasync_call_soon(ua, NULL, test3_check); |
||||
} |
||||
|
||||
static void test3_check(void* arg) { |
||||
(void)arg; |
||||
if (cb_fired) { |
||||
if (cb_ok) { fprintf(stderr, "Test 3: OK — callback CONN_MGR_OK\n"); fflush(stderr); return; } |
||||
fail("callback fired with error"); done(); return; |
||||
} |
||||
/* no callback yet — check if conn_mgr gave up (state reset) */ |
||||
uint8_t st, ty; |
||||
if (conn_mgr_get_status(g_a->conn_mgr, g_b->node_id, &st, &ty) == CONN_MGR_OK && st == CONN_MGR_STATE_DISCONNECTED) { |
||||
/* entry was cleaned up — conn not possible in this test setup with this timing */ |
||||
fprintf(stderr, "Test 3: SKIP — entry cleaned up (conn not yet established in time)\n"); fflush(stderr); |
||||
return; |
||||
} |
||||
uasync_set_timeout(ua, 30, NULL, (timeout_callback_t)test3_check, "t3w"); |
||||
} |
||||
|
||||
static void to_cb(void* arg) { (void)arg; fail("timeout"); done(); } |
||||
|
||||
int main(void) { |
||||
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); |
||||
utun_instance_set_tun_init_enabled(0); |
||||
|
||||
test_mkdtemp(tdir); |
||||
int base = 49000 + (getpid() % 10000); pa = base; pb = base + 1; |
||||
snprintf(ca, sizeof(ca), "%s/a.conf", tdir); snprintf(cb, sizeof(cb), "%s/b.conf", tdir); |
||||
wf(ca, "[global]\ntun_ip=10.97.0.1/24\ntun_ifname=tun95\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", pa); |
||||
wf(cb, "[global]\ntun_ip=10.97.0.2/24\ntun_ifname=tun94\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", pb); |
||||
config_ensure_keys_and_node_id(ca); config_ensure_keys_and_node_id(cb); |
||||
{ struct utun_config* cfa = parse_config(ca); struct utun_config* cfb = parse_config(cb); |
||||
free_config(cfa); free_config(cfb); } |
||||
char *r0 = gv(ca,"priv"), *p0 = gv(ca,"pub"), *p1 = gv(cb,"pub"); |
||||
wf(ca, "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.97.0.1/24\ntun_ifname=tun95\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[client: to_b]\nkeepalive=1\npeer_public_key=%s\nlink=s1:127.0.0.1:%d\n[allowed_keys]\nallow_all=1\n", r0, p0, pa, p1, pb); |
||||
u_free(r0); u_free(p0); u_free(p1); |
||||
|
||||
ua = uasync_create(); |
||||
g_a = utun_instance_create(ua, ca); g_b = utun_instance_create(ua, cb); |
||||
if (!g_a || !g_b) { fprintf(stderr,"create failed\n"); result=2; goto done; } |
||||
utun_instance_init(g_a); utun_instance_init(g_b); |
||||
|
||||
/* Test 3: call conn_mgr BEFORE link is up */ |
||||
uasync_call_soon(ua, NULL, test3_preconnect); |
||||
/* Test 1+2: run after link is established */ |
||||
uasync_call_soon(ua, NULL, test_find_conn); |
||||
uasync_set_timeout(ua, TOTAL_TIMEOUT_TB, NULL, to_cb, "to"); |
||||
{ int el = 0; while (!result && el < TOTAL_TIMEOUT_TB / 10 + 500) { uasync_poll(ua, POLL_MS); el += POLL_MS; } } |
||||
|
||||
done: |
||||
if (g_a) { g_a->running = 0; utun_instance_destroy(g_a); } |
||||
if (g_b) { g_b->running = 0; utun_instance_destroy(g_b); } |
||||
if (ua) uasync_destroy(ua, 0); |
||||
test_unlink(ca); test_unlink(cb); test_rmdir(tdir); |
||||
return (result == 1) ? 0 : 1; |
||||
} |
||||
Loading…
Reference in new issue