feat: P2-2 apply_wal_entry — standby WAL replay with LSN advance and type guards

Upgrade apply_wal_entry() from P0 static no-op stub to the real public
standby replay path:

- Reject oversized payloads (> MAX_WAL_PAYLOAD_BYTES) with -EMSGSIZE
- Skip WAL_CHECKPOINT (100) entries silently — local-only markers, never
  replicated to standbys; current_lsn is NOT advanced on skip
- Skip unknown types (> WAL_ENTRY_TYPE_MAX) with WARN log and rc=0 for
  forward compatibility with WAL written by newer controller versions
- Advance current_lsn to hdr->lsn+1 and update last_crc under p->lock
  for all accepted known entries
- Gate restore_job()/worker_from_capnp() calls on p->cluster != NULL so
  unit tests work without a full cluster wired in
- Add NULL guard returning -EINVAL for missing p or hdr

Add public declaration in persist.h with forward-declare stubs for
restore_job/worker_from_capnp in persist.c (avoids heavy cluster.h chain).

Test suite: 5 new Unity tests in test_repl_apply.c covering all branches.
Weak stubs for restore_job/worker_from_capnp in the test file satisfy the
linker without linking job.c/worker.c's heavy dependency graph.
All 5 new tests pass; full controller suite green.

src/controller/meson.build

@@ -88,6 +88,7 @@ controller_unit_test_src = [
   'testsuite/test_ha_redirect.c',
   'testsuite/test_p1_integration.c',
   'testsuite/test_repl_unit.c',
+  'testsuite/test_repl_apply.c',
   'persist.c',
   'ha.c',
   'repl.c',

src/controller/persist.c

@@ -23,6 +23,14 @@
 #include <unistd.h>
 #include <zlib.h>
 
+/* Forward declarations for P0 restore stubs (defined in job.c / worker.c).
+ * Avoids pulling in the heavy cluster.h / worker.h header chains here.
+ * Full Cap'n Proto wiring happens in P2-4+. */
+extern int restore_job(cluster_t* c, uint64_t jobid,
+                       const void* capnp_payload, size_t len);
+extern int worker_from_capnp(cluster_t* c,
+                             const void* capnp_payload, size_t len);
+
 /* ------------------------------------------------------------------ */
 /* Constants                                                            */
 /* ------------------------------------------------------------------ */
@@ -527,14 +535,68 @@ static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
     return 0;
 }
 
-/* P0 stub: no job/worker deserialization yet — wired in Task 12+/15. */
-static int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr,
-                           const void* payload)
+/* P2-2: Standby WAL replay — apply a single replicated entry to in-memory state.
+ *
+ * No hooks fire on standbys; the hook system is empty until promotion.
+ *
+ * Returns 0 on success (including skip for checkpoint / unknown types),
+ * negative errno on error (NULL args, oversized payload). */
+int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload)
 {
-    (void)p;
-    (void)hdr;
-    (void)payload;
-    /* Full implementation in Task 12+ (hook wiring) and Task 15 (cluster wiring) */
+    if (!p || !hdr)
+        return -EINVAL;
+
+    /* Reject oversized payloads BEFORE any allocation */
+    if (hdr->payload_len > MAX_WAL_PAYLOAD_BYTES) {
+        fprintf(stderr,
+                "apply_wal_entry: payload_len %u exceeds MAX_WAL_PAYLOAD_BYTES %u\n",
+                hdr->payload_len, MAX_WAL_PAYLOAD_BYTES);
+        return -EMSGSIZE;
+    }
+
+    /* WAL_CHECKPOINT (100) is a local-only marker — standbys never receive it
+     * via replication.  Skip silently; do NOT advance LSN. */
+    if (hdr->type == WAL_CHECKPOINT)
+        return 0;
+
+    /* Unknown types above WAL_ENTRY_TYPE_MAX: log WARN and skip for forward
+     * compatibility — an older controller can boot on WAL written by a newer
+     * one without hard-failing on unrecognised entry types. */
+    if (hdr->type > WAL_ENTRY_TYPE_MAX) {
+        fprintf(stderr,
+                "WARN apply_wal_entry: unknown type %u > WAL_ENTRY_TYPE_MAX %u"
+                " (lsn=%" PRIu64 ") — skipping for forward compatibility\n",
+                hdr->type, WAL_ENTRY_TYPE_MAX, hdr->lsn);
+        return 0;
+    }
+
+    /* Advance LSN chain: track the highest replicated LSN seen. */
+    pthread_mutex_lock(&p->lock);
+    if (hdr->lsn + 1 > p->current_lsn)
+        p->current_lsn = hdr->lsn + 1;
+    p->last_crc = hdr->crc;
+    pthread_mutex_unlock(&p->lock);
+
+    /* Entity-creating entries: delegate to existing restore stubs.
+     * Gate on p->cluster != NULL so unit tests without a full cluster work. */
+    switch (hdr->type) {
+    case WAL_JOB_SUBMIT:
+        /* restore_job() is a P0 stub — full Cap'n Proto decode in P2-4+. */
+        if (p->cluster)
+            restore_job(p->cluster, 0 /* jobid decoded from payload in P2-4 */,
+                        payload, hdr->payload_len);
+        break;
+    case WAL_WORKER_CTRL:
+        /* worker_from_capnp() is a P0 stub — full decode in P2-4+. */
+        if (p->cluster)
+            worker_from_capnp(p->cluster, payload, hdr->payload_len);
+        break;
+    default:
+        /* All other known types: LSN/CRC already advanced above.
+         * Full per-type handling wired in P2-4 (Cap'n Proto decode). */
+        break;
+    }
+
     return 0;
 }
 

