feat: P2-1 repl.h/repl.c — replicator lifecycle, slot management, trim_lsn

Add replicator_t structure with per-standby connection slots (standby_conn_t),
lifecycle (repl_init/repl_fini), slot management (repl_add_standby,
repl_remove_standby, repl_expire_disconnect_slots), ACK tracking
(repl_on_ack), WAL trim guard (repl_trim_lsn), and stubs for
repl_broadcast/repl_wait_ack (wired in P2-3/P2-4).

Uses mylock_t (pthread_mutex_t alias) with mylock_lock/mylock_unlock.
snap_interval_s==0 in repl_expire_disconnect_slots forces immediate expiry
to support deterministic unit testing.

5 new Unity tests all pass; 115 pre-existing controller tests unchanged.

src/controller/meson.build

@@ -87,8 +87,10 @@ controller_unit_test_src = [
   'testsuite/test_ha_cluster.c',
   'testsuite/test_ha_redirect.c',
   'testsuite/test_p1_integration.c',
+  'testsuite/test_repl_unit.c',
   'persist.c',
   'ha.c',
+  'repl.c',
 ]
 
 controller_unit_test_deps = [

src/controller/repl.c

@@ -0,0 +1,249 @@
+/*
+ * repl.c — P2-1: Replicator lifecycle, slot management, and WAL trim guard.
+ *
+ * mylock_t is a plain pthread_mutex_t alias (see mylock.h).
+ * API: mylock_init / mylock_fini / mylock_lock / mylock_unlock
+ */
+#include <assert.h>
+#include <errno.h>
+#include <string.h>
+
+#include "repl.h"
+#include "mylock.h"
+
+/* ============================================================
+ * Lifecycle
+ * ============================================================ */
+
+int repl_init(replicator_t *r)
+{
+    if (r == NULL) {
+        return -EINVAL;
+    }
+
+    memset(r, 0, sizeof(*r));
+
+    int rc = mylock_init(&r->lock);
+    if (rc != 0) {
+        return rc;
+    }
+
+    rc = pthread_mutex_init(&r->ack_lock, NULL);
+    if (rc != 0) {
+        mylock_fini(&r->lock);
+        return -rc;
+    }
+
+    rc = pthread_cond_init(&r->ack_cond, NULL);
+    if (rc != 0) {
+        pthread_mutex_destroy(&r->ack_lock);
+        mylock_fini(&r->lock);
+        return -rc;
+    }
+
+    return 0;
+}
+
+void repl_fini(replicator_t *r)
+{
+    if (r == NULL) {
+        return;
+    }
+
+    pthread_cond_destroy(&r->ack_cond);
+    pthread_mutex_destroy(&r->ack_lock);
+    mylock_fini(&r->lock);
+}
+
+/* ============================================================
+ * Standby slot management
+ * ============================================================ */
+
+int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t port)
+{
+    if (r == NULL || ep == NULL || host == NULL) {
+        return -EINVAL;
+    }
+
+    mylock_lock(&r->lock);
+
+    if (r->nstandbys >= MAX_STANDBYS) {
+        mylock_unlock(&r->lock);
+        return -ENOSPC;
+    }
+
+    /* Find first empty slot */
+    int idx = -1;
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        if (r->standbys[i].ep == NULL) {
+            idx = i;
+            break;
+        }
+    }
+
+    if (idx < 0) {
+        /* Should not happen given nstandbys guard above, but be safe */
+        mylock_unlock(&r->lock);
+        return -ENOSPC;
+    }
+
+    standby_conn_t *s = &r->standbys[idx];
+    memset(s, 0, sizeof(*s));
+    s->ep   = ep;
+    s->port = port;
+    strncpy(s->host, host, XCE_MAXNAME_LEN - 1);
+    s->host[XCE_MAXNAME_LEN - 1] = '\0';
+    atomic_store(&s->acked_lsn, 0);
+    atomic_store(&s->in_catch_up, 0);
+
+    r->nstandbys++;
+
+    mylock_unlock(&r->lock);
+    return idx;
+}
+
+void repl_remove_standby(replicator_t *r, repl_ep_t ep)
+{
+    if (r == NULL || ep == NULL) {
+        return;
+    }
+
+    mylock_lock(&r->lock);
+
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        standby_conn_t *s = &r->standbys[i];
+        if (s->ep == ep) {
+            s->last_disconnected_lsn = atomic_load(&s->acked_lsn);
+            s->last_disconnect_time  = time(NULL);
+            s->ep = NULL;
+            r->nstandbys--;
+            break;
+        }
+    }
+
+    mylock_unlock(&r->lock);
+}
+
+void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s)
+{
+    if (r == NULL) {
+        return;
+    }
+
+    mylock_lock(&r->lock);
+
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        standby_conn_t *s = &r->standbys[i];
+        /* Only process disconnected (ep==NULL) slots that still hold a guard LSN */
+        if (s->ep == NULL && s->last_disconnected_lsn != 0) {
+            /* snap_interval_s==0 means expire immediately regardless of wall clock */
+            int expired = (snap_interval_s == 0) ||
+                          (now_s >= s->last_disconnect_time + (time_t)snap_interval_s);
+            if (expired) {
+                memset(s, 0, sizeof(*s));
+            }
+        }
+    }
+
+    mylock_unlock(&r->lock);
+}
+
+/* ============================================================
+ * ACK handling — call from io_worker read callback ONLY
+ * ============================================================ */
+
+void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn)
+{
+    if (r == NULL || ep == NULL) {
+        return;
+    }
+
+    mylock_lock(&r->lock);
+
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        standby_conn_t *s = &r->standbys[i];
+        if (s->ep == ep) {
+            atomic_store(&s->acked_lsn, acked_lsn);
+            break;
+        }
+    }
+
+    mylock_unlock(&r->lock);
+
+    /* Signal any waiter (repl_wait_ack) */
+    pthread_mutex_lock(&r->ack_lock);
+    pthread_cond_signal(&r->ack_cond);
+    pthread_mutex_unlock(&r->ack_lock);
+}
+
+/* ============================================================
+ * Replication send — stub in P2-1; wired in P2-4
+ * ============================================================ */
+
+int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
+                   const void *payload, uint32_t payload_len)
+{
+    (void)r;
+    (void)payload;
+    (void)payload_len;
+
+    /* WAL_CHECKPOINT entries must never be replicated */
+    assert(hdr != NULL);
+    assert(hdr->type != WAL_CHECKPOINT);
+
+    /* Full implementation in P2-4 */
+    return 0;
+}
+
+/* ============================================================
+ * Sync replication — stub in P2-1; wired in P2-3
+ * ============================================================ */
+
+int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms)
+{
+    (void)r;
+    (void)target_lsn;
+    (void)timeout_ms;
+
+    /* Full implementation in P2-3 */
+    return -ENODEV;
+}
+
+/* ============================================================
+ * WAL trim guard
+ * ============================================================ */
+
+uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn,
+                        uint32_t snap_interval_s)
+{
+    (void)snap_interval_s;
+
+    if (r == NULL) {
+        return checkpoint_lsn;
+    }
+
+    uint64_t min_lsn = checkpoint_lsn;
+
+    mylock_lock(&r->lock);
+
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        const standby_conn_t *s = &r->standbys[i];
+
+        if (s->ep != NULL) {
+            /* Active standby: guard at its last acked LSN */
+            uint64_t acked = atomic_load(&s->acked_lsn);
+            if (acked < min_lsn) {
+                min_lsn = acked;
+            }
+        } else if (s->last_disconnected_lsn != 0) {
+            /* Recently disconnected slot still holding a guard */
+            if (s->last_disconnected_lsn < min_lsn) {
+                min_lsn = s->last_disconnected_lsn;
+            }
+        }
+    }
+
+    mylock_unlock(&r->lock);
+
+    return min_lsn;
+}

