feat(config-repl): implement wait_all_with_timeout() for push fan-out

Replaces blocking pthread_join() per thread with a CLOCK_MONOTONIC
deadline + 5ms poll loop. CONF_PUSH_TIMEOUT_MS (500ms) now enforced.
Matches design §6.3 aggregate timeout guarantee.

- Add done field to push_ctx_t; push_thread sets it on exit
- Add static wait_all_with_timeout() before push_config_to_standbys()
- Wire push_config_to_standbys() to use wait_all_with_timeout()
- Add #include <stdatomic.h> to config_repl.c
- Guard push_ctx_t local typedef with #ifndef XCE_TESTING
- Expose push_ctx_t and wait_all_with_timeout_for_test() in header
  under XCE_TESTING for unit test access
- Add test_wait_all_with_timeout_slow_threads (timeout fires, acked=0)
- Add test_wait_all_with_timeout_fast_threads (all complete, acked=3)
- Update design doc §6.3 with actual implementation signature and
  5ms poll interval note

src/controller/config_repl.c

@@ -7,6 +7,7 @@
 #include <limits.h>
 #include <openssl/sha.h>
 #include <pthread.h>
+#include <stdatomic.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -779,6 +780,9 @@ static void send_upgrade_error(endpoint_t *ep, xce_request_t *req,
 
 /* ── push_config_to_standbys ─────────────────────── */
 
+/* push_ctx_t is exposed in config_repl.h under XCE_TESTING so tests can
+ * construct synthetic contexts.  In production builds, define it locally. */
+#ifndef XCE_TESTING
 typedef struct {
     endpoint_t      *ep;
     xce_peerinfo_t  *from;
@@ -788,7 +792,9 @@ typedef struct {
     const uint8_t   *sha256;
     int              acked;
     int              started;
+    int              done;     /* set by push_thread on exit */
 } push_ctx_t;
+#endif
 
 static void *push_thread(void *arg)
 {
@@ -805,9 +811,71 @@ static void *push_thread(void *arg)
     else
         WARN("push_config_to_standbys: push to endpoint %p failed: %d", ctx->ep, rc);
 
+    ctx->done = 1;
     return NULL;
 }
 
+/* wait_all_with_timeout: join all push threads within a CLOCK_MONOTONIC deadline.
+ *
+ * Polls until all started threads have set ctx->done or the deadline expires,
+ * then signals *done = 1 (for future callers that check an early-exit flag),
+ * and joins all started threads (they exit quickly once ctx->done is set).
+ *
+ * Returns the number of threads that had acked before the deadline.
+ */
+static int wait_all_with_timeout(pthread_t *threads, push_ctx_t *ctxs,
+                                  int n, uint32_t timeout_ms, atomic_int *done)
+{
+    struct timespec deadline;
+    clock_gettime(CLOCK_MONOTONIC, &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;
+    }
+
+    /* Poll until deadline or all started threads have set their done flag */
+    for (;;) {
+        struct timespec now;
+        clock_gettime(CLOCK_MONOTONIC, &now);
+        if (now.tv_sec > deadline.tv_sec ||
+            (now.tv_sec == deadline.tv_sec &&
+             now.tv_nsec >= deadline.tv_nsec))
+            break;
+
+        int pending = 0;
+        for (int i = 0; i < n; i++) {
+            if (ctxs[i].started && !ctxs[i].done)
+                pending++;
+        }
+        if (pending == 0)
+            break;
+
+        usleep(5000);   /* 5ms poll */
+    }
+
+    atomic_store(done, 1);   /* signal threads that deadline has passed */
+
+    int acked = 0;
+    for (int i = 0; i < n; i++) {
+        if (ctxs[i].started) {
+            pthread_join(threads[i], NULL);
+            if (ctxs[i].acked)
+                acked++;
+        }
+    }
+    return acked;
+}
+
+#ifdef XCE_TESTING
+int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs,
+                                    int n, uint32_t timeout_ms, atomic_int *done)
+{
+    return wait_all_with_timeout(threads, ctxs, n, timeout_ms, done);
+}
+#endif
+
 int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
                             const uint8_t sha256[32], uint64_t new_tag)
 {
@@ -847,13 +915,11 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
             WARN("push_config_to_standbys: pthread_create failed for standby %d", i);
     }
 
