perf(raft-apply-decoupling): fix O(n^2) apply-pump WAL rescan, measure throughput (task 7.4)

Measured sustained worker-fact throughput against the pre-feature baseline.
Commit throughput is unregressed (~2.5% delta, within noise) and no per-fact
synchronous commit barrier exists, confirming Req 11.4's core claim. But
apply/observable throughput -- the new hop this spec introduces -- regressed
56% with clearly superlinear scaling (2x volume -> 5.2x elapsed).

Root cause: apply_pump_read_entry has no memory across calls -- every LSN
lookup re-lists WAL segments and rescans its segment from byte 0 -- while its
only caller applies committed entries one at a time in strict ascending
order, turning a sequential access pattern into O(n^2) total work.

Fixed with a minimal segment-read cache on persist_t (fd + offset + expected
next LSN, guarded by the existing apply_pump_lock): a fast path reads the
next expected LSN directly at a remembered offset, falling back unchanged to
full rediscovery on any miss. Correctness never depends on the cache, only
performance does. Invalidated on init, fini, recovery, and WAL truncation --
the four points a stale cursor could otherwise misread from.

Two review rounds: round 1 caught a missing lock guard in the recovery
invalidation path and an inaccurate risk claim about the cache/trim
interaction (a real, bounded behavior change deferred to the existing
APPLY_DIVERGENT fence, not "no worse than before") -- both corrected and
re-verified in round 2. Post-fix apply-throughput scales sublinearly (1.46x
elapsed for 2x volume) and exceeds the pre-feature baseline. Zero new test
failures across the pump, correctness-verification, recovery, and truncation
groups.

An intermittent, pre-existing client hang in the synchronous jobsub path and
a trim-floor/apply-lag architectural gap were surfaced but are explicitly
out of scope for this fix -- flagged for separate follow-up.

src/controller/persist.c

@@ -656,6 +656,15 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
     pthread_mutex_init(&p->apply_pump_lock, NULL);
     atomic_store(&p->apply_pump_scheduled, 0);
 
+    /* Apply-pump segment-lookup cache starts invalid (raft-apply-decoupling
+     * task 7.4 perf-fix): no segment/fd cached yet. -1 is always a safe
+     * fallback trigger for apply_pump_read_entry's fast path (see persist.h
+     * doc comment). */
+    p->apply_pump_cache_fd       = -1;
+    p->apply_pump_cache_seg_lsn  = 0;
+    p->apply_pump_cache_off      = 0;
+    p->apply_pump_cache_next_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;
@@ -785,6 +794,15 @@ void persist_fini(persist_t* p)
      * first, same precondition every other persist_t resource here already
      * assumes). */
     pthread_mutex_destroy(&p->apply_pump_lock);
+
+    /* raft-apply-decoupling task 7.4 perf-fix: close any cached segment fd
+     * left open by apply_pump_read_entry. No walk can be in flight here
+     * (same precondition as apply_pump_lock's destroy just above), so no
+     * lock is needed. */
+    if (p->apply_pump_cache_fd >= 0) {
+        close(p->apply_pump_cache_fd);
+        p->apply_pump_cache_fd = -1;
+    }
 }
 
 /* Forward declaration — defined after checkpoint internals */
