#pragma once #include #include #include #include #include #include #include "config.h" #include "wal_types.h" /* wal_entry_hdr_t, wal_entry_type — P2-4: shared with repl.h */ #include "repl.h" /* replicator_t — P2-4: embedded in persist_t */ /* Forward declaration */ typedef struct cluster_s cluster_t; /* ============================================================ * Recovery-integrity verdict (ha-commit-durability task 3.1, Req 2.1) * * The outcome of WAL replay at startup, distinguishing a structurally * intact log from one whose chain is broken. It is about the STRUCTURE * of the local log, NOT about being behind the cluster: a clean log that * simply ends below the cluster's committed high-water mark is still * RECOVER_CLEAN (the freshness/promotion fence handles "behind"). A * corruption-truncated recovery is NEVER reported as a clean recovery. * ============================================================ */ typedef enum { RECOVER_CLEAN = 0, /* WAL forms a valid CRC chain to a clean entry * boundary: clean EOF, pre-allocated-zero tail, * snapshot-sentinel + clean replay, or an empty * first-boot WAL (no entries != corruption). */ RECOVER_TRUNCATED = 1, /* replay stopped on structural corruption: a CRC * chain break, a CRC mismatch, or a short/partial * entry read — the log cannot be reconstructed * contiguously. Marks the member non-promotable * (consumed by the promotion gate, task 4.3). */ } recover_verdict_t; /* ============================================================ * Boot-policy recovery decision (ha-commit-durability task 3.3, Req 2.4) * * The outcome of evaluating a member's boot/promotion-eligibility state * against the Boot Policy Decision Table (design.md "Boot Policy Decision * Table (task 3.2)"). This task (3.3) owns the FATAL terminal; the fence + * catch-up + release path is task 4.3. * * ELIGIBLE — may promote (no fence). Rows 1, 6. * FENCE — ineligible to promote; catch up from a healthy peer, regain * eligibility on success (Req 2.6). Rows 2, 3, 5. A reachable * healthy peer NEVER yields FATAL. * FATAL — fail loud: cannot recover locally AND cannot catch up within * the bounded window. Terminate with the truncation/boundary * logged rather than serve divergent state (Req 2.4). Rows 4, 4b. * * AUTHORITATIVE INVARIANT (design.md): the fence is never a silent indefinite * wait — a truncated/behind member resolves to exactly one of FENCE→ELIGIBLE * or a BOUNDED FATAL. There is no "fenced forever, silently divergent" state. * ============================================================ */ typedef enum { RECOVER_DECISION_ELIGIBLE = 0, RECOVER_DECISION_FENCE = 1, RECOVER_DECISION_FATAL = 2, } recover_decision_t; /* Policy inputs evaluated at boot / promotion-eligibility (design.md table). * Modelled as a plain value so the decision is a pure, unit-testable function * of its inputs — the live catch-up machinery (replication / task 4.3) supplies * these; this task does not build it. */ typedef struct { recover_verdict_t verdict; /* RECOVER_CLEAN / RECOVER_TRUNCATED (task 3.1) */ bool durable_below_hwm; /* local durable position < cluster committed HWM */ bool healthy_peer_reachable; /* a healthy peer is reachable for catch-up */ bool catch_up_succeeded; /* catch-up reached a committed-consistent state * within the bounded window (only meaningful * when a catch-up was attempted) */ uint64_t durable_floor_lsn; /* highest intact replayed LSN (truncation boundary) */ uint64_t cluster_hwm_lsn; /* cluster committed high-water mark, for logging */ } recover_policy_inputs_t; /* ============================================================ * Persistence configuration (Part 2.6) * ============================================================ */ typedef struct { uint32_t wal_sync_ms; /* fsync interval in ms (default 10) */ uint32_t snap_interval_s; /* checkpoint interval in seconds (default 60) */ uint32_t snap_wal_size_mb; /* checkpoint when WAL exceeds this MB (default 64) */ uint32_t snap_keep_n; /* snapshots to retain (default 2) */ uint32_t transient_job_timeout_s; /* seconds before TRANSIENT job auto-terminates (default 300) */ uint32_t finished_retain_s; /* retain finished jobs N seconds (default 3600) */ uint32_t worker_retain_s; /* retain unavailable workers N seconds (default 86400) */ uint32_t sync_timeout_ms; /* sync replication ACK timeout ms (default 50) */ uint32_t max_wal_records; /* checkpoint trigger: records since last snap (default 10000) */ } persist_config_t; static inline persist_config_t persist_config_defaults(void) { persist_config_t c; c.wal_sync_ms = DEFAULT_WAL_SYNC_MS; c.snap_interval_s = DEFAULT_SNAP_INTERVAL_S; c.snap_wal_size_mb = DEFAULT_SNAP_WAL_SIZE_MB; c.snap_keep_n = DEFAULT_SNAP_KEEP_N; c.transient_job_timeout_s = DEFAULT_TRANSIENT_JOB_TIMEOUT_S; c.finished_retain_s = DEFAULT_FINISHED_RETAIN_S; c.worker_retain_s = DEFAULT_WORKER_RETAIN_S; c.sync_timeout_ms = DEFAULT_SYNC_TIMEOUT_MS; c.max_wal_records = DEFAULT_MAX_WAL_RECORDS; return c; } /* * persist_config_set_wal_sync_ms — apply the --wal-sync-ms flush-cadence * override (ha-commit-durability Req 3.1). * * Single override point shared by both persist_config_defaults() callers * (cluster.c phase2 and myapp.c step 6), mirroring the --sync-timeout-ms * convention: override_ms == 0 means "flag not passed" and leaves the * default flush cadence untouched (Req 3.3 — the seam changes timing * controllability only, not the durability contract). A positive value * sets the cadence to exactly that, applied before persistence starts. */ static inline void persist_config_set_wal_sync_ms(persist_config_t *c, uint32_t override_ms) { if (override_ms > 0) c->wal_sync_ms = override_ms; } /* ============================================================ * Key→index mapping for checkpoint serialization (Part 3.8) * ============================================================ */ typedef struct { int index; /* shadow slot index in the frozen zarray */ char key[XCE_MAXNAME_LEN]; /* owned deep copy — not a borrowed kvtab pointer */ size_t key_len; /* length including null terminator */ } slot_key_t; /* ============================================================ * persist_t — concrete definition (exposed for stack/calloc allocation) * The opaque pattern can be restored in a later refactor once the * full API surface is stable. * ============================================================ */ struct persist_s { /* Back-pointer to owning cluster (set by persist_init, used in persist_start) */ cluster_t* cluster; /* WAL state */ int wal_fd; uint64_t current_lsn; uint32_t last_crc; uint32_t last_replicated_crc; uint64_t term; 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]; /* Recovery-integrity verdict (ha-commit-durability task 3.1, Req 2.1). * Set by persist_recover/replay_wal at startup: RECOVER_CLEAN when the * local WAL forms a valid CRC chain to a clean entry boundary (incl. an * empty first-boot WAL, a snapshot-sentinel + clean replay, and a * pre-allocated-zero tail), RECOVER_TRUNCATED when replay stopped on a * structural corruption — a CRC chain break, a CRC mismatch, or a * short/partial entry read — so the on-disk log cannot be reconstructed * contiguously. A node that cleanly ends BELOW the cluster high-water is * still CLEAN here (the freshness/promotion fence catches "behind", task * 4.2/4.3 — not this verdict). 3.1 only CLASSIFIES + sets this; it does * NOT FATAL (task 3.3) and does NOT fence promotion (task 4.3, which * CONSUMES this verdict). Defaults to RECOVER_CLEAN before recovery runs * (a fresh store is clean). Written single-threaded during recovery * before the node participates; read by the promotion gate thereafter. */ recover_verdict_t recovery_verdict; /* Paths (derived from workdir + name via persist_set_workdir) */ char wal_path[4096]; char snap_path[4096]; /* Config */ persist_config_t config; /* Concurrency */ pthread_mutex_t lock; atomic_uint wal_record_count; /* records appended since last checkpoint */ 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; /* fdatasync observability counter (ha-commit-durability task 2.3): * incremented at the standby-path inline sync in * persist_wal_write_replicated, BEFORE that function returns (and thus * before the caller can ACK). A persist-before-ACK regression test reads * this to prove a durable sync ran inside the call — page-cache re-reads * cannot distinguish fsync'd from cache-only, so a direct sync signal is * required. Compiled out of production. */ atomic_uint test_replicated_fdatasync_count; /* Liveness observability (ha-commit-durability task 5.1, Req 1.6): * the thread id of the last flusher fdatasync that advanced durable_lsn. * A test reads this to prove fdatasync runs on the dedicated flusher * thread, NOT on the caller/iomgr/heartbeat thread. Compiled out of * production. */ pthread_t test_last_fdatasync_thread; atomic_int test_last_fdatasync_thread_set; #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 */ /* Durable-advance wake hook (ha-commit-durability task 2.2). * Persistence owns this slot; the commit engine REGISTERS a callback * (ctx = the commit_engine_t) via persist_set_durable_cb. The flusher * invokes it AFTER advancing durable_lsn and AFTER releasing p->lock, * so a COMMIT_MODE_WAIT caller is woken promptly when its entry becomes * durable locally — without it, a commit whose standby ACKs all arrived * before the local fdatasync would stall until the WAIT timeout. The * dependency direction is preserved (commit depends on persist, never * the reverse): persist only stores+calls an opaque function pointer. * NULL until registered (every flush is then a no-op wake). */ void (*on_durable_advance)(void* ctx); void* durable_cb_ctx; /* Checkpoint (stubs for now — wired in Task 10) */ atomic_int checkpoint_requested; pthread_cond_t checkpoint_done; pthread_mutex_t checkpoint_mutex; pthread_t snap_thread; /* Fan-out replicator — P2-4 */ replicator_t repl; }; typedef struct persist_s persist_t; /* Set workdir and controller name before persist_init(). * Derives wal_path and snap_path as //data/{wal,snap}. */ void persist_set_workdir(persist_t* p, const char* workdir, const char* name); /* Lifecycle */ int persist_init(persist_t* p, persist_config_t* config, cluster_t* c); int persist_start(persist_t* p); void persist_stop(persist_t* p); void persist_fini(persist_t* p); int persist_recover(persist_t* p); /* WAL write paths */ int persist_wal_async(persist_t* p, uint32_t type, const void* payload, uint32_t len); int persist_wal_sync_append(persist_t* p, uint32_t type, const void* payload, uint32_t len, int* sync_rc_out); /* Write a REPLICATED entry (received from the primary) into the local WAL * at its GIVEN lsn — used by standbys to durably persist streamed entries * BEFORE acking them (persist-before-ACK; an ACK must imply durability). * * Unlike persist_wal_async it does NOT assign a new lsn (the lsn is supplied * by the primary) and does NOT repl_broadcast (standbys never re-replicate). * It builds a SELF-consistent CRC chain (the streamed entry carries no crc), * fsyncs the segment before returning, and is idempotent on re-stream: if the * lsn is already persisted (lsn+1 <= current_lsn) it is a no-op returning 0. * * WAL_CHECKPOINT is rejected with -EINVAL (never streamed to standbys). * Returns 0 on success (incl. dedup no-op), negative errno on failure. */ int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type, const void* payload, uint32_t len); /* ============================================================ * Write-path commit classification (ha-commit-barrier-rollout) * ============================================================ */ typedef enum { WAL_MODE_WAIT, /* request-shaped: barrier before requester reply */ WAL_MODE_RETAIN, /* fact-shaped: retained until committed, no timeout */ WAL_MODE_LOCAL, /* not replicated (WAL_CHECKPOINT) */ } wal_commit_mode_t; /* Pure classification lookup. detail disambiguates WAL_WORKER_CTRL only * (HOOK_DETAIL_WORKER_JOIN/UPDATE -> WAIT, HOOK_DETAIL_WORKER_DEAD -> * RETAIN). Unknown type: XCE_TESTING assert (fail fast); production logs * ERROR and returns WAL_MODE_RETAIN (never dropped, never confirmed early). */ wal_commit_mode_t wal_classify(uint32_t wal_type, int detail); /* Commit barrier — handler-callable WAIT completion on everything appended * so far (ha-commit-barrier-rollout). Snapshots the appended watermark * (current_lsn - 1, under p->lock — matching persist_wal_sync_append's * read); returns 0 without blocking when committed_lsn already covers it * (or nothing was ever appended). Otherwise delegates to * commit_submit(COMMIT_MODE_WAIT) and returns its code unmodified: * 0 — committed on a majority * -ETIMEDOUT — not committed within the deadline (entries remain * in the log; outcome converges per umbrella 2.5) * -XCE_ERR_NOT_READY — engine admission cap exhausted (retriable) * timeout_ms = 0 falls back to p->config.sync_timeout_ms. n_peers <= 1 * returns 0 immediately (engine self-quorum, standalone preserved). * Call only from vtask threads with no entity locks held. */ 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); /* Durable freshness floor (ha-commit-durability task 3.4, Req 2.3). * * The member's durably-backed last-log index (WAL tip): the highest LSN that * is on stable storage locally. persist_recover re-derives it from the * recovered WAL tip (the highest INTACT replayed LSN) so the post-recovery * value is backed by durable on-disk state, NOT a volatile in-memory * applied_lsn / last_known_committed_lsn. The Raft freshness gate (task 4.2, * ha.c / cluster.c) consumes this so a truncated member cannot over-advertise * a tip it does not durably hold. * * Precision: this is the durable LAST-LOG INDEX, NOT the commit position * (committed_lsn). The post-election safe-commit point still comes from voter * replies, never from this floor (CLAUDE.md). After a truncating restart the * floor equals the lower intact tip, never a stale higher in-memory value and * never the commit position. * * Thin wrapper over persist_durable_lsn: at and after recovery the durable * watermark IS the recovered WAL tip (recovered on-disk entries are genuinely * durable), so a separate field would be redundant. Lock-free atomic read; * safe from any thread. Returns 0 when nothing is durable (empty first-boot * WAL). */ uint64_t persist_durable_floor(persist_t* p); /* Recovery-integrity verdict + flag (ha-commit-durability task 3.1, Req 2.1). * persist_recovery_verdict returns the RECOVER_CLEAN / RECOVER_TRUNCATED * verdict computed by the last persist_recover (RECOVER_CLEAN on a fresh * store before recovery runs). persist_recovery_truncated is the convenience * flag form (true iff the verdict is RECOVER_TRUNCATED) the promotion gate * (task 4.3) consumes to mark the member non-promotable. Both are pure reads * of recovery state established single-threaded at startup. */ recover_verdict_t persist_recovery_verdict(persist_t* p); bool persist_recovery_truncated(persist_t* p); /* Bounded catch-up window (ha-commit-durability task 3.3, Req 2.4). * The FINITE deadline a recovery-truncated / behind member is given to reach a * committed-consistent state from a healthy peer before the boot policy FATALs. * The fence must NEVER be an indefinite wait (design.md authoritative * invariant), so the window is always finite. Default 30s — generous enough for * a healthy-peer catch-up stream yet bounded; the verification mode overrides it * via the inputs/timeout the caller (task 4.3 boot path) supplies. */ #define DEFAULT_RECOVERY_CATCHUP_WINDOW_MS 30000u /* persist_recover_decision (ha-commit-durability task 3.3, Req 2.4). * * Pure decision over the Boot Policy Decision Table: maps the policy inputs to * RECOVER_DECISION_{ELIGIBLE,FENCE,FATAL}. No side effects, no termination — it * only classifies, so it is unit-testable in isolation. The FATAL terminal is * acted on by persist_recover_fatal_if_unrecoverable (which logs + terminates). * * Contract (design.md rows): a reachable healthy peer NEVER returns FATAL * (rows 3, 5); FATAL is returned ONLY when the member is truncated or behind * the HWM, has NO healthy peer, and bounded catch-up did not reach a * committed-consistent state (rows 4, 4b). */ recover_decision_t persist_recover_decision(const recover_policy_inputs_t* in); /* persist_recover_fatal_if_unrecoverable (ha-commit-durability task 3.3, * Req 2.4). * * Evaluate persist_recover_decision over `in`; if the outcome is FATAL, log at * ERROR LOUDLY — INCLUDING the truncation boundary (durable_floor_lsn) and the * cluster high-water mark — and then TERMINATE (fail loud rather than serve * divergent state). Returns the decision so a non-FATAL caller can fence/proceed * (FENCE/ELIGIBLE are acted on by task 4.3). * * Termination routes through persist_fatal_hook: in production the hook is NULL * and the function flushes logs and abort()s (NEVER returns on FATAL); a test * build may install a hook to capture the FATAL decision + boundary log without * the process dying (mirrors the ha_fatal_hook seam in ha_convergence.c). */ recover_decision_t persist_recover_fatal_if_unrecoverable(const recover_policy_inputs_t* in); /* persist_fatal_hook: test-only seam for the FATAL (abort) contract of * persist_recover_fatal_if_unrecoverable (ha-commit-durability task 3.3, * Req 2.4). Mirrors ha_fatal_hook (ha_convergence.c). * * When non-NULL, the FATAL path invokes this INSTEAD OF abort(), so a test * binary can assert the FATAL decision was taken and the truncation boundary * was logged without the process dying. Production leaves it NULL and actually * terminates. Signature: void hook(const char *msg). */ extern void (*persist_fatal_hook)(const char* msg); /* Register a durable-advance wake callback (ha-commit-durability task 2.2). * The group-commit flusher calls fn(ctx) after it advances durable_lsn and * after releasing p->lock, so a consumer (the commit engine) can re-evaluate * its commit majority and wake any COMMIT_MODE_WAIT caller whose entry just * became durable locally. fn MUST NOT take p->lock (it runs after release, on * the flusher thread) — the commit engine's callback takes only ce->lock, * keeping the lock order p->lock → ce->lock cycle-free. Pass fn = NULL to * unregister. Safe to call before persist_start (no flusher running yet). */ void persist_set_durable_cb(persist_t* p, void (*fn)(void* ctx), void* ctx); #ifdef XCE_TESTING /* ============================================================ * Durability-verification test seam (ha-commit-durability Req 3.2, 3.3) * * persist_test_durable_lsn_on_disk — re-read and CRC-validate the WAL * segments from STABLE STORAGE and return the highest fully-persisted * LSN, independent of the in-memory current_lsn / committed_lsn * watermarks. Lets a test assert that an acknowledged operation is * recoverable from disk after an immediate, ungraceful termination — * without relying on the background flush timer (Req 3.2): * * persist_wal_async(p, ...); // append to page cache * persist_flush(p); // fdatasync → durable on disk * lsn = persist_test_durable_lsn_on_disk(p); // highest durable LSN * * It scans the same on-disk format and CRC chain as recovery, stopping * at the first corruption / short read / pre-allocated zero region, so * it reports only entries that are genuinely on stable storage. It * never mutates persist_t state and fires no hooks (Req 3.3 — the seam * is purely inspective; the acknowledgment contract is unchanged). * * Returns the highest durable LSN, or 0 when nothing is durable (empty * first-boot WAL, no segments). Compiled out of production builds: * the xce_controller target is built without -DXCE_TESTING and links * 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); /* Read the standby-path fdatasync counter (task 2.3, persist-before-ACK * regression). Counts the inline fdatasync calls made by * persist_wal_write_replicated. A test snapshots this before the call and * asserts it incremented after the call returns — proving the durable sync * ran inside the persist primitive, BEFORE the caller can send its WalAck. * Lock-free atomic read; compiled out of production. */ unsigned persist_test_replicated_fdatasync_count(persist_t* p); /* Invoke the legacy sync-append commit store (task 2.4) with the durable * precondition applied — the same store persist_wal_sync_append performs on a * majority ACK in its engine-less / LSN-0 branch. Lets a test prove * committed_lsn never advances past durable_lsn on the ACK path without * standing up the replication wire. Compiled out of production. */ void persist_test_sync_append_legacy_commit_store(persist_t* p, uint64_t written_lsn); /* Return the thread id of the last flusher fdatasync that advanced * durable_lsn (task 5.1 / Req 1.6 liveness). A test asserts this is NOT the * caller/iomgr thread — proving durability (fdatasync) work runs on the * dedicated flusher thread and never stalls heartbeat/election handling. * Returns pthread_self() if no flusher sync has advanced durable_lsn yet * (so an unprimed read can never falsely look "off-thread"). Compiled out of * production. */ pthread_t persist_test_last_fdatasync_thread(persist_t* p); #endif /* XCE_TESTING */ /* Checkpoint */ int persist_checkpoint(persist_t* p); /* WAL truncation — remove all entries with LSN > cutoff_lsn. * Idempotent; returns 0 on success, negative errno on I/O error. */ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn); /* Logging helper — never logs payload content (security invariant) */ void persist_log_entry_metadata(const wal_entry_hdr_t* hdr); /* ============================================================ * AFTER_EVENT hook functions — registered by persist_start() * * Signature matches hook_fn_t: int fn(int type, int situation, * int detail, void* data, va_list ap) * cls/data points to the persist_t instance. * * P0: writes a minimal 4-byte payload (detail value) to the WAL. * Full Cap'n Proto serialization is wired in Task 15. * ============================================================ */ int persist_job_hook_fun(int type, int situation, int detail, void* cls, va_list ap); int persist_queue_hook_fun(int type, int situation, int detail, void* cls, va_list ap); int persist_worker_hook_fun(int type, int situation, int detail, void* cls, va_list ap); /* Apply a single WAL entry to in-memory state (standby replay path). * No hooks fire — hook system is empty on standbys until promotion. * Returns 0 on success (including skip for checkpoint/unknown types), * negative errno on error (oversized payload, NULL arguments). */ int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload); /* Load a snapshot file from path, set p->checkpoint_lsn and p->current_lsn. * lsn hint is informational — the actual LSN is read from the file header. * Returns 0 on success, -errno or -EINVAL/-EIO on error. */ int persist_load_snapshot(persist_t* p, const char* path, uint64_t lsn); /* GC retention predicates (P0-5a / P0-8 / P0-13) */ #include #include "xce/types.h" bool job_is_terminal(const job_t* j); bool job_should_retain(const job_t* j, const persist_config_t* cfg); bool worker_should_retain(const xce_worker_t* w, const persist_config_t* cfg);