/* * test_persist_truncate.c — RED-phase tests for persist_truncate_wal_above(). * * Function does not exist yet (task 7.5). Will fail to link until implemented. * * Requirements: 5.6, 5.7 */ #define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #include #include #include "persist.h" #include "wal_types.h" #include "config.h" #include "commit.h" /* raft-apply-decoupling task 1.4: commit_engine_t, * commit_submit_async, commit_outcome_t */ /* ── Forward declaration of function-under-test (not yet implemented) ─────── */ extern int persist_truncate_wal_above(persist_t *p, uint64_t cutoff_lsn); /* ── Helpers ────────────────────────────────────────────────────────────────── */ #define WAL_SEG_FMT "%s/data/wal/seg-%016" PRIx64 ".wal" static persist_t *make_fresh_persist(const char *tmpdir) { 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.snap_keep_n = 2; persist_set_workdir(p, tmpdir, "trunc-test"); if (persist_init(p, &cfg, NULL) != 0) { free(p); return NULL; } if (persist_start(p) != 0) { persist_fini(p); free(p); return NULL; } return p; } static void teardown_persist(persist_t *p) { persist_stop(p); persist_fini(p); free(p); } /* Write a minimal wal_entry_hdr_t with the given lsn to an open fd at offset */ static int write_wal_entry(int fd, uint64_t lsn, off_t offset) { wal_entry_hdr_t hdr; memset(&hdr, 0, sizeof(hdr)); hdr.type = WAL_JOB_SUBMIT; hdr.lsn = lsn; hdr.payload_len = 0; hdr.crc = 0; hdr.prev_crc = 0; ssize_t n = pwrite(fd, &hdr, sizeof(hdr), offset); return (n == (ssize_t)sizeof(hdr)) ? 0 : -EIO; } /* Count .wal files in the WAL directory */ static int count_wal_segments(const char *wal_dir) { DIR *d = opendir(wal_dir); if (!d) return 0; int count = 0; struct dirent *e; while ((e = readdir(d)) != NULL) { size_t len = strlen(e->d_name); if (len > 4 && strcmp(e->d_name + len - 4, ".wal") == 0) count++; } closedir(d); return count; } /* ── Test 1: truncation at segment boundary deletes whole segment ──────────── */ static void test_truncate_at_segment_boundary_deletes_segment(void) { char tmpdir[] = "/tmp/xce-trunc-test-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); /* Write 3 WAL entries so we have a known LSN sequence */ int dummy = 0; persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); uint64_t lsn_before = p->current_lsn; /* Append a 4th entry — this represents an uncommitted entry above cutoff */ persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); /* Truncate above lsn_before-1 (the 3rd entry); 4th should be removed */ int rc = persist_truncate_wal_above(p, lsn_before - 1); TEST_ASSERT_EQUAL_INT(0, rc); /* After truncation, current_lsn (or applied_lsn) should reflect cutoff */ TEST_ASSERT_TRUE(p->current_lsn <= lsn_before); teardown_persist(p); /* Cleanup */ char wal_dir[256]; snprintf(wal_dir, sizeof(wal_dir), "%s/data/wal", tmpdir); printf("test_truncate_at_segment_boundary_deletes_segment: PASS\n"); } /* ── Test 2: truncation mid-segment scans to correct byte offset ───────────── */ static void test_truncate_mid_segment_at_correct_offset(void) { char tmpdir[] = "/tmp/xce-trunc-mid-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); int dummy = 0; /* Write 5 entries, truncate above entry 3 (lsn=2), entries 3-4 gone */ for (int i = 0; i < 5; i++) persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); uint64_t cutoff = 2; /* keep entries with lsn <= 2 */ int rc = persist_truncate_wal_above(p, cutoff); TEST_ASSERT_EQUAL_INT(0, rc); /* applied_lsn / current_lsn must be <= cutoff + 1 */ TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1); teardown_persist(p); printf("test_truncate_mid_segment_at_correct_offset: PASS\n"); } /* ── Test 3: idempotency — calling twice with same cutoff is safe ──────────── */ static void test_truncate_idempotent(void) { char tmpdir[] = "/tmp/xce-trunc-idem-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); int dummy = 0; for (int i = 0; i < 4; i++) persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); uint64_t cutoff = 1; int rc1 = persist_truncate_wal_above(p, cutoff); int rc2 = persist_truncate_wal_above(p, cutoff); TEST_ASSERT_EQUAL_INT(0, rc1); TEST_ASSERT_EQUAL_INT(0, rc2); /* current_lsn consistent after two calls */ TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1); teardown_persist(p); printf("test_truncate_idempotent: PASS\n"); } /* ── Test 4: applied_lsn updated atomically to cutoff_lsn ─────────────────── */ static void test_truncate_updates_applied_lsn(void) { char tmpdir[] = "/tmp/xce-trunc-lsn-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); int dummy = 0; for (int i = 0; i < 6; i++) persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); uint64_t cutoff = 3; int rc = persist_truncate_wal_above(p, cutoff); TEST_ASSERT_EQUAL_INT(0, rc); /* committed_lsn field must exist on persist_t after task 7.5 adds it. * For now, verify current_lsn reflects the truncation. */ TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1); teardown_persist(p); printf("test_truncate_updates_applied_lsn: PASS\n"); } /* ── Test 5: no-op when cutoff_lsn == applied_lsn (nothing to truncate) ───── */ static void test_truncate_noop_at_current_lsn(void) { char tmpdir[] = "/tmp/xce-trunc-noop-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); int dummy = 0; persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); /* Truncate AT current boundary — should be a no-op */ uint64_t cutoff = p->current_lsn - 1; int rc = persist_truncate_wal_above(p, cutoff); TEST_ASSERT_EQUAL_INT(0, rc); /* Nothing should change */ TEST_ASSERT_TRUE(p->current_lsn <= cutoff + 1); teardown_persist(p); printf("test_truncate_noop_at_current_lsn: PASS\n"); } /* ── Test: truncation resolves invalidated commit continuations (task 1.4) ── */ /* Minimal local recorder for commit_submit_async continuations -- mirrors * test_commit_resolve.c's continuation_recorder_t/recorder_fn (kept as a * separate file-static copy; no cross-TU sharing of test statics). */ typedef struct { _Atomic int calls; uint64_t last_lsn; commit_outcome_t last_outcome; int last_rc; } trunc_continuation_recorder_t; static void trunc_recorder_fn(void *ctx, uint64_t lsn, commit_outcome_t outcome, int rc) { trunc_continuation_recorder_t *r = (trunc_continuation_recorder_t *)ctx; r->last_lsn = lsn; r->last_outcome = outcome; r->last_rc = rc; atomic_fetch_add(&r->calls, 1); } /* raft-apply-decoupling task 1.4 (Req 1.3/1.4, 9.3/9.4): a real truncation * through persist_truncate_wal_above must resolve every commit continuation * registered ABOVE the new tail as FAILED -- those entries no longer exist * in the log and can never commit -- while a continuation registered AT OR * BELOW the new tail is left untouched (its entry is still safely in the * log and may still reach majority). * * Uses n_peers=3 with NO standbys registered so majority is permanently * unreachable (majority_candidate_locked's zero-fill keeps committed_lsn at * 0 regardless of local append/durable progress) -- every appended entry * stays genuinely pending, giving a stable LSN range to register * continuations against without racing the commit engine's own majority * evaluation. * * RED: before task 1.4, persist_truncate_wal_above never calls into the * commit engine, so all three continuations stay pending forever (calls == * 0 for all). */ static void test_truncate_above_resolves_pending_continuations_as_failed(void) { char tmpdir[] = "/tmp/xce-trunc-resolve-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); commit_engine_t ce; TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/8)); int dummy = 0; for (int i = 0; i < 5; i++) persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); TEST_ASSERT_EQUAL_UINT64_MESSAGE(6, p->current_lsn, "fixture: 5 entries appended, current_lsn must be 6"); trunc_continuation_recorder_t below = {0}; trunc_continuation_recorder_t above1 = {0}; trunc_continuation_recorder_t above2 = {0}; TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 2, trunc_recorder_fn, &below)); TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 4, trunc_recorder_fn, &above1)); TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 5, trunc_recorder_fn, &above2)); /* Act: truncate above lsn=3 -- entries 4 and 5 are removed. */ int rc = persist_truncate_wal_above(p, 3); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, p->current_lsn, "current_lsn must collapse to cutoff_lsn + 1"); /* Assert: continuations ABOVE the new tail resolve FAILED exactly once. * No vtaskmgr pool is wired on this engine (commit_set_pool never * called), so commit_resolve's dispatch falls back to the inline * fallback -- synchronous with the persist_truncate_wal_above call * above, no polling required. */ TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&above1.calls), "a continuation above the new tail must resolve on truncation"); TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED, (int)above1.last_outcome, "truncation must resolve as FAILED"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, above1.last_rc, "truncation must use the existing -ENOENT vocabulary already " "exercised in test_commit_resolve.c (task 1.2)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, above1.last_lsn, "the recorder must observe the lsn it was registered for"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&above2.calls), "a second continuation above the new tail must also resolve"); TEST_ASSERT_EQUAL_INT_MESSAGE((int)COMMIT_OUTCOME_FAILED, (int)above2.last_outcome, "truncation must resolve as FAILED"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, above2.last_rc, "truncation must use the existing -ENOENT vocabulary"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, above2.last_lsn, "the recorder must observe the lsn it was registered for"); /* Assert: a continuation AT OR BELOW the new tail is left untouched -- * its entry is still in the log and may still reach majority. */ TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&below.calls), "a continuation at/below the new tail must NOT be resolved by " "truncation -- only entries strictly above the new tail were " "removed from the log"); /* bug-1452: persist_stop() MUST run before commit_engine_fini() -- the * fsync_worker flusher thread is still live here and can be mid-flight * calling the durable-advance callback (commit.c's fsync_worker -> * commit_on_durable_advance -> commit_advance_locked) with NO lock held. * commit_engine_fini() destroys ce->lock/ce->cond and nulls ce->p; racing * that against a still-running flusher lets commit_advance_locked observe * ce->p go NULL between its own guard check and its dereference -> SEGV. * persist_stop() joins the flusher thread first, so by the time * commit_engine_fini() runs, nothing can call back into ce. */ persist_stop(p); commit_engine_fini(&ce); teardown_persist(p); printf("test_truncate_above_resolves_pending_continuations_as_failed: PASS\n"); } /* Regression: a no-op truncate (cutoff at/above the current tail) must NOT * fire the truncation wake -- a pending continuation for any lsn must stay * untouched, since nothing was actually removed from the log. */ static void test_truncate_noop_does_not_resolve_pending_continuations(void) { char tmpdir[] = "/tmp/xce-trunc-resolve-noop-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); commit_engine_t ce; TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/8)); int dummy = 0; persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); trunc_continuation_recorder_t rec = {0}; TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&ce, 2, trunc_recorder_fn, &rec)); /* Truncate AT the current boundary -- a no-op (nothing above cutoff). */ uint64_t cutoff = p->current_lsn - 1; int rc = persist_truncate_wal_above(p, cutoff); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&rec.calls), "a no-op truncate (nothing above cutoff_lsn) must not resolve any " "pending continuation"); /* bug-1452: stop the flusher thread before tearing down ce -- see the * comment in test_truncate_above_resolves_pending_continuations_as_failed. */ persist_stop(p); commit_engine_fini(&ce); teardown_persist(p); printf("test_truncate_noop_does_not_resolve_pending_continuations: PASS\n"); } /* ── Concurrent burst (raft-apply-decoupling task 1.5, Req 9.1-9.4) ──────── * Many continuations, registered from real threads racing each other for * ce->lock and split across the truncation cutoff, must ALL resolve exactly * once through the SAME real persist_truncate_wal_above detection path * exercised above: entries above the new tail FAILED/-ENOENT, entries at/ * below left untouched -- proving the bulk resolve holds under genuine * concurrent registration, not just the fixed handful the tests above use. */ #define TRUNC_BURST_ABOVE 6 #define TRUNC_BURST_BELOW 3 #define TRUNC_BURST_TOTAL (TRUNC_BURST_ABOVE + TRUNC_BURST_BELOW) typedef struct { commit_engine_t* ce; uint64_t lsn; int rc; trunc_continuation_recorder_t* rec; } trunc_burst_arg_t; static void* trunc_burst_register_thread(void* p) { trunc_burst_arg_t* a = (trunc_burst_arg_t*)p; a->rc = commit_submit_async(a->ce, a->lsn, trunc_recorder_fn, a->rec); return NULL; } static void test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once(void) { char tmpdir[] = "/tmp/xce-trunc-burst-XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); persist_t *p = make_fresh_persist(tmpdir); TEST_ASSERT_NOT_NULL(p); commit_engine_t ce; TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/TRUNC_BURST_TOTAL + 4)); int dummy = 0; for (int i = 0; i < TRUNC_BURST_TOTAL + 2; i++) persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL); /* Cutoff placed so lsn 1..TRUNC_BURST_BELOW sit at/below the new tail * and lsn (TRUNC_BURST_BELOW+1)..TRUNC_BURST_TOTAL sit above it. */ const uint64_t cutoff = TRUNC_BURST_BELOW; trunc_continuation_recorder_t recs[TRUNC_BURST_TOTAL] = {0}; trunc_burst_arg_t args[TRUNC_BURST_TOTAL]; pthread_t tids[TRUNC_BURST_TOTAL]; /* Register the whole burst -- below-cutoff and above-cutoff LSNs alike * -- CONCURRENTLY from real threads. */ for (int i = 0; i < TRUNC_BURST_TOTAL; i++) { args[i].ce = &ce; args[i].lsn = (uint64_t)(i + 1); args[i].rc = -9999; args[i].rec = &recs[i]; TEST_ASSERT_EQUAL_INT( 0, pthread_create(&tids[i], NULL, trunc_burst_register_thread, &args[i])); } for (int i = 0; i < TRUNC_BURST_TOTAL; i++) { pthread_join(tids[i], NULL); } for (int i = 0; i < TRUNC_BURST_TOTAL; i++) { TEST_ASSERT_EQUAL_INT_MESSAGE( 0, args[i].rc, "every concurrent registration ahead of the truncation must succeed"); } TEST_ASSERT_EQUAL_INT(TRUNC_BURST_TOTAL, ce.inflight_waits); /* Act: a single real truncation above cutoff. The burst above is * already fully registered by the time this fires -- this proves the * BULK resolve path handles a registry populated by real concurrent * registrants correctly, not just the fixed three-continuation case * above. */ int rc = persist_truncate_wal_above(p, cutoff); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT64(cutoff + 1, p->current_lsn); for (int i = 0; i < TRUNC_BURST_TOTAL; i++) { uint64_t lsn = (uint64_t)(i + 1); if (lsn > cutoff) { TEST_ASSERT_EQUAL_INT_MESSAGE( 1, atomic_load(&recs[i].calls), "every continuation above the new tail must resolve exactly " "once on truncation, even when the whole burst was " "registered concurrently"); TEST_ASSERT_EQUAL_INT_MESSAGE( (int)COMMIT_OUTCOME_FAILED, (int)recs[i].last_outcome, "truncation must resolve as FAILED"); TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, recs[i].last_rc, "truncation must use the existing -ENOENT vocabulary"); TEST_ASSERT_EQUAL_UINT64(lsn, recs[i].last_lsn); } else { TEST_ASSERT_EQUAL_INT_MESSAGE( 0, atomic_load(&recs[i].calls), "a continuation at/below the new tail must NOT be resolved " "by truncation, even under concurrent registration"); } } TEST_ASSERT_EQUAL_INT_MESSAGE( TRUNC_BURST_BELOW, ce.inflight_waits, "only the above-cutoff admission slots are released by truncation; " "the below-cutoff ones remain pending"); /* bug-1452: stop the flusher thread before tearing down ce -- see the * comment in test_truncate_above_resolves_pending_continuations_as_failed. */ persist_stop(p); commit_engine_fini(&ce); teardown_persist(p); printf("test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once: " "PASS\n"); } /* ── Test group entry point ────────────────────────────────────────────────── */ void test_persist_truncate(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_truncate_at_segment_boundary_deletes_segment); RUN_TEST(test_truncate_mid_segment_at_correct_offset); RUN_TEST(test_truncate_idempotent); RUN_TEST(test_truncate_updates_applied_lsn); RUN_TEST(test_truncate_noop_at_current_lsn); RUN_TEST(test_truncate_above_resolves_pending_continuations_as_failed); RUN_TEST(test_truncate_noop_does_not_resolve_pending_continuations); RUN_TEST(test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once); }