@@ -3004,6 +3022,72 @@ static int apply_pump_read_entry(persist_t* p, uint64_t lsn,
     if (!p || lsn == 0 || !out_type || !out_payload || !out_len)
         return -EINVAL;
 
+    /* Fast path (raft-apply-decoupling task 7.4 perf-fix): the sole caller
+     * (persist_apply_pump_run) calls this once per LSN in strict ascending
+     * order with no gaps in the common case, and holds apply_pump_lock
+     * across the entire walk -- so the cache fields below need no lock of
+     * their own here (see persist.h doc comment). A cache hit reads the next
+     * header directly at the remembered offset on the already-open fd,
+     * giving amortized O(1) per entry instead of a fresh
+     * opendir/readdir/segment-sort plus a scan-from-zero on every single
+     * call (the O(n^2) regression this fixes). ANY miss below just drops the
+     * cache and falls through to the unchanged slow path -- correctness
+     * never depends on the cache being right, only performance does. */
+    if (p->apply_pump_cache_fd >= 0 && p->apply_pump_cache_next_lsn == lsn) {
+        wal_entry_hdr_t hdr;
+        ssize_t  hn  = pread(p->apply_pump_cache_fd, &hdr, sizeof(hdr),
+                            (off_t)p->apply_pump_cache_off);
+        bool     hit = (hn == (ssize_t)sizeof(hdr)) && (hdr.lsn == lsn);
+        uint8_t* payload = NULL;
+
+        if (hit && hdr.payload_len > 0) {
+            payload = malloc(hdr.payload_len);
+            if (!payload) {
+                hit = false;
+            } else {
+                ssize_t pn = pread(p->apply_pump_cache_fd, payload,
+                                  hdr.payload_len,
+                                  (off_t)(p->apply_pump_cache_off +
+                                          sizeof(hdr)));
+                if (pn != (ssize_t)hdr.payload_len) {
+                    free(payload);
+                    payload = NULL;
+                    hit = false;
+                }
+            }
+        }
+        if (hit) {
+            uint32_t expected_crc =
+                crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE);
+            if (hdr.payload_len > 0)
+                expected_crc = crc32(expected_crc, (const Bytef*)payload,
+                                    hdr.payload_len);
+            if (hdr.crc != expected_crc) {
+                free(payload);
+                payload = NULL;
+                hit = false;
+            }
+        }
+
+        if (hit) {
+            *out_type    = hdr.type;
+            *out_payload = payload;
+            *out_len     = hdr.payload_len;
+            p->apply_pump_cache_off +=
+                (uint32_t)(sizeof(hdr) + hdr.payload_len);
+            p->apply_pump_cache_next_lsn = lsn + 1;
+            return 0;
+        }
+
+        /* Miss (short read/EOF meaning the entry rolled into the next
+         * segment, header mismatch, CRC failure, whatever): don't try to be
+         * clever about advancing to the next segment inline. Close and fall
+         * through to the EXISTING slow path unchanged -- it already
+         * correctly finds the right segment from scratch. */
+        close(p->apply_pump_cache_fd);
+        p->apply_pump_cache_fd = -1;
+    }
+
     DIR* d = opendir(p->wal_path);
     if (!d)
         return -ENOENT;
@@ -3090,12 +3174,29 @@ static int apply_pump_read_entry(persist_t* p, uint64_t lsn,
                 *out_payload = payload;
                 *out_len     = hdr.payload_len;
                 rc = 0;
+
+                /* raft-apply-decoupling task 7.4 perf-fix: populate the
+                 * segment-lookup cache with this fd positioned just past the
+                 * entry we just returned, BEFORE closing it below, so the
+                 * next expected LSN (lsn+1) hits the fast path above instead
+                 * of re-scanning from segment start next time. Every OTHER
+                 * branch (error paths, not-found-in-this-segment
+                 * continuation) closes fd exactly as before -- caching only
+                 * ever happens on this success branch. */
+                p->apply_pump_cache_fd      = fd;
+                p->apply_pump_cache_seg_lsn = seg_lsns[si];
+                p->apply_pump_cache_off     =
+                    off + (uint32_t)(sizeof(hdr) + hdr.payload_len);
+                p->apply_pump_cache_next_lsn = lsn + 1;
+                fd = -1; /* ownership transferred to the cache -- the
+                          * close(fd) below becomes a guarded no-op. */
                 break;
             }
 
             off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
         }
-        close(fd);
+        if (fd >= 0)
+            close(fd);
 
         if (rc == 0 || rc == -ENOMEM || rc == -EIO)
             break; /* found it, or hit a genuine error inside the segment
@@ -3709,6 +3810,29 @@ int persist_recover(persist_t* p)
      * whose own recovery never finished), not a new hazard. */
     atomic_store_explicit(&p->recovery_complete, false, memory_order_release);
 
+    /* raft-apply-decoupling task 7.4 perf-fix (round 2): invalidate the apply
+     * pump's segment-lookup cache before replay -- the cached fd/offset are
+     * only valid relative to the pre-recovery on-disk state, and replay below
+     * can change checkpoint_lsn/current_lsn. Doing this first covers every
+     * return path below (including the early I/O-failure returns), so the
+     * first post-recovery pump kick always falls back to full segment
+     * rediscovery from the correct, recovered position -- never trusting a
+     * stale pre-recovery cursor. Wrapped in apply_pump_lock (matching
+     * persist_truncate_wal_above's identical invalidation site below) so the
+     * cache fields' documented invariant -- "guarded ENTIRELY by
+     * apply_pump_lock, never p->lock" (persist.h) -- holds unconditionally
+     * here too, rather than resting on the implicit fact that no live pump
+     * walk can be racing this boot-time call. No caller of persist_recover()
+     * holds apply_pump_lock (its sole production call site, cluster.c boot
+     * sequence, runs before the pump is ever kicked), so this cannot
+     * deadlock. */
+    pthread_mutex_lock(&p->apply_pump_lock);
+    if (p->apply_pump_cache_fd >= 0) {
+        close(p->apply_pump_cache_fd);
+        p->apply_pump_cache_fd = -1;
+    }
+    pthread_mutex_unlock(&p->apply_pump_lock);
+
     /* --- Step 1: find latest snapshot (highest LSN filename) --- */
     char     snap_file[4096] = "";
     uint64_t snap_lsn        = 0;
