- u_async.c: process immediate_queue BEFORE epoll_wait in uasync_poll - stcp_link.c: defensive NULL etcp_conn/etcp_link in stcp_link_close - utun_instance.c: move stcp_server_list_destroy_all to Phase J (before ETCP), add deferred drain in both Phase J and L - stcp.h/c: STCP_HS_ENC_CLIENT 6→38, STCP_HS_ENC_SERVER 7→39, exchange ed25519_pubkey in both handshake directions - etcp_connections.c: copy peer_ed25519 from STCP to ETCP_CONN in etcp_link_enter_ready_tcp; guard link->etcp in tcp_link_close_cb - chat_sync.c: fix lk->conn NULL dereference for TCP links - test_stcp.c: update stcp_server_create/stcp_client_connect callsites - tools/chat_tcp_test: change transport udp→tcp
293 lines
14 KiB
Markdown
293 lines
14 KiB
Markdown
# DB Sync Protocol v2 — Техническое задание
|
||
|
||
## 1. Проблема
|
||
|
||
chain_hash = SHA256(prev_chain_hash || id || ts || datahash) — хеш-цепочка, где хеш каждой записи зависит от предыдущей.
|
||
|
||
`db_record_insert` (db_sync.c:341-463) при вставке записи в середину сортированного порядка делает каскадный пересчёт chain_hash всех последующих записей — O(N). При двустороннем обмене PUSH'ами получается O(N²).
|
||
|
||
## 2. Решение
|
||
|
||
**Разделить два концерна:**
|
||
- **Сравнение при синхронизации** — перейти на `datahash` (первые 8 байт SHA256(data), позиционно-независимый). Не требует каскада.
|
||
- **Целостность цепочки** — `chain_hash` остаётся в схеме БД, но используется только для однократной стартап-проверки и диагностики.
|
||
|
||
**Синхронизация:** per-peer позиционная (`synced_pos`). При PUSH каскад не делается. При SEND_DATA батч-вставка + один каскад после батча.
|
||
|
||
## 3. Модель данных
|
||
|
||
### 3.1. SI_PEER
|
||
|
||
```c
|
||
struct SI_PEER {
|
||
uint64_t node_id;
|
||
uint32_t synced_pos; // последняя подтверждённо общая позиция (0-based index)
|
||
uint8_t sync_state; // 0=not_synced, 1=syncing, 2=synced
|
||
};
|
||
```
|
||
|
||
- `synced_pos` = индекс последней записи в глобальном порядке `ORDER BY timestamp, datahash`, для которой chain_hash (а после v2 — datahash) подтверждённо совпадает у обоих пиров.
|
||
- При PUSH-вставке записи на позицию P, для всех peer'ов у которых `synced_pos >= P`: `synced_pos = P - 1`.
|
||
- При SEND_DATA: `synced_pos = from + received_count - 1`.
|
||
|
||
### 3.2. Схема БД (без изменений)
|
||
|
||
```sql
|
||
CREATE TABLE "db_sync_<name>_<id>" (
|
||
timestamp INTEGER NOT NULL,
|
||
datahash INTEGER NOT NULL,
|
||
id INTEGER NOT NULL,
|
||
chain_hash BLOB NOT NULL,
|
||
author INTEGER NOT NULL,
|
||
flags INTEGER NOT NULL DEFAULT 0,
|
||
data BLOB,
|
||
author_signature BLOB,
|
||
delivered_peers INTEGER NOT NULL DEFAULT 0,
|
||
delivery_chain TEXT NOT NULL DEFAULT '',
|
||
PRIMARY KEY (timestamp, datahash)
|
||
);
|
||
```
|
||
|
||
## 4. Wire-формат
|
||
|
||
### 4.1. Сообщения, которые меняются
|
||
|
||
#### INIT_SYNC
|
||
|
||
```
|
||
Было: [type:1][count:4][last_chain_hash:32] = 37 байт
|
||
Стало: [type:1][count:4] = 5 байт
|
||
```
|
||
|
||
#### INIT_RESP
|
||
|
||
```
|
||
Было: [type:1][tp:4][chain_at_tp:32][sc:1][(pos:4,chain_hash:32)*sc]
|
||
Стало: [type:1][tp:4][dh_at_tp:8] [sc:1][(pos:4,dh:8)*sc]
|
||
|
||
dh_at_tp — datahash на позиции tp (8 байт вместо 32)
|
||
sparse — datahash вместо chain_hash (8 байт на элемент вместо 32)
|
||
|
||
При sc=0 (exact match): длина 1+4+8+1 = 14 байт (было 37)
|
||
При sc=16 sparse: длина 14 + 16*12 = 206 байт (было 37 + 16*36 = 613)
|
||
```
|
||
|
||
#### SYNC_DONE
|
||
|
||
```
|
||
Было: [type:1][count:4][last_chain_hash:32] = 37 байт
|
||
Стало: [type:1][count:4][last_dh:8] = 13 байт
|
||
```
|
||
|
||
### 4.2. Сообщения, которые НЕ меняются
|
||
|
||
- **PUSH**: `[type:1][id:8][ts:8][dh:8][dlen:4][data][sig_len:1][sig]` — без изменений. `prev_ch` НЕ добавляется.
|
||
- **ACK_PUSH**: `[type:1][dh:8][ts:8]` — без изменений.
|
||
- **REFINE**: `[type:1][from:4][to:4][hc:1][(pos:4,dh:8)*hc]` — уже использует datahash, без изменений.
|
||
- **SEND_DATA**: `[type:1][from:4][count:2][(id:8,ts:8,dh:8,dlen:4,data,sig_len:1,sig)*count]` — без изменений.
|
||
- **ERROR**: `[type:1][code:1]` — без изменений.
|
||
|
||
## 5. Новые/изменённые функции
|
||
|
||
### 5.1. `db_cascade_from`
|
||
|
||
```c
|
||
static void db_cascade_from(struct DB_SYNC_INSTANCE* si, uint32_t from_pos);
|
||
```
|
||
|
||
Выполняет каскадный пересчёт `chain_hash` для записей с позиции `from_pos` до конца таблицы.
|
||
|
||
Алгоритм:
|
||
1. BEGIN IMMEDIATE
|
||
2. SELECT chain_hash записи на позиции `from_pos - 1` (или zero если from_pos = 0) как prev_ch
|
||
3. SELECT id, timestamp, datahash от позиции from_pos до конца
|
||
4. Для каждой: chain_hash = SHA256(prev_ch || id || ts || dh), UPDATE в БД, prev_ch = новый chain_hash
|
||
5. COMMIT
|
||
|
||
Выделяется из текущего кода `db_record_insert` (строки 412-451) в отдельную функцию.
|
||
|
||
### 5.2. `db_record_insert` — новый параметр
|
||
|
||
```c
|
||
static int db_record_insert(struct DB_SYNC_INSTANCE* si,
|
||
uint64_t id, uint64_t ts, uint64_t dh,
|
||
const char* json, size_t jlen,
|
||
const uint8_t* sig, size_t sig_len,
|
||
int do_cascade);
|
||
```
|
||
|
||
- `do_cascade=1` — после INSERT выполняется cascade (строки 412-451). Используется для локальных вставок.
|
||
- `do_cascade=0` — без cascade. Используется для PUSH и SEND_DATA (каскад делается отдельно).
|
||
|
||
### 5.3. `db_verify_chain` (стартап-проверка)
|
||
|
||
```c
|
||
static void db_verify_chain(struct DB_SYNC_INSTANCE* si);
|
||
```
|
||
|
||
Вызывается один раз после `db_sync_instance_add` (перед инициацией sync с пирами).
|
||
|
||
Алгоритм:
|
||
1. Проход записей 0..N-1 в порядке `ORDER BY timestamp, datahash`
|
||
2. Для каждой: chain_hash = SHA256(prev_ch || id || ts || dh)
|
||
3. Сравнить с хранимым в БД
|
||
4. При первом расхождении на позиции P: пересчитать цепочку от P до конца через `db_cascade_from(si, P)`. Завершить.
|
||
|
||
Сложность: O(N) SHA256, однократно при старте.
|
||
|
||
### 5.4. `si_find_pos` (новая)
|
||
|
||
```c
|
||
static uint32_t si_find_pos(struct DB_SYNC_INSTANCE* si,
|
||
uint64_t ts, uint64_t dh);
|
||
```
|
||
|
||
Возвращает позицию (0-based index) записи с ключом `(ts, dh)` в глобальном порядке. Используется после вставки PUSH для корректировки `synced_pos`.
|
||
|
||
### 5.5. `db_datahash_at` — уже существует (стр. 295)
|
||
|
||
Возвращает `datahash` на заданной позиции. Используется для сравнения в INIT_SYNC/INIT_RESP/SYNC_DONE.
|
||
|
||
## 6. Обработчики сообщений
|
||
|
||
### 6.1. PUSH (`db_handle_push`)
|
||
|
||
```
|
||
Было: db_record_insert(si, ..., /* cascade встроен */)
|
||
Стало: db_record_insert(si, ..., /* do_cascade= */ 0)
|
||
uint32_t pos = si_find_pos(si, ts, dh);
|
||
for each peer: if peer->synced_pos >= pos: peer->synced_pos = pos - 1;
|
||
```
|
||
|
||
### 6.2. INIT_SYNC (`db_handle_init_sync`)
|
||
|
||
```
|
||
Изменения:
|
||
- Принимает [count:4] вместо [count:4][chain_hash:32] (проверка len >= 5 вместо 36)
|
||
- Сравнение: datahash вместо chain_hash (8 байт вместо 32)
|
||
- sparse: datahash вместо chain_hash (8 байт на элемент вместо 32)
|
||
- Размер sparse-элемента: 4(pos) + 8(dh) = 12 байт (было 4+32=36)
|
||
- resp буфер: 4096 → достаточно (14 + 16*12 = 206 байт при max sparse)
|
||
```
|
||
|
||
### 6.3. INIT_RESP (`db_handle_init_resp`)
|
||
|
||
```
|
||
Изменения:
|
||
- Принимает [tp:4][dh:8][sc:1][...] вместо [tp:4][ch:32][sc:1][...]
|
||
- Проверка len >= 13 вместо 37
|
||
- Сравнение: datahash вместо chain_hash
|
||
- Пустая БД пира: zero-хеш 8 байт вместо 32
|
||
- Sparse элементы: 12 байт вместо 36
|
||
- При совпадении: synced_pos = tp (в дополнение к sync_state = 2)
|
||
```
|
||
|
||
### 6.4. SEND_DATA (`db_handle_send_data`)
|
||
|
||
```
|
||
Стало:
|
||
struct SI_PEER* sp = si_peer_find(si, src);
|
||
uint32_t fix_from = sp ? sp->synced_pos : 0;
|
||
|
||
for each record:
|
||
db_record_insert(si, ..., /* do_cascade= */ 0);
|
||
|
||
db_cascade_from(si, fix_from);
|
||
|
||
sp->synced_pos = from + received_count - 1;
|
||
sp->sync_state = 2;
|
||
```
|
||
|
||
Каскад от `fix_from` (позиция, которая была synced ДО этого батча) гарантирует, что все chain_hash после этой точки пересчитаны — независимо от того, какие PUSH'и испортили их между синхронизациями.
|
||
|
||
### 6.5. SYNC_DONE (`db_handle_sync_done`)
|
||
|
||
```
|
||
Изменения:
|
||
- Принимает [count:4][last_dh:8] вместо [count:4][last_chain_hash:32] (len >= 13 вместо 36)
|
||
- Сравнение последнего datahash вместо chain_hash
|
||
```
|
||
|
||
## 7. Механика многопировой синхронизации
|
||
|
||
При наличии нескольких пиров с одинаковым `sync_state` (например, syncing=1) выбирается пир с минимальным `synced_pos` — у него самый старый общий префикс. После завершения его синхронизации выбор повторяется.
|
||
|
||
Это гарантирует: двигаемся от самого старого несинхронизированного участка, не прыгая. Новые записи (в хвосте) синхронизируются последними.
|
||
|
||
## 8. Полный сценарий: три участника A, B, C
|
||
|
||
```
|
||
Начальное состояние: все пусты.
|
||
|
||
=== Шаг 1. A создаёт R1 (ts=1000) ===
|
||
A: PUSH R1 → B, C. Все вставляют в конец, cascade 0.
|
||
synced_pos = 0 у всех.
|
||
|
||
=== Шаг 2. B создаёт R2 (ts=2000), C создаёт R3 (ts=1500) ===
|
||
B: [R1(1000), R2(2000)], PUSH R2 → A, C.
|
||
A: R2 в конец. C: R2 в конец.
|
||
C: [R1(1000), R2(2000)]? Нет — C создал R3(1500).
|
||
C: [R1(1000), R3(1500)] локально, потом PUSH R2 → R2.ts=2000 → в конец.
|
||
C: [R1(1000), R3(1500), R2(2000)]
|
||
C: PUSH R3 → A, B.
|
||
A: [R1(1000), R2(2000)] + R3(1500) → между R1 и R2:
|
||
A: [R1, R3, R2] ← R3 в середину, cascade ПРОПУЩЕН
|
||
B: [R1, R3, R2] ← аналогично
|
||
|
||
synced_pos после PUSH R3 на A относительно B:
|
||
A вставил R3 на поз.1 → synced_pos_B: было 1 → стало 0
|
||
(A считает, что совпадает с B только на позиции 0)
|
||
|
||
=== Шаг 3. A инициирует sync с B (чемпион: min synced_pos) ===
|
||
A→B: INIT_SYNC [count=3]
|
||
B→A: INIT_RESP [tp=min(3,3)-1=2, dh_at_tp=H2, sc=2, sparse:{(1,H3),(0,H1)}]
|
||
A проверяет: A@2=R2→dh=H2✓, A@1=R3→dh=H3✓, A@0=R1→dh=H1✓
|
||
→ sync complete, synced_pos_B=2, sync_state=2
|
||
|
||
=== Шаг 4. Рестарт A ===
|
||
db_verify_chain:
|
||
pos 0: R1.ch = SHA256(0||R1) ✓
|
||
pos 1: R3.ch = SHA256(R1.ch||R3) ✓ (посчитан при PUSH без cascade — корректен)
|
||
pos 2: R2.ch = SHA256(R3.ch||R2) ✗ (STALE: был посчитан как SHA256(R1.ch||R2) до PUSH R3)
|
||
→ расхождение на pos=2 → db_cascade_from(2): пересчёт R2.ch = SHA256(R3.ch||R2) ✓
|
||
|
||
=== Шаг 5. A, B, C активно обмениваются ===
|
||
Каждый PUSH: вставка без cascade.
|
||
Периодические sync (по min synced_pos): SEND_DATA + один cascade от synced_pos.
|
||
synced_pos растёт.
|
||
```
|
||
|
||
## 9. Что НЕ трогать
|
||
|
||
- `db_sync_insert_signed` — локальная вставка. ts всегда монотонный, запись в конец, `do_cascade=1`, каскад всегда 0 строк.
|
||
- TTL cleanup (`db_sync_instance_ttl_cb`) — удаление старых записей. Нужен cascade после удаления (уже существующая логика, не меняется).
|
||
- ACK_PUSH, ERROR, REFINE — без изменений.
|
||
- Схема БД — без изменений.
|
||
- `etcp_bind`/эпилог — без изменений.
|
||
|
||
## 10. Сложность операций (итого)
|
||
|
||
| Операция | Каскад | Сложность |
|
||
|----------|--------|-----------|
|
||
| Локальная вставка | 0 строк (конец) | O(1) |
|
||
| PUSH (приём) | нет | O(1) |
|
||
| SEND_DATA (батч N записей) | 1 каскад после батча | O(N + tail) |
|
||
| Стартап-проверка | 1 раз | O(total) |
|
||
| Sync-протокол (INIT_SYNC/INIT_RESP/REFINE) | нет cascade | O(log N) сообщений |
|
||
|
||
## 11. Порядок реализации
|
||
|
||
1. **Выделить `db_cascade_from(si, from_pos)`** из тела `db_record_insert`.
|
||
2. **Добавить параметр `do_cascade`** в `db_record_insert`. При `0` — только INSERT, cascade вызывается отдельно.
|
||
3. **Добавить `synced_pos`** в `struct SI_PEER`. Инициализировать в 0.
|
||
4. **Добавить `si_find_pos(si, ts, dh)`** — поиск позиции записи в глобальном порядке.
|
||
5. **Изменить wire-формат** INIT_SYNC, INIT_RESP, SYNC_DONE (chain_hash → datahash, 32→8 байт).
|
||
6. **Изменить обработчики**:
|
||
- `db_handle_init_sync`: новые размеры, datahash вместо chain_hash
|
||
- `db_handle_init_resp`: новые размеры, datahash вместо chain_hash, обновлять synced_pos
|
||
- `db_handle_send_data`: `do_cascade=0`, `db_cascade_from(fix_from)`, обновить synced_pos
|
||
- `db_handle_push`: `do_cascade=0`, скорректировать все synced_pos
|
||
- `db_handle_sync_done`: обновить synced_pos
|
||
7. **Добавить `db_verify_chain(si)`** — вызвать при старте после `db_sync_instance_add`.
|
||
8. **Многопировая синхронизация**: в `db_sync_peer_check_cb` выбирать пира с min synced_pos вместо первого попавшегося.
|