myxce/docs/superpowers/plans/2026-05-20-config-repl-wire-protocol.md

# Config Replication — Wire Protocol Completion + Bug Fixes

> **SDD Plan** — Subagent-Driven Development
> **Date**: 2026-05-20
> **Goal**: Complete Phase 1 wire protocol (Task 11), fix path bug, implement push timeout.
> **Unblocks**: Standby config fetch, OP_upgradeConfig, Phase 1 design claim becoming true.

---

## Context

Three independent streams of work emerged from the design review and gap analysis:

1. **Task 11 (wire protocol)** — `do_one_config_fetch()` and `handle_config_request()` are
   stubs returning `-ENOSYS` / sending NULL bytes because `repl_config_reply_t.sha256` and
   `.bytes` are `xce_data_t*` (a discriminated cluster-data union), not raw byte carriers.
   The pattern to fix this already exists: `xce_replconfigpush_req_raw()` in `request.c:154`
   uses `capn_new_list8` + `capn_setv8` to bypass the `xce_data_t` path for raw bytes.
   Task 11 applies that same pattern to the reply (primary side) and the request+reply
   decode (standby side + upgrade handler).

2. **Path bug** — `conf_save_config()` uses `make_conf_dir(myconf->workdir, ...)` producing
   `workdir/conf`, but `load_primary_config()` reads from `workdir/name/conf`. Files written
   by an upgrade would be in the wrong location, causing FATAL on restart.

3. **Push timeout** — `push_config_to_standbys()` uses blocking `pthread_join()` with no
   bound. Design §6.3 specifies a 500ms aggregate timeout. `CONF_PUSH_TIMEOUT_MS` is defined
   but unused.

---

## Dependency Map

```
Task A  ──→  Task B (primary reply shim)
(shims)  ──→  Task C (standby fetch)     → all wire protocol done
         ──→  Task D (upgrade wiring)

Task E  (path bug)     — independent, do first
Task F  (push timeout) — independent, do in parallel with Task A
```

**Wave 1 (parallel):** Task A + Task E + Task F
**Wave 2 (parallel, after Task A done):** Task B + Task C + Task D

---

## Task A — `request.c`: Two new raw shims for OP_replConfig

**File:** `src/lib/request.c`
**Header:** add declarations to `src/include/request.h`

**Pattern to copy:** `xce_replconfigpush_req_raw()` at `request.c:154–225`

### A.1 — Implement `xce_replconfig_reply_raw()`

Primary uses this to send `ReplConfigReply` with real bytes (up_to_date=false path).

```c
/*
 * xce_replconfig_reply_raw: send OP_replConfig reply with raw byte arrays.
 *
 * Bypasses xce_data_t for sha256 and bytes fields (same pattern as
 * xce_replconfigpush_req_raw). Called by handle_config_request() when
 * the standby's tag differs from the primary's.
 *
 * ep      — endpoint to reply on
 * from    — primary's peerinfo
 * seq     — request sequence number (for correlation)
 * tag     — primary's version tag
 * sha256  — 32-byte SHA-256 of the config blob
 * config  — raw capnp config blob
 * config_len — byte length of config blob
 */
int xce_replconfig_reply_raw(endpoint_t *ep, xce_peerinfo_t *from,
                              uint64_t seq,
                              uint64_t tag,
                              const uint8_t *sha256, size_t sha256_len,
                              const uint8_t *config, size_t config_len);
```

Implementation mirrors `xce_replconfigpush_req_raw()` but:
- Builds a `RequestReply` (not `Request`) struct
- Sets `opcode = OPCode_replConfig`, `status = 0`
- Sets `data_which = RequestReply_data_replconfigreplymsg`
- Writes `up_to_date = 0`, `version_tag = tag`
- Uses `capn_new_list8` + `capn_setv8` for sha256 and bytes fields
- Calls `write_ReplConfigReply` then `write_RequestReply`
- Enqueues via `endpoint_enq`

### A.2 — Implement `xce_data_extract_raw()`

Decoder shim: extracts raw bytes from a received capnp Data field (stored as `capn_list8`
internally, presented as `xce_data_t*` via the typemap). Used by the standby to read the
reply and by the upgrade handler to read input bytes.

