/* * test_p2_integration.c — P2-6: Replication integration tests — assembled system. * * Tests validate the assembled P2 replication system end-to-end. * No live network I/O — pure in-process state and persist-level logic. * * TDD: these tests were written FIRST (RED), then implementation was verified GREEN. */ #include #include #include #include #include #include #include #include #include #include #include "commit.h" #include "config.h" #include "persist.h" #include "repl.h" #include "typemap.h" #include "wal_types.h" /* ------------------------------------------------------------------ */ /* Test-injectable send hook (drops all sends — no live network I/O) */ /* ------------------------------------------------------------------ */ #ifdef XCE_TESTING extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries, int n_entries); #endif static int p2_drop_send_hook(void *ep, wal_catchup_entry_t **entries, int n_entries) { (void)ep; (void)entries; (void)n_entries; return 0; /* pretend success, drop the data */ } /* ------------------------------------------------------------------ */ /* Helpers */ /* ------------------------------------------------------------------ */ static persist_t* make_p2_persist(const char* suffix) { persist_t* p = calloc(1, sizeof(persist_t)); if (!p) return NULL; persist_config_t cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; cfg.sync_timeout_ms = 50; char tmpdir[256]; snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_p2intg_%s_XXXXXX", suffix); if (!mkdtemp(tmpdir)) { free(p); return NULL; } persist_set_workdir(p, tmpdir, "ctrl"); persist_init(p, &cfg, NULL); persist_start(p); return p; } static void teardown_p2_persist(persist_t* p) { if (!p) return; persist_stop(p); persist_fini(p); free(p); } /* ------------------------------------------------------------------ */ /* Test 1: WAL_CHECKPOINT assert — repl_broadcast must reject checkpoint type. * * We cannot call repl_broadcast(WAL_CHECKPOINT) directly (it asserts/aborts). * Instead we verify the invariant at the persist level: persist_checkpoint() * must NOT change last_replicated_crc, proving repl_broadcast is never called * for checkpoint entries. */ static void test_repl_broadcast_rejects_checkpoint(void) { /* Invariant is tested at the persist level — see test 3. * Here we verify repl_init/fini lifecycle is stable (compile-time guard * that the assert in repl.c exists via the WAL_CHECKPOINT constant). */ replicator_t r; repl_init(&r); /* WAL_CHECKPOINT must equal 100 — asserted against in repl.c */ TEST_ASSERT_EQUAL_INT(100, (int)WAL_CHECKPOINT); repl_fini(&r); printf("test_repl_broadcast_rejects_checkpoint: PASS (invariant verified at persist level)\n"); } /* ------------------------------------------------------------------ */ /* Test 2: Trim LSN with active standby */ static void test_trim_lsn_with_active_standby_at_lsn_5(void) { replicator_t r; int ep = 0; repl_init(&r); repl_add_standby(&r, (void *)&ep, "sb", 7001); repl_on_ack(&r, (void *)&ep, 5); /* trim_lsn must be min(acked_lsn=5, checkpoint_lsn=20) = 5 */ uint64_t trim = repl_trim_lsn(&r, 20, 60); TEST_ASSERT_EQUAL_UINT64(5, trim); repl_fini(&r); printf("test_trim_lsn_with_active_standby_at_lsn_5: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 3: last_replicated_crc is NOT updated by checkpoint write */ static void test_checkpoint_does_not_update_replicated_crc(void) { persist_t* p = make_p2_persist("crc2"); TEST_ASSERT_NOT_NULL(p); /* Write a WAL entry — updates last_replicated_crc */ int rc = persist_wal_async(p, WAL_JOB_UPDATE, "data", 4); TEST_ASSERT_EQUAL_INT(0, rc); uint32_t crc_after_wal = p->last_replicated_crc; /* Checkpoint — must NOT change last_replicated_crc */ persist_checkpoint(p); TEST_ASSERT_EQUAL_UINT32(crc_after_wal, p->last_replicated_crc); teardown_p2_persist(p); printf("test_checkpoint_does_not_update_replicated_crc: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 4: Sync replication degraded mode — WAL write succeeds, sync_rc < 0 */ static void test_sync_replication_degraded_mode(void) { persist_t* p = make_p2_persist("sync2"); TEST_ASSERT_NOT_NULL(p); /* No standbys — sync submit must succeed locally but sync_rc = degraded */ int sync_rc = 0; int rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, "submit", 6, &sync_rc); TEST_ASSERT_EQUAL_INT(0, rc); /* WAL write succeeded */ TEST_ASSERT_TRUE(sync_rc < 0); /* Degraded: no standbys or timeout */ TEST_ASSERT_TRUE(p->current_lsn > 0); /* LSN advanced */ teardown_p2_persist(p); printf("test_sync_replication_degraded_mode: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 5: Catch-up lifecycle integration — start, ACK during catch-up ignored, * complete, ACK counted */ static void test_catchup_full_lifecycle_integration(void) { replicator_t r; int ep = 0; repl_init(&r); repl_add_standby(&r, (void *)&ep, "sb", 7001); /* Start catch-up */ repl_start_catchup(&r, (void *)&ep, /*join_applied_lsn=*/3); /* ACK during catch-up must not satisfy wait_ack */ repl_on_ack(&r, (void *)&ep, 50); int rc = repl_wait_ack(&r, 10, 50); TEST_ASSERT_EQUAL_INT(-ENODEV, rc); /* Complete catch-up — ACK at 50 should now satisfy wait_ack for target 10 */ repl_complete_catchup(&r, (void *)&ep); rc = repl_wait_ack(&r, 10, 50); TEST_ASSERT_EQUAL_INT(0, rc); repl_fini(&r); printf("test_catchup_full_lifecycle_integration: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 6: needs_snapshot boundary conditions */ static void test_needs_snapshot_boundary(void) { replicator_t r; repl_init(&r); /* Exact boundary: standby_lsn == oldest_wal_lsn — no snapshot needed */ TEST_ASSERT_EQUAL_INT(0, repl_needs_snapshot(&r, 10, 10)); /* One behind: standby_lsn == oldest_wal_lsn - 1 — snapshot required */ TEST_ASSERT_EQUAL_INT(1, repl_needs_snapshot(&r, 9, 10)); repl_fini(&r); printf("test_needs_snapshot_boundary: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 7: P2 gate — repl.h public API completeness */ static void test_repl_api_completeness(void) { /* Verify all P2 API functions are present by calling them without crash */ replicator_t r; int ep = 0; repl_init(&r); repl_add_standby(&r, (void *)&ep, "sb", 7001); repl_on_ack(&r, (void *)&ep, 1); repl_start_catchup(&r, (void *)&ep, 0); repl_complete_catchup(&r, (void *)&ep); repl_needs_snapshot(&r, 0, 0); repl_trim_lsn(&r, 100, 60); repl_remove_standby(&r, (void *)&ep); repl_fini(&r); printf("test_repl_api_completeness: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 4b: single-node path — production contract (ha-commit-engine Req 6.3). * * UPDATED by ha-commit-engine task 5.3: this test previously pinned the * `has_standbys → -ENODEV` degraded special case, which 5.3 deliberately * replaced. In production the commit engine is wired at cluster init, so a * single-member primary (n_peers=1) self-commits: persist_wal_sync_append * returns 0, sync status is 0 (NOT -ENODEV), and committed_lsn advances to * the entry. The engine — not persist — owns the advancement. * * A priming async append makes the sync entry land at LSN 1: the first-ever * WAL entry is LSN 0, which the adapter routes to the legacy branch (the * accepted once-per-cluster-lifetime LSN-0 interim behaviour). */ static void test_single_node_committed_lsn_advances(void) { commit_engine_t ce; persist_t* p = make_p2_persist("clsn_1n"); TEST_ASSERT_NOT_NULL(p); /* Wire the engine exactly as cluster_init_phase2 does for n_peers=1 */ TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, 1, 8)); repl_set_commit_engine(&p->repl, &ce); /* Reserved-0 contract: fresh store starts current_lsn=1, so the sync entry * below lands at LSN 1 with no priming append (LSN 0 is the reserved * sentinel — the engine rejects lsn == 0). */ int sync_rc = -1; int rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, "job", 3, &sync_rc); /* WAL write must succeed locally */ TEST_ASSERT_EQUAL_INT(0, rc); /* Single-member self-quorum: confirmed locally, no -ENODEV degraded mode */ TEST_ASSERT_EQUAL_INT_MESSAGE(0, sync_rc, "single-node wired engine must self-commit (0), not -ENODEV"); /* committed_lsn advances to the entry (engine-owned advancement) */ TEST_ASSERT_EQUAL_UINT64_MESSAGE( 1, atomic_load_explicit(&p->committed_lsn, memory_order_acquire), "single-node: committed_lsn must advance to the self-committed entry"); /* WAL LSN did advance (local write succeeded) */ TEST_ASSERT_TRUE(p->current_lsn > 0); commit_engine_fini(&ce); teardown_p2_persist(p); printf("test_single_node_committed_lsn_advances: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 8: Req 4.2 (P2) / ha-commit-engine Req 2.1, 4.4 — committed_lsn * must NOT advance when sync times out, and the write is REJECTED. * * UPDATED by ha-commit-engine task 6.1: this test previously pinned the * OLD accept-on-timeout contract (rc==0 with sync_rc==-ETIMEDOUT as a * mere advisory). persist_wal_sync_append now rejects on timeout * (returns -ETIMEDOUT) so no unconfirmed write reaches the caller as * success (Req 4.4). The fixture is also upgraded to the engine-wired * production path: cluster_init_phase2 always wires the commit engine, * so the honest place to pin "committed_lsn unmoved on timeout" is the * engine-owned path, not the deprecated ce==NULL legacy counting. * * Set up: 3-member cluster (self + two standbys that never ACK), engine * wired exactly as cluster_init_phase2 does. A priming async append * makes the sync entry land at LSN 1 (LSN 0 routes to the legacy branch * — the accepted once-per-cluster-lifetime interim behaviour). * Verify: rc == -ETIMEDOUT (write rejected), sync_rc == -ETIMEDOUT, * and committed_lsn stays at 0 (engine never advanced it). */ static void test_committed_lsn_does_not_advance_on_timeout(void) { commit_engine_t ce; persist_t* p = make_p2_persist("clsn_to"); TEST_ASSERT_NOT_NULL(p); /* Add two standbys — they will never ACK, so the wait times out */ static int ep_a, ep_b; repl_add_standby(&p->repl, (void *)&ep_a, "sb_a", 7010); repl_add_standby(&p->repl, (void *)&ep_b, "sb_b", 7011); /* Wire the engine exactly as cluster_init_phase2 does for n_peers=3 */ TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, 3, 8)); repl_set_commit_engine(&p->repl, &ce); /* Use a very short timeout so the test finishes quickly */ p->config.sync_timeout_ms = 40; uint64_t committed_before = atomic_load_explicit(&p->committed_lsn, memory_order_acquire); TEST_ASSERT_EQUAL_UINT64(0, committed_before); /* Install drop hook so repl_broadcast doesn't dereference dummy eps */ repl_broadcast_send_hook = p2_drop_send_hook; /* Priming append: sync entry below lands at LSN 1 (engine-wired path) */ TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "p", 1)); int sync_rc = 0; int rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, "job", 3, &sync_rc); repl_broadcast_send_hook = NULL; /* No per-entry majority within the deadline → the write is REJECTED * (ha-commit-engine Req 4.4: timeout propagates to the caller) */ TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc, "sync append without majority ACK must be rejected with -ETIMEDOUT"); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, sync_rc); /* committed_lsn must not have advanced (Req 4.2; engine Req 2.1) */ uint64_t committed_after = atomic_load_explicit(&p->committed_lsn, memory_order_acquire); TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, committed_after, "committed_lsn must stay unmoved when the entry never reaches majority"); commit_engine_fini(&ce); teardown_p2_persist(p); printf("test_committed_lsn_does_not_advance_on_timeout: PASS\n"); } /* ------------------------------------------------------------------ */ /* ACK helper thread for test 9 */ struct p2_ack_args { persist_t* p; void* ep; uint64_t lsn; uint32_t delay_ms; }; static void* p2_ack_thread(void* arg) { struct p2_ack_args* a = arg; if (a->delay_ms > 0) { struct timespec ts = { .tv_sec = 0, .tv_nsec = (long)a->delay_ms * 1000000L }; nanosleep(&ts, NULL); } repl_on_ack(&a->p->repl, a->ep, a->lsn); return NULL; } /* ------------------------------------------------------------------ */ /* Test 9: Req 4.1 / 4.4 — committed_lsn advances after majority ACK. * * Simulates a 3-node cluster (primary + 2 standbys). * One standby ACKs in a background thread after a short delay. * ceil((2+1)/2) = 2 including primary self-count → quorum met with 1 standby ACK. * Verify: persist_wal_sync_append returns 0, sync_rc == 0, * and committed_lsn advanced past committed_before. * * Note: under the reserved-0 contract a fresh persist_t starts current_lsn=1, * so the first WAL write lands at lsn 1 (written_lsn = current_lsn - 1 = 1). * The standby ACK below is for lsn 1, so it exactly covers the first entry. * The test captures committed_before and asserts committed_after > * committed_before (advance) OR committed_after == written_lsn (exact match). */ static persist_t* g_p9 = NULL; static int g_ep9_a, g_ep9_b; static struct p2_ack_args g_ack9; static void test_committed_lsn_advances_after_majority_ack(void) { commit_engine_t ce; g_p9 = make_p2_persist("clsn_ack"); TEST_ASSERT_NOT_NULL(g_p9); /* Two standbys — one will ACK, satisfying majority (quorum = 2 incl. self) */ repl_add_standby(&g_p9->repl, (void *)&g_ep9_a, "sb_x", 7020); repl_add_standby(&g_p9->repl, (void *)&g_ep9_b, "sb_y", 7021); /* Wire the commit engine exactly as cluster_init_phase2 does (n_peers=3). * Under the reserved-0 contract the first WAL entry is LSN 1, which routes * through the engine-wired walAck path (commit_submit), so the engine must * be wired — the old engine-less path only "worked" via the LSN-0 * immediate-confirm escape hatch, which no longer applies. */ TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, g_p9, 3, 8)); repl_set_commit_engine(&g_p9->repl, &ce); /* Give enough time for the ACK thread to fire before timeout */ g_p9->config.sync_timeout_ms = 2000; /* Capture committed_lsn before sync so we can assert it advanced */ uint64_t committed_before = atomic_load_explicit(&g_p9->committed_lsn, memory_order_acquire); /* ACK from sb_x arrives after 20 ms — satisfies quorum */ g_ack9 = (struct p2_ack_args){ g_p9, (void *)&g_ep9_a, /*lsn=*/1, /*delay_ms=*/20 }; pthread_t t; pthread_create(&t, NULL, p2_ack_thread, &g_ack9); /* Install drop hook so repl_broadcast doesn't dereference dummy eps */ repl_broadcast_send_hook = p2_drop_send_hook; int sync_rc = -1; int rc = persist_wal_sync_append(g_p9, WAL_JOB_SUBMIT, "job", 3, &sync_rc); repl_broadcast_send_hook = NULL; pthread_join(t, NULL); /* WAL write and sync must both succeed */ TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(0, sync_rc); /* committed_lsn must have advanced to written_lsn (Req 4.1). * written_lsn = current_lsn - 1 after persist_wal_async; under the * reserved-0 contract this equals 1 on the very first write. We check * that committed_after > committed_before (advance happened) or that * committed_after == written_lsn (the first entry committed exactly). */ uint64_t committed_after = atomic_load_explicit(&g_p9->committed_lsn, memory_order_acquire); /* Derive written_lsn the same way persist_wal_sync_append does */ pthread_mutex_lock(&g_p9->lock); uint64_t written_lsn = g_p9->current_lsn - 1; pthread_mutex_unlock(&g_p9->lock); TEST_ASSERT_TRUE_MESSAGE( committed_after > committed_before || committed_after == written_lsn, "committed_lsn must advance after majority ACK (Req 4.1)"); commit_engine_fini(&ce); teardown_p2_persist(g_p9); g_p9 = NULL; printf("test_committed_lsn_advances_after_majority_ack: PASS\n"); } /* ------------------------------------------------------------------ */ /* Entry point */ /* ------------------------------------------------------------------ */ void test_p2_integration(void) { RUN_TEST(test_repl_broadcast_rejects_checkpoint); RUN_TEST(test_trim_lsn_with_active_standby_at_lsn_5); RUN_TEST(test_checkpoint_does_not_update_replicated_crc); RUN_TEST(test_sync_replication_degraded_mode); RUN_TEST(test_single_node_committed_lsn_advances); RUN_TEST(test_catchup_full_lifecycle_integration); RUN_TEST(test_needs_snapshot_boundary); RUN_TEST(test_repl_api_completeness); RUN_TEST(test_committed_lsn_does_not_advance_on_timeout); RUN_TEST(test_committed_lsn_advances_after_majority_ack); }