#include #include #include #include #include #include #include #include /* Use the same Unity include pattern as other controller tests */ #include #include "repl.h" #include "config.h" #include "typemap.h" #include "wal_types.h" #include "controller.capnp.h" /* OPCode_walAck */ /* ============================================================ * Test seam for repl_broadcast send path (task 1.1) * Declared as extern here; defined in repl.c under XCE_TESTING. * ============================================================ */ #ifdef XCE_TESTING extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries, int n_entries); #endif static void test_repl_init_zeroes_nstandbys(void) { replicator_t r; TEST_ASSERT_EQUAL_INT(0, repl_init(&r)); TEST_ASSERT_EQUAL_INT(0, r.nstandbys); repl_fini(&r); printf("test_repl_init_zeroes_nstandbys: PASS\n"); } static void test_repl_add_remove_standby(void) { replicator_t r; /* Use a dummy int as the endpoint pointer — repl.c only compares ep pointers */ int dummy_ep = 0; repl_init(&r); int idx = repl_add_standby(&r, (void*)&dummy_ep, "standby1", 7001); TEST_ASSERT_TRUE(idx >= 0); TEST_ASSERT_EQUAL_INT(1, r.nstandbys); repl_remove_standby(&r, (void*)&dummy_ep); TEST_ASSERT_EQUAL_INT(0, r.nstandbys); repl_fini(&r); printf("test_repl_add_remove_standby: PASS\n"); } static void test_repl_add_standby_max(void) { replicator_t r; int eps[MAX_STANDBYS + 1]; repl_init(&r); for (int i = 0; i < MAX_STANDBYS; i++) { int rc = repl_add_standby(&r, (void*)&eps[i], "sb", 7000 + i); TEST_ASSERT_TRUE(rc >= 0); } /* One over limit must fail */ int rc = repl_add_standby(&r, (void*)&eps[MAX_STANDBYS], "sb_extra", 9999); TEST_ASSERT_TRUE(rc < 0); repl_fini(&r); printf("test_repl_add_standby_max: PASS\n"); } static void test_repl_trim_lsn_no_standbys(void) { replicator_t r; repl_init(&r); uint64_t trim = repl_trim_lsn(&r, 42, 60); TEST_ASSERT_EQUAL_UINT64(42, trim); repl_fini(&r); printf("test_repl_trim_lsn_no_standbys: PASS\n"); } static void test_repl_trim_lsn_disconnected_slot_expiry(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); repl_on_ack(&r, (void*)&dummy_ep, 10); repl_remove_standby(&r, (void*)&dummy_ep); /* Immediately after disconnect: trim guard holds at last_disconnected_lsn=10 */ uint64_t trim = repl_trim_lsn(&r, 50, 60); TEST_ASSERT_EQUAL_UINT64(10, trim); /* After expiry (snap_interval_s=0 forces immediate expiry): trim == checkpoint */ repl_expire_disconnect_slots(&r, 999999, 0); trim = repl_trim_lsn(&r, 50, 0); TEST_ASSERT_EQUAL_UINT64(50, trim); repl_fini(&r); printf("test_repl_trim_lsn_disconnected_slot_expiry: PASS\n"); } /* ============================================================ * P2-3: repl_wait_ack() tests * ============================================================ */ /* Test 1: repl_wait_ack returns -ENODEV when no standbys connected */ static void test_wait_ack_no_standbys_returns_enodev(void) { replicator_t r; repl_init(&r); int rc = repl_wait_ack(&r, /*target_lsn=*/10, /*timeout_ms=*/50); TEST_ASSERT_EQUAL_INT(-ENODEV, rc); repl_fini(&r); printf("test_wait_ack_no_standbys_returns_enodev: PASS\n"); } /* Test 2: repl_wait_ack returns 0 when standby already ACKed target LSN */ static void test_wait_ack_already_acked(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); /* Pre-set acked_lsn to 20 (already past target 10) */ repl_on_ack(&r, (void*)&dummy_ep, 20); int rc = repl_wait_ack(&r, 10, 50); TEST_ASSERT_EQUAL_INT(0, rc); repl_fini(&r); printf("test_wait_ack_already_acked: PASS\n"); } /* Test 3: repl_wait_ack times out when standby never ACKs */ static void test_wait_ack_timeout(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); /* acked_lsn stays at 0 — never reaches target 100 */ struct timespec start, end; clock_gettime(CLOCK_MONOTONIC, &start); int rc = repl_wait_ack(&r, 100, 100 /* 100ms timeout */); clock_gettime(CLOCK_MONOTONIC, &end); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc); /* Verify we actually waited ~100ms (not returned immediately) */ long elapsed_ms = (end.tv_sec - start.tv_sec) * 1000 + (end.tv_nsec - start.tv_nsec) / 1000000; TEST_ASSERT_TRUE(elapsed_ms >= 80); /* at least 80ms */ repl_fini(&r); printf("test_wait_ack_timeout: PASS\n"); } /* Test 4: ACK from background thread wakes repl_wait_ack */ struct ack_args { replicator_t *r; void *ep; uint64_t lsn; int delay_ms; }; static void* ack_sender_thread(void* arg) { struct ack_args* a = arg; struct timespec ts = { 0, (long)a->delay_ms * 1000000L }; nanosleep(&ts, NULL); repl_on_ack(a->r, a->ep, a->lsn); return NULL; } static void test_wait_ack_woken_by_background_ack(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001); struct ack_args args = { &r, (void*)&dummy_ep, 15, 30 /* 30ms delay */ }; pthread_t th; pthread_create(&th, NULL, ack_sender_thread, &args); /* Wait for LSN 15 with 200ms timeout — ACK arrives after 30ms */ int rc = repl_wait_ack(&r, 15, 200); pthread_join(th, NULL); TEST_ASSERT_EQUAL_INT(0, rc); repl_fini(&r); printf("test_wait_ack_woken_by_background_ack: PASS\n"); } /* ============================================================ * Task 1.1: repl_broadcast send path tests * ============================================================ */ /* Shared hook state for send-path tests */ static int g_hook_call_count; static void *g_hook_last_ep; static uint64_t g_hook_last_lsn; static uint8_t g_hook_last_entry_type; static const void *g_hook_last_payload; static size_t g_hook_last_payload_len; static int g_hook_return_val; /* 0 = success, negative = error */ static int test_broadcast_hook(void *ep, wal_catchup_entry_t **entries, int n_entries) { g_hook_call_count++; g_hook_last_ep = ep; if (n_entries >= 1 && entries && entries[0]) { g_hook_last_lsn = entries[0]->lsn; g_hook_last_entry_type = entries[0]->entry_type; g_hook_last_payload = entries[0]->raw_payload; g_hook_last_payload_len = entries[0]->raw_payload_len; } return g_hook_return_val; } static void reset_hook_state(void) { g_hook_call_count = 0; g_hook_last_ep = NULL; g_hook_last_lsn = 0; g_hook_last_entry_type = 0; g_hook_last_payload = NULL; g_hook_last_payload_len = 0; g_hook_return_val = 0; } /* Test: no standbys — broadcast returns 0, hook not called */ static void test_broadcast_no_standbys_returns_zero(void) { #ifdef XCE_TESTING replicator_t r; repl_init(&r); reset_hook_state(); repl_broadcast_send_hook = test_broadcast_hook; wal_entry_hdr_t hdr = { .lsn = 10, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0xAA, 0xBB }; int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(0, g_hook_call_count); repl_broadcast_send_hook = NULL; repl_fini(&r); printf("test_broadcast_no_standbys_returns_zero: PASS\n"); #else printf("test_broadcast_no_standbys_returns_zero: SKIPPED (no XCE_TESTING)\n"); #endif } /* Test: one standby — hook called once with correct lsn, type, payload */ static void test_broadcast_one_standby_sends_entry(void) { #ifdef XCE_TESTING replicator_t r; int dummy_ep = 0; repl_init(&r); repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); reset_hook_state(); repl_broadcast_send_hook = test_broadcast_hook; wal_entry_hdr_t hdr = { .lsn = 42, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 1, 2, 3, 4 }; int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(1, g_hook_call_count); TEST_ASSERT_EQUAL_PTR((void *)&dummy_ep, g_hook_last_ep); TEST_ASSERT_EQUAL_UINT64(42, g_hook_last_lsn); TEST_ASSERT_EQUAL_UINT8(WAL_JOB_SUBMIT, g_hook_last_entry_type); TEST_ASSERT_EQUAL_PTR(payload, g_hook_last_payload); TEST_ASSERT_EQUAL_UINT(sizeof(payload), g_hook_last_payload_len); repl_broadcast_send_hook = NULL; repl_fini(&r); printf("test_broadcast_one_standby_sends_entry: PASS\n"); #else printf("test_broadcast_one_standby_sends_entry: SKIPPED (no XCE_TESTING)\n"); #endif } /* Test: standby in catch-up is skipped — hook not called for it */ static void test_broadcast_skips_catchup_standby(void) { #ifdef XCE_TESTING replicator_t r; int ep1 = 0, ep2 = 1; repl_init(&r); repl_add_standby(&r, (void *)&ep1, "sb1", 7001); repl_add_standby(&r, (void *)&ep2, "sb2", 7002); /* Mark sb1 as in catch-up */ repl_start_catchup(&r, (void *)&ep1, 0); reset_hook_state(); repl_broadcast_send_hook = test_broadcast_hook; wal_entry_hdr_t hdr = { .lsn = 99, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0xFF }; int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); TEST_ASSERT_EQUAL_INT(0, rc); /* Only sb2 should be sent to */ TEST_ASSERT_EQUAL_INT(1, g_hook_call_count); TEST_ASSERT_EQUAL_PTR((void *)&ep2, g_hook_last_ep); repl_broadcast_send_hook = NULL; repl_fini(&r); printf("test_broadcast_skips_catchup_standby: PASS\n"); #else printf("test_broadcast_skips_catchup_standby: SKIPPED (no XCE_TESTING)\n"); #endif } /* Test: standby with pending_ack_count > MAX_PENDING_ACKS is evicted to catch-up */ static void test_broadcast_evicts_overloaded_standby(void) { #ifdef XCE_TESTING replicator_t r; int dummy_ep = 0; repl_init(&r); int idx = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); /* Manually set pending_ack_count just above the limit */ r.standbys[idx].pending_ack_count = MAX_PENDING_ACKS + 1; reset_hook_state(); repl_broadcast_send_hook = test_broadcast_hook; wal_entry_hdr_t hdr = { .lsn = 5, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0x01 }; int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); TEST_ASSERT_EQUAL_INT(0, rc); /* Evicted — hook must NOT be called */ TEST_ASSERT_EQUAL_INT(0, g_hook_call_count); /* Standby must now be in catch-up */ TEST_ASSERT_NOT_EQUAL(0, atomic_load_explicit(&r.standbys[idx].in_catch_up, memory_order_acquire)); repl_broadcast_send_hook = NULL; repl_fini(&r); printf("test_broadcast_evicts_overloaded_standby: PASS\n"); #else printf("test_broadcast_evicts_overloaded_standby: SKIPPED (no XCE_TESTING)\n"); #endif } /* Test (Task 2.3): repl_add_standby is idempotent for the same ep. * Calling it twice with the same endpoint must return the same index, * keep nstandbys==1, and not crash or corrupt state. * Req 1.3: If repl_add_standby is called with an endpoint already * registered, the system shall treat it as idempotent (no-op or update-in-place). */ static void test_repl_add_standby_idempotent(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); int idx1 = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); TEST_ASSERT_TRUE(idx1 >= 0); TEST_ASSERT_EQUAL_INT(1, r.nstandbys); /* Second call with the same ep — must be a no-op */ int idx2 = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); TEST_ASSERT_TRUE(idx2 >= 0); /* nstandbys must still be 1 (not 2) */ TEST_ASSERT_EQUAL_INT(1, r.nstandbys); /* Must return the same slot index */ TEST_ASSERT_EQUAL_INT(idx1, idx2); repl_fini(&r); printf("test_repl_add_standby_idempotent: PASS\n"); } /* ============================================================ * Task 4.1: WAL ACK reply handler — Req 4.3 * * These tests verify the behaviour that repl_ack_decode_and_dispatch * implements in ha.c: on a successful WalAckReply it calls * repl_on_ack(&c->persist.repl, ep, acked_lsn), which must: * a) update standby's acked_lsn * b) signal ack_cond so repl_wait_majority_ack unblocks * * We test repl_on_ack directly (the exact call the handler makes). * ============================================================ */ /* ACK sender thread args reused from above — ack_args struct already defined */ /* Test: repl_on_ack (handler ACK path) updates acked_lsn and unblocks * repl_wait_majority_ack. Req 4.3: when the primary receives a WAL ACK * from a standby it shall update that standby's acked_lsn in the replicator * and signal any waiter on the ACK condition variable. */ static void test_wal_ack_reply_updates_acked_lsn_and_signals(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); const uint64_t target_lsn = 77; /* Background thread simulates the handler calling repl_on_ack after 30 ms */ struct ack_args args = { &r, (void *)&dummy_ep, target_lsn, 30 }; pthread_t th; pthread_create(&th, NULL, ack_sender_thread, &args); /* repl_wait_majority_ack: 1 active standby, quorum=(1+2)/2=1, * acked_total = 1(self) + 1(standby) = 2 >= 1 → returns 0. */ int rc = repl_wait_majority_ack(&r, 0, 200, (uint8_t)OPCode_walAck); pthread_join(th, NULL); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "repl_wait_majority_ack must return 0 after repl_on_ack signals (Req 4.3)"); /* Verify acked_lsn was actually updated in the slot */ int found = 0; for (int i = 0; i < MAX_STANDBYS; i++) { if (r.standbys[i].ep == (void *)&dummy_ep) { uint64_t al = atomic_load_explicit(&r.standbys[i].acked_lsn, memory_order_acquire); TEST_ASSERT_EQUAL_UINT64_MESSAGE(target_lsn, al, "acked_lsn must equal the LSN passed to repl_on_ack (Req 4.3)"); found = 1; break; } } TEST_ASSERT_TRUE_MESSAGE(found, "standby slot must be present after repl_on_ack"); repl_fini(&r); printf("test_wal_ack_reply_updates_acked_lsn_and_signals: PASS\n"); } /* Test: NACK path — handler does NOT call repl_on_ack, so acked_lsn stays 0. * Req 4.3 (negative): only a success==1 reply triggers the update; success==0 * is a no-op on the replicator. * * Quorum analysis for this scenario: * n_active = 1 (one standby registered, not in catch-up) * quorum = (1 + 2) / 2 = 1 * walAck: acked_total = 1 (self) + 0 (standby acked_lsn == 0, not > 0) = 1 * 1 >= quorum(1) → repl_wait_majority_ack returns 0 IMMEDIATELY. * * The primary self-count alone satisfies a 1-standby quorum, so the function * does NOT time out even when the standby never ACKs. The NACK invariant is * that acked_lsn remains 0 (not that the wait times out). */ static void test_wal_nack_does_not_update_acked_lsn(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); /* No repl_on_ack call — simulates handler receiving success==0 NACK */ /* acked_lsn must remain 0 */ uint64_t al = 0; for (int i = 0; i < MAX_STANDBYS; i++) { if (r.standbys[i].ep == (void *)&dummy_ep) { al = atomic_load_explicit(&r.standbys[i].acked_lsn, memory_order_acquire); break; } } TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, al, "acked_lsn must remain 0 when NACK path skips repl_on_ack (Req 4.3)"); /* With 1 standby and no standby ACK, primary self-count (1) already meets * quorum (1), so repl_wait_majority_ack returns 0, not -ETIMEDOUT. * A timeout would only occur if quorum required at least one standby ACK, * which only happens when n_active >= 2 (e.g. 2+ standbys registered). */ int rc = repl_wait_majority_ack(&r, 0, 50, (uint8_t)OPCode_walAck); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "repl_wait_majority_ack must return 0 with 1 standby: " "self-count(1) >= quorum(1) even without standby ACK (Req 4.3)"); repl_fini(&r); printf("test_wal_nack_does_not_update_acked_lsn: PASS\n"); } /* Test: send failure (hook returns -EIO) is tolerated — broadcast returns 0 */ static void test_broadcast_send_failure_continues(void) { #ifdef XCE_TESTING replicator_t r; int ep1 = 0, ep2 = 1; repl_init(&r); repl_add_standby(&r, (void *)&ep1, "sb1", 7001); repl_add_standby(&r, (void *)&ep2, "sb2", 7002); reset_hook_state(); g_hook_return_val = -EIO; /* First call fails */ repl_broadcast_send_hook = test_broadcast_hook; wal_entry_hdr_t hdr = { .lsn = 77, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0xDE, 0xAD }; int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); /* broadcast should still return 0 — errors are WARN+continue */ TEST_ASSERT_EQUAL_INT(0, rc); /* Both standbys attempted */ TEST_ASSERT_EQUAL_INT(2, g_hook_call_count); repl_broadcast_send_hook = NULL; repl_fini(&r); printf("test_broadcast_send_failure_continues: PASS\n"); #else printf("test_broadcast_send_failure_continues: SKIPPED (no XCE_TESTING)\n"); #endif } /* ============================================================ * Task 5.3: Registration lifecycle — Req 1.1, 1.2, 1.3 * * Comprehensive test covering all three lifecycle steps in sequence: * Step 1 (Req 1.1): Register ep → nstandbys==1, standbys[idx].ep==ep * Step 2 (Req 1.3): Re-register same ep → idempotent, nstandbys==1, same idx * Step 3 (Req 1.2): Deregister ep → nstandbys==0, standbys[idx].ep==NULL * ============================================================ */ static void test_repl_registration_lifecycle(void) { replicator_t r; int dummy_ep = 0; repl_init(&r); /* Step 1: Register ep — Req 1.1 */ int idx = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); TEST_ASSERT_TRUE_MESSAGE(idx >= 0, "repl_add_standby must return a valid slot index"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, r.nstandbys, "nstandbys must be 1 after first registration (Req 1.1)"); TEST_ASSERT_EQUAL_PTR_MESSAGE((void *)&dummy_ep, r.standbys[idx].ep, "standbys[idx].ep must equal registered ep (Req 1.1)"); /* Step 2: Re-register same ep — idempotent, Req 1.3 */ int idx2 = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); TEST_ASSERT_TRUE_MESSAGE(idx2 >= 0, "duplicate repl_add_standby must return valid index"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, r.nstandbys, "nstandbys must remain 1 after duplicate registration (Req 1.3)"); TEST_ASSERT_EQUAL_INT_MESSAGE(idx, idx2, "duplicate registration must return the same slot index (Req 1.3)"); /* Step 3: Deregister ep — Req 1.2 */ repl_remove_standby(&r, (void *)&dummy_ep); TEST_ASSERT_EQUAL_INT_MESSAGE(0, r.nstandbys, "nstandbys must be 0 after deregistration (Req 1.2)"); TEST_ASSERT_NULL_MESSAGE(r.standbys[idx].ep, "standbys[idx].ep must be NULL after deregistration (Req 1.2)"); repl_fini(&r); printf("test_repl_registration_lifecycle: PASS\n"); } void test_repl_unit(void) { UNITY_BEGIN(); RUN_TEST(test_repl_init_zeroes_nstandbys); RUN_TEST(test_repl_add_remove_standby); RUN_TEST(test_repl_add_standby_max); RUN_TEST(test_repl_trim_lsn_no_standbys); RUN_TEST(test_repl_trim_lsn_disconnected_slot_expiry); RUN_TEST(test_wait_ack_no_standbys_returns_enodev); RUN_TEST(test_wait_ack_already_acked); RUN_TEST(test_wait_ack_timeout); RUN_TEST(test_wait_ack_woken_by_background_ack); /* Task 2.3: idempotent re-registration */ RUN_TEST(test_repl_add_standby_idempotent); /* Task 1.1: broadcast send path */ RUN_TEST(test_broadcast_no_standbys_returns_zero); RUN_TEST(test_broadcast_one_standby_sends_entry); RUN_TEST(test_broadcast_skips_catchup_standby); RUN_TEST(test_broadcast_evicts_overloaded_standby); RUN_TEST(test_broadcast_send_failure_continues); /* Task 4.1: WAL ACK reply handler — Req 4.3 */ RUN_TEST(test_wal_ack_reply_updates_acked_lsn_and_signals); RUN_TEST(test_wal_nack_does_not_update_acked_lsn); /* Task 5.3: registration lifecycle — Req 1.1, 1.2, 1.3 */ RUN_TEST(test_repl_registration_lifecycle); UNITY_END(); }