```c
/*
 * xce_data_extract_raw: extract raw bytes from a capnp Data field.
 *
 * The sha256/bytes fields in ReplConfigReply and UpgradeConfigRequest are
 * typed as xce_data_t* in the typemap — the underlying capnp pointer is a
 * capn_list8 (List(UInt8)). This function reads the bytes out using
 * capn_getv8().
 *
 * data     — the xce_data_t* from the decoded message (may be NULL)
 * out      — heap-allocated buffer written here (caller must free)
 * out_len  — byte count written here
 *
 * Returns 0 on success, -ENODATA if data is NULL/empty, -ENOMEM on alloc failure.
 */
int xce_data_extract_raw(const xce_data_t *data,
                          uint8_t **out, size_t *out_len);
```

Implementation:
- If `data == NULL` → return `-ENODATA`
- The capnp `Data` type stores as `capn_list8`; access via the raw `capn_ptr` embedded in
  `xce_data_t` (needs a union-cast or a new accessor — check how `capnp_repl_envelope_t.payload`
  is decoded in existing code for the pattern)
- `malloc(len)`, `capn_getv8(list, 0, out, len)`, set `*out_len`

### A.3 — Implement `xce_replconfig_req_raw()`

Standby uses this to send a `OP_replConfig` request over the standalone SSL socket.

```c
/*
 * xce_replconfig_req_raw: send OP_replConfig request over an SSL socket.
 *
 * Used by do_one_config_fetch() on a raw SSL connection (no iomgr).
 * Serialises ReplConfigRequest{version_tag}, writes to ssl_fd.
 *
 * Returns bytes written on success, negative errno on failure.
 */
int xce_replconfig_req_raw(SSL *ssl, uint64_t version_tag);
```

Implementation: use `capn_init_malloc` + `write_ReplConfigRequest` +
`capn_write_fd(ssl_fd)` pattern (see how `xce_encode` works in `request.c`).

### A.4 — Implement `xce_replconfig_reply_decode_raw()`

Standby uses this to decode the primary's reply from the raw SSL socket.

```c
/*
 * xce_replconfig_reply_decode_raw: read and decode ReplConfigReply from SSL.
 *
 * Reads exactly one capnp message, decodes into:
 *   out_up_to_date — set to 1 if tag matched (no bytes follow)
 *   out_tag        — primary's version tag
 *   out_sha256     — heap-alloc'd 32 bytes (caller frees); NULL if up_to_date
 *   out_bytes      — heap-alloc'd config blob (caller frees); NULL if up_to_date
 *   out_len        — byte length of config blob
 *
 * Returns 0 on success, negative errno on failure.
 */
int xce_replconfig_reply_decode_raw(SSL *ssl,
                                     int *out_up_to_date,
                                     uint64_t *out_tag,
                                     uint8_t **out_sha256,
                                     uint8_t **out_bytes,
                                     size_t   *out_len);
```

### A.5 — Tests

Add to `src/controller/testsuite/test_config_repl.c`:

- `test_replconfig_reply_raw_roundtrip` — encode reply with 32-byte sha256 and N-byte blob,
  write to a `socketpair`, decode back, verify sha256 and bytes match byte-for-byte
- `test_replconfig_req_raw_roundtrip` — encode request with version_tag, decode back, verify
- `test_data_extract_raw_null` — NULL data → `-ENODATA`
- `test_data_extract_raw_bytes` — encode Data field, extract, verify content

### A.6 — Commit message

```
feat(config-repl): add raw capnp shims for OP_replConfig wire protocol

xce_replconfig_reply_raw(), xce_replconfig_req_raw(),
xce_replconfig_reply_decode_raw(), xce_data_extract_raw() —
same pattern as existing xce_replconfigpush_req_raw(). Unblocks
do_one_config_fetch() and handle_config_request() wiring.
```

---

## Task B — `config_repl.c`: Wire `handle_config_request()` reply bytes

**File:** `src/controller/config_repl.c`
**Depends on:** Task A

Replace the NULL-bytes stub in `handle_config_request()` (lines 650–655).

**Before (stub):**
```c
rmsg->sha256 = NULL;
rmsg->bytes  = NULL;
xce_reply(ep, c->myhost, &reply);
free(bytes);
free(rmsg);
```

**After:**
Call `xce_replconfig_reply_raw()` directly instead of `xce_reply()`. This sidesteps the
`xce_data_t` path entirely — the reply is serialised with raw bytes inline.

