14 KiB

Raw Blame History

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

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. Схема БД (без изменений)

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`

static void db_cascade_from(struct DB_SYNC_INSTANCE* si, uint32_t from_pos);

Выполняет каскадный пересчёт chain_hash для записей с позиции from_pos до конца таблицы.

Алгоритм:

BEGIN IMMEDIATE
SELECT chain_hash записи на позиции from_pos - 1 (или zero если from_pos = 0) как prev_ch
SELECT id, timestamp, datahash от позиции from_pos до конца
Для каждой: chain_hash = SHA256(prev_ch || id || ts || dh), UPDATE в БД, prev_ch = новый chain_hash
COMMIT

Выделяется из текущего кода db_record_insert (строки 412-451) в отдельную функцию.

5.2. `db_record_insert` — новый параметр

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` (стартап-проверка)

static void db_verify_chain(struct DB_SYNC_INSTANCE* si);

Вызывается один раз после db_sync_instance_add (перед инициацией sync с пирами).

Алгоритм:

Проход записей 0..N-1 в порядке ORDER BY timestamp, datahash
Для каждой: chain_hash = SHA256(prev_ch || id || ts || dh)
Сравнить с хранимым в БД
При первом расхождении на позиции P: пересчитать цепочку от P до конца через db_cascade_from(si, P). Завершить.

Сложность: O(N) SHA256, однократно при старте.

5.4. `si_find_pos` (новая)

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. Порядок реализации

Выделить db_cascade_from(si, from_pos) из тела db_record_insert.
Добавить параметр do_cascade в db_record_insert. При 0 — только INSERT, cascade вызывается отдельно.
Добавить synced_pos в struct SI_PEER. Инициализировать в 0.
Добавить si_find_pos(si, ts, dh) — поиск позиции записи в глобальном порядке.
Изменить wire-формат INIT_SYNC, INIT_RESP, SYNC_DONE (chain_hash → datahash, 32→8 байт).
Изменить обработчики:
- 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
Добавить db_verify_chain(si) — вызвать при старте после db_sync_instance_add.
Многопировая синхронизация: в db_sync_peer_check_cb выбирать пира с min synced_pos вместо первого попавшегося.

14 KiB Raw Blame History