-    int acked = 0;
-    for (int i = 0; i < n; i++) {
-        if (!ctxs[i].started)
-            continue;
-        pthread_join(threads[i], NULL);
-        if (ctxs[i].acked) acked++;
-    }
+    atomic_int push_done;
+    atomic_init(&push_done, 0);
+
+    int acked = wait_all_with_timeout(threads, ctxs, n,
+                                       CONF_PUSH_TIMEOUT_MS, &push_done);
 
     INFO("Config pushed to %d/%d standbys", acked, n);
 

src/controller/config_repl.h

@@ -128,6 +128,9 @@ int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req);
 char *sanitise_cn(const char *src, size_t maxlen);
 
 #ifdef XCE_TESTING
+#include <pthread.h>
+#include <stdatomic.h>
+
 /*
  * load_primary_config_for_test: test-only wrapper that exercises the path
  * resolution logic of load_primary_config() without calling FATAL.
@@ -135,4 +138,31 @@ char *sanitise_cn(const char *src, size_t maxlen);
  * otherwise. Does NOT perform Cap'n Proto decode or SHA-256 computation.
  */
 int load_primary_config_for_test(cluster_t *c);
+
+/*
+ * push_ctx_t: exposed under XCE_TESTING so unit tests can construct
+ * synthetic push contexts to drive wait_all_with_timeout_for_test().
+ * In production builds this type is private to config_repl.c.
+ */
+typedef struct {
+    endpoint_t      *ep;
+    xce_peerinfo_t  *from;
+    uint64_t         new_tag;
+    const uint8_t   *bytes;
+    size_t           blen;
+    const uint8_t   *sha256;
+    int              acked;
+    int              started;
+    int              done;
+} push_ctx_t;
+
+/*
+ * wait_all_with_timeout_for_test: test-only shim exposing the static
+ * wait_all_with_timeout() helper for unit tests.
+ * Joins all started push threads within timeout_ms (CLOCK_MONOTONIC).
+ * Sets *done = 1 when deadline expires. Returns number of acked threads.
+ */
+int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs,
+                                    int n, uint32_t timeout_ms,
+                                    atomic_int *done);
 #endif

src/controller/testsuite/test_config_repl.c

@@ -5,6 +5,8 @@
 #include <errno.h>
 #include <fcntl.h>
 #include <openssl/sha.h>
+#include <pthread.h>
+#include <stdatomic.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -482,6 +484,111 @@ static void test_conf_save_config_path_consistency(void)
     printf("test_conf_save_config_path_consistency: PASS\n");
 }
 
+/* ── wait_all_with_timeout (XCE_TESTING shim) ───── */
+
+#ifdef XCE_TESTING
+
+/* Thread args for timeout tests */
+typedef struct {
+    atomic_int *stop;   /* poll this; exit when set to allow pthread_join to return */
+    push_ctx_t *ctx;
+} timeout_test_arg_t;
+
+/* Polls stop flag at 5ms intervals; exits when stop is set WITHOUT setting acked.
+ * This simulates a synchronous push that respects the done signal. */
+static void *slow_push_thread(void *arg)
+{
+    timeout_test_arg_t *a = (timeout_test_arg_t *)arg;
+    while (!atomic_load(a->stop))
+        usleep(5000);
+    /* Thread exits without setting acked — deadline fired before completion */
+    a->ctx->done = 1;
+    return NULL;
+}
+
+static void *fast_push_thread(void *arg)
+{
+    push_ctx_t *ctx = (push_ctx_t *)arg;
+    ctx->acked = 1;
+    ctx->done  = 1;
+    return NULL;
+}
+
+static void test_wait_all_with_timeout_slow_threads(void)
+{
+    /* 2 threads that poll the done flag and exit without acking.
+     * Timeout is 100ms. Expect: acked==0, done==1, returns within ~200ms. */
+    const int N = 2;
+    pthread_t          threads[2];
+    push_ctx_t         ctxs[2];
+    timeout_test_arg_t args[2];
+    atomic_int         done;
+
+    memset(ctxs, 0, sizeof(ctxs));
+    atomic_init(&done, 0);
+
+    for (int i = 0; i < N; i++) {
+        ctxs[i].started = 1;
+        args[i].stop    = &done;
+        args[i].ctx     = &ctxs[i];
+        pthread_create(&threads[i], NULL, slow_push_thread, &args[i]);
+    }
+
+    struct timespec t0, t1;
+    clock_gettime(CLOCK_MONOTONIC, &t0);
+    int acked = wait_all_with_timeout_for_test(threads, ctxs, N, 100, &done);
+    clock_gettime(CLOCK_MONOTONIC, &t1);
+
+    long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000
+                    + (t1.tv_nsec - t0.tv_nsec) / 1000000;
+
+    /* Threads did not ack within the 100ms timeout */
+    TEST_ASSERT_EQUAL_INT(0, acked);
+    /* done flag must be set by wait_all_with_timeout */
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&done));
+    /* Total elapsed should be close to the 100ms timeout + small join overhead */
+    TEST_ASSERT_LESS_THAN(400, (int)elapsed_ms);
+
+    printf("test_wait_all_with_timeout_slow_threads: PASS (elapsed=%ldms acked=%d)\n",
+           elapsed_ms, acked);
+}
+
+static void test_wait_all_with_timeout_fast_threads(void)
+{
+    /* 3 threads that complete immediately; timeout is 200ms.
+     * Expect: all 3 acked, returns quickly (well under 200ms). */
+    const int N = 3;
+    pthread_t  threads[3];
+    push_ctx_t ctxs[3];
+    atomic_int done;
+
+    memset(ctxs, 0, sizeof(ctxs));
+    atomic_init(&done, 0);
+
+    for (int i = 0; i < N; i++) {
+        ctxs[i].started = 1;
+        pthread_create(&threads[i], NULL, fast_push_thread, &ctxs[i]);
+    }
+
+    struct timespec t0, t1;
+    clock_gettime(CLOCK_MONOTONIC, &t0);
+    int acked = wait_all_with_timeout_for_test(threads, ctxs, N, 200, &done);
+    clock_gettime(CLOCK_MONOTONIC, &t1);
+
+    long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000
+                    + (t1.tv_nsec - t0.tv_nsec) / 1000000;
+
+    TEST_ASSERT_EQUAL_INT(N, acked);
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&done));
+    /* Should have returned long before the 200ms timeout */
+    TEST_ASSERT_LESS_THAN(200, (int)elapsed_ms);
+
+    printf("test_wait_all_with_timeout_fast_threads: PASS (elapsed=%ldms acked=%d)\n",
+           elapsed_ms, acked);
+}
+
+#endif /* XCE_TESTING */
+
 /* ── dispatch ───────────────────────────────────── */
 
 void test_config_repl(void)
