lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ha-term-safety): append config markers via unified log + derive committed config (task 4.2)

Browse source 
Wires do_upgrade_config_with_bytes to append WAL_CONFIG_MARKER through
raft_log_append (live term-stamped consensus record), demotes committed_config_tag
to a non-authoritative cache, and derives the committed config from the latest
committed marker. Preserves the derive across checkpoint compaction via an optional,
backward-compatible config-ref tail in the snapshot sentinel.
This commit is contained in:
Rongsong Shen 2026-06-22 10:34:10 +00:00
parent f447ad4183
commit c3201a2b1e
7 changed files with 883 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

252
src/controller/config_repl.c
View file

@ -88,6 +88,225 @@ int config_marker_decode(const uint8_t *buf, size_t len,
return 0;
}
/* ── Config marker append + committed-config derive (task 4.2) ───────────────
*
* A configuration change is ordered + committed through the SAME (term,lsn) log
* as job state. config_append_marker routes the marker through the single
* primary append entry point (ha_primary_log_append → raft_log_append), so the
* library stamps the live term and assigns the LSN, fires on_persist_entry, and
* the marker inherits the same commit rule and (term,lsn) reconcile/truncate as
* a job entry. The committed configuration is DERIVED from the latest committed
* marker within the committed prefix (config_committed_marker) — there is no
* separate authoritative config commit index. */
int config_append_marker(cluster_t *c, uint64_t version_tag,
const uint8_t sha256[32], uint64_t *out_lsn)
{
if (!c || !sha256)
return -EINVAL;
uint8_t payload[WAL_CONFIG_MARKER_PAYLOAD_LEN];
size_t plen = 0;
int rc = config_marker_encode(version_tag, sha256, payload,
sizeof(payload), &plen);
if (rc != 0)
return rc;
/* Single primary append entry point: stamps term (R7.3), assigns LSN,
* commits through the same barrier (R7.11), reconciles by (term,lsn) (R7.6).
* The config BYTES never enter this payload (R7.2/R7.4). */
return ha_primary_log_append(c, WAL_CONFIG_MARKER, payload,
(uint32_t)plen, out_lsn);
}
/* scan_latest_committed_marker: walk the local WAL segment files and find the
* HIGHEST-LSN WAL_CONFIG_MARKER whose LSN is <= committed_lsn (the committed
* prefix). On a match, decode its {version_tag, sha256} into the out params.
*
* Returns 0 on a found+decoded marker; -ENOENT when no committed marker exists;
* -EIO on an unreadable WAL directory. An uncommitted marker (LSN >
* committed_lsn) is deliberately skipped so a config change is not the committed
* config until its marker reaches the committed prefix (R7.7). */
static int scan_latest_committed_marker(const char *wal_path,
uint64_t committed_lsn,
uint64_t *version_tag,
uint8_t sha256_out[32])
{
if (committed_lsn == 0)
return -ENOENT; /* nothing committed → no committed config marker */
DIR *d = opendir(wal_path);
if (!d)
return -EIO;
/* Enumerate segment start-LSNs ascending so we replay in LSN order and the
* last marker we see at LSN <= committed_lsn is the latest committed one. */
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent *de;
while ((de = readdir(d)) != NULL &&
nseg < (int)(sizeof(seg_lsns) / sizeof(*seg_lsns))) {
uint64_t sl = 0;
if (sscanf(de->d_name, "seg-%016" SCNx64 ".wal", &sl) == 1)
seg_lsns[nseg++] = sl;
}
closedir(d);
if (nseg == 0)
return -ENOENT;
/* insertion sort ascending */
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j + 1] = seg_lsns[j]; j--; }
seg_lsns[j + 1] = key;
}
int found = 0;
uint64_t found_tag = 0;
uint8_t found_sha[32];
for (int si = 0; si < nseg; si++) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
"%s/seg-%016" PRIx64 ".wal", wal_path, seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0)
continue;
uint32_t off = 0;
for (;;) {
wal_entry_hdr_t hdr;
ssize_t n = pread(fd, &hdr, sizeof(hdr), (off_t)off);
if (n < (ssize_t)sizeof(hdr)) break;
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break;
if (hdr.lsn > committed_lsn)
break; /* past the committed prefix — stop scanning this seg */
if (hdr.type == WAL_CONFIG_MARKER &&
hdr.payload_len >= WAL_CONFIG_MARKER_PAYLOAD_LEN) {
uint8_t buf[WAL_CONFIG_MARKER_PAYLOAD_LEN];
ssize_t pr = pread(fd, buf, sizeof(buf),
(off_t)(off + sizeof(hdr)));
if (pr == (ssize_t)sizeof(buf)) {
uint64_t tag = 0;
if (config_marker_decode(buf, sizeof(buf), &tag,
found_sha) == 0) {
found_tag = tag;
found = 1; /* later markers overwrite — latest wins */
}
}
}
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
if (!found)
return -ENOENT;
if (version_tag)
*version_tag = found_tag;
if (sha256_out)
memcpy(sha256_out, found_sha, 32);
return 0;
}
int config_committed_marker(const cluster_t *c, uint64_t *version_tag,
uint8_t sha256_out[32])
{
if (!c)
return -EINVAL;
uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
int rc = scan_latest_committed_marker(c->persist.wal_path, committed,
version_tag, sha256_out);
if (rc == 0 || rc == -EINVAL)
return rc;
/* Compaction fallback (task 4.2 remediation r1): a checkpoint can trim the
* WAL segment holding the latest committed marker, so the live scan above
* returns -ENOENT even though a committed config exists. The checkpoint
* stamped that marker's {version_tag, sha256} reference into the snapshot
* sentinel (persist.cfgref_*, read back at recovery), so fall back to it.
* This keeps the derived committed config stable across compaction and lets
* recovery activate the latest committed marker (R7.7/R7.10) even when no
* marker survives in the live WAL. */
if (c->persist.cfgref_present) {
if (version_tag)
*version_tag = c->persist.cfgref_tag;
if (sha256_out)
memcpy(sha256_out, c->persist.cfgref_sha, 32);
return 0;
}
return rc; /* genuinely no committed config marker anywhere */
}
/* config_commit_change (ha-term-safety task 4.2 — LIVE wiring): order a config
* change through the unified (term,lsn) log as the CONSENSUS RECORD and make the
* derived committed marker authoritative, retiring committed_config_tag to a
* non-authoritative cache.
*
* 1. Append a WAL_CONFIG_MARKER ({version_tag, sha256} — NO bytes, R7.2/R7.4)
* via config_append_marker → ha_primary_log_append → raft_log_append. The
* library stamps the live term (R7.3), assigns the LSN, and the marker
* inherits the same (term,lsn) reconcile/truncate as a job entry (R7.6).
* This is the ORDERING authority: config_committed_marker derives the
* committed config from it (R7.7).
* 2. Deliver the config BYTES out-of-band and wait for the existing
* majority-ack barrier via push_config_to_standbys (R7.5/R7.11 — byte
* delivery/verify/activate stays task 4.3; the barrier MECHANISM is
* unchanged). push_config_to_standbys advances committed_config_tag as the
* CACHE on success; consumers read the derived marker as the authority.
*
* The marker append precedes the barrier so the consensus record is in the log
* before the change is acknowledged committed. A marker-append failure is fatal
* (fail-loud): the upgrade cannot be ordered, so it returns the append rc and
* the byte barrier is not attempted.
*
* Returns 0 on a committed config change; the barrier rc (-ETIMEDOUT/-EINVAL)
* when the change never reaches majority; a negative append rc on marker-append
* failure. *acked_out (may be NULL) receives the out-of-band transport push
* count (unchanged from push_config_to_standbys' acked semantics). */
int config_commit_change(cluster_t *c, const uint8_t *bytes, size_t blen,
const uint8_t sha256[32], uint64_t version_tag,
int *acked_out)
{
if (acked_out)
*acked_out = 0;
if (!c || !sha256)
return -EINVAL;
/* 1. Append the consensus record: the WAL_CONFIG_MARKER (ordering authority
* for the derived committed config). Fail-loud on append failure — the
* change cannot be ordered, so it is not committed. */
uint64_t marker_lsn = 0;
int rc = config_append_marker(c, version_tag, sha256, &marker_lsn);
if (rc != 0) {
ERROR("config_commit_change: marker append failed (rc=%d, tag=%016"
PRIx64 ") — config change not ordered", rc, version_tag);
return rc;
}
/* 2. Out-of-band byte delivery + the existing majority-ack barrier (the
* commit barrier mechanism is unchanged). On success it advances the
* committed_config_tag CACHE; the derived marker (config_committed_marker)
* remains the authority consumers read. */
rc = push_config_to_standbys(c, bytes, blen, sha256, version_tag, acked_out);
if (rc != 0) {
WARN("config_commit_change: config change did not reach majority"
" (rc=%d, marker_lsn=%" PRIu64 ", tag=%016" PRIx64 ")",
rc, marker_lsn, version_tag);
return rc;
}
INFO("Config change committed (marker_lsn=%" PRIu64 ", tag=%016" PRIx64 ")",
marker_lsn, version_tag);
return 0;
}
/* SSL context for xce_probe_config_tag.
* Stored as void* to avoid SSL_CTX type at this point in the translation unit. */
static void *s_probe_ssl_ctx = NULL;
@ -1012,7 +1231,14 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen,
* OP_upgradeConfigAck, passing the in-flight push tag for per-push
* evaluation (entry_lsn carries the tag — task 5.2). On quorum:
* advance committed_config_tag atomically (Req 8.5 / 4.3 — the
* advance stays inside this success branch only). */
* advance stays inside this success branch only).
*
* ha-term-safety task 4.2: committed_config_tag is now a NON-AUTHORITATIVE
* CACHE. The authoritative committed configuration is DERIVED from the
* latest committed WAL_CONFIG_MARKER (config_committed_marker, R7.7); this
* store only keeps legacy consumers (status, heartbeat) consistent with the
* derived value. The marker is the ordering/consensus record; this barrier
* remains the out-of-band byte-delivery quorum. */
int rc = repl_wait_majority_ack(&c->persist.repl, new_tag,
CONF_PUSH_TIMEOUT_MS,
(uint8_t)OPCode_upgradeConfigAck);
@ -1154,17 +1380,21 @@ static int do_upgrade_config_with_bytes(cluster_t *c, endpoint_t *ep,
/* 6. Update in-memory generation */
c->myconf->config_generation = new_gen;
/* 7. Push to standbys — the majority-ACK barrier result is
* authoritative for the upgrade outcome (barrier-rollout Req 4.1/4.2):
* non-zero means the new tag never reached quorum, so the requester
* gets an explicit error and the cluster stays on the old committed
* tag (no success ACK, no restart — Req 4.3/4.4). Disk save and
* generation bump have already happened (pre-existing sequence);
* convergence of a half-pushed config remains the OP_replConfig
/* 7. Commit the config change through the unified (term,lsn) log
* (ha-term-safety task 4.2): config_commit_change appends a
* WAL_CONFIG_MARKER as the CONSENSUS RECORD, pushes the bytes out-of-band,
* and waits for the MARKER to reach the committed prefix (the majority
* requirement is now "marker committed" — R7.1/R7.11). The committed config
* is DERIVED from that marker (config_committed_marker); committed_config_tag
* is only a cache refreshed inside the success branch. A non-zero rc means
* the marker never committed, so the requester gets an explicit error and
* the cluster stays on the old committed config (no success ACK, no
* restart). Disk save and generation bump already happened (pre-existing
* sequence); convergence of a half-pushed config remains the OP_replConfig
* anti-entropy behavior. */
int acked = 0;
rc = push_config_to_standbys(c, out_bytes, out_len, computed,
c->myconf->config_version_tag, &acked);
rc = config_commit_change(c, out_bytes, out_len, computed,
c->myconf->config_version_tag, &acked);
free(out_bytes);
if (rc != 0) {
@ -1172,7 +1402,7 @@ static int do_upgrade_config_with_bytes(cluster_t *c, endpoint_t *ep,
* status so the admin tool sees WHY the commit failed (task 5.2
* r1, reviewer finding 3). */
send_upgrade_error(ep, req, c, rc,
"Config push failed to reach majority quorum"
"Config marker failed to reach committed prefix"
" -- upgrade not committed, no restart");
return rc;
}

71
src/controller/config_repl.h
View file

@ -73,6 +73,77 @@ int config_marker_encode(uint64_t version_tag, const uint8_t sha256[32],
int config_marker_decode(const uint8_t *buf, size_t len,
uint64_t *version_tag, uint8_t sha256_out[32]);
/* ── Config marker append / committed-config derive (task 4.2) ───────
*
* A configuration change is ordered and committed through the SAME replicated
* (term,lsn) log as job state: a lightweight WAL_CONFIG_MARKER carrying only a
* reference to the configuration ({version_tag, sha256}). The committed
* configuration is DERIVED from the latest committed marker within the committed
* prefix — there is no separate authoritative config commit index (R7.1/R7.7).
*/
/*
* config_append_marker: append a config-change marker (version_tag + sha256,
* NO config bytes — R7.2/R7.4) to the unified replicated log via the single
* primary append entry point (ha_primary_log_append → raft_log_append), so the
* marker inherits the live-term stamp (R7.3), the same commit rule (R7.11), and
* the same (term,lsn) reconcile/truncate (R7.6) as a job-state entry. The
* configuration bytes travel out-of-band on the config-push transport (R7.5),
* never inside the marker.
*
* c — cluster (must be the leader for a real raft append)
* version_tag — the config version tag (be64 of the first 8 sha256 bytes)
* sha256 — the full 32-byte content hash (the reference)
* out_lsn — (may be NULL) receives the assigned 1-based LSN on success
*
* Returns 0 on success; -EINVAL on a NULL argument; otherwise the append rc
* propagated from ha_primary_log_append.
*/
int config_append_marker(cluster_t *c, uint64_t version_tag,
const uint8_t sha256[32], uint64_t *out_lsn);
/*
* config_committed_marker: derive the committed configuration reference from the
* LATEST WAL_CONFIG_MARKER at or below committed_lsn (R7.7). Scans the committed
* prefix of the local WAL; an uncommitted marker (LSN > committed_lsn) is
* deliberately ignored, so a config change is not the committed config until its
* marker reaches the committed prefix.
*
* c — cluster
* version_tag — (may be NULL) receives the latest committed marker's tag
* sha256_out — (may be NULL) receives the latest committed marker's 32-byte hash
*
* Returns 0 on success; -ENOENT when no committed config marker exists;
* -EINVAL on a NULL cluster; -EIO on a WAL read error.
*/
int config_committed_marker(const cluster_t *c, uint64_t *version_tag,
uint8_t sha256_out[32]);
/*
* config_commit_change: order + commit a config change through the unified
* (term,lsn) log (ha-term-safety task 4.2 LIVE wiring). Appends a
* WAL_CONFIG_MARKER as the consensus record, pushes the config bytes
* out-of-band on the existing config-push transport (R7.5 — byte delivery /
* hash-verify / activate is task 4.3), and waits for the MARKER's LSN to reach
* the committed prefix via the same WAL commit barrier job state uses (R7.11).
* On commit it refreshes committed_config_tag as a NON-AUTHORITATIVE cache that
* mirrors the DERIVED committed marker (config_committed_marker is the
* authority — R7.7). do_upgrade_config_with_bytes uses this instead of the
* retired term-blind committed_config_tag push/ack/advance.
*
* c — cluster (must be leader for the marker append)
* bytes/blen — re-encoded config bytes (pushed out-of-band, never in the WAL)
* sha256 — content hash (the marker reference)
* version_tag — config version tag
* acked_out — (may be NULL) receives the out-of-band transport push count
*
* Returns 0 once the marker is committed; the barrier rc (-ETIMEDOUT/-EINVAL)
* when it never commits; a negative append rc on append failure.
*/
int config_commit_change(cluster_t *c, const uint8_t *bytes, size_t blen,
const uint8_t sha256[32], uint64_t version_tag,
int *acked_out);
/* ── Phase 1: Storage ────────────────────────────────────────── */
/*

146
src/controller/persist.c
View file

@ -1009,6 +1009,19 @@ pthread_t persist_test_last_fdatasync_thread(persist_t* p)
return p->test_last_fdatasync_thread;
}
/* Force a fresh WAL segment at the current LSN (test seam — see persist.h).
* Takes p->lock for the same single-writer serialization the append path uses,
* so the segment swap cannot race a concurrent write. */
int persist_test_force_new_segment(persist_t* p)
{
if (!p)
return -EINVAL;
pthread_mutex_lock(&p->lock);
int rc = rotate_segment(p);
pthread_mutex_unlock(&p->lock);
return rc;
}
/* ------------------------------------------------------------------ *
* Durability-verification test seam (ha-commit-durability Req 3.2/3.3)*
* *
@ -1208,10 +1221,18 @@ static int write_snapshot(persist_t* p)
if (fd < 0)
return -errno;
/* Sentinel header: magic (8 bytes) + checkpoint_lsn (8 bytes) */
char header[16];
/* Sentinel: magic (8) + checkpoint_lsn (8) + committed-config reference
* (1 present + 8 tag + 32 sha = 41 bytes; ha-term-safety task 4.2 r1).
* The config-ref tail is OPTIONAL on read-back, so an older 16-byte
* sentinel still loads (load_snapshot reads the tail best-effort). The
* tag is little-endian on this arch, matching the checkpoint_lsn field. */
char header[16 + 1 + 8 + 32];
memset(header, 0, sizeof(header));
memcpy(header, "XCESNAP\0", 8);
memcpy(header + 8, &p->checkpoint_lsn, 8);
header[16] = (char)(p->cfgref_present ? 1 : 0);
memcpy(header + 17, &p->cfgref_tag, 8);
memcpy(header + 25, p->cfgref_sha, 32);
ssize_t nw = write(fd, header, sizeof(header));
fdatasync(fd);
@ -1230,6 +1251,83 @@ static int write_snapshot(persist_t* p)
return 0;
}
/* scan_latest_committed_config_ref (ha-term-safety task 4.2 remediation r1):
* walk the local WAL segments and capture the HIGHEST-LSN WAL_CONFIG_MARKER
* whose LSN is <= committed_lsn into p->cfgref_{present,tag,sha}. This is the
* SAME committed-prefix marker config_committed_marker derives, cached durably
* so a later checkpoint can trim the marker's segment without losing the
* derived committed-config reference (compaction safety). The marker payload is
* a fixed 40-byte reference (8-byte big-endian version_tag + 32-byte sha256 —
* mirrors config_marker_encode; the config BYTES are never in the WAL), decoded
* inline here so persist stays free of a config_repl layering dependency. */
static void scan_latest_committed_config_ref(persist_t* p)
{
uint64_t committed = atomic_load_explicit(&p->committed_lsn,
memory_order_acquire);
if (committed == 0)
return; /* nothing committed → keep any prior cached ref unchanged */
DIR* d = opendir(p->wal_path);
if (!d)
return;
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent* e;
while ((e = readdir(d)) != NULL &&
nseg < (int)(sizeof(seg_lsns) / sizeof(seg_lsns[0]))) {
uint64_t sl = 0;
if (sscanf(e->d_name, "seg-%016" SCNx64 ".wal", &sl) == 1)
seg_lsns[nseg++] = sl;
}
closedir(d);
if (nseg == 0)
return;
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j + 1] = seg_lsns[j]; j--; }
seg_lsns[j + 1] = key;
}
for (int si = 0; si < nseg; si++) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path), WAL_SEG_FMT, p->wal_path,
seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0)
continue;
uint32_t off = 0;
for (;;) {
wal_entry_hdr_t hdr;
ssize_t n = pread(fd, &hdr, sizeof(hdr), (off_t)off);
if (n < (ssize_t)sizeof(hdr)) break;
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break;
if (hdr.lsn > committed) break; /* past committed prefix */
if (hdr.type == WAL_CONFIG_MARKER &&
hdr.payload_len >= WAL_CONFIG_MARKER_PAYLOAD_LEN) {
uint8_t buf[WAL_CONFIG_MARKER_PAYLOAD_LEN];
ssize_t pr = pread(fd, buf, sizeof(buf),
(off_t)(off + sizeof(hdr)));
if (pr == (ssize_t)sizeof(buf)) {
p->cfgref_tag =
((uint64_t)buf[0] << 56) | ((uint64_t)buf[1] << 48) |
((uint64_t)buf[2] << 40) | ((uint64_t)buf[3] << 32) |
((uint64_t)buf[4] << 24) | ((uint64_t)buf[5] << 16) |
((uint64_t)buf[6] << 8) | ((uint64_t)buf[7]);
memcpy(p->cfgref_sha, buf + 8, 32);
p->cfgref_present = 1; /* later markers overwrite — latest wins */
}
}
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
}
/* Delete WAL segment files whose start LSN is strictly less than the
* currently-open segment's start LSN. We parse the active segment
* filename (seg_path) to get its start LSN and never delete that file,
@ -1385,8 +1483,14 @@ int persist_checkpoint(persist_t* p)
p->checkpoint_lsn = p->current_lsn;
pthread_mutex_unlock(&p->lock);
/* 3. Serialize in-memory state to a snapshot file
* (P0: no live kvmgrs — writes sentinel header only) */
/* 3a. Capture the latest committed config-marker reference BEFORE trimming
* (ha-term-safety task 4.2 r1): the trim below can unlink the segment
* holding a committed WAL_CONFIG_MARKER, so the derived committed config
* must be cached durably (into the sentinel) before that happens. */
scan_latest_committed_config_ref(p);
/* 3b. Serialize in-memory state to a snapshot file
* (P0: no live kvmgrs — writes sentinel header + cached config ref) */
rc = write_snapshot(p);
if (rc != 0)
goto done;
@ -1417,7 +1521,8 @@ static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
if (fd < 0)
return -errno;
char header[16];
/* magic(8) + checkpoint_lsn(8) + optional config-ref tail(1+8+32). */
char header[16 + 1 + 8 + 32];
ssize_t n = read(fd, header, sizeof(header));
close(fd);
@ -1433,6 +1538,17 @@ static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
p->checkpoint_lsn = stored_lsn;
p->current_lsn = stored_lsn;
/* Optional committed-config reference (ha-term-safety task 4.2 r1): present
* only when the sentinel carries the 41-byte tail. An older 16-byte sentinel
* (n == 16) leaves the cached ref untouched (cfgref_present stays 0 from
* persist_init), so a member that never checkpointed a config marker simply
* has no fallback reference — the live WAL scan is authoritative. */
if (n >= 16 + 1 + 8 + 32 && header[16] != 0) {
p->cfgref_present = 1;
memcpy(&p->cfgref_tag, header + 17, 8);
memcpy(p->cfgref_sha, header + 25, 32);
}
return 0;
}
@ -1541,6 +1657,18 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
* replay), not here — recognised explicitly rather than via default. */
break;
case WAL_CONFIG_MARKER:
/* Config-change marker (ha-term-safety task 4.2): a {version_tag,
* sha256} reference, NEVER the config bytes (R7.4). It carries NO
* entity state, so replay performs no restore here — the committed
* configuration is DERIVED from the latest committed marker on demand
* (config_committed_marker), and the bytes are obtained + hash-verified
* out-of-band before activation (task 4.3). Recognised explicitly so
* the per-LSN term index (built above this switch's caller) records its
* term and on_get_entry_term / (term,lsn) reconcile treat it as an
* ordinary log entry. */
break;
default:
break;
}
@ -2223,6 +2351,14 @@ wal_commit_mode_t wal_classify(uint32_t wal_type, int detail)
* retained until committed (ha-commit-barrier-rollout pre-declared
* row: promotion | internal | RETAIN; guard forces classification) */
return WAL_MODE_RETAIN;
case WAL_CONFIG_MARKER:
/* config-change marker (ha-term-safety task 4.2): a committable
* (term,lsn) log entry like a job-state fact — replicated and retained
* until committed; it commits through the SAME barrier as job state
* (R7.1/R7.11), so it is RETAIN, NOT a no-op or a LOCAL marker. The
* requester barrier on a config upgrade is owned by the upgrade path
* (do_upgrade_config_with_bytes), not the per-entry WAIT hook. */
return WAL_MODE_RETAIN;
case WAL_CHECKPOINT:
/* local segment marker — never replicated, by design */
return WAL_MODE_LOCAL;

