feat: P2-3 repl_wait_ack() with pthread_cond_timedwait + ACK deadlock prevention

Implements sync replication wait: primary blocks until at least one non-catching-up
standby ACKs the target LSN, or times out. Uses pthread_cond_timedwait with
CLOCK_REALTIME deadline. Inner loop rechecks all standbys after each wakeup to
guard against spurious wakeups.

ACK deadlock prevention: repl_on_ack() is called exclusively from the io_worker
read callback (not vtaskmgr), so it only does atomic_store + pthread_cond_signal —
safe to call while vcpu threads block in repl_wait_ack().

4 new TDD tests (all green, 9 total in test_repl_unit):
- test_wait_ack_no_standbys_returns_enodev
- test_wait_ack_already_acked
- test_wait_ack_timeout (verifies >=80ms elapsed)
- test_wait_ack_woken_by_background_ack (pthread thread signals after 30ms)

src/controller/repl.c

@@ -201,12 +201,61 @@ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
 
 int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms)
 {
-    (void)r;
-    (void)target_lsn;
-    (void)timeout_ms;
+    if (!r) return -EINVAL;
 
-    /* Full implementation in P2-3 */
-    return -ENODEV;
+    /* Check whether any non-catching-up standby exists */
+    mylock_lock(&r->lock);
+    int has_standby = 0;
+    for (int i = 0; i < MAX_STANDBYS; i++) {
+        if (r->standbys[i].ep != NULL &&
+            !atomic_load_explicit(&r->standbys[i].in_catch_up,
+                                  memory_order_acquire)) {
+            has_standby = 1;
+            break;
+        }
+    }
+    mylock_unlock(&r->lock);
+
+    if (!has_standby) return -ENODEV;
+
+    /* Compute absolute deadline for pthread_cond_timedwait (CLOCK_REALTIME) */
+    struct timespec deadline;
+    clock_gettime(CLOCK_REALTIME, &deadline);
+    deadline.tv_sec  += (long)(timeout_ms / 1000);
+    deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
+    if (deadline.tv_nsec >= 1000000000L) {
+        deadline.tv_sec++;
+        deadline.tv_nsec -= 1000000000L;
+    }
+
+    pthread_mutex_lock(&r->ack_lock);
+    int rc = -ETIMEDOUT;
+    while (1) {
+        /* Recheck every standby after each wakeup (spurious-wakeup guard) */
+        mylock_lock(&r->lock);
+        for (int i = 0; i < MAX_STANDBYS; i++) {
+            standby_conn_t *s = &r->standbys[i];
+            if (s->ep == NULL) continue;
+            if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire))
+                continue;
+            if (atomic_load_explicit(&s->acked_lsn, memory_order_acquire)
+                    >= target_lsn) {
+                rc = 0;
+                break;
+            }
+        }
+        mylock_unlock(&r->lock);
+        if (rc == 0) break;
+
+        int wait_rc = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock,
+                                             &deadline);
+        if (wait_rc == ETIMEDOUT) {
+            rc = -ETIMEDOUT;
+            break;
+        }
+    }
+    pthread_mutex_unlock(&r->ack_lock);
+    return rc;
 }
 
 /* ============================================================

src/controller/testsuite/test_repl_unit.c

@@ -1,7 +1,10 @@
 #include <assert.h>
+#include <errno.h>
 #include <stdint.h>
 #include <string.h>
 #include <stdio.h>
+#include <pthread.h>
+#include <time.h>
 /* Use the same Unity include pattern as other controller tests */
 #include <unity/unity.h>
 #include "repl.h"
@@ -74,6 +77,89 @@ static void test_repl_trim_lsn_disconnected_slot_expiry(void) {
     printf("test_repl_trim_lsn_disconnected_slot_expiry: PASS\n");
 }
 