@@ -5115,8 +5239,25 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
      * releasing p->lock: the callback (the commit engine) takes only
      * ce->lock, preserving the cycle-free p->lock -> ce->lock edge already
      * established by the durable-advance wake. */
-    if (rc == 0 && after < before && p->on_truncate != NULL) {
-        p->on_truncate(p->truncate_cb_ctx, cutoff_lsn);
+    if (rc == 0 && after < before) {
+        /* raft-apply-decoupling task 7.4 perf-fix: a real truncation may
+         * ftruncate()/unlink() segment files the apply-pump's segment-lookup
+         * cache still points into -- invalidate it so a subsequent
+         * apply_pump_read_entry can never read truncated-away data. p->lock
+         * is already released above and apply_pump_lock is never held by
+         * this call path (persist_apply_pump_run is the only place that
+         * holds it, during its own walk), so no lock-order hazard here. */
+        pthread_mutex_lock(&p->apply_pump_lock);
+        if (p->apply_pump_cache_fd >= 0) {
+            close(p->apply_pump_cache_fd);
+            p->apply_pump_cache_fd = -1;
+        }
+        p->apply_pump_cache_next_lsn = 0;
+        pthread_mutex_unlock(&p->apply_pump_lock);
+
+        if (p->on_truncate != NULL) {
+            p->on_truncate(p->truncate_cb_ctx, cutoff_lsn);
+        }
     }
     return rc;
 }

src/include/persist.h

@@ -433,6 +433,52 @@ struct persist_s {
      * pump; see design.md "Restart Replay & Verification"). */
     pthread_mutex_t apply_pump_lock;
     _Atomic int     apply_pump_scheduled;
+
+    /* raft-apply-decoupling task 7.4 perf-fix: apply_pump_read_entry's
+     * segment-lookup cache. apply_pump_read_entry's only caller
+     * (persist_apply_pump_run) calls it once per LSN in strict ascending
+     * order with no gaps in the common case; without this cache every single
+     * lookup redid a full opendir/readdir/segment-sort plus a pread scan
+     * from byte offset 0, making N contiguous applies O(N^2). These fields
+     * cache the currently-open segment fd + byte offset positioned to read
+     * whichever LSN the pump is expected to ask for next, giving amortized
+     * O(1) per entry on the common contiguous-advance path.
+     *
+     * Guarded ENTIRELY by apply_pump_lock (never p->lock) -- the pump's sole
+     * caller already holds apply_pump_lock across the whole walk, including
+     * every apply_pump_read_entry call, so no additional locking is needed
+     * here. apply_pump_cache_fd == -1 means "no cache" -- this is always a
+     * safe, correct fallback trigger (falls through to full segment
+     * rediscovery), never a bug to fall back to. Reset to -1 in persist_init,
+     * persist_fini, persist_recover (a stale pre-recovery cursor must never
+     * be trusted), and persist_truncate_wal_above (a real truncation may
+     * ftruncate()/unlink() the segment the cache still points into).
+     *
+     * Known, accepted, bounded tradeoff vs. checkpoint-driven trim
+     * (trim_wal_segments): trim only unlinks segments below its own trim
+     * floor, and POSIX unlink-keeps-fd-alive semantics mean an fd this cache
+     * already has open into a segment stays readable even after trim unlinks
+     * it on disk. If the pump is lagging behind checkpoint/trim when trim
+     * fires -- reachable in this feature, since local-apply-lagging-behind-
+     * commit is raft-apply-decoupling's entire premise -- the cache can keep
+     * successfully reading that stale, deleted-on-disk data and silently
+     * advance last_applied_lsn past it. Before this cache existed, every
+     * lookup did a fresh opendir/readdir, so the equivalent situation failed
+     * loudly and immediately with -ENOENT on the very next lookup. This is a
+     * real behavior change -- NOT "no worse than before" -- but it is bounded:
+     * a crash before the next checkpoint leaves in-memory apply state ahead
+     * of what disk can prove on restart, and the existing APPLY_DIVERGENT
+     * boot fence (tasks 5.1/5.2, persist_recover) catches exactly this at the
+     * next restart and fences the node loudly rather than serve wrong state.
+     * So the loud failure is deferred from "immediate, at read-time" to
+     * "delayed, at next-restart-boot-time," not eliminated. Redesigning
+     * trim/cache invalidation to close this gap at read-time is a distinct,
+     * pre-existing architectural item (tracked separately) and explicitly
+     * out of scope for this perf fix. */
+    int             apply_pump_cache_fd;
+    uint64_t        apply_pump_cache_seg_lsn;
+    uint32_t        apply_pump_cache_off;
+    uint64_t        apply_pump_cache_next_lsn;
 #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