From cc83caf5a0e75c6dc11d4dfc9394a95ac20119c6 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 18 Jun 2026 06:06:55 +0000 Subject: [PATCH] feat(ha-commit-durability): demand-driven group-commit flusher + durable_lsn watermark (task 2.1) --- .kiro/specs/ha-commit-durability/tasks.md | 3 +- src/controller/persist.c | 294 ++++++++++++++++-- .../testsuite/test_commit_durability.c | 266 ++++++++++++++++ src/include/persist.h | 31 ++ 4 files changed, 567 insertions(+), 27 deletions(-) diff --git a/.kiro/specs/ha-commit-durability/tasks.md b/.kiro/specs/ha-commit-durability/tasks.md index 72a71336..0f36ad29 100644 --- a/.kiro/specs/ha-commit-durability/tasks.md +++ b/.kiro/specs/ha-commit-durability/tasks.md @@ -24,7 +24,7 @@ - _Depends: 1.1, 1.2_ - [ ] 2. Durable commit mechanism (acknowledgment ⇒ majority persistence) -- [ ] 2.1 Introduce durable-LSN tracking via a group-commit flusher +- [x] 2.1 Introduce durable-LSN tracking via a group-commit flusher - Repurpose the background fsync thread from a timer into a demand-driven flusher: an append signals it; it performs one `fdatasync` over all pending appends, advances a durable-LSN watermark, and wakes the commit evaluation. - Guard against lost wakeups: the signal is level-triggered on the appended-vs-synced gap so an entry is never considered durable before its `fdatasync` (gating item). - Observable: a unit test shows one `fdatasync` advances the durable-LSN past several appended LSNs under load, and the durable-LSN never exceeds the synced tip. @@ -123,3 +123,4 @@ - 1.1: Config plumbing routes through `myconf_t.wal_sync_ms` + `persist_config_set_wal_sync_ms()` (0-sentinel = keep `DEFAULT_WAL_SYNC_MS`), NOT a new config.h constant (design File Structure Plan keeps `DEFAULT_WAL_SYNC_MS`). The override is applied at BOTH `persist_init` sites — `myapp_run` step-6 (production path) and `cluster_init_phase2` (guarded by `wal_path[0]=='\0'`, skipped in production). They are mutually exclusive; apply any new persist_config override at both sites or it is silently dropped. Pre-existing latent bug (do NOT fix here): `--sync-timeout-ms` is applied only in phase2, so it is ignored on the production startup path — relevant if a later task relies on `sync_timeout_ms`. - 1.2: Test seam = `persist_test_durable_lsn_on_disk(persist_t*)` (decl `persist.h`, impl `persist.c`, all `#ifdef XCE_TESTING`) — re-opens WAL segment files from disk and CRC-validates the chain (read-only, state-free) to return the highest fully-persisted LSN; new test `testsuite/test_commit_durability.c` (registered in `meson.build` + `controller_test.c`). Production `xce_controller` links without it (proven via `nm`). DRY NOTE: ~115 lines of segment-scan/CRC-chain logic are duplicated from the static `replay_wal`; tasks 3.1 (recovery verdict) and 3.4 (durable floor from WAL tip) touch the SAME scan/CRC logic — consider extracting a shared read-only `wal_scan_durable_lsn`-style helper there and routing this seam through it (avoid a third copy). `XCE_TESTING` is defined only for `controller_unit_test` (`src/controller/meson.build:167`). - 1.3: Deterministic mode = `bash src/tools/test-commit-durability.sh --deterministic` proves the durability GAP via strace, NOT joblist data-loss. SCOPE CLARIFICATION (review-ratified): single-host SIGKILL committed-job loss is physically NON-deterministic — `persist_wal_async` pwrites inline (page cache survives process death) and standbys fdatasync-before-WalAck (persist.c:515), so a committed job is always durable off the primary; a data-loss run flips GREEN/RED/RED. The deterministic slice: with `--wal-sync-ms 30000` the primary acks "jobsubmit ok" with ZERO primary fdatasync in the submit→ack window (exec-wrap `strace -f -tt -e trace=fdatasync,fsync`; PTRACE_ATTACH `-p` is yama-blocked). RED = in-window primary syncs==0 (15/15 across 5 runs; primary=0 vs standby=2). This is a valid RED→GREEN gate: the task-2.x fsync-before-ack fix makes the primary count ≥1 ⇒ GREEN. Original racy soak retained as the default mode. Boundary: only `src/tools/test-commit-durability.sh` (flag plumbed via deploy-ha.sh start.sh `$*` passthrough — no C edits). For task 5.2's E2E, prefer this strace gap-proof over joblist-loss for a deterministic durability assertion. +- 2.1: `persist_t.durable_lsn` (`_Atomic uint64_t`, accessor `persist_durable_lsn()`) = prefix watermark advanced ONLY after a successful `fdatasync`. `fsync_worker` is now a demand-driven group-commit flusher: `persist_wal_async` pwrites under `p->lock`, releases, then signals `fsync_cond`. Lost-wakeup guard is LEVEL-triggered on `flusher_work_pending_locked` (`durable_lsn < current_lsn-1`), re-checked under `p->lock` before wait and after each flush (reviewer-confirmed sound). Invariants: monotonic, `durable_lsn <= current_lsn-1`, clamped on truncate (`clamp_durable_lsn_locked` at both truncate paths). fail-loud rc-gate: `durable_lsn` does NOT advance on `fdatasync` error (all 4 sites: flusher main/fallback, `persist_stop`, `persist_flush`) — logs ERROR. Test fault-injection seam: `#ifdef XCE_TESTING` toggle `persist_test_set_force_fsync_fail()` (production-excluded, nm-verified). NOT yet consumed by commit (that is 2.2). FOLLOW-UP (non-blocking, out of scope for 2.1): bound the flusher retry on SUSTAINED fdatasync failure (backoff / rate-limited logging) — currently tight-spins (releases `p->lock` each iter, so liveness OK, but CPU/log burn on a degrading disk). NOTE: a HANGING (not erroring) fdatasync blocks appends since it runs under `p->lock` — pre-existing, not 2.1's boundary. diff --git a/src/controller/persist.c b/src/controller/persist.c index 8e283b40..555d38f9 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -167,41 +167,196 @@ static int rotate_segment(persist_t* p) } /* ------------------------------------------------------------------ */ -/* Dedicated fsync pthread (fires every wal_sync_ms milliseconds) */ +/* Demand-driven group-commit flusher (ha-commit-durability task 2.1) */ +/* */ +/* Repurposes the old fixed-interval fsync timer into a flusher driven */ +/* by the append-vs-synced gap. On each pass it batches one fdatasync */ +/* over all entries appended since the last sync, advances durable_lsn */ +/* to the synced tip, and wakes the commit evaluation. --wal-sync-ms */ +/* now bounds the MAX batching latency (a no-further-append entry is */ +/* still synced within the cadence), not a durability gap. */ +/* */ +/* LOST-WAKEUP / CORRECTNESS REASONING (gating item): */ +/* */ +/* The "work pending" condition is LEVEL-triggered on two persistent */ +/* variables, NOT an edge: pending ⇔ durable_lsn < current_lsn - 1 */ +/* (flusher_work_pending below, evaluated under p->lock). Because it */ +/* is a comparison of stored state — not a transient flag consumed on */ +/* read — a signal that races the flusher's wait can never be lost: */ +/* */ +/* - persist_wal_async advances current_lsn under p->lock, THEN */ +/* signals fsync_cond. If the flusher is mid-flush when the append */ +/* lands, current_lsn has already moved past the tip the flusher */ +/* captured, so after that flush durable_lsn < current_lsn - 1 */ +/* STILL holds; the flusher re-checks the level under p->lock on */ +/* its next loop iteration and flushes again (it does NOT wait */ +/* first). The missed signal is therefore harmless — the level */ +/* itself carries the pending work forward. */ +/* - If the append's signal arrives while the flusher sits in */ +/* pthread_cond_wait, it wakes it. If it arrives in the tiny window */ +/* after the flusher decided "no work" but before it re-enters the */ +/* wait, the flusher re-evaluates the level under p->lock before */ +/* waiting (we re-check inside the fsync_mutex-held region by */ +/* reading current_lsn/durable_lsn under p->lock first), so it */ +/* will not block while work is pending. */ +/* */ +/* "NEVER durable before its fdatasync": the flusher snapshots */ +/* tip = current_lsn - 1 AND calls fdatasync WHILE HOLDING p->lock. */ +/* persist_wal_async also pwrites under p->lock, so no append can */ +/* extend the segment between the tip snapshot and the fdatasync — by */ +/* the time fdatasync returns, every byte for LSNs <= tip is on stable */ +/* storage. durable_lsn is only then advanced to tip (monotonic max). */ +/* Releasing p->lock and broadcasting commit-eval happens after the */ +/* advance, so a WAIT caller that observes durable_lsn >= its LSN is */ +/* guaranteed the bytes are synced. */ +/* */ +/* Lock discipline: p->lock and fsync_mutex are NEVER held together */ +/* (fsync_mutex guards only the cond wait; the flush section takes */ +/* p->lock alone). This keeps fdatasync off any lock the iomgr / */ +/* heartbeat path needs and preserves liveness (Req 1.6). */ /* ------------------------------------------------------------------ */ +/* Test-only fault-injection probe (ha-commit-durability task 2.1 error path). + * Returns true when the test has forced fdatasync to be treated as failed. + * Compiled to a constant `false` in production (no -DXCE_TESTING), so the + * fail-loud gate below collapses to the pure rc != 0 check with zero overhead + * and no dependency on the test-only struct field. */ +static inline bool flusher_fsync_forced_fail(persist_t* p) +{ +#ifdef XCE_TESTING + return atomic_load_explicit(&p->test_force_fsync_fail, + memory_order_acquire) != 0; +#else + (void)p; + return false; +#endif +} + +/* Returns true iff there is un-synced appended data (level predicate). + * MUST be called with p->lock held. current_lsn - 1 is the appended tip + * (current_lsn is the NEXT lsn to assign); LSN 0 reserved, so a fresh + * store has current_lsn == 1 → tip 0 → no pending work. */ +static bool flusher_work_pending_locked(persist_t* p) +{ + uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0; + uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire); + return dur < tip; +} + +/* Perform one group-commit flush: snapshot the appended tip, fdatasync the + * live segment, and advance durable_lsn to the synced tip. Takes p->lock + * for the whole snapshot+sync+advance so no concurrent append can extend + * the segment between the tip read and the fdatasync. Returns the new + * durable_lsn (unchanged if there was nothing to sync). */ +static uint64_t flusher_flush_once(persist_t* p) +{ + pthread_mutex_lock(&p->lock); + + uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0; + uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire); + + if (dur >= tip || p->wal_fd < 0) { + /* Nothing new to sync (or no segment yet). Still flush the fd for a + * clean-shutdown / fallback pass, but do not advance the watermark. + * The watermark is already <= tip here, so a failed fallback sync + * cannot make it claim more than is durable — but we still surface + * the error loudly so a degrading disk is never silent (Req 1.1). */ + if (p->wal_fd >= 0) { + int rc = fdatasync(p->wal_fd); + if (rc != 0 || flusher_fsync_forced_fail(p)) { + int e = (rc != 0) ? errno : EIO; + ERROR("persist: flusher fallback fdatasync FAILED (errno=%d %s)" + " — durable_lsn held at %" PRIu64 "; WAL not advanced", + e, strerror(e), dur); + } + } + pthread_mutex_unlock(&p->lock); + return dur; + } + + /* fdatasync under p->lock: appends also hold p->lock to pwrite, so the + * segment cannot grow between this tip snapshot and the sync. Every byte + * for LSNs <= tip is therefore on stable storage when fdatasync returns + * SUCCESSFULLY. */ + int rc = fdatasync(p->wal_fd); + + /* FAIL-LOUD GATE (Req 1.1/1.2): advance durable_lsn ONLY on a successful + * sync. On EIO/ENOSPC the bytes are NOT on stable storage, so leaving the + * watermark unchanged keeps the non-durable LSNs from being counted toward + * the commit majority. We do NOT retry into an advanced watermark — the + * level predicate stays pending and the next pass retries the sync. */ + if (rc != 0 || flusher_fsync_forced_fail(p)) { + int e = (rc != 0) ? errno : EIO; + ERROR("persist: flusher fdatasync FAILED (errno=%d %s) — durable_lsn " + "held at %" PRIu64 " (tip=%" PRIu64 " NOT made durable); commit " + "majority will not count these LSNs", + e, strerror(e), dur, tip); + pthread_mutex_unlock(&p->lock); + return dur; + } + + /* Monotonic advance to the synced tip — never past it (the invariant + * durable_lsn <= current_lsn - 1 is preserved because tip == current_lsn-1 + * was read under this same lock). One fdatasync advances durable_lsn past + * however many LSNs were appended since the last sync (batching). */ + if (tip > dur) + atomic_store_explicit(&p->durable_lsn, tip, memory_order_release); + + pthread_mutex_unlock(&p->lock); + return tip; +} + static void* fsync_worker(void* arg) { persist_t* p = arg; - /* Sleep the configured interval via a timed condvar wait rather than an - * uninterruptible nanosleep, so persist_stop wakes us immediately instead - * of blocking its pthread_join for up to a full wal_sync_ms (5s in tests - * that set a slow interval — was the controller-suite timeout cause). */ pthread_mutex_lock(&p->fsync_mutex); while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) { - long ms = (long)p->config.wal_sync_ms; - struct timespec dl; - - clock_gettime(CLOCK_REALTIME, &dl); - dl.tv_sec += ms / 1000; - dl.tv_nsec += (ms % 1000) * 1000000L; - if (dl.tv_nsec >= 1000000000L) { - dl.tv_sec += 1; - dl.tv_nsec -= 1000000000L; - } - pthread_cond_timedwait(&p->fsync_cond, &p->fsync_mutex, &dl); - - /* Stop requested during the wait: exit before the next fdatasync. */ - if (atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) - break; - - /* fdatasync under p->lock — never held together with fsync_mutex. */ - pthread_mutex_unlock(&p->fsync_mutex); + /* Re-evaluate the level predicate BEFORE waiting so a signal that + * raced in just before we entered the wait is not needed for + * correctness — if work is already pending we skip the wait entirely + * and flush. (Reads current_lsn/durable_lsn under p->lock.) */ pthread_mutex_lock(&p->lock); - if (p->wal_fd >= 0) - fdatasync(p->wal_fd); + bool pending = flusher_work_pending_locked(p); pthread_mutex_unlock(&p->lock); + + if (!pending) { + /* No pending work: wait for an append-signal OR the cadence + * deadline (bounded fallback flush so a lone no-further-append + * entry is still synced within wal_sync_ms — Req 3.1). */ + long ms = (long)p->config.wal_sync_ms; + struct timespec dl; + clock_gettime(CLOCK_REALTIME, &dl); + dl.tv_sec += ms / 1000; + dl.tv_nsec += (ms % 1000) * 1000000L; + if (dl.tv_nsec >= 1000000000L) { + dl.tv_sec += 1; + dl.tv_nsec -= 1000000000L; + } + pthread_cond_timedwait(&p->fsync_cond, &p->fsync_mutex, &dl); + + if (atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) + break; + } + + /* Flush section: never hold fsync_mutex across the fdatasync (the + * flush takes p->lock alone). Release fsync_mutex, flush, re-acquire. + * After the flush we loop and re-check the level — if a new append + * landed during the flush, the level is still pending and we flush + * again without needing its signal (lost-wakeup-safe). */ + pthread_mutex_unlock(&p->fsync_mutex); + flusher_flush_once(p); + + /* COMMIT-EVAL WAKE (ordering fixed here; consumer wired by task 2.2): + * durable_lsn is advanced INSIDE flusher_flush_once (atomic store with + * release) BEFORE we return here. A WAIT caller's commit evaluation + * reads durable_lsn via an acquire load, so once it is signalled it + * observes the advanced watermark. Task 2.2 adds the explicit + * commit-eval condvar broadcast at THIS point (after the durable_lsn + * advance) — the "advance THEN wake" ordering this loop establishes is + * what makes that wake lost-wakeup-safe. Nothing consumes durable_lsn + * for commit yet, so no wake is needed in 2.1. */ + pthread_mutex_lock(&p->fsync_mutex); } pthread_mutex_unlock(&p->fsync_mutex); @@ -255,6 +410,11 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) atomic_store(&p->wal_record_count, 0); atomic_store(&p->checkpoint_in_progress, 0); + /* Durable watermark starts at 0 ("nothing fdatasync'd"): mirrors the + * LSN-0-reserved committed_lsn convention. It advances only on the + * flusher thread after a successful fdatasync (ha-commit-durability 2.1). */ + atomic_store(&p->durable_lsn, 0); + /* Create WAL and snapshot directories */ if (mkdir_p(p->wal_path, 0700) != 0) return -errno; if (mkdir_p(p->snap_path, 0700) != 0) return -errno; @@ -317,7 +477,30 @@ void persist_stop(persist_t* p) pthread_mutex_lock(&p->lock); if (p->wal_fd >= 0) { - fdatasync(p->wal_fd); + /* Final synchronous flush-or-exit on stop: fdatasync everything + * appended so far and advance durable_lsn to the synced tip so a + * post-stop read reflects the durable state (the flusher thread is + * already joined and will not advance it again). + * + * FAIL-LOUD GATE (Req 1.1/1.2): advance the watermark ONLY when the + * final sync SUCCEEDS — on EIO/ENOSPC the bytes are not on stable + * storage, so durable_lsn must stay where it is and the failure is + * surfaced loudly. */ + int rc = fdatasync(p->wal_fd); + if (rc == 0 && !flusher_fsync_forced_fail(p)) { + uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0; + uint64_t dur = atomic_load_explicit(&p->durable_lsn, + memory_order_acquire); + if (tip > dur) + atomic_store_explicit(&p->durable_lsn, tip, + memory_order_release); + } else { + int e = (rc != 0) ? errno : EIO; + ERROR("persist: final fdatasync on stop FAILED (errno=%d %s) — " + "durable_lsn left unchanged; appended-but-unsynced entries " + "are NOT durable", + e, strerror(e)); + } close(p->wal_fd); p->wal_fd = -1; } @@ -412,6 +595,17 @@ int persist_wal_async(persist_t* p, uint32_t type, pthread_mutex_unlock(&p->lock); + /* Signal the group-commit flusher (ha-commit-durability 2.1): current_lsn + * advanced and the entry's bytes are pwritten, so durable_lsn < tip now + * holds (the level predicate). The signal is sent AFTER releasing p->lock + * and is level-triggered — if it races the flusher and is "lost", the + * flusher re-checks durable_lsn < current_lsn-1 under p->lock and flushes + * anyway (see fsync_worker). We do NOT fdatasync inline on this path; the + * flusher batches the sync across all pending appends. */ + pthread_mutex_lock(&p->fsync_mutex); + pthread_cond_signal(&p->fsync_cond); + pthread_mutex_unlock(&p->fsync_mutex); + /* WAL record-count checkpoint trigger: schedule an async checkpoint once * the record count reaches the configured threshold. The fetch_add is * outside the lock — atomic_fetch_add is sufficient for the counter; the @@ -532,11 +726,43 @@ int persist_flush(persist_t* p) { pthread_mutex_lock(&p->lock); int rc = (p->wal_fd >= 0) ? fdatasync(p->wal_fd) : 0; + + /* persist_flush is a synchronous durability point (tests, clean shutdown): + * it fdatasync'd everything appended so far, so advance durable_lsn to the + * synced tip too. Without this, an explicit-flush-only path (slow flusher + * cadence) would leave durable_lsn lagging even though the bytes are on + * disk. Monotonic max; preserves durable_lsn <= current_lsn - 1 (tip read + * under the same p->lock). Gated on rc == 0 (fail-loud, Req 1.1/1.2): a + * failed flush must not advance the watermark. */ + if (rc == 0 && !flusher_fsync_forced_fail(p)) { + uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0; + uint64_t dur = atomic_load_explicit(&p->durable_lsn, + memory_order_acquire); + if (tip > dur) + atomic_store_explicit(&p->durable_lsn, tip, memory_order_release); + } else if (rc != 0 || flusher_fsync_forced_fail(p)) { + int e = (rc != 0) ? errno : EIO; + ERROR("persist: persist_flush fdatasync FAILED (errno=%d %s) — " + "durable_lsn not advanced; flushed bytes are NOT durable", + e, strerror(e)); + } pthread_mutex_unlock(&p->lock); return rc; } +uint64_t persist_durable_lsn(persist_t* p) +{ + return atomic_load_explicit(&p->durable_lsn, memory_order_acquire); +} + #ifdef XCE_TESTING +void persist_test_set_force_fsync_fail(persist_t* p, int fail) +{ + if (p) + atomic_store_explicit(&p->test_force_fsync_fail, fail ? 1 : 0, + memory_order_release); +} + /* ------------------------------------------------------------------ * * Durability-verification test seam (ha-commit-durability Req 3.2/3.3)* * * @@ -1823,6 +2049,20 @@ bool worker_should_retain(const xce_worker_t* w, const persist_config_t* cfg) * Idempotent: calling twice with same cutoff_lsn returns 0 both times. * ============================================================ */ +/* Clamp durable_lsn down to `tip` when the log tip recedes below it (truncate + * / divergent-suffix discard). Without this, durable_lsn could exceed + * current_lsn - 1 after a truncate, violating its invariant and letting the + * commit majority count self at an LSN whose bytes were just removed. MUST be + * called with p->lock held (mirrors the committed_lsn clamp in the truncate + * paths). durable_lsn only ever decreases here — the sole exception to its + * monotonic-non-decreasing rule (ha-commit-durability 2.1 State Management). */ +static void clamp_durable_lsn_locked(persist_t* p, uint64_t tip) +{ + uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire); + if (dur > tip) + atomic_store_explicit(&p->durable_lsn, tip, memory_order_release); +} + int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn) { pthread_mutex_lock(&p->lock); @@ -1902,6 +2142,7 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn) p->last_crc = WAL_CRC_NONE; atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release); + clamp_durable_lsn_locked(p, cutoff_lsn); pthread_mutex_unlock(&p->lock); return 0; } @@ -1980,6 +2221,7 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn) p->current_lsn = cutoff_lsn + 1; p->last_crc = new_last_crc; atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release); + clamp_durable_lsn_locked(p, cutoff_lsn); pthread_mutex_unlock(&p->lock); return 0; diff --git a/src/controller/testsuite/test_commit_durability.c b/src/controller/testsuite/test_commit_durability.c index 6c3b416a..33c4df55 100644 --- a/src/controller/testsuite/test_commit_durability.c +++ b/src/controller/testsuite/test_commit_durability.c @@ -1,4 +1,6 @@ #define _GNU_SOURCE +#include +#include #include #include #include @@ -172,6 +174,265 @@ static void test_durable_floor_survives_process_loss(void) printf("test_durable_floor_survives_process_loss: PASS\n"); } +/* ================================================================== * + * task 2.1 — durable-LSN tracking via a group-commit flusher * + * * + * These tests pin the invariants of persist_t.durable_lsn, the * + * prefix watermark advanced by the demand-driven flusher (the * + * repurposed fsync_worker): * + * - durable_lsn never exceeds the synced tip (<= current_lsn - 1); * + * - durable_lsn is monotonic non-decreasing; * + * - one fdatasync advances durable_lsn past MULTIPLE appended LSNs * + * (batching / group-commit); * + * - durable_lsn advances ONLY after a successful fdatasync; * + * - under concurrent append load the invariants always hold. * + * ------------------------------------------------------------------ */ + +/* Spin until the flusher catches durable_lsn up to the appended tip, + * or a generous deadline elapses. Returns the observed durable_lsn. */ +static uint64_t wait_durable_reaches(persist_t* p, uint64_t target_lsn, + int max_ms) +{ + for (int i = 0; i < max_ms; i++) { + uint64_t d = persist_durable_lsn(p); + if (d >= target_lsn) + return d; + usleep(1000); /* 1 ms */ + } + return persist_durable_lsn(p); +} + +/* ------------------------------------------------------------------ * + * Test 4: a fresh store has durable_lsn == 0 (nothing synced yet). * + * ------------------------------------------------------------------ */ +static void test_durable_lsn_initial_zero(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_durlsn0_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + persist_t* p = make_durability_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + TEST_ASSERT_EQUAL_UINT64(0, persist_durable_lsn(p)); + + teardown_durability_persist(p); + printf("test_durable_lsn_initial_zero: PASS\n"); +} + +/* ------------------------------------------------------------------ * + * Test 5 (core / batching, Req 1.1): several appends, then the * + * demand-driven flusher advances durable_lsn past ALL of them with a * + * single fdatasync — and never past the appended tip. Uses the fast * + * cadence so the append-signal (not the timer) drives the flush, but * + * the slow-cadence helper would converge too via the bounded fallback.* + * ------------------------------------------------------------------ */ +static void test_durable_lsn_batches_multiple_appends(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_durlsn1_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + persist_t* p = make_durability_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + /* LSN 0 reserved → these take 1,2,3,4,5; current_lsn ends at 6. */ + for (int i = 0; i < 5; i++) + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "x", 1)); + + uint64_t tip = p->current_lsn - 1; /* 5 */ + TEST_ASSERT_EQUAL_UINT64(5, tip); + + uint64_t d = wait_durable_reaches(p, tip, 5000); + + /* The flusher advanced durable_lsn past several appended LSNs with + * one fdatasync (batching), reaching the synced tip. */ + TEST_ASSERT_EQUAL_UINT64(tip, d); + /* Invariant: never exceeds the synced tip. */ + TEST_ASSERT_TRUE(d <= p->current_lsn - 1); + /* The on-disk seam agrees: those LSNs are genuinely fdatasync'd. */ + TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d); + + teardown_durability_persist(p); + printf("test_durable_lsn_batches_multiple_appends: PASS\n"); +} + +/* ------------------------------------------------------------------ * + * Test 6 (advances-only-after-fdatasync + monotonic): the durable * + * floor reported by the on-disk seam must always be >= durable_lsn — * + * i.e. durable_lsn never claims an LSN whose bytes are not actually * + * on stable storage. Also asserts monotonic non-decreasing across * + * successive append+flush rounds. * + * ------------------------------------------------------------------ */ +static void test_durable_lsn_never_exceeds_synced_tip(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_durlsn2_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + persist_t* p = make_durability_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + uint64_t prev = persist_durable_lsn(p); + + for (int round = 0; round < 4; round++) { + for (int i = 0; i < 3; i++) + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_UPDATE, "y", 1)); + + uint64_t tip = p->current_lsn - 1; + uint64_t d = wait_durable_reaches(p, tip, 5000); + + /* Monotonic non-decreasing. */ + TEST_ASSERT_TRUE(d >= prev); + /* Never exceeds the synced tip. */ + TEST_ASSERT_TRUE(d <= p->current_lsn - 1); + /* Advances only after a real fdatasync: the independent on-disk + * scan must confirm every claimed-durable LSN is on disk. */ + TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d); + + prev = d; + } + + teardown_durability_persist(p); + printf("test_durable_lsn_never_exceeds_synced_tip: PASS\n"); +} + +/* ------------------------------------------------------------------ * + * Test 7 (UNDER LOAD, Req 1.6 / lost-wakeup): many appends from * + * multiple threads racing the flusher. Asserts the invariants hold * + * throughout (durable_lsn <= synced tip, never past disk) AND that * + * the flusher eventually catches durable_lsn up to the final tip — * + * i.e. no append is left permanently un-synced (no lost wakeup). * + * ------------------------------------------------------------------ */ +#define LOAD_THREADS 4 +#define LOAD_PER_THREAD 200 + +typedef struct { + persist_t* p; + int n; +} load_arg_t; + +static void* load_appender(void* arg) +{ + load_arg_t* a = arg; + for (int i = 0; i < a->n; i++) { + /* Ignore rc: a full-segment rotate or transient EIO is not the + * subject under test; the invariant checks below tolerate it. */ + (void)persist_wal_async(a->p, WAL_JOB_UPDATE, "z", 1); + /* Sample the invariant mid-flight: durable_lsn must NEVER exceed + * the appended tip, even while a concurrent flush is in progress. */ + uint64_t d = persist_durable_lsn(a->p); + uint64_t cur = atomic_load_explicit( + (const _Atomic uint64_t*)&a->p->current_lsn, memory_order_acquire); + /* current_lsn is read without the lock here purely as a coarse + * upper bound; a torn read can only make `cur` smaller, so the + * assertion stays conservative. durable_lsn is a prefix watermark + * that can never outrun the last assigned LSN. */ + TEST_ASSERT_TRUE(d == 0 || d <= cur); + } + return NULL; +} + +static void test_durable_lsn_invariants_under_load(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_durlsn3_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + persist_t* p = make_durability_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + pthread_t th[LOAD_THREADS]; + load_arg_t args[LOAD_THREADS]; + for (int i = 0; i < LOAD_THREADS; i++) { + args[i].p = p; + args[i].n = LOAD_PER_THREAD; + TEST_ASSERT_EQUAL_INT(0, pthread_create(&th[i], NULL, + load_appender, &args[i])); + } + for (int i = 0; i < LOAD_THREADS; i++) + pthread_join(th[i], NULL); + + /* Final tip after all appends settled. */ + uint64_t tip = p->current_lsn - 1; + + /* No lost wakeup: the flusher must eventually advance durable_lsn up + * to the final appended tip (every entry gets synced). */ + uint64_t d = wait_durable_reaches(p, tip, 10000); + TEST_ASSERT_EQUAL_UINT64(tip, d); + TEST_ASSERT_TRUE(d <= p->current_lsn - 1); + TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d); + + teardown_durability_persist(p); + printf("test_durable_lsn_invariants_under_load: PASS\n"); +} + +/* ------------------------------------------------------------------ * + * Test 8 (ERROR PATH, Req 1.1/1.2 — fail-loud): when fdatasync FAILS, * + * durable_lsn must NOT advance. Bytes that did not reach stable * + * storage must never be counted durable — otherwise the member would * + * contribute a non-durable LSN toward the commit majority. * + * * + * Mechanism: the XCE_TESTING fault-injection toggle forces every * + * fdatasync site to treat the sync as failed. We append entries while * + * the toggle is ON and assert durable_lsn stays put across the full * + * cadence window (the flusher runs but its advance is gated on rc==0). * + * Then we clear the toggle and confirm the flusher recovers — durable_ * + * lsn advances to the tip once syncs succeed again. * + * ------------------------------------------------------------------ */ +static void test_durable_lsn_holds_on_fsync_failure(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_durlsn4_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + persist_t* p = make_durability_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + /* First, append one entry and let it become durable normally, so the + * watermark has a known non-zero baseline that must HOLD during the + * failure window. */ + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1)); + uint64_t baseline = wait_durable_reaches(p, 1, 5000); + TEST_ASSERT_EQUAL_UINT64(1, baseline); + + /* Now inject fdatasync failure and append more entries. */ + persist_test_set_force_fsync_fail(p, 1); + + for (int i = 0; i < 5; i++) + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1)); + + uint64_t tip = p->current_lsn - 1; /* 6 */ + TEST_ASSERT_EQUAL_UINT64(6, tip); + + /* Give the flusher ample time to run several passes (cadence is 5s but + * the append signal fires it immediately) AND an explicit flush attempt. + * Under injected failure, durable_lsn MUST remain at the baseline — the + * new bytes are not on stable storage, so they are not durable. */ + (void)persist_flush(p); /* also gated on rc==0 — must not advance either */ + for (int i = 0; i < 200; i++) { + TEST_ASSERT_EQUAL_UINT64(baseline, persist_durable_lsn(p)); + usleep(1000); /* 1 ms; 200 ms total — many flusher signal cycles */ + } + /* Still parked at the baseline after the window. */ + TEST_ASSERT_EQUAL_UINT64(baseline, persist_durable_lsn(p)); + TEST_ASSERT_TRUE(persist_durable_lsn(p) <= p->current_lsn - 1); + + /* Recovery: clear the fault and confirm the flusher catches up — proving + * the failure path did not corrupt the watermark, only held it. */ + persist_test_set_force_fsync_fail(p, 0); + /* Nudge the flusher with one more append so the level predicate signals. */ + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "c", 1)); + uint64_t new_tip = p->current_lsn - 1; /* 7 */ + uint64_t d = wait_durable_reaches(p, new_tip, 5000); + TEST_ASSERT_EQUAL_UINT64(new_tip, d); + TEST_ASSERT_TRUE(persist_test_durable_lsn_on_disk(p) >= d); + + teardown_durability_persist(p); + printf("test_durable_lsn_holds_on_fsync_failure: PASS\n"); +} + #endif /* XCE_TESTING */ /* ------------------------------------------------------------------ */ @@ -183,6 +444,11 @@ void test_commit_durability(void) RUN_TEST(test_durable_floor_empty); RUN_TEST(test_durable_floor_after_flush); RUN_TEST(test_durable_floor_survives_process_loss); + RUN_TEST(test_durable_lsn_initial_zero); + RUN_TEST(test_durable_lsn_batches_multiple_appends); + RUN_TEST(test_durable_lsn_never_exceeds_synced_tip); + RUN_TEST(test_durable_lsn_invariants_under_load); + RUN_TEST(test_durable_lsn_holds_on_fsync_failure); #else printf("test_commit_durability: skipped (no XCE_TESTING)\n"); #endif diff --git a/src/include/persist.h b/src/include/persist.h index f30b1cf2..2ae70fdf 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -87,6 +87,13 @@ struct persist_s { uint32_t seg_offset; uint64_t checkpoint_lsn; _Atomic uint64_t committed_lsn; /* highest LSN known committed; invariant: <= current_lsn - 1 */ + _Atomic uint64_t durable_lsn; /* highest LSN whose bytes are fdatasync'd locally (group-commit + * prefix watermark). Advanced ONLY by the flusher after a + * successful fdatasync. Invariants: monotonic non-decreasing + * (except clamp-on-truncate); durable_lsn <= current_lsn - 1. + * The commit majority counts self at this LSN, not the + * appended tip (ha-commit-durability Req 1.1/1.2; consumed by + * task 2.2). */ uint64_t total_wal_bytes; char seg_path[4096]; @@ -103,6 +110,14 @@ struct persist_s { atomic_int checkpoint_in_progress; /* CAS guard: 0=idle, 1=scheduled/running */ atomic_int fsync_stop; atomic_int fsync_thread_running; +#ifdef XCE_TESTING + /* Fault-injection seam (ha-commit-durability task 2.1 error-path test): + * when non-zero, every fdatasync site treats the sync as FAILED — it + * still calls the real fdatasync but forces a non-zero rc, so durable_lsn + * must NOT advance and the failure is logged loudly. Compiled out of + * production (xce_controller is built without -DXCE_TESTING). */ + atomic_int test_force_fsync_fail; +#endif pthread_t fsync_thread; pthread_mutex_t fsync_mutex; /* guards the fsync timer wait */ pthread_cond_t fsync_cond; /* signaled by persist_stop for prompt join */ @@ -184,6 +199,13 @@ int persist_commit_barrier(persist_t* p, uint32_t timeout_ms); /* Force WAL fdatasync (for tests and clean shutdown) */ int persist_flush(persist_t* p); +/* Highest LSN whose bytes are durably fdatasync'd locally — the + * group-commit prefix watermark advanced by the demand-driven flusher + * (ha-commit-durability task 2.1). The commit majority self-contribution + * caps at this value (task 2.2 consumer). Lock-free atomic read; safe from + * any thread. Returns 0 when nothing is yet durable. */ +uint64_t persist_durable_lsn(persist_t* p); + #ifdef XCE_TESTING /* ============================================================ * Durability-verification test seam (ha-commit-durability Req 3.2, 3.3) @@ -211,6 +233,15 @@ int persist_flush(persist_t* p); * without this symbol. * ============================================================ */ uint64_t persist_test_durable_lsn_on_disk(persist_t* p); + +/* Fault-injection toggle for the fdatasync error path (task 2.1). + * When `fail` is non-zero, every fdatasync site in persist.c forces a + * failed sync result: the real fdatasync still runs, but durable_lsn is + * left UNCHANGED and the failure is logged loudly — proving an acknowledged + * entry is never counted durable on a sync error (Req 1.1/1.2, fail-loud). + * Lock-free atomic write; safe to flip from a test thread at any time. + * Compiled out of production (xce_controller links without it). */ +void persist_test_set_force_fsync_fail(persist_t* p, int fail); #endif /* XCE_TESTING */ /* Checkpoint */