**Steps:**
- [ ] Replace the `xce_reply()` path in `handle_config_request()` with
      `xce_replconfig_reply_raw(ep, c->myhost, req->seq, my_tag, recomputed, 32, bytes, blen)`
- [ ] Remove the `rmsg` calloc / free (no longer needed)
- [ ] Keep the `up_to_date = true` fast path (tag match) using `xce_reply()` — only
      the data-carrying path needs the raw shim
- [ ] Add test: `test_handle_config_request_sends_bytes` — mock a request with mismatched tag,
      verify the reply written to a socketpair contains non-NULL sha256 and bytes

**Commit:**
```
fix(config-repl): wire handle_config_request() to send real bytes via raw shim
```

---

## Task C — `config_repl.c`: Wire `do_one_config_fetch()` full SSL path

**File:** `src/controller/config_repl.c`
**Depends on:** Task A, Task B (primary must reply before standby can fetch)

Replace the `-ENOSYS` stub in `do_one_config_fetch()` (lines 674–683).

**Implementation outline:**

```c
static int do_one_config_fetch(cluster_t *c, const char *host, uint32_t port,
                               uint64_t our_tag)
{
    // 1. Open raw SSL connection (reuse probe_thread pattern from ha.c)
    //    socket() + fcntl(O_NONBLOCK) + connect() → poll() → SSL_connect()
    //    Use c->ha.ssl_ctx (stashed in Phase 1)

    // 2. Send OP_replConfig request
    //    xce_replconfig_req_raw(ssl, our_tag)

    // 3. Decode reply
    //    xce_replconfig_reply_decode_raw(ssl, &up_to_date, &tag, &sha256, &bytes, &len)

    // 4. If up_to_date: return 0 (tag matched, nothing to save)

    // 5. Verify SHA-256: SHA256(bytes, len, computed); memcmp(computed, sha256, 32)
    //    On mismatch: free + return -EBADMSG (retry)

    // 6. Save to disk
    //    conf_save_config(c->myconf, bytes, len, sha256)

    // 7. Update myconf fields
    //    config_version_tag, config_sha256, cluster_config (re-decode), config_generation

    // 8. Log success, free buffers, close SSL, return 0
}
```

**SSL connection setup** — copy the non-blocking connect pattern from `probe_thread()` in
`ha.c`. Key constraints:
- Check `atomic_load(&election_done)` between poll iterations (same as probe_thread)
- Use `c->ha.ssl_ctx` — same `SSL_CTX` as election probes
- On SSL errors: treat as fetch failure (return -errno), caller retries

**Steps:**
- [ ] Extract a `static int connect_ssl_raw(SSL_CTX*, const char*, uint32_t, SSL**)` helper
      (reusable by future heartbeat listener implementation)
- [ ] Implement full `do_one_config_fetch()` body per the outline above
- [ ] Wire `load_cluster_config_from_file()` call to re-decode the saved config into
      `myconf->cluster_config` after saving (reuse `load_primary_config()` path)
- [ ] Test: `test_do_one_config_fetch_integration` — requires two processes or socketpair mock;
      use `XCE_TESTING` guard to inject a mock reply

**Commit:**
```
feat(config-repl): wire do_one_config_fetch() — full SSL roundtrip with SHA-256 verify
```

---

## Task D — `config_repl.c`: Wire `handle_upgrade_config()` to `do_upgrade_config_with_bytes()`

**File:** `src/controller/config_repl.c`
**Depends on:** Task A (`xce_data_extract_raw()`)

Replace the `-ENOSYS` stub in `handle_upgrade_config()` (lines 1032–1049).

**Current (gated stub):**
```c
#if CONF_ENABLE_UPGRADE_CONFIG
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
    send_upgrade_error(ep, req, c, "OP_upgradeConfig byte extraction requires Task 11...");
    return -ENOSYS;
}
```