34
src/controller/testsuite/test_commit_barrier_rollout.c
View file

@ -277,6 +277,7 @@ static void test_classify_is_total_over_wal_entry_type(void)
{ WAL_JOB_PREASON, HOOK_DETAIL_CLRPREASON, "WAL_JOB_PREASON" },
{ WAL_QUEUE_CTRL, HOOK_DETAIL_QUEUE_UPDATE, "WAL_QUEUE_CTRL" },
{ WAL_WORKER_CTRL, HOOK_DETAIL_WORKER_DEAD, "WAL_WORKER_CTRL" },
{ WAL_CONFIG_MARKER, 0, "WAL_CONFIG_MARKER" },
{ WAL_CHECKPOINT, 0, "WAL_CHECKPOINT" },
};
size_t i;
@ -1167,6 +1168,8 @@ static zarray_t g_cb5_standbys;
static myconf_t g_cb5_myconf;
static char g_cb5_tmpdir[256];
static int g_cb5_have_tmpdir;
static char g_cb5_waldir[256]; /* started-persist WAL dir (task 4.2) */
static int g_cb5_have_waldir;
static void cb5_setup(void)
{
@ -1179,6 +1182,7 @@ static void cb5_setup(void)
memset(&g_cb5_req_priv, 0, sizeof(g_cb5_req_priv));
memset(&g_cb5_myconf, 0, sizeof(g_cb5_myconf));
g_cb5_have_tmpdir = 0;
g_cb5_have_waldir = 0;
g_cb5_c.myhost = &g_cb5_c.myhost_info;
g_cb5_c.myhost_info.name = (char *)"ctrl-cb5";
@ -1197,9 +1201,22 @@ static void cb5_setup(void)
}
g_cb5_c.ha.standby_endpoints = &g_cb5_standbys;
/* Replicator registry mirrors the push endpoints; wired engine carries
* n_peers=3 for the configAck majority denominator. */
TEST_ASSERT_EQUAL_INT(0, repl_init(&g_cb5_c.persist.repl));
/* Started persist so the live config-change path (config_commit_change →
* config_append_marker → ha_primary_log_append → persist_wal_async) can
* append the WAL_CONFIG_MARKER consensus record (ha-term-safety task 4.2).
* persist_init creates the replicator registry; we then mirror the push
* endpoints and wire the n_peers=3 engine (configAck majority denominator).
* A dedicated tmpdir keeps the WAL isolated from the myconf workdir. */
{
persist_config_t pcfg = persist_config_defaults();
pcfg.wal_sync_ms = 5000;
snprintf(g_cb5_waldir, sizeof(g_cb5_waldir), "/tmp/xce_cb5wal_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_cb5_waldir));
g_cb5_have_waldir = 1;
persist_set_workdir(&g_cb5_c.persist, g_cb5_waldir, "ctrl-cb5");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_cb5_c.persist, &pcfg, &g_cb5_c));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_cb5_c.persist));
}
TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl,
&g_cb5_sb_ep[0], "cb5_a", 7500) >= 0);
TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl,
@ -1251,7 +1268,9 @@ static void cb5_teardown(void)
cb5_drain_ep(&g_cb5_sb_ep[1]);
cb5_drain_ep(&g_cb5_req_ep);
commit_engine_fini(&g_cb5_eng);
repl_fini(&g_cb5_c.persist.repl);
/* persist_fini calls repl_fini internally (started-persist fixture). */
persist_stop(&g_cb5_c.persist);
persist_fini(&g_cb5_c.persist);
zarray_fini(&g_cb5_standbys);
cbuf_fini(&g_cb5_c.wbuf);
free(g_cb5_myconf.conffile);
@ -1263,6 +1282,13 @@ static void cb5_teardown(void)
system(cmd);
g_cb5_have_tmpdir = 0;
}
if (g_cb5_have_waldir) {
char cmd[300];
snprintf(cmd, sizeof(cmd), "rm -rf %s", g_cb5_waldir);
system(cmd);
g_cb5_have_waldir = 0;
}
}
/* ── Reply observation (raw capnp parse of ep->outq packets) ────────────── */

