2495 lines
86 KiB
Markdown
2495 lines
86 KiB
Markdown
> **ARCHIVED — superseded.** This plan proposed a priority-based `am_i_primary()` election (non-blocking connect/poll probing with xorshift jitter) and a standalone single-ACK `OP_replConfigPush` fan-out (500ms timeout, no WAL involvement) for config replication. What actually shipped: primary election is quorum-based Raft (`ha_start_election_async`, replacing the now-deleted `am_i_primary`/`am_i_primary_quorum`/`probe_ctx_t` — see closed specs `election-iomgr-refactor`, `raft-log-replication`, `raft-commit-safety`, `ha-term-safety`), and config changes now travel as `WAL_CONFIG_MARKER` entries inside the Raft/WAL commit path (`ha-term-safety` task 4.1, `ha-commit-engine`) rather than an out-of-band push. The HEAD-file/SHA-256 versioned storage layer (`conf_save_config`, `conf_read_current_filename`, `config_repl.c`) did ship largely as described here and is still current.
|
||
|
||
# Config Replication Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** Implement config replication so standbys fetch `ClusterConfig` from the primary on startup, eliminating manual per-node provisioning.
|
||
|
||
**Architecture:** Split `cluster_init()` into Phase 1 (SSL/HA init, no config) and Phase 2 (iomgr/kvmgr/persist, config required). The primary loads config from `conf/current` (HEAD-file pointer → versioned `cluster.bin.<sha256>` file). Standbys fetch via `OP_replConfig` before Phase 2. SHA-256 is used as both version tag and integrity token. `am_i_primary()` replaced with real election probing (non-blocking connect + poll per candidate).
|
||
|
||
**Tech Stack:** C99, OpenSSL SHA-256, POSIX threads (non-blocking connect + poll), Cap'n Proto codecgen, Unity test framework, Meson build.
|
||
|
||
---
|
||
|
||
## File Map
|
||
|
||
| File | Action | What changes |
|
||
|------|--------|--------------|
|
||
| `src/schema/controller.capnp` | Modify | Add 3 opcodes + 6 structs + `configGeneration` field + `sender_priority` to Heartbeat |
|
||
| `src/include/myconf.h` | Modify | Add 6 fields to `myconf_t` |
|
||
| `src/include/xce/cluster_def.h` | Modify | Add 4 fields to `cluster_t` |
|
||
| `src/controller/ha.h` | Modify | Add `ssl_ctx`, `jitter_seed`, `standby_endpoints`, `election_priority_step_ms`, `election_jitter_max_ms` to `ha_state_t`; add `probe_ctx_t`; new API declarations |
|
||
| `src/include/persist.h` | Modify | Add `max_wal_records` to `persist_config_t`; add `wal_record_count`, `checkpoint_in_progress` to `persist_s` |
|
||
| `src/include/byteorder.h` | Create | `be64toh()` portability fallback |
|
||
| `src/controller/config_repl.h` | Create | Public API for config_repl.c |
|
||
| `src/controller/config_repl.c` | Create | All config storage, SHA-256, fetch, upgrade, push functions |
|
||
| `src/controller/main.c` | Modify | HEAD-file pointer resolution; add `--conf-keep-n`, `--conf-stale-timeout` flags; heap-alloc conffile |
|
||
| `src/include/myconf.h` | Modify | `myconf_fini()` declaration |
|
||
| `src/controller/myconf.c` | Modify | Add `load_primary_config()`, `myconf_fini()`; remove `get_default_conf()` static global |
|
||
| `src/controller/cluster.c` | Modify | Split `cluster_init()` into Phase 1 + Phase 2; add opcode dispatch for new ops; `cluster_run()` checks `restart_requested`; `handle_endpoint_error` prunes `standby_endpoints` |
|
||
| `src/controller/myapp.c` | Modify | Restructure to call Phase 1, `am_i_primary()`, config load/fetch, Phase 2; handle restart |
|
||
| `src/controller/ha.c` | Modify | Replace `am_i_primary()` stub with real election probing; add `probe_thread`, heartbeat listener, `ha_check_heartbeat_nocluster` priority comparison |
|
||
| `src/controller/join.c` | Modify | Implement `handle_controller_join()`; register in `standby_endpoints` |
|
||
| `src/include/typemap.h` | Modify | Add new union arms to `xce_request_t` and `xce_reply_t` |
|
||
| `src/lib/request.c` | Modify | Add encode/decode shims for 6 new message types |
|
||
| `src/controller/persist.c` | Modify | Add `schedule_checkpoint()` triggered by `max_wal_records` |
|
||
| `src/controller/meson.build` | Modify | Add `config_repl.c`, compile-time defines |
|
||
| `src/controller/testsuite/test_config_repl.c` | Create | Unit tests for Phase 1 + Phase 2 config functions |
|
||
| `src/controller/testsuite/test_ha_role.c` | Modify | Extend with election probing tests |
|
||
| `src/controller/testsuite/controller_test.c` | Modify | Register new test suites |
|
||
|
||
---
|
||
|
||
## Task 1: Schema additions
|
||
|
||
**Files:**
|
||
- Modify: `src/schema/controller.capnp`
|
||
|
||
- [ ] **Step 1: Add new OPCode entries**
|
||
|
||
In `controller.capnp`, after `replCatchUpDone @28;`, add:
|
||
|
||
```capnp
|
||
replConfig @29;
|
||
replConfigPush @30;
|
||
upgradeConfig @31;
|
||
```
|
||
|
||
- [ ] **Step 2: Add new message structs**
|
||
|
||
After the `MasterCandidate` struct, add:
|
||
|
||
```capnp
|
||
struct ReplConfigRequest $C.mapname("repl_config_req_t") {
|
||
versionTag @0: UInt64 $C.mapname("version_tag");
|
||
}
|
||
|
||
struct ReplConfigReply $C.mapname("repl_config_reply_t") {
|
||
upToDate @0: Bool $C.mapname("up_to_date");
|
||
versionTag @1: UInt64 $C.mapname("version_tag");
|
||
sha256 @2: Data $C.mapname("sha256");
|
||
bytes @3: Data $C.mapname("bytes");
|
||
}
|
||
|
||
struct ReplConfigPushRequest $C.mapname("repl_config_push_req_t") {
|
||
versionTag @0: UInt64 $C.mapname("version_tag");
|
||
sha256 @1: Data $C.mapname("sha256");
|
||
bytes @2: Data $C.mapname("bytes");
|
||
}
|
||
|
||
struct ReplConfigPushReply $C.mapname("repl_config_push_reply_t") {
|
||
accepted @0: Bool $C.mapname("accepted");
|
||
versionTag @1: UInt64 $C.mapname("version_tag");
|
||
}
|
||
|
||
struct UpgradeConfigRequest $C.mapname("upgrade_config_req_t") {
|
||
bytes @0: Data $C.mapname("bytes");
|
||
sha256 @1: Data $C.mapname("sha256");
|
||
}
|
||
|
||
struct UpgradeConfigReply $C.mapname("upgrade_config_reply_t") {
|
||
versionTag @0: UInt64 $C.mapname("version_tag");
|
||
error @1: Text $C.mapname("error");
|
||
ackedCount @2: UInt32 $C.mapname("acked_count");
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 3: Add union arms to Request.body**
|
||
|
||
In `struct Request`, after `hostctrlmsg @21: HostControl;`:
|
||
|
||
```capnp
|
||
replConfigRequestMsg @22: ReplConfigRequest;
|
||
replConfigPushRequestMsg @23: ReplConfigPushRequest;
|
||
upgradeConfigRequestMsg @24: UpgradeConfigRequest;
|
||
```
|
||
|
||
- [ ] **Step 4: Add union arms to RequestReply.data**
|
||
|
||
In `struct RequestReply`, after `clusterconf @7: ClusterConf;`:
|
||
|
||
```capnp
|
||
replConfigReplyMsg @8: ReplConfigReply;
|
||
replConfigPushReplyMsg @9: ReplConfigPushReply;
|
||
upgradeConfigReplyMsg @10: UpgradeConfigReply;
|
||
```
|
||
|
||
- [ ] **Step 5: Add configGeneration to ClusterConfig**
|
||
|
||
In `struct ClusterConfig`, after the last existing field, add:
|
||
|
||
```capnp
|
||
configGeneration @20: UInt64 $C.mapname("config_generation");
|
||
```
|
||
|
||
- [ ] **Step 6: Add sender_priority to a new Heartbeat struct**
|
||
|
||
Add a new struct (the schema currently has no Heartbeat struct — `OP_heartbeat` uses the nop/void path):
|
||
|
||
```capnp
|
||
struct HeartbeatMsg $C.mapname("heartbeat_msg_t") {
|
||
senderPriority @0: UInt32 $C.mapname("sender_priority");
|
||
term @1: UInt64 $C.mapname("term");
|
||
}
|
||
```
|
||
|
||
Then in `Request.body`, add:
|
||
|
||
```capnp
|
||
heartbeatmsg @25: HeartbeatMsg;
|
||
```
|
||
|
||
- [ ] **Step 7: Build to verify schema compiles**
|
||
|
||
```bash
|
||
cd /worktrees/persistent_ha/src && nix develop --command bash -c "meson setup build --reconfigure 2>&1 | tail -20 && ninja -C build src/schema/controller.capnp.h 2>&1 | tail -20"
|
||
```
|
||
|
||
Expected: no capnp parse errors, generated header updated.
|
||
|
||
- [ ] **Step 8: Commit**
|
||
|
||
```bash
|
||
git add src/schema/controller.capnp
|
||
git commit -m "feat: add replConfig/replConfigPush/upgradeConfig opcodes and message structs to schema"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 2: Struct field additions
|
||
|
||
**Files:**
|
||
- Modify: `src/include/myconf.h`
|
||
- Modify: `src/include/xce/cluster_def.h`
|
||
- Modify: `src/controller/ha.h`
|
||
- Modify: `src/include/persist.h`
|
||
- Create: `src/include/byteorder.h`
|
||
|
||
- [ ] **Step 1: Add fields to myconf_t**
|
||
|
||
In `src/include/myconf.h`, after `xce_cluster_config_t peers_cfg;`:
|
||
|
||
```c
|
||
uint64_t config_version_tag; /* first 8 bytes of SHA-256, host order */
|
||
uint8_t config_sha256[32]; /* full SHA-256 of conffile bytes */
|
||
int conf_keep_n; /* versioned file retention count (default 2) */
|
||
uint64_t config_generation; /* monotonic audit counter (§4.4) */
|
||
time_t conf_stale_timeout; /* seconds before saved config is flagged stale (default 86400) */
|
||
int config_stale; /* set when saved config is older than conf_stale_timeout */
|
||
```
|
||
|
||
Also add `myconf_fini()` declaration at the bottom of the file:
|
||
|
||
```c
|
||
void myconf_fini(myconf_t *myconf);
|
||
```
|
||
|
||
- [ ] **Step 2: Add fields to cluster_t**
|
||
|
||
In `src/include/xce/cluster_def.h`, inside `struct cluster_s`, after `ha_state_t ha;`:
|
||
|
||
```c
|
||
_Atomic int restart_requested; /* set by OP_upgradeConfig handler */
|
||
pthread_t heartbeat_listener_tid; /* pre-Phase-2 SSL listener thread */
|
||
_Atomic int heartbeat_listener_running;
|
||
_Atomic int lost_and_found_job_count; /* incremented during persist_recover() */
|
||
```
|
||
|
||
- [ ] **Step 3: Add fields to ha_state_t and ha_config_t**
|
||
|
||
In `src/controller/ha.h`:
|
||
|
||
(a) In `ha_config_t`, after `max_promotion_retries`:
|
||
```c
|
||
uint32_t election_jitter_max_ms; /* random jitter added to election timeout */
|
||
uint32_t election_priority_step_ms; /* extra ms per priority level */
|
||
```
|
||
|
||
(b) Add new default macros after the existing defaults:
|
||
```c
|
||
#define HA_DEFAULT_ELECTION_JITTER_MAX_MS 200
|
||
#define HA_DEFAULT_ELECTION_PRIORITY_STEP_MS 500
|
||
```
|
||
|
||
(c) In `ha_state_t`, after `watchdog_thread`:
|
||
```c
|
||
void *ssl_ctx; /* SSL_CTX* cast; avoids pulling in openssl headers */
|
||
uint64_t jitter_seed; /* xorshift64 PRNG state for election jitter */
|
||
zarray_t *standby_endpoints; /* connected standby controller endpoints */
|
||
```
|
||
|
||
(d) Add `probe_ctx_t` struct declaration before `ha_state_t`:
|
||
```c
|
||
typedef struct {
|
||
struct xce_candidate_s *candidate; /* peer to probe */
|
||
void *ssl_ctx; /* SSL_CTX* (void* to avoid openssl header) */
|
||
uint32_t timeout_ms;
|
||
atomic_int *election_done;
|
||
int got_ack;
|
||
pthread_t thread;
|
||
} probe_ctx_t;
|
||
```
|
||
|
||
(e) Add new API declarations at the bottom:
|
||
```c
|
||
int ha_is_primary(const ha_state_t *ha);
|
||
void start_heartbeat_listener(cluster_t *c);
|
||
void stop_heartbeat_listener(cluster_t *c);
|
||
```
|
||
|
||
- [ ] **Step 4: Add fields to persist_config_t and persist_s**
|
||
|
||
In `src/include/persist.h`, in `persist_config_t` after `sync_timeout_ms`:
|
||
```c
|
||
uint32_t max_wal_records; /* checkpoint trigger: records since last snap (default 10000) */
|
||
```
|
||
|
||
In `persist_config_defaults()`, add:
|
||
```c
|
||
c.max_wal_records = 10000;
|
||
```
|
||
|
||
In `struct persist_s` after `pthread_mutex_t lock;`:
|
||
```c
|
||
_Atomic uint32_t wal_record_count; /* records since last checkpoint */
|
||
_Atomic int checkpoint_in_progress; /* CAS guard for schedule_checkpoint */
|
||
```
|
||
|
||
- [ ] **Step 5: Create byteorder.h**
|
||
|
||
Create `src/include/byteorder.h`:
|
||
|
||
```c
|
||
#pragma once
|
||
/* Portable be64toh() for non-glibc or pre-2.25 targets.
|
||
* On Linux glibc >= 2.9 this is already available from <endian.h>.
|
||
* On other platforms we provide a fallback. */
|
||
#if defined(__linux__)
|
||
# include <endian.h>
|
||
#elif defined(__APPLE__)
|
||
# include <machine/endian.h>
|
||
# include <libkern/OSByteOrder.h>
|
||
# define be64toh(x) OSSwapBigToHostInt64(x)
|
||
#else
|
||
# include <stdint.h>
|
||
static inline uint64_t be64toh(uint64_t x) {
|
||
# if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
|
||
return __builtin_bswap64(x);
|
||
# else
|
||
return x;
|
||
# endif
|
||
}
|
||
#endif
|
||
```
|
||
|
||
- [ ] **Step 6: Build to verify all struct changes compile**
|
||
|
||
```bash
|
||
cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build 2>&1 | grep -E 'error:|warning:|ninja:' | head -30"
|
||
```
|
||
|
||
Expected: only warnings about new fields being unused — no errors.
|
||
|
||
- [ ] **Step 7: Commit**
|
||
|
||
```bash
|
||
git add src/include/myconf.h src/include/xce/cluster_def.h src/controller/ha.h src/include/persist.h src/include/byteorder.h
|
||
git commit -m "feat: add config replication fields to myconf_t, cluster_t, ha_state_t, persist_t"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 3: config_repl.h + config_repl.c (Phase 1 storage functions)
|
||
|
||
**Files:**
|
||
- Create: `src/controller/config_repl.h`
|
||
- Create: `src/controller/config_repl.c` (Phase 1 functions only)
|
||
- Create: `src/controller/testsuite/test_config_repl.c` (Phase 1 tests)
|
||
|
||
- [ ] **Step 1: Write failing tests for Phase 1 storage functions**
|
||
|
||
Create `src/controller/testsuite/test_config_repl.c`:
|
||
|
||
```c
|
||
/* test_config_repl.c — Unity tests for config replication (Phase 1) */
|
||
#include "unity.h"
|
||
#include <fcntl.h>
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <string.h>
|
||
#include <sys/stat.h>
|
||
#include <unistd.h>
|
||
#include "config_repl.h"
|
||
|
||
static char tmpdir[256];
|
||
|
||
void setUp(void) {
|
||
snprintf(tmpdir, sizeof(tmpdir), "/tmp/test_config_repl_%d", (int)getpid());
|
||
mkdir(tmpdir, 0700);
|
||
char conf_dir[300];
|
||
snprintf(conf_dir, sizeof(conf_dir), "%s/conf", tmpdir);
|
||
mkdir(conf_dir, 0700);
|
||
}
|
||
|
||
void tearDown(void) {
|
||
char cmd[300];
|
||
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
|
||
system(cmd);
|
||
}
|
||
|
||
/* test: conf_read_current_filename() returns -ENOENT when conf/current absent */
|
||
void test_conf_read_current_filename_missing(void) {
|
||
char out[256] = {0};
|
||
int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
|
||
TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
|
||
}
|
||
|
||
/* test: conf_save_config() writes versioned file + HEAD pointer; read back matches */
|
||
void test_conf_save_and_read_current(void) {
|
||
const uint8_t data[] = "fake capnproto config blob";
|
||
size_t len = sizeof(data) - 1;
|
||
uint8_t sha256[32];
|
||
SHA256(data, len, sha256);
|
||
|
||
myconf_t myconf = {0};
|
||
myconf.workdir = tmpdir;
|
||
myconf.name = "ctrl0";
|
||
myconf.conf_keep_n = 2;
|
||
|
||
int rc = conf_save_config(&myconf, data, len, sha256);
|
||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||
|
||
/* conf/current pointer file must exist */
|
||
char ptr_path[512];
|
||
snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
|
||
TEST_ASSERT_EQUAL_INT(0, access(ptr_path, F_OK));
|
||
|
||
/* Reading back the current filename must succeed */
|
||
char out[256] = {0};
|
||
rc = conf_read_current_filename(tmpdir, out, sizeof(out));
|
||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||
TEST_ASSERT_NOT_NULL(strstr(out, "cluster.bin."));
|
||
}
|
||
|
||
/* test: conf_save_config() then conf_read_current_filename() parses 12-hex suffix */
|
||
void test_conf_current_filename_format(void) {
|
||
const uint8_t data[] = "another fake blob for hex test";
|
||
size_t len = sizeof(data) - 1;
|
||
uint8_t sha256[32];
|
||
SHA256(data, len, sha256);
|
||
|
||
myconf_t myconf = {0};
|
||
myconf.workdir = tmpdir;
|
||
myconf.name = "ctrl0";
|
||
myconf.conf_keep_n = 2;
|
||
conf_save_config(&myconf, data, len, sha256);
|
||
|
||
char out[256] = {0};
|
||
conf_read_current_filename(tmpdir, out, sizeof(out));
|
||
|
||
/* Must start with "cluster.bin." */
|
||
TEST_ASSERT_EQUAL_INT(0, strncmp(out, "cluster.bin.", 12));
|
||
/* Suffix must be exactly 12 hex chars */
|
||
const char *hex = out + 12;
|
||
TEST_ASSERT_EQUAL_INT(12, (int)strlen(hex));
|
||
TEST_ASSERT_EQUAL_INT(12, (int)strspn(hex, "0123456789abcdef"));
|
||
}
|
||
|
||
/* test: conf_trim_old_versions() keeps only conf_keep_n files */
|
||
void test_conf_trim_keep_n(void) {
|
||
myconf_t myconf = {0};
|
||
myconf.workdir = tmpdir;
|
||
myconf.name = "ctrl0";
|
||
myconf.conf_keep_n = 2;
|
||
uint8_t sha256[32];
|
||
|
||
/* Write 3 different config versions */
|
||
for (int i = 0; i < 3; i++) {
|
||
char data[32];
|
||
snprintf(data, sizeof(data), "config version %d", i);
|
||
SHA256((uint8_t*)data, strlen(data), sha256);
|
||
usleep(10000); /* ensure distinct mtime */
|
||
conf_save_config(&myconf, (uint8_t*)data, strlen(data), sha256);
|
||
}
|
||
|
||
/* Count cluster.bin.* files (excluding .tmp) */
|
||
char conf_dir[300];
|
||
snprintf(conf_dir, sizeof(conf_dir), "%s/conf", tmpdir);
|
||
int count = 0;
|
||
DIR *d = opendir(conf_dir);
|
||
struct dirent *e;
|
||
while ((e = readdir(d)) != NULL) {
|
||
if (strncmp(e->d_name, "cluster.bin.", 12) == 0 &&
|
||
strstr(e->d_name, ".tmp") == NULL)
|
||
count++;
|
||
}
|
||
closedir(d);
|
||
TEST_ASSERT_EQUAL_INT(2, count);
|
||
}
|
||
|
||
/* test: zero-tag sentinel — standby with no config sends tag=0 */
|
||
void test_conf_zero_tag_is_no_config_sentinel(void) {
|
||
/* When conf/current is absent, compute_config_tag() returns 0 */
|
||
uint64_t tag = conf_get_saved_tag(tmpdir);
|
||
TEST_ASSERT_EQUAL_UINT64(0, tag);
|
||
}
|
||
|
||
/* test: conf_read_current_filename() returns -EINVAL on path separator in pointer */
|
||
void test_conf_current_filename_rejects_path_separator(void) {
|
||
char ptr_path[300];
|
||
snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
|
||
FILE *f = fopen(ptr_path, "w");
|
||
fprintf(f, "cluster.bin/../evil\n");
|
||
fclose(f);
|
||
|
||
char out[256] = {0};
|
||
int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
|
||
TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
|
||
}
|
||
|
||
/* test: empty conf/current treated as ENODATA */
|
||
void test_conf_empty_pointer_file_returns_enodata(void) {
|
||
char ptr_path[300];
|
||
snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
|
||
/* Create empty file */
|
||
int fd = open(ptr_path, O_WRONLY|O_CREAT|O_TRUNC, 0600);
|
||
close(fd);
|
||
|
||
char out[256] = {0};
|
||
|