+/* ============================================================
+ * P2-3: repl_wait_ack() tests
+ * ============================================================ */
+
+/* Test 1: repl_wait_ack returns -ENODEV when no standbys connected */
+static void test_wait_ack_no_standbys_returns_enodev(void) {
+    replicator_t r;
+    repl_init(&r);
+    int rc = repl_wait_ack(&r, /*target_lsn=*/10, /*timeout_ms=*/50);
+    TEST_ASSERT_EQUAL_INT(-ENODEV, rc);
+    repl_fini(&r);
+    printf("test_wait_ack_no_standbys_returns_enodev: PASS\n");
+}
+
+/* Test 2: repl_wait_ack returns 0 when standby already ACKed target LSN */
+static void test_wait_ack_already_acked(void) {
+    replicator_t r;
+    int dummy_ep = 0;
+    repl_init(&r);
+    repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001);
+    /* Pre-set acked_lsn to 20 (already past target 10) */
+    repl_on_ack(&r, (void*)&dummy_ep, 20);
+    int rc = repl_wait_ack(&r, 10, 50);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    repl_fini(&r);
+    printf("test_wait_ack_already_acked: PASS\n");
+}
+
+/* Test 3: repl_wait_ack times out when standby never ACKs */
+static void test_wait_ack_timeout(void) {
+    replicator_t r;
+    int dummy_ep = 0;
+    repl_init(&r);
+    repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001);
+    /* acked_lsn stays at 0 — never reaches target 100 */
+    struct timespec start, end;
+    clock_gettime(CLOCK_MONOTONIC, &start);
+    int rc = repl_wait_ack(&r, 100, 100 /* 100ms timeout */);
+    clock_gettime(CLOCK_MONOTONIC, &end);
+    TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc);
+    /* Verify we actually waited ~100ms (not returned immediately) */
+    long elapsed_ms = (end.tv_sec - start.tv_sec) * 1000
+                    + (end.tv_nsec - start.tv_nsec) / 1000000;
+    TEST_ASSERT_TRUE(elapsed_ms >= 80);  /* at least 80ms */
+    repl_fini(&r);
+    printf("test_wait_ack_timeout: PASS\n");
+}
+
+/* Test 4: ACK from background thread wakes repl_wait_ack */
+struct ack_args {
+    replicator_t *r;
+    void         *ep;
+    uint64_t      lsn;
+    int           delay_ms;
+};
+
+static void* ack_sender_thread(void* arg) {
+    struct ack_args* a = arg;
+    struct timespec ts = { 0, (long)a->delay_ms * 1000000L };
+    nanosleep(&ts, NULL);
+    repl_on_ack(a->r, a->ep, a->lsn);
+    return NULL;
+}
+
+static void test_wait_ack_woken_by_background_ack(void) {
+    replicator_t r;
+    int dummy_ep = 0;
+    repl_init(&r);
+    repl_add_standby(&r, (void*)&dummy_ep, "sb", 7001);
+
+    struct ack_args args = { &r, (void*)&dummy_ep, 15, 30 /* 30ms delay */ };
+    pthread_t th;
+    pthread_create(&th, NULL, ack_sender_thread, &args);
+
+    /* Wait for LSN 15 with 200ms timeout — ACK arrives after 30ms */
+    int rc = repl_wait_ack(&r, 15, 200);
+    pthread_join(th, NULL);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    repl_fini(&r);
+    printf("test_wait_ack_woken_by_background_ack: PASS\n");
+}
+
 void test_repl_unit(void) {
     UNITY_BEGIN();
     RUN_TEST(test_repl_init_zeroes_nstandbys);
@@ -81,5 +167,9 @@ void test_repl_unit(void) {
     RUN_TEST(test_repl_add_standby_max);
     RUN_TEST(test_repl_trim_lsn_no_standbys);
     RUN_TEST(test_repl_trim_lsn_disconnected_slot_expiry);
+    RUN_TEST(test_wait_ack_no_standbys_returns_enodev);
+    RUN_TEST(test_wait_ack_already_acked);
+    RUN_TEST(test_wait_ack_timeout);
+    RUN_TEST(test_wait_ack_woken_by_background_ack);
     UNITY_END();
 }