@@ -506,5 +613,7 @@ void test_config_repl(void)
     RUN_TEST(test_conf_save_config_path_consistency);
 #ifdef XCE_TESTING
     RUN_TEST(test_load_primary_config_missing_returns_error);
+    RUN_TEST(test_wait_all_with_timeout_slow_threads);
+    RUN_TEST(test_wait_all_with_timeout_fast_threads);
 #endif
 }

src/docs/config-replication-design.md

@@ -850,19 +850,22 @@ that the config may not have reached any standby.
 for (int i = 0; i < n_standbys; i++)
     pthread_create(&push_threads[i], NULL, push_config_thread, &push_ctx[i]);
 
-/* Single aggregate wait — 500ms total, not 500ms per standby.
-	 * Uses CONF_PUSH_TIMEOUT_MS (default 500ms), distinct from probe_connect_timeout_ms
-	 * used in election probing (§9.2). */
-wait_all_with_timeout(push_threads, n_standbys, CONF_PUSH_TIMEOUT_MS);
-
-/* Cancel still-running threads via atomic stop flag (same pattern as probe_thread) */
-atomic_store(&push_done, 1);
-for (int i = 0; i < n_standbys; i++)
-    pthread_join(push_threads[i], NULL);
+/* Single aggregate wait — 500ms total, not 500ms per standby. */
+atomic_int push_done;
+atomic_init(&push_done, 0);
+int acked = wait_all_with_timeout(threads, ctxs, n,
+                                   CONF_PUSH_TIMEOUT_MS, &push_done);
 
 LOG("Config pushed to %d/%d standbys; restarting", acked, n_standbys);
 ```
 
+Implemented signature (in `config_repl.c`):
+```c
+static int wait_all_with_timeout(pthread_t *threads, push_ctx_t *ctxs,
+                                  int n, uint32_t timeout_ms, atomic_int *done);
+```
+Uses a 5ms poll loop against `ctx->done` flags until the CLOCK_MONOTONIC deadline, then sets `*done = 1` and joins all started threads. The `done` flag allows thread implementations that check it to exit early; `push_thread` sets `ctx->done = 1` on natural completion, so joins return immediately for threads that finished within the timeout.
+
 Offline standbys recover on their next reconnect via the normal pull mechanism
 (§5.2) — their `conf/current` will have the old tag, the new primary will detect
 the mismatch and push via `OP_replConfig`.