**After:**
```c
#if CONF_ENABLE_UPGRADE_CONFIG
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
    /* Extract raw bytes from xce_data_t fields using Task A shim */
    uint8_t *in_bytes = NULL; size_t in_len = 0;
    uint8_t *in_sha   = NULL; size_t sha_len = 0;

    int rc = xce_data_extract_raw(req->body.upgradeconfigrequestmsg->bytes,
                                   &in_bytes, &in_len);
    if (rc != 0) { send_upgrade_error(..., "bytes field missing"); return rc; }

    rc = xce_data_extract_raw(req->body.upgradeconfigrequestmsg->sha256,
                               &in_sha, &sha_len);
    if (rc != 0 || sha_len != 32) {
        free(in_bytes);
        send_upgrade_error(..., "sha256 field missing or wrong length");
        return -EINVAL;
    }

    rc = do_upgrade_config_with_bytes(c, ep, req, in_bytes, in_len, in_sha);
    free(in_bytes);
    free(in_sha);
    return rc;
}
```

**Steps:**
- [ ] Replace `handle_upgrade_config()` body with `xce_data_extract_raw()` calls + delegate
      to `do_upgrade_config_with_bytes()`
- [ ] Remove the `Task 11 dependency` comments from `do_upgrade_config_with_bytes()`
- [ ] Test: existing `test_upgrade_config_sha256_mismatch` and `test_upgrade_config_invalid_capnp`
      in `test_config_repl.c` should now exercise real code paths instead of the ENOSYS stub

**Commit:**
```
feat(config-repl): wire handle_upgrade_config() via xce_data_extract_raw() shim
```

---

## Task E — `config_repl.c`: Fix `conf_save_config()` path bug

**File:** `src/controller/config_repl.c`
**Independent** — no dependencies, fix this first.

**Root cause:** `conf_save_config()` calls `make_conf_dir(myconf->workdir, ...)` → produces
`workdir/conf`. But `load_primary_config()` reads from `workdir/name/conf`. Files would be
written to the wrong location after a config upgrade restart.

**Fix:**

```c
int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len,
                     const uint8_t sha256[32])
{
    /* effective_dir = workdir/name — must match load_primary_config() path */
    char effective_dir[4096];
    snprintf(effective_dir, sizeof(effective_dir), "%s/%s",
             myconf->workdir, myconf->name);

    char conf_dir[4096];
    make_conf_dir(effective_dir, conf_dir, sizeof(conf_dir));
    // ... rest unchanged, replace all make_conf_dir(myconf->workdir, ...) with conf_dir
```

Also fix `conf_trim_old_versions()` call at the bottom of `conf_save_config()`:
```c
    // Before: conf_trim_old_versions(myconf->workdir, myconf->conf_keep_n);
    // After:
    conf_trim_old_versions(effective_dir, myconf->conf_keep_n);
```

And fix `conffile` asprintf path at the end of `conf_save_config()`:
```c
    // Before: asprintf(&myconf->conffile, "%s/cluster.bin.%s", conf_dir, hex_tag)
    // (conf_dir was wrong — missing /name)
    // After: conf_dir now correctly is effective_dir/conf, so this is already right
    // if conf_dir is built from effective_dir above
```

**Steps:**
- [ ] Add `effective_dir` construction at top of `conf_save_config()`
- [ ] Replace all `myconf->workdir` path uses in the function with `effective_dir`
- [ ] Fix `conf_trim_old_versions()` call to pass `effective_dir` not `myconf->workdir`
- [ ] Add test: `test_conf_save_config_path_consistency` — call `conf_save_config()` with
      a `myconf` where `workdir="/tmp/w"` and `name="ctrl01"`, verify the written file is at
      `/tmp/w/ctrl01/conf/cluster.bin.*` not `/tmp/w/conf/cluster.bin.*`
- [ ] Update §3.2 of `config-replication-design.md`: add one sentence defining
      `effective_dir = ${workdir}/${name}`

**Commit:**
```
fix(config-repl): conf_save_config() must use workdir/name not workdir

load_primary_config() reads from workdir/name/conf — conf_save_config()
was writing to workdir/conf. Latent bug: would have corrupted upgrade
restart cycle once Task 11 wired the upgrade path.
```

---

## Task F — `config_repl.c`: Implement `wait_all_with_timeout()` + wire push timeout

**File:** `src/controller/config_repl.c`
**Independent** — no dependencies.

**Design spec (§6.3):** single aggregate 500ms timeout for all standby push threads.

**Implementation** — use a deadline + polling loop with the atomic stop flag pattern from
`probe_thread()`:

