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

Connect MEMBERS to staged receive and publish only actual version changes

Browse Source
This commit is contained in:
evgeny committed 2026-11-04 00:43:50 +01:00
1 parent 81bba75d02
commit fcff5c626b
1 file changed
+41 -7
Download Patch File Download Diff File Expand all files Collapse all files
src/chat/member_sync.c
+41 -7
View File
@@ -1201,6 +1201,8 @@ static int member_decode(const uint8_t* p, size_t len, struct member_wire* w, si
/* Поля опубликованной записи, включая локальные настройки: бэкап сохраняется без потерь. */
#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"
#define MEMBER_SIGNED_FIELDS "node_id,x25519_pubkey,ed25519_pubkey,join_sig,join_ts,update_sig,update_ts,userinfo," \
"signed_by,signature,adm_tags,adm_tags_sig,source"
struct member_pull {
char channel[21];
@@ -1209,7 +1211,7 @@ struct member_pull {
uint64_t* changed;
size_t count;
struct group_policy_update* admission;
int committed, cleaned;
int committed, cleaned, policy_changed;
};
struct member_saved {
@@ -1326,15 +1328,27 @@ static int member_stage_page(void* ctx, void* state, uint64_t prefix, int after_
((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;
if (off != len || memcmp(next->bytes, key.bytes, MT_KEY_SIZE) || (!count && more)) goto invalid;
char sql[768]; sqlite3_stmt* st = NULL; int rc;
struct group_policy policy, previous_policy = {0};
if (group_policy_decode(db, pull->channel, data + 2, &policy)) goto invalid;
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);
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_bind_blob(st, 1, data + 2, GROUP_POLICY_WIRE_SIZE, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_ROW || (sqlite3_column_bytes(st, 0) &&
(sqlite3_column_bytes(st, 0) != GROUP_POLICY_WIRE_SIZE ||
group_policy_decode(db, pull->channel, sqlite3_column_blob(st, 0), &previous_policy)))) goto rollback;
int final = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_DONE || final != SQLITE_OK || sqlite3_changes(db) != 1) goto rollback;
if (final != SQLITE_OK) goto rollback;
if (policy.version > previous_policy.version || (policy.version == previous_policy.version && policy.all_members < previous_policy.all_members)) {
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);
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)",
@@ -1405,7 +1419,8 @@ static int member_commit_pull(void* ctx, void* state) {
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;
pull->policy_changed = group_policy_apply(db, pull->channel, &policy);
if (pull->policy_changed < 0) goto fail;
}
rc = sqlite3_finalize(st); st = NULL; if (rc != SQLITE_OK) goto fail;
stage = "incoming list";
@@ -1461,6 +1476,14 @@ static int member_commit_pull(void* ctx, void* state) {
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 = "changed versions";
u_free(pull->changed); pull->changed = NULL; pull->count = 0;
snprintf(sql, sizeof(sql), "SELECT node_id FROM (SELECT " MEMBER_SIGNED_FIELDS " FROM peers_%s WHERE record_state=0"
" EXCEPT SELECT " MEMBER_SIGNED_FIELDS " FROM peers_%s WHERE record_state=2 AND sync_id=%lld)"
" UNION SELECT node_id FROM (SELECT " MEMBER_SIGNED_FIELDS " FROM peers_%s WHERE record_state=2 AND sync_id=%lld"
" EXCEPT SELECT " MEMBER_SIGNED_FIELDS " FROM peers_%s WHERE record_state=0)",
pull->channel, pull->channel, (long long)pull->id, pull->channel, (long long)pull->id, pull->channel);
if (member_receive_ids(db, sql, &pull->changed, &pull->count)) goto fail;
stage = "admission";
if (group_policy_project(inst, pull->channel, &pull->admission)) goto fail;
stage = "derived data";
@@ -1501,13 +1524,19 @@ 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);
if (pull->count || pull->policy_changed) merkle_sync_changed(inst, MT_DATA_MEMBERS, pull->channel);
group_policy_publish(inst, admission);
for (size_t i = 0; i < pull->count; i++) {
if (topo_node_sqlite_channel_get(_db(inst), pull->channel, NULL, 0, NULL, NULL, NULL, NULL)) break;
member_sync_record_committed(inst, pull->channel, pull->changed[i], MS_APPLY_CHANGED | MS_APPLY_FEEDBACK);
if (topo_node_sqlite_channel_get(_db(inst), pull->channel, NULL, 0, NULL, NULL, NULL, NULL)) break;
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);
else if (!found) {
radio_forget_member_mute(inst, chat_channel_id(pull->channel), pull->changed[i]);
_fire_props_changed(inst, pull->changed[i], NULL, pull->channel);
}
}
}
@@ -1663,6 +1692,11 @@ static const struct merkle_sync_data_ops g_member_ops = {
.validate_peer = _member_validate_peer,
.finish = member_finish,
.on_result = member_policy_result,
.begin_pull = member_begin_pull,
.stage_page = member_stage_page,
.commit_pull = member_commit_pull,
.publish_pull = member_publish_pull,
.abort_pull = member_abort_pull,
};
/* ── Public API ── */