src/controller/repl.h

@@ -0,0 +1,59 @@
+#pragma once
+/*
+ * repl.h — P2-1: Replicator structures and public API.
+ * Fan-out TCP replication via existing iomgr/SSL endpoint pipeline.
+ * ACKs processed in io_worker read callback (NOT vtaskmgr) to prevent deadlock.
+ */
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdint.h>
+#include <time.h>
+#include "config.h"
+#include "mylock.h"
+#include "persist.h"   /* wal_entry_hdr_t */
+
+/* Endpoint is an opaque pointer — full type in netio.h (avoid heavy include) */
+typedef void* repl_ep_t;
+
+/* Per-standby connection slot */
+typedef struct {
+    repl_ep_t          ep;              /* NULL = slot empty */
+    _Atomic uint64_t   acked_lsn;      /* Last LSN ACKed by this standby */
+    _Atomic int        in_catch_up;    /* Non-zero while catching up */
+    uint64_t           join_applied_lsn; /* LSN standby reported on reconnect */
+    char               host[XCE_MAXNAME_LEN];
+    uint32_t           port;
+    uint64_t           last_disconnected_lsn;
+    time_t             last_disconnect_time;
+} standby_conn_t;
+
+/* Replicator — embedded in persist_t or cluster_t (P2-4) */
+typedef struct {
+    standby_conn_t  standbys[MAX_STANDBYS];
+    int             nstandbys;          /* Active standby count */
+    mylock_t        lock;               /* Protects standbys[], nstandbys */
+    pthread_cond_t  ack_cond;           /* Signaled when any ACK arrives */
+    pthread_mutex_t ack_lock;           /* Guards ack_cond */
+} replicator_t;
+
+/* Lifecycle */
+int  repl_init(replicator_t *r);
+void repl_fini(replicator_t *r);
+
+/* Standby management */
+int  repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t port);
+void repl_remove_standby(replicator_t *r, repl_ep_t ep);
+void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s);
+
+/* Replication send (stub in P2-1; wired in P2-4) */
+int  repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
+                    const void *payload, uint32_t payload_len);
+
+/* ACK handling — call from io_worker read callback ONLY */
+void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn);
+
+/* Sync replication (stub in P2-1; wired in P2-3) */
+int  repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms);
+
+/* WAL trim guard */
+uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, uint32_t snap_interval_s);