```c
/* wait_all_with_timeout: join all push threads within a CLOCK_MONOTONIC deadline.
 *
 * Sets atomic_int *done to 1 when the deadline expires, signalling threads
 * to exit their push_thread loop. Then joins all threads (they exit quickly
 * once *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  += timeout_ms / 1000;
    deadline.tv_nsec += (timeout_ms % 1000) * 1000000L;
    if (deadline.tv_nsec >= 1000000000L) {
        deadline.tv_sec++;
        deadline.tv_nsec -= 1000000000L;
    }

    /* Poll until deadline or all threads done */
    while (1) {
        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;

        /* Check if all started threads have finished */
        int pending = 0;
        for (int i = 0; i < n; i++) {
            if (ctxs[i].started && !ctxs[i].acked && !ctxs[i].done)
                pending++;
        }
        if (pending == 0) break;

        usleep(5000);   /* 5ms poll — low overhead, tight enough for 500ms window */
    }

    atomic_store(done, 1);   /* signal threads to exit their push_thread loop */

    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;
}
```

**Wire into `push_config_to_standbys()`:**
- [ ] Add `atomic_int push_done` initialized to 0
- [ ] Add `int done` flag to `push_ctx_t` (set by `push_thread` when it completes)
- [ ] Update `push_thread()` to set `ctx->done = 1` on exit (success or failure)
- [ ] Replace the blocking `pthread_join` loop with `wait_all_with_timeout()`
- [ ] Pass `CONF_PUSH_TIMEOUT_MS` as the timeout

**Test:**
- [ ] `test_push_config_timeout` — launch N push threads that sleep 2s each, call
      `wait_all_with_timeout()` with 100ms timeout, verify it returns within ~200ms
      (not 2s × N), and `acked` is 0
- [ ] `test_push_config_all_ack` — all threads ACK immediately, verify timeout is not hit

**Update design doc §6.3:** replace the `wait_all_with_timeout()` pseudocode stub with the
actual implementation signature and add a note explaining the 5ms poll interval.

**Commit:**
```
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.
```

---

## Execution Order (SDD Waves)

### Wave 1 — Parallel (no dependencies between tasks)

Launch three subagents simultaneously:

| Subagent | Task | ETA |
|---|---|---|
| Subagent-1 | Task A — raw shims in `request.c` | ~2h |
| Subagent-2 | Task E — path bug fix + design update | ~30m |
| Subagent-3 | Task F — push timeout implementation | ~45m |

### Wave 2 — After Task A completes (parallel)

| Subagent | Task | Prerequisite |
|---|---|---|
| Subagent-4 | Task B — wire `handle_config_request()` | Task A done |
| Subagent-5 | Task C — wire `do_one_config_fetch()` | Task A done |
| Subagent-6 | Task D — wire `handle_upgrade_config()` | Task A done |

### Wave 3 — Final integration

- Run full test suite: `ninja -C build test`
- Verify `impl-status.md` table entries updated
- Update design doc status header: `Status: Phase 1 wire protocol complete; Phase 2 gated`

---

## Files Changed Summary

| File | Tasks | Change |
|---|---|---|
| `src/lib/request.c` | A | +4 new shim functions (~120 LOC) |
| `src/include/request.h` | A | +4 declarations |
| `src/controller/config_repl.c` | B,C,D,E,F | ~150 LOC changes |
| `src/controller/testsuite/test_config_repl.c` | A,B,C,D,E,F | +~10 tests |
| `src/docs/config-replication-design.md` | E,F | §3.2 effective_dir definition; §6.3 timeout impl |
| `src/docs/config-replication-impl-status.md` | all | Update task status rows |

**Total estimated net-new LOC:** ~280 (request.c shims ~120, config_repl.c wiring ~100,
tests ~60).

---

## Definition of Done

- [ ] `ninja -C build test` passes with 0 failures
- [ ] `do_one_config_fetch()` no longer has `-ENOSYS` stub comment
- [ ] `handle_config_request()` sends non-NULL sha256 and bytes in reply
- [ ] `handle_upgrade_config()` calls `do_upgrade_config_with_bytes()` (not ENOSYS)
- [ ] `conf_save_config()` path verified by new test against `workdir/name/conf`
- [ ] `push_config_to_standbys()` returns within `CONF_PUSH_TIMEOUT_MS` even if standbys hang
- [ ] Design doc §3.2 defines `effective_dir`, §6.3 timeout mechanism documented
- [ ] `impl-status.md` header updated to reflect Phase 1 wire protocol complete