204 lines
10 KiB
C
204 lines
10 KiB
C
/* Реальные callbacks, БД, lifecycle и uasync; наблюдаем только границу запуска member_sync.
|
|
* Private-реализация включена как в test_merkle_protocol: публичный API не расширяем ради теста. */
|
|
#include "chat/member_sync.h"
|
|
#include "transport_layer/etcp_api.h"
|
|
#include "../lib/u_async.h"
|
|
#include "../lib/ll_queue.h"
|
|
#include "test_clock.h"
|
|
#include <assert.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
|
|
static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg);
|
|
static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry);
|
|
static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name);
|
|
#define member_sync_start observe_start
|
|
#define etcp_send discard_send
|
|
#define uasync_set_timeout schedule_timeout
|
|
#include "../src/chat/chat_sync.c"
|
|
#undef member_sync_start
|
|
#undef etcp_send
|
|
#undef uasync_set_timeout
|
|
|
|
struct fixture {
|
|
struct UASYNC* ua;
|
|
struct UTUN_INSTANCE* inst;
|
|
struct ETCP_CONN* peers[3];
|
|
char channel[2][21];
|
|
uint64_t gid[2];
|
|
};
|
|
|
|
static struct { uint64_t peer, channel; } starts[128];
|
|
static unsigned start_count;
|
|
static int fail_timer;
|
|
static struct fixture* injecting;
|
|
|
|
static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg) {
|
|
(void)inst; (void)cb; (void)arg;
|
|
assert(start_count < sizeof(starts) / sizeof(starts[0]));
|
|
starts[start_count].peer = peer; starts[start_count++].channel = strtoull(channel, NULL, 10);
|
|
if (injecting) {
|
|
struct fixture* f = injecting; injecting = NULL;
|
|
cs_on_conn_down(f->peers[1], ETCP_CBK_EVENT_DOWN, f->inst->chat_sync);
|
|
cs_schedule_peer_sync(f->inst->chat_sync, f->peers[2]->peer_node_id, "during_flush");
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry) {
|
|
(void)conn;
|
|
queue_dgram_free(entry); queue_entry_free(entry);
|
|
return 0;
|
|
}
|
|
|
|
static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name) {
|
|
if (fail_timer && !strcmp(name, "cs_sync")) { fail_timer = 0; return NULL; }
|
|
return uasync_set_timeout(ua, tb, arg, cb, name);
|
|
}
|
|
|
|
static void run_ms(struct fixture* f, unsigned ms) { test_clock_run_for(f->ua, ms * 10); }
|
|
|
|
/* Два канала: A состоит только в первом, B в обоих, C только во втором. */
|
|
static void setup(struct fixture* f) {
|
|
memset(f, 0, sizeof(*f)); start_count = 0; fail_timer = 0; injecting = NULL;
|
|
f->ua = uasync_create(); assert(f->ua);
|
|
f->inst = utun_instance_create_from_str(f->ua,
|
|
"[global]\nmy_node_name=schedule test\n[server: udp]\naddr=127.0.0.1:0\ntype=public\n"
|
|
"[ntp]\nenabled=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n");
|
|
assert(f->inst && utun_core_start(f->inst) == 0 && chat_service_start(f->inst) == 0);
|
|
chat_core_create_channel_auto(f->inst, "schedule 1"); chat_core_create_channel_auto(f->inst, "schedule 2");
|
|
unsigned channels = 0;
|
|
for (struct ll_entry* e = f->inst->topo_groups->group_list->head; e; e = e->next) {
|
|
struct TOPO_GROUP* group = (struct TOPO_GROUP*)e;
|
|
assert(group->group_type == TOPO_GROUP_TYPE_CHAT && channels < 2);
|
|
snprintf(f->channel[channels], sizeof(f->channel[channels]), "%llu", (unsigned long long)group->group_id);
|
|
f->gid[channels++] = group->group_id;
|
|
}
|
|
assert(channels == 2);
|
|
for (unsigned i = 0; i < 3; i++) {
|
|
struct SC_MYKEYS keys; assert(sc_generate_keypair(&keys) == SC_OK);
|
|
struct ETCP_CONN* conn = etcp_connection_create(f->inst, "schedule peer"); assert(conn);
|
|
conn->peer_node_id = sc_derive_node_id_from_pubkey(keys.public_key);
|
|
conn->initialized = 1; conn->links_up = 1; conn->state = 1;
|
|
assert(sc_init_ctx(&conn->crypto_ctx, &f->inst->my_keys) == SC_OK);
|
|
assert(sc_set_peer_public_key(&conn->crypto_ctx, keys.public_key, SC_PEER_PUBKEY_BIN) == SC_OK);
|
|
assert(etcp_conn_index_peer(conn) == 0); f->peers[i] = conn;
|
|
uint8_t signature[64] = {0};
|
|
for (unsigned ch = 0; ch < 2; ch++) {
|
|
if ((i == 0 && ch == 1) || (i == 2 && ch == 0)) continue;
|
|
assert(topo_node_sqlite_member_block_put(f->inst->topo_sqlite_db, f->channel[ch], conn->peer_node_id,
|
|
signature, 1, signature, 1, keys.public_key, keys.public_key, "{}", NULL, 0, signature) == 0);
|
|
}
|
|
}
|
|
/* Кеш намеренно не обновляем: задания должны видеть актуальное членство в БД. */
|
|
}
|
|
|
|
static void cleanup(struct fixture* f) {
|
|
utun_instance_destroy(f->inst); uasync_poll(f->ua, 0);
|
|
assert(f->ua->timer_alloc_count == f->ua->timer_free_count);
|
|
assert(f->ua->socket_alloc_count == f->ua->socket_free_count);
|
|
uasync_destroy(f->ua, 0);
|
|
}
|
|
|
|
static unsigned peer_starts(struct fixture* f, unsigned peer) {
|
|
unsigned count = 0;
|
|
for (unsigned i = 0; i < start_count; i++) if (starts[i].peer == f->peers[peer]->peer_node_id) count++;
|
|
return count;
|
|
}
|
|
|
|
static void test_reinit_isolation(void) {
|
|
struct fixture f; setup(&f);
|
|
for (unsigned i = 0; i < 3; i++) etcp_fire_conn_status(f.peers[i], ETCP_CONN_STATUS_UP);
|
|
assert(!start_count); run_ms(&f, 1100);
|
|
assert(start_count == 4 && peer_starts(&f, 0) == 1 && peer_starts(&f, 1) == 2 && peer_starts(&f, 2) == 1);
|
|
start_count = 0;
|
|
for (unsigned i = 0; i < 12; i++) {
|
|
etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_REINIT); run_ms(&f, 200);
|
|
assert(!peer_starts(&f, 1) && !peer_starts(&f, 2));
|
|
}
|
|
run_ms(&f, 1100); assert(start_count == 1 && starts[0].channel == f.gid[0]);
|
|
run_ms(&f, 4000); assert(start_count == 1);
|
|
cleanup(&f); puts("PASS: repeated REINIT is isolated; unchanged peers and idle service start no new rounds");
|
|
}
|
|
|
|
static void test_coalesce_and_deadline(void) {
|
|
struct fixture f; setup(&f);
|
|
etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
for (unsigned i = 0; i < 7; i++) {
|
|
run_ms(&f, 100); etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_REINIT);
|
|
etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
}
|
|
run_ms(&f, 400);
|
|
assert(start_count == 3 && peer_starts(&f, 0) == 1 && peer_starts(&f, 1) == 2 && !peer_starts(&f, 2));
|
|
cleanup(&f); puts("PASS: duplicate pairs coalesce and other peers cannot move the first deadline");
|
|
}
|
|
|
|
static void test_sleep_and_cancel(void) {
|
|
struct fixture f; setup(&f);
|
|
etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
f.peers[0]->peer_sleep_phase = 1; utun_fire_peer_sleep_cbk(f.inst, f.peers[0]->peer_node_id, 1);
|
|
run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 1) == 2);
|
|
start_count = 0; f.peers[0]->peer_sleep_phase = 0;
|
|
utun_fire_peer_sleep_cbk(f.inst, f.peers[0]->peer_node_id, 0);
|
|
etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); run_ms(&f, 1100);
|
|
assert(start_count == 1 && peer_starts(&f, 0) == 1);
|
|
start_count = 0;
|
|
etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
f.peers[0]->links_up = 0; etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_DOWN);
|
|
run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 1) == 2);
|
|
start_count = 0;
|
|
etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
char sql[160]; snprintf(sql, sizeof(sql), "UPDATE peers_%s SET deleted=1 WHERE node_id=%lld", f.channel[0],
|
|
(long long)f.peers[1]->peer_node_id);
|
|
assert(sqlite3_exec(f.inst->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
|
|
run_ms(&f, 1100); assert(start_count == 1 && starts[0].channel == f.gid[1]);
|
|
cleanup(&f); puts("PASS: sleep, DOWN and changed membership cancel only the affected requests");
|
|
}
|
|
|
|
static void test_join_existing_transport(void) {
|
|
struct fixture f; setup(&f);
|
|
cs_send_join_ready(f.inst->chat_sync, f.channel[0], f.peers[0]->peer_node_id);
|
|
assert(!start_count); run_ms(&f, 1100);
|
|
assert(start_count == 1 && starts[0].peer == f.peers[0]->peer_node_id && starts[0].channel == f.gid[0]);
|
|
start_count = 0;
|
|
cs_handle_join_ready(f.inst->chat_sync, f.peers[1]->peer_node_id, f.channel[0], (uint8_t*)&f.gid[0], 8);
|
|
run_ms(&f, 1100);
|
|
assert(start_count == 1 && starts[0].peer == f.peers[1]->peer_node_id && starts[0].channel == f.gid[0]);
|
|
cleanup(&f); puts("PASS: both JOIN_READY sides start only their pair without a new transport UP");
|
|
}
|
|
|
|
static void test_reentrant_cancel(void) {
|
|
struct fixture f; setup(&f);
|
|
etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
injecting = &f; run_ms(&f, 1100);
|
|
assert(start_count == 1 && peer_starts(&f, 0) == 1);
|
|
run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 2) == 1 && !peer_starts(&f, 1));
|
|
cleanup(&f); puts("PASS: callbacks can cancel remaining pairs and defer newly queued work");
|
|
}
|
|
|
|
static void test_lifecycle_and_timer_failure(void) {
|
|
struct fixture f; setup(&f);
|
|
fail_timer = 1; etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP);
|
|
assert(!f.inst->chat_sync->sync_pending && !f.inst->chat_sync->sync_timer);
|
|
etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
chat_sync_on_channel_deleted(f.inst, f.channel[0]); run_ms(&f, 1100);
|
|
assert(start_count == 1 && starts[0].channel == f.gid[1]);
|
|
start_count = 0; etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
|
|
chat_service_stop(f.inst); run_ms(&f, 1100); assert(!start_count);
|
|
assert(chat_service_start(f.inst) == 0); assert(!start_count);
|
|
run_ms(&f, 1100); assert(start_count == 4);
|
|
cleanup(&f); puts("PASS: timer failure recovers; channel/stop cancellation and startup retain correct scope");
|
|
}
|
|
|
|
int main(void) {
|
|
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
|
|
debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG);
|
|
test_clock_init(1000);
|
|
size_t baseline = u_get_allocated_count();
|
|
test_reinit_isolation(); test_coalesce_and_deadline(); test_sleep_and_cancel();
|
|
test_join_existing_transport(); test_reentrant_cancel(); test_lifecycle_and_timer_failure();
|
|
test_clock_finish();
|
|
assert(u_get_allocated_count() == baseline);
|
|
puts("ALL PASS: targeted chat synchronization scheduling"); return 0;
|
|
}
|