src/controller/testsuite/controller_test.c

@@ -23,6 +23,7 @@ extern void test_ha_role(void);
 extern void test_ha_cluster(void);
 extern void test_ha_redirect(void);
 extern void test_p1_integration(void);
+extern void test_repl_unit(void);
 
 void setUp(void) { return; }
 
@@ -115,6 +116,7 @@ int main(int argc, char* argv[])
     TEST(test_ha_cluster);
     TEST(test_ha_redirect);
     TEST(test_p1_integration);
+    TEST(test_repl_unit);
 
     return UNITY_END();
 }

src/controller/testsuite/test_repl_unit.c

@@ -0,0 +1,85 @@
+#include <assert.h>
+#include <stdint.h>
+#include <string.h>
+#include <stdio.h>
+/* Use the same Unity include pattern as other controller tests */
+#include <unity/unity.h>
+#include "repl.h"
+#include "config.h"
+
+static void test_repl_init_zeroes_nstandbys(void) {
+    replicator_t r;
+    TEST_ASSERT_EQUAL_INT(0, repl_init(&r));
+    TEST_ASSERT_EQUAL_INT(0, r.nstandbys);
+    repl_fini(&r);
+    printf("test_repl_init_zeroes_nstandbys: PASS\n");
+}
+
+static void test_repl_add_remove_standby(void) {
+    replicator_t r;
+    /* Use a dummy int as the endpoint pointer — repl.c only compares ep pointers */
+    int dummy_ep = 0;
+    repl_init(&r);
+    int idx = repl_add_standby(&r, (void*)&dummy_ep, "standby1", 7001);
+    TEST_ASSERT_TRUE(idx >= 0);
+    TEST_ASSERT_EQUAL_INT(1, r.nstandbys);
+    repl_remove_standby(&r, (void*)&dummy_ep);
+    TEST_ASSERT_EQUAL_INT(0, r.nstandbys);
+    repl_fini(&r);
+    printf("test_repl_add_remove_standby: PASS\n");
+}
+
+static void test_repl_add_standby_max(void) {
+    replicator_t r;
+    int eps[MAX_STANDBYS + 1];
+    repl_init(&r);
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        int rc = repl_add_standby(&r, (void*)&eps[i], "sb", 7000 + i);
+        TEST_ASSERT_TRUE(rc >= 0);
+    }
+    /* One over limit must fail */
+    int rc = repl_add_standby(&r, (void*)&eps[MAX_STANDBYS], "sb_extra", 9999);
+    TEST_ASSERT_TRUE(rc < 0);
+    repl_fini(&r);
+    printf("test_repl_add_standby_max: PASS\n");
+}
+
+static void test_repl_trim_lsn_no_standbys(void) {
+    replicator_t r;
+    repl_init(&r);
+    uint64_t trim = repl_trim_lsn(&r, 42, 60);
+    TEST_ASSERT_EQUAL_UINT64(42, trim);
+    repl_fini(&r);
+    printf("test_repl_trim_lsn_no_standbys: PASS\n");
+}
+
+static void test_repl_trim_lsn_disconnected_slot_expiry(void) {
+    replicator_t r;
+    int dummy_ep = 0;
+    repl_init(&r);
+    repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001);
+    repl_on_ack(&r, (void*)&dummy_ep, 10);
+    repl_remove_standby(&r, (void*)&dummy_ep);
+
+    /* Immediately after disconnect: trim guard holds at last_disconnected_lsn=10 */
+    uint64_t trim = repl_trim_lsn(&r, 50, 60);
+    TEST_ASSERT_EQUAL_UINT64(10, trim);
+
+    /* After expiry (snap_interval_s=0 forces immediate expiry): trim == checkpoint */
+    repl_expire_disconnect_slots(&r, 999999, 0);
+    trim = repl_trim_lsn(&r, 50, 0);
+    TEST_ASSERT_EQUAL_UINT64(50, trim);
+
+    repl_fini(&r);
+    printf("test_repl_trim_lsn_disconnected_slot_expiry: PASS\n");
+}
+
+void test_repl_unit(void) {
+    UNITY_BEGIN();
+    RUN_TEST(test_repl_init_zeroes_nstandbys);
+    RUN_TEST(test_repl_add_remove_standby);
+    RUN_TEST(test_repl_add_standby_max);
+    RUN_TEST(test_repl_trim_lsn_no_standbys);
+    RUN_TEST(test_repl_trim_lsn_disconnected_slot_expiry);
+    UNITY_END();
+}