370
src/controller/testsuite/test_config_repl.c
View file

@ -21,11 +21,13 @@
#include <capnp_c.h>
#include "commit.h"
#include "config_repl.h"
#include "controller.capnp.h"
#include "ha.h"
#include "myconf.h"
#include "persist.h"
#include "repl.h"
#include "request.h"
#include "typemap.h"
#include "wal_types.h"
@ -590,6 +592,366 @@ static void test_config_marker_encode_short_buffer_fails(void)
printf("test_config_marker_encode_short_buffer_fails: PASS\n");
}
/* ── Config marker append + committed-config derive (task 4.2) ────────
*
* These bring a node up as the raft LEADER so config_append_marker routes the
* WAL_CONFIG_MARKER through ha_primary_log_append → raft_log_append (the same
* unified-log path job state uses), and assert the committed config is DERIVED
* from the latest committed marker within the committed prefix (R7.1/7.3/7.6/
* 7.7/7.11). Mirrors make_cluster/make_leader from test_term_safety_adapter.c. */
static cluster_t *t42_make_cluster(const char *tmpdir)
{
cluster_t *c = calloc(1, sizeof(cluster_t));
if (!c) return NULL;
persist_config_t cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.snap_keep_n = 2;
persist_set_workdir(&c->persist, tmpdir, "ctrl-test");
if (persist_init(&c->persist, &cfg, c) != 0) { free(c); return NULL; }
if (persist_start(&c->persist) != 0) {
persist_fini(&c->persist);
free(c);
return NULL;
}
c->ha.raft_c = c;
return c;
}
static void t42_teardown(cluster_t *c)
{
if (!c) return;
persist_stop(&c->persist);
persist_fini(&c->persist);
free(c);
}
/* Bring the node up as the raft LEADER at `term` (so raft_log_append is enabled
* — it returns -EPERM off the leader; see make_leader in the adapter suite). */
static void t42_make_leader(cluster_t *c, uint64_t term, int n_peers)
{
atomic_store(&c->ha.role, CTRL_PRIMARY);
atomic_store(&c->ha.term, term);
ha_init_raft(&c->ha, c, n_peers);
c->ha.raft.term = term;
c->ha.raft.role = RAFT_LEADER;
}
/* Fill an sha256 with a deterministic pattern; tag = be64 of first 8 bytes,
* matching the config_repl version_tag convention (config_version_tag). */
static uint64_t t42_make_ref(uint8_t sha256[32], uint8_t seed)
{
for (int i = 0; i < 32; i++)
sha256[i] = (uint8_t)(seed + i);
uint64_t tag;
memcpy(&tag, sha256, 8);
return be64toh(tag);
}
/* (a) Appending a config change writes a WAL_CONFIG_MARKER through the unified
* log: the entry lands at an LSN with the live election term stamped, and its
* payload decodes to {version_tag, sha256} (R7.1/7.3). */
static void test_config_marker_append_via_unified_log(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_cr42_append_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t T = 7;
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, T, /*n_peers*/1);
uint8_t sha[32];
uint64_t tag = t42_make_ref(sha, 0x10);
uint64_t lsn = 0;
TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &lsn));
/* Library assigned LSN 1 and stamped the live term for that entry. */
TEST_ASSERT_EQUAL_UINT64(1, lsn);
TEST_ASSERT_EQUAL_UINT64(1, c->ha.raft.log_len);
TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1));
t42_teardown(c);
printf("test_config_marker_append_via_unified_log: PASS\n");
}
/* (b) After the marker COMMITS (committed_lsn covers its LSN), the derived
* committed-config reference == that marker's {version_tag, sha256} (R7.7). */
static void test_committed_marker_derives_after_commit(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_cr42_derive_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, /*term*/4, /*n_peers*/1);
/* A job entry then a config marker, so the marker is NOT at LSN 1. */
wal_jobstate_t js = { .jobid = 1, .state = 1 };
uint64_t jlsn = 0;
TEST_ASSERT_EQUAL_INT(0,
ha_primary_log_append(c, WAL_JOB_UPDATE, &js, sizeof(js), &jlsn));
uint8_t sha[32];
uint64_t tag = t42_make_ref(sha, 0x40);
uint64_t mlsn = 0;
TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &mlsn));
TEST_ASSERT_EQUAL_UINT64(2, mlsn);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
/* Drive committed_lsn to the marker's LSN (the commit advance source). */
atomic_store_explicit(&c->persist.committed_lsn, mlsn, memory_order_release);
uint64_t got_tag = 0;
uint8_t got_sha[32];
memset(got_sha, 0, sizeof(got_sha));
TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha));
TEST_ASSERT_EQUAL_HEX64(tag, got_tag);
TEST_ASSERT_EQUAL_HEX8_ARRAY(sha, got_sha, 32);
t42_teardown(c);
printf("test_committed_marker_derives_after_commit: PASS\n");
}
/* (c) An UNCOMMITTED marker (LSN > committed_lsn) does NOT change the derived
* committed config: with committed_lsn at the first marker, a later second
* marker is invisible until committed (R7.7/7.11). */
static void test_uncommitted_marker_not_yet_committed_config(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_cr42_uncommit_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, /*term*/5, /*n_peers*/1);
uint8_t sha1[32], sha2[32];
uint64_t tag1 = t42_make_ref(sha1, 0x10);
uint64_t tag2 = t42_make_ref(sha2, 0x80);
uint64_t l1 = 0, l2 = 0;
TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag1, sha1, &l1));
TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag2, sha2, &l2));
TEST_ASSERT_EQUAL_UINT64(1, l1);
TEST_ASSERT_EQUAL_UINT64(2, l2);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
/* Only the FIRST marker is committed. The second is uncommitted. */
atomic_store_explicit(&c->persist.committed_lsn, l1, memory_order_release);
uint64_t got_tag = 0;
uint8_t got_sha[32];
TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha));
/* Derived config is the FIRST (committed) marker, NOT the uncommitted one. */
TEST_ASSERT_EQUAL_HEX64(tag1, got_tag);
TEST_ASSERT_EQUAL_HEX8_ARRAY(sha1, got_sha, 32);
/* With NOTHING committed, there is no committed config marker at all. */
atomic_store_explicit(&c->persist.committed_lsn, 0ULL, memory_order_release);
TEST_ASSERT_EQUAL_INT(-ENOENT, config_committed_marker(c, &got_tag, got_sha));
t42_teardown(c);
printf("test_uncommitted_marker_not_yet_committed_config: PASS\n");
}
/* (d) A divergent config marker at a conflicting (term,lsn) is truncated by the
* reconcile path exactly like a job entry: on_truncate_from(index) removes the
* marker (and everything above), leaving the committed prefix intact. This
* proves the marker is an ORDINARY log entry under (term,lsn) matching (R7.6).
*
* We append a committed job at LSN 1, then a divergent marker at LSN 2, drive
* committed_lsn to 1, then truncate from LSN 2 (the conflict point) and confirm
* the marker is gone while the committed job prefix survives. */
static void test_divergent_marker_truncated_like_job_entry(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_cr42_trunc_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, /*term*/6, /*n_peers*/1);
/* LSN 1: committed job entry. LSN 2: a (divergent) config marker. */
wal_jobstate_t js = { .jobid = 9, .state = 2 };
uint64_t jlsn = 0;
TEST_ASSERT_EQUAL_INT(0,
ha_primary_log_append(c, WAL_JOB_UPDATE, &js, sizeof(js), &jlsn));
TEST_ASSERT_EQUAL_UINT64(1, jlsn);
uint8_t sha[32];
uint64_t tag = t42_make_ref(sha, 0x55);
uint64_t mlsn = 0;
TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &mlsn));
TEST_ASSERT_EQUAL_UINT64(2, mlsn);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
/* The committed prefix is LSN 1 (the job). The marker at LSN 2 is the
* divergent suffix. The derived committed config is therefore NONE yet. */
atomic_store_explicit(&c->persist.committed_lsn, 1ULL, memory_order_release);
uint64_t got_tag = 0;
uint8_t got_sha[32];
TEST_ASSERT_EQUAL_INT(-ENOENT, config_committed_marker(c, &got_tag, got_sha));
/* Reconcile truncates from the conflict point (LSN 2) exactly as it would a
* job entry — the SAME on_truncate_from seam used by (term,lsn) matching. */
TEST_ASSERT_EQUAL_INT(0,
ha_cb_truncate_from_for_test(&c->ha, /*index*/2));
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
/* The marker is gone (term index shrank); the committed job prefix is intact. */
TEST_ASSERT_EQUAL_UINT64(0, persist_entry_term(&c->persist, 2));
TEST_ASSERT_EQUAL_UINT64(6, persist_entry_term(&c->persist, 1));
/* Still no committed config marker — the divergent one never committed. */
TEST_ASSERT_EQUAL_INT(-ENOENT, config_committed_marker(c, &got_tag, got_sha));
t42_teardown(c);
printf("test_divergent_marker_truncated_like_job_entry: PASS\n");
}
/* (e) INTEGRATION (Finding 1 remediation): drive a config change through the
* LIVE commit path (config_commit_change — the helper do_upgrade_config_with_bytes
* now calls) and assert (i) a WAL_CONFIG_MARKER lands at the LIVE term, and
* (ii) once the marker COMMITS through the real WAL commit barrier, the DERIVED
* committed config (config_committed_marker) equals it. This proves the live
* config-change path appends + commits + derives the marker — not the retired
* term-blind committed_config_tag push/ack/advance.
*
* A single-member commit engine is wired so config_commit_change's WAL barrier
* (repl_wait_majority_ack/OPCode_walAck → commit_submit) self-quorums and
* advances committed_lsn for real, exactly as the production single-node path. */
static void test_config_commit_change_live_path_commits_and_derives(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_cr42_live_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t T = 9;
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, T, /*n_peers*/1);
/* Single-member commit engine: the marker's WAL LSN self-quorums, so
* driving commit_submit on it advances committed_lsn through the REAL commit
* machinery — the same path that commits the marker in production once a
* majority acks the WAL stream (no manual committed_lsn poke). */
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&c->commit, &c->persist, /*n_peers*/1, /*n_vcpus*/8));
repl_set_commit_engine(&c->persist.repl, &c->commit);
uint8_t sha[32];
uint64_t tag = t42_make_ref(sha, 0x30);
/* The LIVE entry point do_upgrade_config_with_bytes now calls: append the
* WAL_CONFIG_MARKER consensus record + out-of-band byte push + the existing
* majority barrier. With no standbys (single member) the byte barrier
* self-quorums and the call returns 0. */
int acked = -1;
TEST_ASSERT_EQUAL_INT(0,
config_commit_change(c, (const uint8_t *)"cfg-bytes", 9, sha, tag, &acked));
/* (i) The marker landed at LSN 1 with the LIVE election term stamped. */
TEST_ASSERT_EQUAL_UINT64(1, c->ha.raft.log_len);
TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1));
/* No standbys registered → zero out-of-band pushes (barrier owns the verdict). */
TEST_ASSERT_EQUAL_INT(0, acked);
/* Make the marker durable (advances durable_lsn) then commit its WAL LSN
* through the real commit engine (self-quorum), exactly as a majority WAL
* ack does in production — this advances committed_lsn so the derive sees
* the marker in the committed prefix. */
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
TEST_ASSERT_EQUAL_INT(0,
commit_submit(&c->commit, /*marker LSN*/1, COMMIT_MODE_WAIT, 1000));
TEST_ASSERT_TRUE(
atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire) >= 1);
/* (ii) The DERIVED committed config equals the change just committed. */
uint64_t got_tag = 0;
uint8_t got_sha[32];
TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha));
TEST_ASSERT_EQUAL_HEX64(tag, got_tag);
TEST_ASSERT_EQUAL_HEX8_ARRAY(sha, got_sha, 32);
/* committed_config_tag is now a CACHE mirroring the derived marker. */
TEST_ASSERT_EQUAL_HEX64(tag,
atomic_load_explicit(&c->ha.committed_config_tag, memory_order_acquire));
commit_engine_fini(&c->commit);
t42_teardown(c);
printf("test_config_commit_change_live_path_commits_and_derives: PASS\n");
}
/* (f) COMPACTION SAFETY (Finding 2 remediation): a checkpoint that TRIMS the WAL
* segment holding a committed marker must NOT lose the derived committed config.
* Without the sentinel-carried reference + fallback, the live-WAL scan regresses
* to -ENOENT after the trim. We commit a marker, force a NEW segment + checkpoint
* (which trims the old segment containing the marker), and assert the derive
* STILL returns the committed reference (R7.7/R7.10). */
static void test_committed_marker_survives_checkpoint_trim(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_cr42_trim_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t T = 11;
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, T, /*n_peers*/1);
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&c->commit, &c->persist, /*n_peers*/1, /*n_vcpus*/8));
repl_set_commit_engine(&c->persist.repl, &c->commit);
uint8_t sha[32];
uint64_t tag = t42_make_ref(sha, 0x70);
/* Append the marker (segment 0, the lowest-LSN segment) and commit its WAL
* LSN through the real engine (self-quorum) so it enters the committed
* prefix — the production marker-commit path. */
int acked = -1;
TEST_ASSERT_EQUAL_INT(0,
config_commit_change(c, (const uint8_t *)"cfg", 3, sha, tag, &acked));
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
TEST_ASSERT_EQUAL_INT(0,
commit_submit(&c->commit, /*marker LSN*/1, COMMIT_MODE_WAIT, 1000));
uint64_t committed =
atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire);
TEST_ASSERT_TRUE(committed >= 1);
/* Sanity: before any trim, the live-WAL scan finds the marker. */
uint64_t got_tag = 0;
uint8_t got_sha[32];
TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha));
TEST_ASSERT_EQUAL_HEX64(tag, got_tag);
/* Roll to a NEW segment so the checkpoint's trim has an older segment (the
* one holding the marker) to unlink — the compaction hazard. */
TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(&c->persist));
/* Checkpoint: stamps the committed-marker reference into the sentinel, then
* trims the older segment (which physically held the marker). */
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&c->persist));
/* The marker's segment is gone, so the live-WAL scan alone would regress to
* -ENOENT; the derive MUST fall back to the durable sentinel reference. */
got_tag = 0;
memset(got_sha, 0, sizeof(got_sha));
TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha));
TEST_ASSERT_EQUAL_HEX64(tag, got_tag);
TEST_ASSERT_EQUAL_HEX8_ARRAY(sha, got_sha, 32);
commit_engine_fini(&c->commit);
t42_teardown(c);
printf("test_committed_marker_survives_checkpoint_trim: PASS\n");
}
/* ── wait_all_with_timeout (XCE_TESTING shim) ───── */
#ifdef XCE_TESTING
@ -2351,6 +2713,14 @@ void test_config_repl(void)
RUN_TEST(test_config_marker_encode_decode_roundtrip);
RUN_TEST(test_config_marker_decode_truncated_fails);
RUN_TEST(test_config_marker_encode_short_buffer_fails);
/* Task 4.2: append marker via unified log + derive committed config */
RUN_TEST(test_config_marker_append_via_unified_log);
RUN_TEST(test_committed_marker_derives_after_commit);
RUN_TEST(test_uncommitted_marker_not_yet_committed_config);
RUN_TEST(test_divergent_marker_truncated_like_job_entry);
/* Task 4.2 r1: live wiring + compaction-safe derive */
RUN_TEST(test_config_commit_change_live_path_commits_and_derives);
RUN_TEST(test_committed_marker_survives_checkpoint_trim);
#ifdef XCE_TESTING
RUN_TEST(test_load_primary_config_missing_returns_error);
RUN_TEST(test_wait_all_with_timeout_slow_threads);