Exercise concurrent member versions, fallback, SQL rollback and restart
This commit is contained in:
evgeny committed
1 parent
1aa528d352
commit
7319f900c4
3 files changed
+142
-3
No files matched your search
@@ -1613,22 +1613,23 @@ static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer, co
|
||||
static int _member_validate_peer(void* ctx, const char* ns, uint64_t peer) {
|
||||
struct UTUN_INSTANCE* inst = ctx;
|
||||
struct ETCP_CONN* conn = inst ? instance_find_conn(inst, peer) : NULL;
|
||||
return conn && conn->initialized && conn->links_up && conn->state != 2 && !conn->close_requested &&
|
||||
return ns && conn && conn->initialized && conn->links_up && conn->state != 2 && !conn->close_requested &&
|
||||
topo_group_chat_allowed(inst, chat_channel_id(ns), peer) &&
|
||||
topo_node_sqlite_member_key_allowed(_db(inst), ns, conn->peer_pubkey);
|
||||
topo_node_sqlite_member_key_allowed(_db(inst), ns, conn->crypto_ctx.peer_public_key);
|
||||
}
|
||||
|
||||
/* Разрешение только спросить: канал выбран локально, ключ источника уже связан с его JOIN.
|
||||
* Никаких страниц/хешей в ответ этот режим не выдаёт; источник сам проверяет наш допуск. */
|
||||
static int member_validate_request(void* ctx, const char* ns, uint64_t peer) {
|
||||
struct UTUN_INSTANCE* inst = ctx;
|
||||
if (!ns) return 0;
|
||||
struct TOPO_GROUP* group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, chat_channel_id(ns)) : NULL;
|
||||
struct ETCP_CONN* conn = inst ? instance_find_conn(inst, peer) : NULL;
|
||||
if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT || group->stopping || strcmp(group->channel_id, ns) ||
|
||||
!conn || !conn->initialized || !conn->links_up || conn->state == 2 || conn->close_requested) return 0;
|
||||
struct member_saved record;
|
||||
int found = member_read_saved(_db(inst), ns, peer, 0, 0, &record);
|
||||
if (found != 1 || record.source || record.malformed || memcmp(record.x, conn->peer_pubkey, 32) ||
|
||||
if (found != 1 || record.source || record.malformed || memcmp(record.x, conn->crypto_ctx.peer_public_key, 32) ||
|
||||
sc_derive_node_id_from_pubkey(record.x) != peer || _verify_join_sig(_db(inst), ns, &record.rec) != 1) return 0;
|
||||
uint8_t key[32]; enum ms_author_status status = member_sync_author_key(inst, ns, peer, key);
|
||||
return status == MS_AUTHOR_ACTIVE || status == MS_AUTHOR_PENDING;
|
||||
|
||||
@@ -67,6 +67,7 @@ check_PROGRAMS = \
|
||||
test_multicast \
|
||||
test_group_ownership \
|
||||
test_merkle_protocol \
|
||||
test_member_versions \
|
||||
test_chat_sync_schedule \
|
||||
test_ncd_config \
|
||||
test_node_snapshot \
|
||||
@@ -1004,6 +1005,9 @@ test_chat_feedback_LDFLAGS = -Wl,--wrap=sqlite3_exec -Wl,--wrap=sqlite3_prepare_
|
||||
test_merkle_protocol_SOURCES = test_merkle_protocol.c
|
||||
test_merkle_protocol_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib
|
||||
test_merkle_protocol_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
|
||||
test_member_versions_SOURCES = test_member_versions.c
|
||||
test_member_versions_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib
|
||||
test_member_versions_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
|
||||
test_chat_sync_schedule_SOURCES = test_chat_sync_schedule.c $(test_clock_sources)
|
||||
test_chat_sync_schedule_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib
|
||||
test_chat_sync_schedule_LDFLAGS = $(test_clock_ldflags)
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
/* Проверенные/новые/старые версии в настоящей SQLite: конкурентные пиры, отказ и отмена. */
|
||||
#include "../src/chat/member_sync.c"
|
||||
#include <assert.h>
|
||||
|
||||
struct identity { uint64_t id; uint8_t x[32], ed[32], secret[32], join[64], tree[64]; };
|
||||
static struct identity identities[5];
|
||||
static uint8_t group_secret[32] = {91}, group_public[32], group_x[32] = {92};
|
||||
|
||||
static void public_key(const uint8_t secret[32], uint8_t public[32]) {
|
||||
EVP_PKEY* key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, secret, 32); assert(key);
|
||||
size_t bytes = 32; assert(EVP_PKEY_get_raw_public_key(key, public, &bytes) == 1); EVP_PKEY_free(key);
|
||||
}
|
||||
|
||||
static void init_identity(unsigned i) {
|
||||
struct identity* p = &identities[i]; p->x[0] = i + 11; p->secret[0] = i + 31;
|
||||
public_key(p->secret, p->ed); p->id = sc_derive_node_id_from_pubkey(p->x);
|
||||
uint8_t message[112]; assert(member_sync_build_join_msg(group_x, group_public, p->id, p->x, 1, message, sizeof(message)) == 112);
|
||||
assert(sc_ed25519_sign(p->secret, message, sizeof(message), p->join) == SC_OK);
|
||||
assert(!member_sync_sign_join(group_secret, p->join, p->tree));
|
||||
}
|
||||
|
||||
static void profile(struct UTUN_INSTANCE* inst, unsigned person, uint64_t version) {
|
||||
struct identity* p = &identities[person]; uint8_t message[256], signature[64]; char info[64];
|
||||
snprintf(info, sizeof(info), "{\"name\":\"profile-%llu\"}", (unsigned long long)version);
|
||||
int bytes = member_sync_build_update_msg(p->join, version, info, message, sizeof(message)); assert(bytes > 0);
|
||||
assert(sc_ed25519_sign(p->secret, message, bytes, signature) == SC_OK);
|
||||
assert(member_sync_put(inst, "1001", p->id, p->x, p->ed, p->join, 1, signature, version, info,
|
||||
NULL, NULL, 0, 0, p->tree) >= 0);
|
||||
}
|
||||
|
||||
static void setup(struct UTUN_INSTANCE* inst, struct UASYNC* ua, unsigned self) {
|
||||
memset(inst, 0, sizeof(*inst)); inst->node_id = identities[self].id; inst->ua = ua;
|
||||
assert(sqlite3_open(":memory:", &inst->topo_sqlite_db) == SQLITE_OK);
|
||||
assert(!topo_node_sqlite_init(inst->topo_sqlite_db));
|
||||
assert(!topo_node_sqlite_channel_put(inst->topo_sqlite_db, "1001", "versions", identities[0].id,
|
||||
group_x, NULL, group_public, NULL, (uint8_t[64]){0}));
|
||||
assert(!member_sync_init(inst));
|
||||
for (unsigned i = 0; i < 3; i++) profile(inst, i, 1);
|
||||
}
|
||||
|
||||
static sqlite3_int64 scalar(struct UTUN_INSTANCE* inst, const char* sql) {
|
||||
sqlite3_stmt* st = NULL; assert(sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL) == SQLITE_OK);
|
||||
assert(sqlite3_step(st) == SQLITE_ROW); sqlite3_int64 value = sqlite3_column_int64(st, 0);
|
||||
assert(sqlite3_finalize(st) == SQLITE_OK); return value;
|
||||
}
|
||||
|
||||
static uint64_t version(struct UTUN_INSTANCE* inst) {
|
||||
struct member_saved record; assert(member_read_saved(inst->topo_sqlite_db, "1001", identities[2].id, 0, 0, &record) == 1);
|
||||
return record.rec.update_ts;
|
||||
}
|
||||
|
||||
static struct member_pull* receive(struct UTUN_INSTANCE* source, struct UTUN_INSTANCE* target, unsigned person, uint64_t peer) {
|
||||
void* state = NULL; assert(!member_begin_pull(target, "1001", peer, 1, &state));
|
||||
uint64_t prefix = merkle_sync_level_prefix(identities[person].id, MT_MAX_LEVEL);
|
||||
uint8_t bytes[MT_PAGE_SIZE]; size_t length = sizeof(bytes); int more; merkle_key_t after = {{0}}, next;
|
||||
assert(!_member_get_page(source, "1001", prefix, 0, &after, bytes, &length, &next, &more) && !more);
|
||||
assert(!member_stage_page(target, state, prefix, 0, &after, &next, more, bytes, length)); return state;
|
||||
}
|
||||
|
||||
static void accept(struct UTUN_INSTANCE* inst, struct member_pull* pull) {
|
||||
assert(!member_commit_pull(inst, pull)); member_publish_pull(inst, pull); member_abort_pull(inst, pull);
|
||||
assert(sqlite3_get_autocommit(inst->topo_sqlite_db));
|
||||
}
|
||||
|
||||
static void corrupt(struct UTUN_INSTANCE* inst) {
|
||||
char sql[192]; snprintf(sql, sizeof(sql), "UPDATE peers_1001 SET update_sig=zeroblob(64)||X'01' WHERE node_id=%lld AND record_state=0",
|
||||
(long long)identities[2].id);
|
||||
assert(sqlite3_exec(inst->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
|
||||
}
|
||||
|
||||
int main(void) {
|
||||
debug_config_init(); debug_set_level(DEBUG_LEVEL_INFO);
|
||||
size_t baseline = u_get_allocated_count(); struct UASYNC* ua = uasync_create(); assert(ua);
|
||||
public_key(group_secret, group_public);
|
||||
for (unsigned i = 0; i < 5; i++) init_identity(i);
|
||||
struct UTUN_INSTANCE source, target; setup(&source, ua, 1); setup(&target, ua, 0);
|
||||
uint8_t original[32], staged[32]; assert(!merkle_sync_read_hash(&target, MT_DATA_MEMBERS, "1001", 0, 0, original));
|
||||
profile(&source, 2, 2);
|
||||
struct member_pull* a = receive(&source, &target, 2, identities[1].id);
|
||||
assert(version(&target) == 1 && member_sync_count(&target, "1001") == 3);
|
||||
assert(!merkle_sync_read_hash(&target, MT_DATA_MEMBERS, "1001", 0, 0, staged) && !memcmp(original, staged, 32));
|
||||
assert(scalar(&target, "SELECT COUNT(*) FROM nodes WHERE name='profile-2'") == 0);
|
||||
profile(&source, 2, 3);
|
||||
struct member_pull* b = receive(&source, &target, 2, identities[3].id);
|
||||
assert(a->id != b->id && scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state=1") == 4);
|
||||
accept(&target, b); assert(version(&target) == 3);
|
||||
assert(scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state=1") == 2);
|
||||
accept(&target, a); assert(version(&target) == 3);
|
||||
assert(!scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state<>0"));
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS two peers: isolated staging, no early profile/hash change, stale completion preserves newer commit");
|
||||
|
||||
profile(&source, 2, 4); a = receive(&source, &target, 2, identities[1].id); b = receive(&source, &target, 2, identities[3].id);
|
||||
member_abort_pull(&target, a);
|
||||
assert(scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state=1") == 2);
|
||||
accept(&target, b); assert(version(&target) == 4);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS cancelling one peer leaves the other exchange intact");
|
||||
|
||||
profile(&source, 2, 5);
|
||||
char sql[256]; snprintf(sql, sizeof(sql), "UPDATE peers_1001 SET update_sig=randomblob(64) WHERE node_id=%lld AND record_state=0",
|
||||
(long long)identities[2].id);
|
||||
assert(sqlite3_exec(source.topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
|
||||
a = receive(&source, &target, 2, identities[1].id); accept(&target, a); assert(version(&target) == 4);
|
||||
assert(sqlite3_exec(target.topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
|
||||
a = receive(&source, &target, 2, identities[1].id); accept(&target, a);
|
||||
assert(!topo_node_sqlite_member_in_channel(target.topo_sqlite_db, "1001", identities[2].id));
|
||||
assert(member_sync_count(&target, "1001") == 2 && topo_node_sqlite_member_allowed(target.topo_sqlite_db, "1001", identities[0].id));
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS bad new keeps good old; both invalid removes only that member");
|
||||
|
||||
profile(&source, 2, 6); a = receive(&source, &target, 2, identities[1].id); accept(&target, a); assert(version(&target) == 6);
|
||||
profile(&source, 2, 7); a = receive(&source, &target, 2, identities[1].id);
|
||||
snprintf(sql, sizeof(sql), "CREATE TRIGGER fail_identity BEFORE INSERT ON nodes WHEN NEW.node_id=%lld"
|
||||
" BEGIN SELECT RAISE(ABORT,'injected identity failure'); END", (long long)identities[2].id);
|
||||
assert(sqlite3_exec(target.topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
|
||||
assert(member_commit_pull(&target, a) < 0 && version(&target) == 6);
|
||||
member_publish_pull(&target, a); member_abort_pull(&target, a);
|
||||
assert(!scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state<>0"));
|
||||
assert(sqlite3_exec(target.topo_sqlite_db, "DROP TRIGGER fail_identity", NULL, NULL, NULL) == SQLITE_OK);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS SQL failure rolls back versions, identities and admission together");
|
||||
|
||||
a = receive(&source, &target, 2, identities[1].id); sqlite3_int64 previous = a->id;
|
||||
member_sync_destroy(&target); assert(!member_sync_init(&target));
|
||||
assert(version(&target) == 6 && !scalar(&target, "SELECT COUNT(*) FROM peers_1001 WHERE record_state<>0"));
|
||||
b = receive(&source, &target, 2, identities[3].id); assert(b->id > previous);
|
||||
member_abort_pull(&target, a); accept(&target, b); assert(version(&target) == 7);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS restart removes abandoned versions without reusing exchange identifiers");
|
||||
|
||||
corrupt(&target); profile(&source, 2, 8);
|
||||
a = receive(&source, &target, 2, identities[1].id); accept(&target, a); assert(version(&target) == 8);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS valid new version repairs a malformed previous version");
|
||||
member_sync_destroy(&source); member_sync_destroy(&target);
|
||||
assert(sqlite3_close(source.topo_sqlite_db) == SQLITE_OK && sqlite3_close(target.topo_sqlite_db) == SQLITE_OK);
|
||||
uasync_destroy(ua, 0); assert(u_get_allocated_count() == baseline);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "ALL PASS member version isolation and recovery"); return 0;
|
||||
}
|
||||
Reference in new issue
Block a user