20 KiB
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:
-
Task 11 (wire protocol) —
do_one_config_fetch()andhandle_config_request()are stubs returning-ENOSYS/ sending NULL bytes becauserepl_config_reply_t.sha256and.bytesarexce_data_t*(a discriminated cluster-data union), not raw byte carriers. The pattern to fix this already exists:xce_replconfigpush_req_raw()inrequest.c:154usescapn_new_list8+capn_setv8to bypass thexce_data_tpath for raw bytes. Task 11 applies that same pattern to the reply (primary side) and the request+reply decode (standby side + upgrade handler). -
Path bug —
conf_save_config()usesmake_conf_dir(myconf->workdir, ...)producingworkdir/conf, butload_primary_config()reads fromworkdir/name/conf. Files written by an upgrade would be in the wrong location, causing FATAL on restart. -
Push timeout —
push_config_to_standbys()uses blockingpthread_join()with no bound. Design §6.3 specifies a 500ms aggregate timeout.CONF_PUSH_TIMEOUT_MSis 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).
/*
* 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(notRequest) 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_setv8for sha256 and bytes fields - Calls
write_ReplConfigReplythenwrite_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.
/*
* 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
Datatype stores ascapn_list8; access via the rawcapn_ptrembedded inxce_data_t(needs a union-cast or a new accessor — check howcapnp_repl_envelope_t.payloadis 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.
/*
* 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.
/*
* 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 asocketpair, decode back, verify sha256 and bytes match byte-for-bytetest_replconfig_req_raw_roundtrip— encode request with version_tag, decode back, verifytest_data_extract_raw_null— NULL data →-ENODATAtest_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):
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 inhandle_config_request()withxce_replconfig_reply_raw(ep, c->myhost, req->seq, my_tag, recomputed, 32, bytes, blen) - Remove the
rmsgcalloc / free (no longer needed) - Keep the
up_to_date = truefast path (tag match) usingxce_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:
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— sameSSL_CTXas 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 intomyconf->cluster_configafter saving (reuseload_primary_config()path) - Test:
test_do_one_config_fetch_integration— requires two processes or socketpair mock; useXCE_TESTINGguard 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):
#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:
#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 withxce_data_extract_raw()calls + delegate todo_upgrade_config_with_bytes() - Remove the
Task 11 dependencycomments fromdo_upgrade_config_with_bytes() - Test: existing
test_upgrade_config_sha256_mismatchandtest_upgrade_config_invalid_capnpintest_config_repl.cshould 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:
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():
// 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():
// 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_dirconstruction at top ofconf_save_config() - Replace all
myconf->workdirpath uses in the function witheffective_dir - Fix
conf_trim_old_versions()call to passeffective_dirnotmyconf->workdir - Add test:
test_conf_save_config_path_consistency— callconf_save_config()with amyconfwhereworkdir="/tmp/w"andname="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 definingeffective_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():
/* 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_doneinitialized to 0 - Add
int doneflag topush_ctx_t(set bypush_threadwhen it completes) - Update
push_thread()to setctx->done = 1on exit (success or failure) - Replace the blocking
pthread_joinloop withwait_all_with_timeout() - Pass
CONF_PUSH_TIMEOUT_MSas the timeout
Test:
test_push_config_timeout— launch N push threads that sleep 2s each, callwait_all_with_timeout()with 100ms timeout, verify it returns within ~200ms (not 2s × N), andackedis 0test_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.mdtable 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 testpasses with 0 failuresdo_one_config_fetch()no longer has-ENOSYSstub commenthandle_config_request()sends non-NULL sha256 and bytes in replyhandle_upgrade_config()callsdo_upgrade_config_with_bytes()(not ENOSYS)conf_save_config()path verified by new test againstworkdir/name/confpush_config_to_standbys()returns withinCONF_PUSH_TIMEOUT_MSeven if standbys hang- Design doc §3.2 defines
effective_dir, §6.3 timeout mechanism documented impl-status.mdheader updated to reflect Phase 1 wire protocol complete