claude skills

2026-05-29 15:00:06 +08:00 · 2026-05-29 15:00:06 +08:00 · 9bbff7b016
commit 9bbff7b016
parent 94bf5d94b8
40 changed files with 4072 additions and 34 deletions
--- a/docs/superpowers/plans/2026-05-20-config-repl-wire-protocol.md
+++ b/docs/superpowers/plans/2026-05-20-config-repl-wire-protocol.md
@ -0,0 +1,549 @@
+# 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