src/controller/testsuite/controller_test.c

@@ -24,6 +24,7 @@ extern void test_ha_cluster(void);
 extern void test_ha_redirect(void);
 extern void test_p1_integration(void);
 extern void test_repl_unit(void);
+extern void test_repl_apply(void);
 
 void setUp(void) { return; }
 
@@ -117,6 +118,7 @@ int main(int argc, char* argv[])
     TEST(test_ha_redirect);
     TEST(test_p1_integration);
     TEST(test_repl_unit);
+    TEST(test_repl_apply);
 
     return UNITY_END();
 }

src/controller/testsuite/test_repl_apply.c

@@ -0,0 +1,197 @@
+/*
+ * test_repl_apply.c — P2-2: Tests for apply_wal_entry() standby WAL replay.
+ *
+ * TDD RED phase: written BEFORE the implementation.
+ * These tests verify:
+ *   - LSN advances past replayed entry
+ *   - WAL_CHECKPOINT entries are skipped (rc=0, lsn unchanged)
+ *   - Oversized payloads are rejected (rc != 0)
+ *   - Unknown types (> WAL_ENTRY_TYPE_MAX) are skipped with rc=0
+ */
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "config.h"
+#include "persist.h"
+
+/* ------------------------------------------------------------------ */
+/* Weak stubs for P0 entity-restore functions (job.c / worker.c).      */
+/* The unit test binary does not link those translation units — they   */
+/* pull in the full cluster.h / netio.h chain.  apply_wal_entry() only */
+/* calls these when p->cluster != NULL; tests use NULL cluster, so the */
+/* weak stubs are never reached.  The linker needs the symbols to link. */
+/* ------------------------------------------------------------------ */
+typedef struct cluster_s cluster_t;
+
+__attribute__((weak))
+int restore_job(cluster_t* c, uint64_t jobid,
+                const void* capnp_payload, size_t len)
+{
+    (void)c; (void)jobid; (void)capnp_payload; (void)len;
+    return 0;
+}
+
+__attribute__((weak))
+int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len)
+{
+    (void)c; (void)capnp_payload; (void)len;
+    return 0;
+}
+
+/* ------------------------------------------------------------------ */
+/* Test helpers                                                         */
+/* ------------------------------------------------------------------ */
+
+static persist_t* make_test_persist_for_apply(void)
+{
+    persist_t* p = calloc(1, sizeof(persist_t));
+    if (!p) {
+        fprintf(stderr, "calloc failed\n");
+        exit(1);
+    }
+    persist_config_t cfg = persist_config_defaults();
+    cfg.wal_sync_ms = 5000; /* slow fsync — we don't need it in unit tests */
+
+    char tmpdir[256];
+    snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_apply_XXXXXX");
+    if (!mkdtemp(tmpdir)) {
+        fprintf(stderr, "mkdtemp failed\n");
+        free(p);
+        exit(1);
+    }
+
+    persist_set_workdir(p, tmpdir, "ctrl-apply");
+    persist_init(p, &cfg, NULL); /* NULL cluster — entity stubs gate on p->cluster */
+    persist_start(p);
+    return p;
+}
+
+static void destroy_test_persist(persist_t* p)
+{
+    persist_stop(p);
+    persist_fini(p);
+    free(p);
+}
+
+/* ------------------------------------------------------------------ */
+/* Tests                                                                */
+/* ------------------------------------------------------------------ */
+
+static void test_apply_wal_entry_advances_lsn(void)
+{
+    persist_t* p = make_test_persist_for_apply();
+
+    wal_entry_hdr_t hdr;
+    memset(&hdr, 0, sizeof(hdr));
+    hdr.type        = WAL_JOB_UPDATE;
+    hdr.lsn         = 42;
+    hdr.term        = 1;
+    hdr.payload_len = 4;
+    hdr.crc         = 0xDEADBEEF;
+    hdr.prev_crc    = WAL_CRC_NONE;
+
+    uint8_t payload[4] = {1, 2, 3, 4};
+
+    int rc = apply_wal_entry(p, &hdr, payload);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    /* current_lsn must have advanced to at least lsn+1 */
+    TEST_ASSERT_TRUE(p->current_lsn > 42);
+
+    /* last_crc must reflect the entry's crc */
+    TEST_ASSERT_EQUAL_UINT32(0xDEADBEEF, p->last_crc);
+
+    destroy_test_persist(p);
+    printf("test_apply_wal_entry_advances_lsn: PASS\n");
+}
+
+static void test_apply_wal_entry_skips_checkpoint(void)
+{
+    persist_t* p = make_test_persist_for_apply();
+    uint64_t lsn_before = p->current_lsn;
+
+    wal_entry_hdr_t hdr;
+    memset(&hdr, 0, sizeof(hdr));
+    hdr.type        = WAL_CHECKPOINT;
+    hdr.lsn         = 100;
+    hdr.term        = 1;
+    hdr.payload_len = 0;
+
+    int rc = apply_wal_entry(p, &hdr, NULL);
+    TEST_ASSERT_EQUAL_INT(0, rc); /* skip = success, not error */
+
+    /* Checkpoint entries are local-only markers — standbys never receive them.
+     * current_lsn must NOT advance: skip means no-op. */
+    TEST_ASSERT_EQUAL_UINT64(lsn_before, p->current_lsn);
+
+    destroy_test_persist(p);
+    printf("test_apply_wal_entry_skips_checkpoint: PASS\n");
+}
+
+static void test_apply_wal_entry_rejects_oversized(void)
+{
+    persist_t* p = make_test_persist_for_apply();
+
+    wal_entry_hdr_t hdr;
+    memset(&hdr, 0, sizeof(hdr));
+    hdr.type        = WAL_JOB_SUBMIT;
+    hdr.lsn         = 1;
+    hdr.term        = 1;
+    hdr.payload_len = MAX_WAL_PAYLOAD_BYTES + 1; /* too large */
+
+    int rc = apply_wal_entry(p, &hdr, NULL);
+    TEST_ASSERT_NOT_EQUAL(0, rc); /* must be rejected */
+
+    destroy_test_persist(p);
+    printf("test_apply_wal_entry_rejects_oversized: PASS\n");
+}
+
+static void test_apply_wal_entry_skips_unknown_type(void)
+{
+    persist_t* p = make_test_persist_for_apply();
+    uint64_t lsn_before = p->current_lsn;
+
+    /*
+     * Type 200 is above WAL_ENTRY_TYPE_MAX (100).
+     * Spec: log WARN and skip with rc=0 (forward compatibility).
+     */
+    wal_entry_hdr_t hdr;
+    memset(&hdr, 0, sizeof(hdr));
+    hdr.type        = 200;
+    hdr.lsn         = 5;
+    hdr.term        = 1;
+    hdr.payload_len = 0;
+
+    int rc = apply_wal_entry(p, &hdr, NULL);
+    TEST_ASSERT_EQUAL_INT(0, rc); /* unknown type: skip not error */
+
+    /* LSN must NOT advance for a skipped-unknown type */
+    TEST_ASSERT_EQUAL_UINT64(lsn_before, p->current_lsn);
+
+    destroy_test_persist(p);
+    printf("test_apply_wal_entry_skips_unknown_type: PASS\n");
+}
+
+static void test_apply_wal_entry_null_guard(void)
+{
+    /* Both NULL inputs must return non-zero (EINVAL) */
+    int rc = apply_wal_entry(NULL, NULL, NULL);
+    TEST_ASSERT_NOT_EQUAL(0, rc);
+    printf("test_apply_wal_entry_null_guard: PASS\n");
+}
+
+/* ------------------------------------------------------------------ */
+/* Suite entry point (registered in controller_test.c)                 */
+/* ------------------------------------------------------------------ */
+
+void test_repl_apply(void)
+{
+    RUN_TEST(test_apply_wal_entry_advances_lsn);
+    RUN_TEST(test_apply_wal_entry_skips_checkpoint);
+    RUN_TEST(test_apply_wal_entry_rejects_oversized);
+    RUN_TEST(test_apply_wal_entry_skips_unknown_type);
+    RUN_TEST(test_apply_wal_entry_null_guard);
+}

src/include/persist.h

@@ -167,6 +167,12 @@ int persist_queue_hook_fun(int type, int situation, int detail,
 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);
+
 /* GC retention predicates (P0-5a / P0-8 / P0-13) */
 #include <stdbool.h>
 #include "xce/types.h"