jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Stage each member exchange separately and validate versions before atomic publication

Browse Source
This commit is contained in:
evgeny committed 2026-11-04 00:42:39 +01:00
1 parent 592b6e1b75
commit 81bba75d02
2 files changed
+314
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/member_sync.c
+313
View File
@@ -1198,6 +1198,319 @@ static int member_decode(const uint8_t* p, size_t len, struct member_wire* w, si
return 0;
}
/* Поля опубликованной записи, включая локальные настройки: бэкап сохраняется без потерь. */
#define MEMBER_SAVED_FIELDS "node_id,x25519_pubkey,ed25519_pubkey,join_sig,join_ts,update_sig,update_ts,userinfo," \
"creator_sig,signed_by,signature,adm_tags,adm_tags_sig,node_type,node_RTT,connected,storage,deleted,local_nick,radio_muted,admitted,source"
struct member_pull {
char channel[21];
sqlite3_int64 id; /* ID служебной строки node_id=0; уникален среди всех одновременных приёмов канала. */
uint64_t peer;
uint64_t* changed;
size_t count;
struct group_policy_update* admission;
int committed, cleaned;
};
struct member_saved {
struct ms_member_rec rec;
uint8_t x[32], ed[32], join[64], update[64], owner[64], tree[64];
char info[MS_USERINFO_MAX + 1], tags[MS_ADM_TAGS_MAX + 1];
int source, malformed;
};
/* Снимок строки не держит SQLite pointers через изменения таблицы и callbacks. */
static int member_read_saved(sqlite3* db, const char* ns, uint64_t node, int state, sqlite3_int64 sync, struct member_saved* out) {
char sql[512]; sqlite3_stmt* st = NULL; memset(out, 0, sizeof(*out));
snprintf(sql, sizeof(sql), "SELECT x25519_pubkey,ed25519_pubkey,join_sig,join_ts,update_sig,update_ts,userinfo,"
"adm_tags,adm_tags_sig,signed_by,signature,source FROM peers_%s WHERE node_id=? AND record_state=? AND sync_id=?", ns);
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node);
if (rc == SQLITE_OK) rc = sqlite3_bind_int(st, 2, state);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, sync);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int found = rc == SQLITE_ROW;
if (found) {
const int cols[] = {0, 1, 2, 4, 8, 10}, sizes[] = {32, 32, 64, 64, 64, 64};
uint8_t* blobs[] = {out->x, out->ed, out->join, out->update, out->owner, out->tree};
for (size_t i = 0; i < 6; i++) {
int bytes = sqlite3_column_bytes(st, cols[i]);
if (bytes == sizes[i]) memcpy(blobs[i], sqlite3_column_blob(st, cols[i]), (size_t)bytes);
else if (bytes || i < 2) out->malformed = 1;
}
for (int i = 0; i < 2; i++) {
int col = i ? 7 : 6, bytes = sqlite3_column_bytes(st, col), max = i ? MS_ADM_TAGS_MAX : MS_USERINFO_MAX;
const char* text = (const char*)sqlite3_column_text(st, col);
if (bytes > max || (text && strlen(text) != (size_t)bytes)) out->malformed = 1;
else if (text) memcpy(i ? out->tags : out->info, text, (size_t)bytes + 1);
}
out->rec = (struct ms_member_rec){.node_id=node,.x25519=out->x,.ed25519=out->ed,.join_sig=out->join,
.join_ts=(uint64_t)sqlite3_column_int64(st,3),.update_sig=out->update,.update_ts=(uint64_t)sqlite3_column_int64(st,5),
.userinfo=out->info,.adm_tags=out->tags,.adm_tags_sig=out->owner,
.signed_by=(uint64_t)sqlite3_column_int64(st,9),.signature=out->tree};
out->source = sqlite3_column_int(st, 11);
}
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member version read ch=%s node=%016llx state=%d rc=%d: %s",
ns, (unsigned long long)node, state, rc, sqlite3_errmsg(db)); return -1;
}
return found;
}
/* Удаляет только строки своего приёма; завершение/отмена соседнего обмена ему не мешает. */
static int member_pull_cleanup(struct UTUN_INSTANCE* inst, struct member_pull* pull) {
if (pull->cleaned) return 0;
char sql[192];
snprintf(sql, sizeof(sql), "DELETE FROM peers_%s WHERE sync_id=%lld AND record_state IN (1,2)",
pull->channel, (long long)pull->id);
if (sqlite3_exec(_db(inst), sql, NULL, NULL, NULL) != SQLITE_OK) {
if (topo_node_sqlite_channel_get(_db(inst), pull->channel, NULL, 0, NULL, NULL, NULL, NULL) == TOPO_SQLITE_LOOKUP_MISSING) {
pull->cleaned = 1; return 0;
}
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive cleanup ch=%s sync=%lld: %s",
pull->channel, (long long)pull->id, sqlite3_errmsg(_db(inst))); return -1;
}
pull->cleaned = 1; return 0;
}
static void member_abort_pull(void* ctx, void* state) {
struct member_pull* pull = state;
member_pull_cleanup(ctx, pull);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "member receive released ch=%s sync=%lld committed=%d",
pull->channel, (long long)pull->id, pull->committed);
group_policy_discard(pull->admission); u_free(pull->changed); u_free(pull);
}
/* ID выдаёт AUTOINCREMENT самой таблицы: перезапуск и параллельные пиры не переиспользуют номер приёма. */
static int member_begin_pull(void* ctx, const char* ns, uint64_t peer, uint64_t round, void** out) {
struct UTUN_INSTANCE* inst = ctx; sqlite3* db = _db(inst); char sql[512];
*out = NULL;
if (!db || !chat_channel_id(ns) || strlen(ns) > 20 || !sqlite3_get_autocommit(db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive begin: invalid namespace/transaction"); return MT_ERR_IO;
}
struct member_pull* pull = u_calloc(1, sizeof(*pull));
if (!pull) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive allocation ch=%s", ns); return MT_ERR_IO; }
strcpy(pull->channel, ns); pull->peer = peer;
if (sqlite3_exec(db, "SAVEPOINT member_receive_begin", NULL, NULL, NULL) != SQLITE_OK) goto error;
snprintf(sql, sizeof(sql), "INSERT INTO peers_%s(node_id,record_state,x25519_pubkey,ed25519_pubkey,join_sig,join_ts)"
" VALUES(0,1,zeroblob(32),zeroblob(32),zeroblob(64),0)", ns);
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) goto rollback;
pull->id = sqlite3_last_insert_rowid(db);
snprintf(sql, sizeof(sql), "UPDATE peers_%s SET sync_id=id WHERE id=%lld", ns, (long long)pull->id);
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK || sqlite3_changes(db) != 1 ||
sqlite3_exec(db, "RELEASE member_receive_begin", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "member receive begin ch=%s peer=%016llx sync=%lld round=%llu",
ns, (unsigned long long)peer, (long long)pull->id, (unsigned long long)round);
*out = pull; return 0;
rollback:
if (sqlite3_exec(db, "ROLLBACK TO member_receive_begin; RELEASE member_receive_begin", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive begin rollback ch=%s: %s", ns, sqlite3_errmsg(db));
error:
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive begin ch=%s: %s", ns, sqlite3_errmsg(db)); u_free(pull); return MT_ERR_IO;
}
/* Проверяет рамку и курсор; подписи и права проверяются по полному набору после приёма. */
static int member_stage_page(void* ctx, void* state, uint64_t prefix, int after_valid, const merkle_key_t* after,
const merkle_key_t* next, int more, const uint8_t* data, size_t len) {
struct member_pull* pull = state; sqlite3* db = _db(ctx);
if (!data || len < 2 + GROUP_POLICY_WIRE_SIZE) goto invalid;
unsigned count = ((unsigned)data[0] << 8) | data[1]; size_t off = 2 + GROUP_POLICY_WIRE_SIZE;
if (count > (len - off) / 285) goto invalid;
merkle_key_t previous = after_valid ? *after : (merkle_key_t){{0}}, key = {{0}};
for (unsigned i = 0; i < count; i++) {
struct member_wire record; size_t used;
if (member_decode(data + off, len - off, &record, &used) < 0 || !record.rec.node_id) goto invalid;
member_write64(key.bytes, record.rec.node_id);
if (merkle_sync_level_prefix(record.rec.node_id, MT_MAX_LEVEL) != prefix ||
((i || after_valid) && memcmp(key.bytes, previous.bytes, MT_KEY_SIZE) <= 0)) goto invalid;
previous = key; off += used;
}
if (off != len || (count && memcmp(next->bytes, key.bytes, MT_KEY_SIZE)) || (!count && more)) goto invalid;
char sql[768]; sqlite3_stmt* st = NULL; int rc;
if (sqlite3_exec(db, "SAVEPOINT member_receive_page", NULL, NULL, NULL) != SQLITE_OK) goto error;
snprintf(sql, sizeof(sql), "UPDATE peers_%s SET sync_policy=? WHERE id=%lld AND record_state=1", pull->channel, (long long)pull->id);
rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, data + 2, GROUP_POLICY_WIRE_SIZE, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int final = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_DONE || final != SQLITE_OK || sqlite3_changes(db) != 1) goto rollback;
snprintf(sql, sizeof(sql), "INSERT INTO peers_%s(node_id,record_state,sync_id,previous_id,sync_record,"
"x25519_pubkey,ed25519_pubkey,join_sig,join_ts) VALUES(?1,1,%lld,"
"(SELECT id FROM peers_%s WHERE node_id=?1 AND record_state=0),?2,zeroblob(32),zeroblob(32),zeroblob(64),0)",
pull->channel, (long long)pull->id, pull->channel);
rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL); off = 2 + GROUP_POLICY_WIRE_SIZE;
for (unsigned i = 0; rc == SQLITE_OK && i < count; i++) {
struct member_wire record; size_t used;
if (member_decode(data + off, len - off, &record, &used) < 0) { rc = SQLITE_CORRUPT; break; }
rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)record.rec.node_id);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, data + off, (int)used, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_DONE && sqlite3_changes(db) == 1) rc = sqlite3_reset(st);
off += used;
}
final = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_OK || final != SQLITE_OK || sqlite3_exec(db, "RELEASE member_receive_page", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "member receive staged ch=%s sync=%lld records=%u bytes=%zu",
pull->channel, (long long)pull->id, count, len); return 0;
rollback:
sqlite3_finalize(st);
if (sqlite3_exec(db, "ROLLBACK TO member_receive_page; RELEASE member_receive_page", NULL, NULL, NULL) != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive page rollback ch=%s: %s", pull->channel, sqlite3_errmsg(db));
error:
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive page ch=%s sync=%lld: %s", pull->channel, (long long)pull->id, sqlite3_errmsg(db));
return MT_ERR_IO;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "member receive malformed page ch=%s sync=%lld prefix=%016llx bytes=%zu",
pull->channel, (long long)pull->id, (unsigned long long)prefix, len); return MT_ERR_DATA;
}
/* Список ID копируется до изменения строк; NULL при пустом результате. */
static int member_receive_ids(sqlite3* db, const char* sql, uint64_t** out, size_t* count) {
sqlite3_stmt* st = NULL; uint64_t* ids = NULL; size_t n = 0, capacity = 0;
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
if (n == capacity) {
size_t next = capacity ? capacity * 2 : 32;
if (next > SIZE_MAX / sizeof(*ids)) { rc = SQLITE_TOOBIG; break; }
uint64_t* resized = u_realloc(ids, next * sizeof(*ids));
if (!resized) { rc = SQLITE_NOMEM; break; } ids = resized; capacity = next;
}
ids[n++] = (uint64_t)sqlite3_column_int64(st, 0);
}
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive ID snapshot failed rc=%d: %s", rc, sqlite3_errmsg(db)); u_free(ids); return -1;
}
*out = ids; *count = n; return 0;
}
/* Сравнивает с текущей опубликованной версией, а не с previous_id, полученным в начале обмена. */
static int member_commit_pull(void* ctx, void* state) {
struct UTUN_INSTANCE* inst = ctx; struct member_pull* pull = state; sqlite3* db = _db(inst);
char sql[2048]; sqlite3_stmt* st = NULL; const char* stage = "begin";
uint64_t* incoming = NULL; size_t count = 0; int result = MT_ERR_IO, rc;
if (!sqlite3_get_autocommit(db) || sqlite3_exec(db, "BEGIN IMMEDIATE /* member receive */", NULL, NULL, NULL) != SQLITE_OK) goto fail;
stage = "backup latest";
snprintf(sql, sizeof(sql), "INSERT INTO peers_%s(" MEMBER_SAVED_FIELDS ",record_state,sync_id,previous_id)"
" SELECT " MEMBER_SAVED_FIELDS ",2,%lld,id FROM peers_%s WHERE record_state=0", pull->channel, (long long)pull->id, pull->channel);
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) goto fail;
stage = "policy";
snprintf(sql, sizeof(sql), "SELECT sync_policy FROM peers_%s WHERE id=%lld AND record_state=1", pull->channel, (long long)pull->id);
rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_ROW) goto fail;
if (sqlite3_column_bytes(st, 0)) {
struct group_policy policy;
if (sqlite3_column_bytes(st, 0) != GROUP_POLICY_WIRE_SIZE || group_policy_decode(db, pull->channel, sqlite3_column_blob(st, 0), &policy)) {
result = MT_ERR_DATA; goto fail;
}
if (group_policy_apply(db, pull->channel, &policy) < 0) goto fail;
}
rc = sqlite3_finalize(st); st = NULL; if (rc != SQLITE_OK) goto fail;
stage = "incoming list";
snprintf(sql, sizeof(sql), "SELECT node_id FROM peers_%s WHERE record_state=1 AND sync_id=%lld AND node_id<>0 ORDER BY node_id<0,node_id",
pull->channel, (long long)pull->id);
if (member_receive_ids(db, sql, &incoming, &count)) goto fail;
for (size_t i = 0; i < count; i++) {
stage = "merge incoming";
uint8_t bytes[MT_PAGE_SIZE]; size_t length = 0;
snprintf(sql, sizeof(sql), "SELECT sync_record FROM peers_%s WHERE node_id=%lld AND record_state=1 AND sync_id=%lld",
pull->channel, (long long)incoming[i], (long long)pull->id);
rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) <= 0 || sqlite3_column_bytes(st, 0) > MT_PAGE_SIZE) goto fail;
length = (size_t)sqlite3_column_bytes(st, 0); memcpy(bytes, sqlite3_column_blob(st, 0), length);
rc = sqlite3_finalize(st); st = NULL; if (rc != SQLITE_OK) goto fail;
struct member_wire record; size_t used;
if (member_decode(bytes, length, &record, &used) || used != length) { result = MT_ERR_DATA; goto fail; }
if (!record.rec.join_sig && !topo_node_sqlite_member_get_join(db, pull->channel, record.rec.node_id, record.local_join, &record.rec.join_ts))
record.rec.join_sig = record.local_join;
rc = member_apply_record(inst, pull->channel, pull->peer, &record.rec, 1);
if (rc < 0 && rc != MT_ERR_DATA) goto fail;
if (rc == MT_ERR_DATA) DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "member receive rejected candidate ch=%s sync=%lld node=%016llx",
pull->channel, (long long)pull->id, (unsigned long long)incoming[i]);
}
stage = "verify and fallback";
snprintf(sql, sizeof(sql), "SELECT node_id FROM peers_%s WHERE record_state=0 ORDER BY node_id<0,node_id", pull->channel);
if (member_receive_ids(db, sql, &pull->changed, &pull->count)) goto fail;
for (size_t i = 0; i < pull->count; i++) {
struct member_saved current, backup;
uint64_t node = pull->changed[i];
if (member_read_saved(db, pull->channel, node, 0, 0, &current) != 1) goto fail;
if (current.source) continue;
int verified = current.malformed ? 0 : member_sync_verify_local_record(db, pull->channel, &current.rec);
if (verified < 0) goto fail;
if (verified) continue;
int found = member_read_saved(db, pull->channel, node, 2, pull->id, &backup);
if (found < 0) goto fail;
int valid_backup = found && !backup.malformed ? member_sync_verify_local_record(db, pull->channel, &backup.rec) : 0;
if (valid_backup < 0) goto fail;
if (topo_node_sqlite_member_del(db, pull->channel, node)) goto fail;
if (valid_backup) {
snprintf(sql, sizeof(sql), "UPDATE peers_%s SET record_state=0,sync_id=0,previous_id=NULL"
" WHERE node_id=%lld AND record_state=2 AND sync_id=%lld", pull->channel, (long long)node, (long long)pull->id);
if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK || sqlite3_changes(db) != 1) goto fail;
}
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "member receive decision ch=%s sync=%lld node=%016llx action=%s",
pull->channel, (long long)pull->id, (unsigned long long)node, valid_backup ? "restore latest backup" : "remove invalid versions");
}
stage = "invitation chains";
/* Только существующие итоговые записи: доказанно неверные версии могли быть удалены. */
size_t proof_count = 0;
for (size_t i = 0; i < count; i++) if (topo_node_sqlite_member_in_channel(db, pull->channel, incoming[i])) incoming[proof_count++] = incoming[i];
rc = group_policy_verify_members(inst, pull->channel, incoming, proof_count);
if (rc != 1) { result = rc < 0 ? MT_ERR_IO : MT_ERR_CONFLICT; goto fail; }
stage = "admission";
if (group_policy_project(inst, pull->channel, &pull->admission)) goto fail;
stage = "derived data";
for (size_t i = 0; i < pull->count; i++) {
struct member_saved current; uint64_t node = pull->changed[i];
int found = member_read_saved(db, pull->channel, node, 0, 0, &current);
if (found < 0) goto fail;
if (found && !current.source && topo_node_sqlite_member_allowed(db, pull->channel, node)) {
char name[128] = ""; json_flat_get(current.info, "name", name, sizeof(name));
if (topo_node_sqlite_node_update_verified(db, node, name, current.x, current.ed,
current.rec.update_ts ? current.rec.update_ts : current.rec.join_ts, ntp_time_get_seconds(inst)) < 0) goto fail;
}
if (topo_node_sqlite_nodeinfo_updated(db, node) < 0 || chat_feedback_member_update(inst, node) < 0 ||
merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, pull->channel, node) < 0) goto fail;
}
if (merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, pull->channel, 0) < 0) goto fail;
stage = "cleanup";
if (member_pull_cleanup(inst, pull)) goto fail;
stage = "commit";
if (sqlite3_exec(db, "COMMIT /* member receive */", NULL, NULL, NULL) != SQLITE_OK) goto fail;
pull->committed = 1;
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "member receive committed ch=%s peer=%016llx sync=%lld received=%zu checked=%zu self_allowed=%d",
pull->channel, (unsigned long long)pull->peer, (long long)pull->id, count, pull->count,
topo_node_sqlite_member_allowed(db, pull->channel, inst->node_id));
u_free(incoming); return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive rejected ch=%s sync=%lld stage=%s result=%d: %s",
pull->channel, (long long)pull->id, stage, result, sqlite3_errmsg(db));
sqlite3_finalize(st);
if (!sqlite3_get_autocommit(db) && sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member receive rollback failed ch=%s", pull->channel); inst->running = 0;
}
pull->cleaned = 0; u_free(incoming); return result;
}
/* Контекст уже отсоединён от Merkle; отзыв допуска может свободно закрывать его сессию. */
static void member_publish_pull(void* ctx, void* state) {
struct UTUN_INSTANCE* inst = ctx; struct member_pull* pull = state;
if (!pull->committed) return;
struct group_policy_update* admission = pull->admission; pull->admission = NULL;
merkle_sync_changed(inst, MT_DATA_MEMBERS, pull->channel);
group_policy_publish(inst, admission);
for (size_t i = 0; i < pull->count; i++) {
member_sync_record_committed(inst, pull->channel, pull->changed[i], MS_APPLY_CHANGED | MS_APPLY_FEEDBACK);
struct member_saved current;
int found = member_read_saved(_db(inst), pull->channel, pull->changed[i], 0, 0, &current);
if (found == 1) _fire_apply_cbk(inst, pull->channel, current.rec.node_id, current.x, current.rec.signed_by, current.tree);
}
}
static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer, const uint8_t* data, size_t len) {
struct UTUN_INSTANCE* inst = ctx;
sqlite3* db = _db(inst);
src/routing_layer/topo_node_sqlite.c
+1
View File
@@ -575,6 +575,7 @@ int topo_node_sqlite_channel_put(sqlite3* db, const char* channel_id,
" sync_id INTEGER NOT NULL DEFAULT 0," /* 0 у опубликованных записей, id маркера приёма у временных */
" previous_id INTEGER," /* прежняя версия на момент получения; не основание для отката новой */
" sync_policy BLOB," /* подписанная политика только у маркера приёма (node_id=0) */
" sync_record BLOB," /* исходная новая запись; не читается обычными сервисами */
" x25519_pubkey BLOB NOT NULL," /* X25519 публичный ключ (32 байта) */
" ed25519_pubkey BLOB NOT NULL," /* Ed25519 публичный ключ (32 байта) — верификация join_sig/update_sig */
" join_sig BLOB NOT NULL," /* Ed25519 подпись присоединения (PEERS_JOIN_SIG_SIZE=64).