feat: P2-4 wire repl_broadcast into persist_wal_async/sync_append; last_replicated_crc

- Move repl.h from controller/ to include/ (shared header) to break
  persist.h ↔ repl.h circular include
- Extract wal_entry_hdr_t + wal_entry_type into include/wal_types.h so
  repl.h can include only what it needs without pulling in all of persist.h
- Add replicator_t repl field to persist_t (embedded by value, P2-4)
- Wire repl_init/repl_fini into persist_init/persist_fini lifecycle
- Wire repl_broadcast into persist_wal_async after each WAL write;
  last_replicated_crc updated here (not in persist_write_checkpoint_marker)
- Wire repl_wait_ack into persist_wal_sync_append; returns 0 on local
  WAL success; sync_rc_out carries -ENODEV/-ETIMEDOUT for degraded mode
- Add 4 TDD tests in test_persist_repl.c: structural repl field check,
  async no-standbys succeeds, sync degraded mode rc, checkpoint CRC invariant

src/controller/meson.build

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

src/controller/persist.c

@@ -168,6 +168,7 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
     pthread_mutex_init(&p->lock, NULL);
     pthread_mutex_init(&p->checkpoint_mutex, NULL);
     pthread_cond_init(&p->checkpoint_done, NULL);
+    repl_init(&p->repl);   /* P2-4: fan-out replicator lifecycle */
 
     fprintf(stderr,
             "INFO: WAL/snapshot stored unencrypted in %s — "
@@ -214,6 +215,7 @@ void persist_stop(persist_t* p)
 
 void persist_fini(persist_t* p)
 {
+    repl_fini(&p->repl);   /* P2-4: fan-out replicator lifecycle */
     pthread_mutex_destroy(&p->lock);
     pthread_mutex_destroy(&p->checkpoint_mutex);
     pthread_cond_destroy(&p->checkpoint_done);
@@ -274,8 +276,15 @@ int persist_wal_async(persist_t* p, uint32_t type,
     p->seg_offset          += entry_size;
     p->total_wal_bytes     += entry_size;
     p->last_crc             = hdr.crc;
-    /* For single-controller P0 there is no replication; track same value */
-    p->last_replicated_crc  = hdr.crc;
+
+    /* Replicate to standbys — never replicate WAL_CHECKPOINT entries.
+     * last_replicated_crc is distinct from last_crc:
+     * last_crc includes checkpoint markers; last_replicated_crc skips them.
+     * This maintains a continuous CRC chain for standbys (design doc Part 2.3).
+     * Note: WAL_CHECKPOINT is blocked by the assert at function entry, so
+     * every entry that reaches here is a replicable entry. */
+    p->last_replicated_crc = hdr.crc;
+    repl_broadcast(&p->repl, &hdr, payload, len);
 
     /* Rotate segment when full */
     if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) {
@@ -845,9 +854,26 @@ int persist_wal_sync_append(persist_t* p, uint32_t type,
                             int* sync_rc_out)
 {
     int rc = persist_wal_async(p, type, payload, len);
+    if (rc != 0) {
+        if (sync_rc_out)
+            *sync_rc_out = rc;
+        return rc;
+    }
+
+    /* For sync replication (WAL_JOB_SUBMIT): wait for at least one standby ACK.
+     * We use current_lsn - 1 because persist_wal_async already incremented it. */
+    uint64_t written_lsn;
+    pthread_mutex_lock(&p->lock);
+    written_lsn = p->current_lsn - 1;
+    pthread_mutex_unlock(&p->lock);
+
+    int sync_err = repl_wait_ack(&p->repl, written_lsn,
+                                 p->config.sync_timeout_ms);
     if (sync_rc_out)
-        *sync_rc_out = rc;
-    return rc;
+        *sync_rc_out = sync_err;
+    /* -ETIMEDOUT = degraded mode (local-only durability); -ENODEV = no standbys.
+     * Both are acceptable — primary continues operating. */
+    return 0;
 }
 
 void persist_log_entry_metadata(const wal_entry_hdr_t* hdr)

src/controller/testsuite/controller_test.c

@@ -25,6 +25,7 @@ extern void test_ha_redirect(void);
 extern void test_p1_integration(void);
 extern void test_repl_unit(void);
 extern void test_repl_apply(void);
+extern void test_persist_repl(void);
 
 void setUp(void) { return; }
 
@@ -119,6 +120,7 @@ int main(int argc, char* argv[])
     TEST(test_p1_integration);
     TEST(test_repl_unit);
     TEST(test_repl_apply);
+    TEST(test_persist_repl);
 
     return UNITY_END();
 }

src/controller/testsuite/test_persist_repl.c

@@ -0,0 +1,145 @@
+/*
+ * test_persist_repl.c — P2-4: Tests for repl_broadcast wired into WAL write paths.
+ *
+ * TDD RED phase: these tests must FAIL before implementation.
+ * After implementation they must all PASS.
+ *
+ * Tests verify:
+ *  1. persist_t has repl field (structural/compile-time check)
+ *  2. persist_wal_async succeeds with zero standbys
+ *  3. persist_wal_sync_append returns 0 (WAL success) but sets sync_rc < 0 (degraded)
+ *  4. last_replicated_crc NOT updated by WAL_CHECKPOINT (checkpoint marker)
+ */
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#include <unity/unity.h>
+#include "persist.h"
+#include "repl.h"
+
+/* ------------------------------------------------------------------ */
+/* Helpers                                                              */
+/* ------------------------------------------------------------------ */
+
+static persist_t* make_repl_persist(const char* suffix)
+{
+    persist_t* p = calloc(1, sizeof(persist_t));
+    persist_config_t cfg = persist_config_defaults();
+    cfg.wal_sync_ms      = 5000;
+    cfg.sync_timeout_ms  = 50;   /* short timeout so sync tests don't hang */
+
+    char tmpdir[256];
+    snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_repl_%s_XXXXXX", suffix);
+    if (!mkdtemp(tmpdir)) {
+        free(p);
+        return NULL;
+    }
+
+    persist_set_workdir(p, tmpdir, "ctrl");
+    persist_init(p, &cfg, NULL);
+    persist_start(p);
+    return p;
+}
+
+static void teardown_repl_persist(persist_t* p)
+{
+    if (!p) return;
+    persist_stop(p);
+    persist_fini(p);
+    free(p);
+}
+
+/* ------------------------------------------------------------------ */
+/* Tests                                                                */
+/* ------------------------------------------------------------------ */
+
+/*
+ * Structural: persist_t must contain a repl field of type replicator_t.
+ * This is a compile-time check — accessing p.repl.nstandbys must compile.
+ */
+static void test_persist_has_repl_field(void)
+{
+    persist_t p;
+    memset(&p, 0, sizeof(p));
+    /* Accessing repl.nstandbys is the compile-time guard */
+    TEST_ASSERT_EQUAL_INT(0, p.repl.nstandbys);
+    printf("test_persist_has_repl_field: PASS\n");
+}
+
+/*
+ * WAL async must succeed when there are zero standbys.
+ * repl_broadcast is a stub; no replication occurs, but write must not fail.
+ */
+static void test_wal_async_no_standbys_succeeds(void)
+{
+    persist_t* p = make_repl_persist("async");
+    TEST_ASSERT_NOT_NULL(p);
+
+    int rc = persist_wal_async(p, WAL_JOB_UPDATE, "payload", 7);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    teardown_repl_persist(p);
+    printf("test_wal_async_no_standbys_succeeds: PASS\n");
+}
+
+/*
+ * persist_wal_sync_append must return 0 (WAL write succeeded locally) even
+ * when there are no standbys (degraded mode).  The sync_rc_out parameter
+ * must be set to a negative value to signal degraded mode
+ * (-ENODEV = no standbys, or -ETIMEDOUT = timeout).
+ */
+static void test_wal_sync_no_standbys_returns_enodev_in_sync_rc(void)
+{
+    persist_t* p = make_repl_persist("sync");
+    TEST_ASSERT_NOT_NULL(p);
+
+    int sync_rc = 0;
+    int rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, "job_payload", 11,
+                                     &sync_rc);
+    /* WAL write itself must succeed */
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    /* sync_rc: -ENODEV (no standbys) or -ETIMEDOUT — either indicates degraded */
+    TEST_ASSERT_TRUE(sync_rc < 0);
+
+    teardown_repl_persist(p);
+    printf("test_wal_sync_no_standbys_returns_enodev_in_sync_rc: PASS\n");
+}
+
+/*
+ * WAL_CHECKPOINT (via persist_checkpoint) must NOT update last_replicated_crc.
+ * Checkpoint markers are local-only; the replication CRC chain must remain
+ * unbroken for standbys (design doc Part 2.3).
+ */
+static void test_last_replicated_crc_not_updated_for_checkpoint(void)
+{
+    persist_t* p = make_repl_persist("crc");
+    TEST_ASSERT_NOT_NULL(p);
+
+    uint32_t crc_before = p->last_replicated_crc;
+
+    /* persist_checkpoint writes a WAL_CHECKPOINT marker then triggers a snap.
+     * Neither the marker write nor snap must update last_replicated_crc. */
+    persist_checkpoint(p);
+
+    /* last_replicated_crc must be unchanged from before the checkpoint */
+    TEST_ASSERT_EQUAL_UINT32(crc_before, p->last_replicated_crc);
+
+    teardown_repl_persist(p);
+    printf("test_last_replicated_crc_not_updated_for_checkpoint: PASS\n");
+}
+
+/* ------------------------------------------------------------------ */
+/* Entry point                                                          */
+/* ------------------------------------------------------------------ */
+
+void test_persist_repl(void)
+{
+    RUN_TEST(test_persist_has_repl_field);
+    RUN_TEST(test_wal_async_no_standbys_succeeds);
+    RUN_TEST(test_wal_sync_no_standbys_returns_enodev_in_sync_rc);
+    RUN_TEST(test_last_replicated_crc_not_updated_for_checkpoint);
+}

src/include/persist.h

@@ -6,44 +6,12 @@
 #include <stdint.h>
 
 #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;
 
-/* ============================================================
- * WAL entry header — 40 bytes, 8-byte aligned (Part 2.2)
- * ============================================================ */
-typedef struct {
-    uint64_t term;        /* Election term (incremented on failover) */
-    uint64_t lsn;         /* Log sequence number within this term */
-    uint64_t timestamp;   /* Epoch microseconds */
-    uint32_t type;        /* wal_entry_type enum value */
-    uint32_t payload_len; /* Bytes of payload following this header */
-    uint32_t crc;         /* CRC32(header[0..WAL_HDR_CRC_SIZE-1] + payload) */
-    uint32_t prev_crc;    /* CRC of previous record; WAL_CRC_NONE for first in segment */
-} wal_entry_hdr_t;
-_Static_assert(sizeof(wal_entry_hdr_t) == 40,
-               "wal_entry_hdr_t must be 40 bytes for on-disk format stability");
-
-/* ============================================================
- * WAL entry types (Part 2.3)
- * ============================================================ */
-typedef enum {
-    WAL_JOB_SUBMIT  = 1,  /* SYNC replication */
-    WAL_JOB_START   = 2,
-    WAL_JOB_FINISH  = 3,
-    WAL_JOB_FAIL    = 4,
-    WAL_JOB_STOP    = 5,
-    WAL_JOB_RESUME  = 6,
-    WAL_JOB_UPDATE  = 7,
-    WAL_JOB_RUSAGE  = 8,
-    WAL_JOB_CKPT    = 9,
-    WAL_JOB_PREASON = 10,
-    WAL_QUEUE_CTRL  = 20,
-    WAL_WORKER_CTRL = 30,
-    WAL_CHECKPOINT  = 100, /* Local WAL marker only — never replicated */
-} wal_entry_type;
-
 /* ============================================================
  * Persistence configuration (Part 2.6)
  * ============================================================ */
@@ -120,6 +88,9 @@ struct persist_s {
     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;
 

src/include/repl.h

@@ -10,7 +10,7 @@
 #include <time.h>
 #include "config.h"
 #include "mylock.h"
-#include "persist.h"   /* wal_entry_hdr_t */
+#include "wal_types.h"  /* wal_entry_hdr_t — P2-4: use wal_types.h to break circular include */
 
 /* Endpoint is an opaque pointer — full type in netio.h (avoid heavy include) */
 typedef void* repl_ep_t;
@@ -27,8 +27,9 @@ typedef struct {
     time_t             last_disconnect_time;
 } standby_conn_t;
 
-/* Replicator — embedded in persist_t or cluster_t (P2-4) */
-typedef struct {
+/* Replicator — embedded in persist_t or cluster_t (P2-4).
+ * Named tag replicator_s matches the forward declaration in persist.h. */
+typedef struct replicator_s {
     standby_conn_t  standbys[MAX_STANDBYS];
     int             nstandbys;          /* Active standby count */
     mylock_t        lock;               /* Protects standbys[], nstandbys */

src/include/wal_types.h

@@ -0,0 +1,41 @@
+#pragma once
+/*
+ * wal_types.h — WAL on-disk structures shared between persist.h and repl.h.
+ * Extracted in P2-4 to break the persist.h ↔ repl.h circular include.
+ */
+#include <stdint.h>
+#include "config.h"
+
+/* ============================================================
+ * WAL entry header — 40 bytes, 8-byte aligned (Part 2.2)
+ * ============================================================ */
+typedef struct {
+    uint64_t term;        /* Election term (incremented on failover) */
+    uint64_t lsn;         /* Log sequence number within this term */
+    uint64_t timestamp;   /* Epoch microseconds */
+    uint32_t type;        /* wal_entry_type enum value */
+    uint32_t payload_len; /* Bytes of payload following this header */
+    uint32_t crc;         /* CRC32(header[0..WAL_HDR_CRC_SIZE-1] + payload) */
+    uint32_t prev_crc;    /* CRC of previous record; WAL_CRC_NONE for first in segment */
+} wal_entry_hdr_t;
+_Static_assert(sizeof(wal_entry_hdr_t) == 40,
+               "wal_entry_hdr_t must be 40 bytes for on-disk format stability");
+
+/* ============================================================
+ * WAL entry types (Part 2.3)
+ * ============================================================ */
+typedef enum {
+    WAL_JOB_SUBMIT  = 1,  /* SYNC replication */
+    WAL_JOB_START   = 2,
+    WAL_JOB_FINISH  = 3,
+    WAL_JOB_FAIL    = 4,
+    WAL_JOB_STOP    = 5,
+    WAL_JOB_RESUME  = 6,
+    WAL_JOB_UPDATE  = 7,
+    WAL_JOB_RUSAGE  = 8,
+    WAL_JOB_CKPT    = 9,
+    WAL_JOB_PREASON = 10,
+    WAL_QUEUE_CTRL  = 20,
+    WAL_WORKER_CTRL = 30,
+    WAL_CHECKPOINT  = 100, /* Local WAL marker only — never replicated */
+} wal_entry_type;