diff --git a/src/chat/member_sync.c b/src/chat/member_sync.c index 308dece00..da4c41bb7 100644 --- a/src/chat/member_sync.c +++ b/src/chat/member_sync.c @@ -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, ¤t); 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 ── */