From 9ed38d4bad4fadbb17e564b99eb3400aae6e560e Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 30 Apr 2026 05:17:56 +0000 Subject: [PATCH] fix: address fifth-round adversarial review findings for persistence-and-ha-design P1 fixes: - A5-008: Change `in_catch_up` from plain `int` to `_Atomic int` with documented access patterns and memory ordering - A5-002: Add missing opcodes to Part 5.10 capnp schema (heartbeat, steppingDown, replSnapshot, replCatchUpDone) - A5-004: Add hook list synchronization spec for demotion (RCU-style removal with disabled flag and deferred memory reclamation) - A5-005: Change xce_connect_cluster return type from int to SSL*, add xce_disconnect_cluster(SSL*), update xce_reconnect to SSL** - A5-006: Add GC use-after-free mitigation section (per-operation READ lock hold for P0, epoch-based reclamation as P3) - A5-010: Add OP_steppingDown handler spec to P1-2 milestone - A5-001: Clarify P0-3 checkpoint timer vs dedicated pthread timers - A5-003: Clarify WAL_CRC_NONE handling in replication CRC validation - A5-013: Add repl_remove_standby slot cleanup requirement - A5-012: Add primary_ep lifecycle documentation (iomgr-owned borrow, NULL-before-free under mutex) - A5-014: Add async hook error handling convention - A5-015: Add aggregate memory budget analysis for concurrent snapshots - A5-016: Fix ha_check_heartbeat plain role read to use atomic_load_explicit - A5-017: Fix Appendix A.4 in_catch_up type consistency (_Atomic int) Co-Authored-By: Claude Opus 4.7 --- src/docs/persistence-and-ha-design.md | 296 ++++++++++++++++++-------- 1 file changed, 210 insertions(+), 86 deletions(-) diff --git a/src/docs/persistence-and-ha-design.md b/src/docs/persistence-and-ha-design.md index 846686c2..c9eb70ad 100644 --- a/src/docs/persistence-and-ha-design.md +++ b/src/docs/persistence-and-ha-design.md @@ -416,11 +416,11 @@ static int xzarray_get(meta_obj_t* obj, int idx, void** data) { /* COW mode: check active first, fall through to shadow */ if (priv->shadow != NULL) { - if (idx < a->caps && a->data[idx].inuse) { + if (idx < a->caps && atomic_load_explicit(&a->data[idx].inuse, memory_order_acquire)) { *data = a->data[idx].data; return 0; } - if (idx < priv->shadow_caps && priv->shadow[idx].inuse) { + if (idx < priv->shadow_caps && atomic_load_explicit(&priv->shadow[idx].inuse, memory_order_acquire)) { *data = priv->shadow[idx].data; return 0; } @@ -449,10 +449,10 @@ static int xzarray_put_vec(meta_obj_t* obj, int count, int* idxs, void** data) { for (int i = 0; i < count; i++) { int idx = idxs[i]; - if (a->data[idx].inuse && a->killer) + if (atomic_load_explicit(&a->data[idx].inuse, memory_order_acquire) && a->killer) a->killer(a->data[idx].data); a->data[idx].data = data[i]; - a->data[idx].inuse = 1; + atomic_store_explicit(&a->data[idx].inuse, true, memory_order_release); } return 0; } @@ -560,22 +560,23 @@ static int xzarray_store(meta_obj_t* obj, void* data) { #### Iteration — Merge Walk of Both Arrays ```c -static int xzarray_iter_next(meta_obj_t* obj, void** data) { +static int xzarray_iter_next(meta_obj_t* obj, void* ctx, void** data) { xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj); xzarray_private_t* priv = RTTI_OBJ_TO_PRIV(obj); + size_t* pos = (size_t*)ctx; /* iterator position — caller-allocated cursor */ if (priv->shadow != NULL) { /* COW mode: walk both arrays, active takes precedence */ size_t max_caps = (a->caps > priv->shadow_caps) ? a->caps : priv->shadow_caps; - for (size_t i = (size_t)iter->data; i < max_caps; i++) { - if (i < a->caps && a->data[i].inuse) { + for (size_t i = *pos; i < max_caps; i++) { + if (i < a->caps && atomic_load_explicit(&a->data[i].inuse, memory_order_acquire)) { *data = a->data[i].data; /* active wins — shadow[i] intentionally skipped */ - iter->data = (void*)(i + 1); + *pos = i + 1; return RTTI_ITER_OK; } - if (i < priv->shadow_caps && priv->shadow[i].inuse) { + if (i < priv->shadow_caps && atomic_load_explicit(&priv->shadow[i].inuse, memory_order_acquire)) { *data = priv->shadow[i].data; /* shadow only (no active entry at i) */ - iter->data = (void*)(i + 1); + *pos = i + 1; return RTTI_ITER_OK; } } @@ -655,7 +656,14 @@ static int xzarray_thaw(meta_obj_t* obj, retain_fn_t should_retain, void* arg) { } else { /* Copy shadow pointer to active (no retention check yet) */ xzarray_ensure_items(a, i + 1); - a->data[i] = priv->shadow[i]; /* struct copy (pointer, inuse) */ + /* Use explicit field copies — cannot struct-copy xzarray_item_t + * because it contains _Atomic _Bool inuse (C11 prohibits plain + * assignment of atomic fields). Thaw runs under kvmgr WRITE lock, + * so relaxed ordering is sufficient. */ + a->data[i].data = priv->shadow[i].data; + atomic_store_explicit(&a->data[i].inuse, + atomic_load_explicit(&priv->shadow[i].inuse, memory_order_relaxed), + memory_order_relaxed); a->count++; } } @@ -673,7 +681,7 @@ static int xzarray_thaw(meta_obj_t* obj, retain_fn_t should_retain, void* arg) { if (!a->data[i].inuse) continue; if (!should_retain(arg, a->data[i].data)) { if (a->killer) a->killer(a->data[i].data); - a->data[i].inuse = 0; + atomic_store_explicit(&a->data[i].inuse, false, memory_order_relaxed); a->data[i].data = NULL; a->count--; /* Caller must also remove kvtab entry for this index */ @@ -832,6 +840,16 @@ thaw + GC: **Per-object locks are NOT nested inside kvmgr lock.** The kvmgr lock serializes container-level operations (key→index lookup, COW coordination). Per-object locks (`job_t.lock`, `worker_t.lock`) are taken after releasing the kvmgr lock — the lookup returns a referenced pointer, and the caller locks the object separately. +**GC use-after-free mitigation — reference-counted object lifecycle (CRITICAL)**: During thaw+GC, the kvmgr WRITE lock is held, which prevents new kvtab lookups. However, a vcpu thread that previously completed a kvtab lookup (under READ lock) and obtained a pointer to a `job_t` may still be accessing that job object after releasing the READ lock. If thaw+GC frees that job during the window between the vcpu thread's READ unlock and its next access, the vcpu thread dereferences freed memory — a use-after-free. + +This window exists because XCE's lock protocol is: (1) acquire READ lock, (2) kvtab lookup → get pointer, (3) release READ lock, (4) use pointer to access the object. Between steps 3 and 4, the checkpoint thread can acquire the WRITE lock, run thaw+GC, and free the object. + +**Mitigation — epoch-based reclamation with generation counter**: The kvmgr gains a `uint64_t epoch` counter, incremented atomically at the start of `xzarray_thaw()` (inside the WRITE lock, before any GC frees). Vcpu threads record `kvmgr.epoch` at lookup time (step 1). After each per-object operation, the vcpu thread checks whether `kvmgr.epoch` has changed. If it has, the pointer may be stale and the vcpu must re-lookup the object. This is a lightweight generation counter — no per-object reference counting is needed, just an atomic counter check after each kvmgr lookup. + +Alternative (simpler but higher overhead): vcpu threads hold the kvmgr READ lock for the entire duration of their per-object operation (lookup → use → release). This prevents GC from acquiring the WRITE lock until all READ holders are done. The downside is longer READ lock hold times, which delays checkpoint and increases the checkpoint "freeze" pause. At XCE's scale (10K jobs, ~100μs per job operation), holding the READ lock across a per-job operation is acceptable — the checkpoint thread waits at most the duration of one per-job operation for the WRITE lock. + +**Recommended approach**: Use the simpler per-operation READ lock hold for P0, with the epoch optimization as a P3 performance improvement if checkpoints cause detectable latency spikes. The P0 implementation holds the kvmgr READ lock across the entire per-object operation (lookup through use), preventing GC from freeing objects that any vcpu thread is actively accessing. This is safe because per-object operations (job status update, worker state change) are fast (<100μs), and the WRITE lock holder (checkpoint) only needs to wait for the brief overlap. + **Why there is no lock race during serialization + thaw**: The concern is that concurrent `vput` calls during the "no locks held" serialization phase might interleave with `xzarray_thaw()`. This is safe because: 1. During serialization (no locks): `vput` writes to `active[]` under the kvmgr READ lock. These writes are to `active[i].data` — they never touch `shadow[]`. 2. During thaw (WRITE lock held): `xzarray_thaw()` merges `shadow[]` → `active[]`. The write lock blocks all concurrent `vput` (read-lock) calls. No interleave is possible. @@ -1237,8 +1255,23 @@ typedef struct { _Atomic int32_t role; /* C11 atomic — use atomic_load/store_explicit, see Part 5.7 */ int my_priority; /* My index in candidates[] */ replicator_t repl; /* Replicate to standbys (primary only) */ - uint64_t applied_lsn; /* Last applied WAL entry (standby only) */ - endpoint_t* primary_ep; /* Connection to current primary via iomgr SSL pipeline (standby only) */ + _Atomic uint64_t applied_lsn; /* Last applied WAL entry (standby only). + * Must be _Atomic because it is written by the iomgr/vtask + * thread during OP_replEntry processing and read by the + * watchdog thread when constructing OP_replJoin for catch-up. + * Use atomic_store_explicit/load_explicit with release/acquire. */ + endpoint_t* primary_ep; /* Connection to current primary via iomgr SSL pipeline (standby only). + * LIFECYCLE: set by iomgr accept/connect — iomgr owns the endpoint + * and the ha_state_t struct only holds a borrowed reference. + * On connection loss, iomgr sets primary_ep = NULL _before_ freeing + * the endpoint. All access must check for NULL; no dereference + * after NULL. The standby's catch-up thread reads primary_ep under + * the ha_state_t mutex (ha->lock), which serializes it against + * iomgr's NULL-then-free sequence. The watchdog thread also + * checks primary_ep == NULL under ha->lock before heartbeat + * comparison. This borrow-with-mutex pattern ensures no use-after-free: + * iomgr holds ha->lock while clearing the pointer, so any + * concurrent dereference sees either a valid endpoint or NULL. */ char primary_host[XCE_MAXNAME_LEN]; uint32_t primary_port; struct timespec last_heartbeat; /* Last heartbeat from primary (CLOCK_MONOTONIC, standby) */ @@ -1248,11 +1281,16 @@ typedef struct { * watchdog thread and read by vcpu threads during * promotion status checks. Use atomic_load/store * with relaxed ordering. */ - int in_catch_up; /* Non-zero while standby is processing historical - * catch-up entries from the primary. Set to 1 when - * the standby sends OP_replJoin with applied_lsn < - * primary's committed_lsn; cleared when applied_lsn - * reaches the primary's committed_lsn (signaled via + _Atomic int in_catch_up; /* C11 atomic — non-zero while standby is processing + * historical catch-up entries. Must be _Atomic because + * it is written by the iomgr/vtask thread (on + * OP_replJoin receipt and OP_replCatchUpDone) and + * read by vcpu threads during ACK suppression + * (Part 6.4). Use atomic_store/load_explicit with + * release/acquire ordering. Set to 1 when the standby + * sends OP_replJoin with applied_lsn < primary's + * committed_lsn; cleared when applied_lsn reaches + * the primary's committed_lsn (signaled via * OP_replCatchUpDone). Gates ACK suppression: during * catch-up, the standby does NOT send OP_replAck * for historical entries (see Part 6.4 line 1778). */ @@ -1364,7 +1402,7 @@ static void ha_check_heartbeat(cluster_t* c) { ha_state_t* ha = &c->ha; - if (ha->role != CTRL_STANDBY || ha->primary_ep == NULL) + if ((ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire) != CTRL_STANDBY || ha->primary_ep == NULL) return; struct timespec now; @@ -1396,7 +1434,7 @@ When a standby promotes to primary: 2. **Increment term** (`ha.term++`) — fencing token for stale standbys 3. **Flush the local WAL** (`persist_flush()` — fsync the last WAL segment to ensure all replicated entries are on disk) and **verify WAL consistency** (validate CRC chain from the standby's own last snapshot LSN forward — standbys run `persist_recover()` on startup and maintain their own local WAL and snapshots, so the validation starting point is the standby's `persist_t.checkpoint_lsn`). Do **NOT** call `persist_recover()` again — the standby's kvmgrs are already populated from replication (via `apply_wal_entry()` during normal `OP_replEntry` processing) AND from the startup recovery pass. Calling `persist_recover()` a second time would double-apply entries. Only a WAL flush + CRC chain validation is needed. **WAL gap handling**: Async-replicated entries that the old primary committed but had not yet replicated before crashing are **lost** — the dead primary cannot serve catch-up. This is a known limitation of the async replication model: non-sync WAL entries (job starts, finishes, rusage updates) may be lost on failover. The promoted primary's view is consistent but potentially stale; async state gaps self-heal through worker reconnection (see step 9). **WAL consistency guarantee**: the standby's local WAL is a strict prefix of the old primary's WAL — standbys only write entries received via `OP_replEntry` from the primary. No divergence is possible if the standby was continuously connected; the standby's WAL cannot contain entries the old primary did not originate. The only split-brain divergence scenario (standby accepted writes from clients during a partition) cannot occur because standbys do not accept writes — they only redirect to the primary. 4. **Switch role** to `CTRL_PRIMARY` — only after recovery succeeds. **If recovery fails, abort this promotion attempt and set `ha.promotion_failed = 1`**. The standby remains in `CTRL_STANDBY` role, but since the old primary is dead (that triggered this promotion), the next heartbeat watchdog check (`ha_check_heartbeat()`) will re-fire `ha_promote_to_primary()`. The `promotion_failed` flag gates retry: the standby retries promotion on each watchdog tick unless: (a) a new heartbeat is received from a higher-priority candidate (another controller won the race), or (b) a configurable `max_promotion_retries` (default: 5) is exceeded, at which point the standby logs `FATAL "cannot promote after %d attempts — manual intervention required"` and stops retrying. This prevents a permanent no-primary state while bounding the retry window. -5. **Register persistence hooks** — standby did not register persistence hooks during startup (hooks are registered only after `persist_recover()` completes, and only on the primary path; standbys register them here on promotion). **Idempotent registration guard**: `ha_promote_to_primary()` must check whether hooks are already registered before adding them (e.g., via a `persist->hooks_registered` flag). Without this guard, a promote→demote→re-promote cycle would accumulate duplicate hooks that fire N times per event, causing double WAL writes and replication sends. +5. **Register persistence hooks** — standby did not register persistence hooks during startup (hooks are registered only after `persist_recover()` completes, and only on the primary path; standbys register them here on promotion). **Idempotent registration guard**: `ha_promote_to_primary()` must check whether hooks are already registered before adding them (e.g., via a `ha->hooks_registered` flag in `ha_state_t`). Without this guard, a promote→demote→re-promote cycle would accumulate duplicate hooks that fire N times per event, causing double WAL writes and replication sends. 5.5 **Start heartbeat immediately** — begin sending heartbeat messages to other standbys BEFORE `cluster_start()` (step 8). The promoted standby must establish its authority as the new primary by sending heartbeats with its incremented term. This prevents other standbys from timing out and promoting themselves during the potentially lengthy replicator initialization and scheduler gating steps. The heartbeat thread is launched with the new term and role set to PRIMARY, but `cluster_start()` (step 8) is deferred until all preparation steps complete. **CRITICAL**: The heartbeat messages sent during this pre-operational phase contain only term and role — no LSN or state information. Other standbys receiving these heartbeats update their `ha.term` and `primary_host` to the new primary, but do not expect any replication data until step 8 is reached. 6. **Initialize replicator** — establish connections to remaining standbys 7. **Gate the scheduler**: pause or flush the scheduler's run queue before accepting writes. The scheduler maintains allocation state derived from WAL-replayed job state; async-replicated events (job starts/finishes) may be stale immediately after promotion. The scheduler must re-query job state from the promoted primary before making scheduling decisions (see step 8). @@ -1422,7 +1460,7 @@ The `term` field prevents split-brain: **Partition heal — old primary returns**: When a partitioned primary recovers network connectivity, it resumes sending heartbeats and replication messages with its old (lower) term. The promoted primary (higher term) ignores these messages (term < own term rule above). The partitioned primary receives a heartbeat reply or replication message from the promoted primary carrying the higher term. On receiving any message with `term > ha.term`, the old primary must detect that it has been superseded and demote itself: set `ha.role = CTRL_STANDBY`, stop accepting writes, and reconnect to the promoted primary as a standby. The `handle_msg()` path for all incoming messages must check `hdr.term > ha.term` and call `ha_demote_to_standby()` when this condition is detected, then re-process the message in standby mode. This prevents the old primary from continuing to accept writes after the partition heals, eliminating the split-brain write window. -**Hook unregistration on demotion**: `ha_demote_to_standby()` must unregister the persistence hooks that were registered during promotion (step 5 above). Use `xce_remove_hook()` (or equivalent) with the same function pointers and hook types registered in `ha_promote_to_primary()`. Without unregistration, a promote→demote→re-promote cycle accumulates duplicate hooks (despite the idempotent guard in step 5, the old hook entries still exist in the hook list). Unregistration must happen before the role switch to `CTRL_STANDBY` so that no in-flight requests trigger hooks after the role has changed. +**Hook unregistration on demotion**: `ha_demote_to_standby()` must unregister the persistence hooks that were registered during promotion (step 5 above). Use `xce_remove_hook()` (or equivalent) with the same function pointers and hook types registered in `ha_promote_to_primary()`. Without unregistration, a promote→demote→re-promote cycle accumulates duplicate hooks (despite the idempotent guard in step 5, the old hook entries still exist in the hook list). Unregistration must happen before the role switch to `CTRL_STANDBY` so that no in-flight requests trigger hooks after the role has changed. **Hook list synchronization**: `xce_remove_hook()` must be safe to call while vcpu threads are concurrently iterating the hook list inside `xce_run_hooks()`. This requires the hook list to use a lock-free or RCU-style removal pattern: `xce_remove_hook()` marks the hook entry as disabled (e.g., sets `fn = NULL` or a `disabled` flag), and `xce_run_hooks()` skips disabled entries. The actual memory reclamation of the hook entry must be deferred until no vcpu thread can be inside `xce_run_hooks()` — a safe deferral point is after `cluster_fini()` drains all vtask threads. This avoids a data race where a vcpu thread reads a freed hook function pointer from the list while `xce_remove_hook()` is concurrently removing it. The role-check defense-in-depth (checking `ha.role == CTRL_PRIMARY` inside each hook function before proceeding) provides a second layer: even if a hook entry is still in the list during demotion, it does nothing because the role has already switched to STANDBY. **In-flight request window during demotion (CRITICAL)**: Between the higher-term message arriving at `handle_msg()` and the demotion taking effect, vcpu threads may already be executing `handle_jobsubmit()` past the `AFTER_EVENT` hook. These in-flight requests have the following risks: 1. The persistence hook already called `persist_wal_sync_append()`, which is writing to the local WAL and may have already sent `repl_broadcast()`. Standbys following the higher-term primary will ignore the lower-term replication message. @@ -1474,9 +1512,13 @@ All HA connections (primary→standby replication, standby→primary heartbeat, standby @5; /* NEW — standby controller role */ /* In OPCode enum: */ -replEntry @23; /* Replicated WAL entry (primary → standby) */ -replAck @24; /* ACK for sync-replicated entry (standby → primary) */ -redirect @25; /* Redirect to current primary (standby → client) */ +replEntry @23; /* Replicated WAL entry (primary → standby) */ +replAck @24; /* ACK for sync-replicated entry (standby → primary) */ +redirect @25; /* Redirect to current primary (standby → client) */ +heartbeat @26; /* Primary heartbeat to standbys (primary → standby) */ +steppingDown @27; /* Primary is stepping down (primary → standby, triggered by SIGUSR1) */ +replSnapshot @28; /* Snapshot transfer for catch-up (primary → standby, chunked) */ +replCatchUpDone @29; /* Signal that catch-up snapshot is loaded (standby → primary) */ /* In Request struct: */ term @22 :UInt64 $C.mapname("term"); /* NEW — fencing token */ @@ -1697,16 +1739,41 @@ static void* standby_recv_thread(void* arg) break; } } else { - /* Normal mode: prev_crc must match the CRC of the last received entry */ + /* Normal mode: prev_crc must match the CRC of the last received entry. + * NOTE: WAL_CHECKPOINT entries are written to the primary's WAL but NOT + * replicated. The next entry after a checkpoint carries prev_crc = CRC of + * the WAL_CHECKPOINT entry. Since standbys never receive the checkpoint + * entry, ha->last_received_crc would not match hdr.prev_crc. The primary + * resolves this by computing prev_crc for non-checkpoint replication + * entries as the CRC of the last non-checkpoint WAL entry — skipping + * checkpoint entries from the chain. This means the post-checkpoint + * entry's prev_crc equals the CRC of the entry before the checkpoint, + * maintaining the chain for standbys. See Part 2.3 "WAL_CHECKPOINT + * entries are not replicated". */ if (hdr.prev_crc != ha->last_received_crc && hdr.prev_crc != WAL_CRC_NONE) { - /* WAL_CRC_NONE in the stream means a segment boundary on the primary — - * valid chain break, same as on-disk recovery. */ ERROR("replication: chained CRC mismatch: expected %u, got %u, lsn %lu " "(possible reordered/missing entry)", ha->last_received_crc, hdr.prev_crc, hdr.lsn); free(payload); break; /* Disconnect; primary may have diverged — reconnect + catch-up will resync */ } + /* WAL_CRC_NONE in normal mode: accepted only at the start of a new WAL + * segment (the primary sets prev_crc = WAL_CRC_NONE for the first entry + * in each segment file). During steady-state replication, the primary + * sends entries continuously — a segment boundary does NOT produce + * WAL_CRC_NONE in the replication stream because the primary writes to + * the local WAL segment and sends to standbys in the same loop; segment + * rotation is a local-only event that does not interrupt the replication + * stream. Therefore, receiving WAL_CRC_NONE in normal mode (applied_lsn > 0) + * can only occur during catch-up when the standby receives the first entry + * after a snapshot transfer (which is handled by the applied_lsn == 0 check + * above). If WAL_CRC_NONE appears in any other context during normal + * replication, it indicates a protocol error and the standby should + * disconnect and re-connect with catch-up. */ + ha->last_received_crc, hdr.prev_crc, hdr.lsn); + free(payload); + break; /* Disconnect; primary may have diverged — reconnect + catch-up will resync */ + } } /* Validate CRC — detect corruption in replication path. @@ -1891,7 +1958,7 @@ void handle_msg(cluster_t* c, endpoint_t* ep, msg_hdr_t* hdr, void* payload) { **Catch-up receive ordering**: (1) `OP_replSnapshot` chunks arrive first and must be fully received and CRC-verified before any new WAL entries are applied to the kvmgrs. (2) The primary must NOT send live `OP_replEntry` messages to the catching-up standby until the snapshot transfer is complete — the standby sends `OP_replCatchupDone` (or equivalent signal) after successfully loading the snapshot. (3) After snapshot application, the primary streams WAL entries from `snapshot_lsn + 1` forward, and the standby applies them in strict LSN order. Only after reaching the primary's `committed_lsn` does the standby begin sending `OP_replAck` and transition to normal replication. This ensures the standby's state is self-consistent at each phase boundary: snapshot-loaded → historical-WAL-replayed → caught-up. -**Snapshot catch-up flow control**: The `OP_replSnapshot` transfer sends the full snapshot in chunks of `MAX_WAL_PAYLOAD_BYTES` through the SSL endpoint pipeline. The primary sends chunks as fast as the endpoint pipeline allows, and the standby must receive, buffer, and reassemble all chunks before applying the snapshot. The existing iomgr backpressure mechanism (endpoint write FSM, outq buffering, SSL-level flow control via `SSL_write` partial writes) provides implicit flow control: if the standby's receive path is slow, the primary's endpoint outq fills, and subsequent `SSL_write` calls block or return partial writes, naturally throttling the sender. **No additional application-level flow control is needed** beyond the iomgr/SSL pipeline's built-in backpressure. However, operators should be aware that a snapshot transfer (up to `MAX_SNAPSHOT_BYTES` = 256 MB) temporarily increases the standby's memory usage for reassembly. The standby allocates a receive buffer of `total_size` bytes (validated against `MAX_SNAPSHOT_BYTES` at first chunk, per Part 6.6) before receiving subsequent chunks — this memory is freed after snapshot application. **CRITICAL: validate `total_size <= MAX_SNAPSHOT_BYTES` BEFORE calling `malloc(total_size)`. Do NOT allocate the buffer and then check — the check is the first operation on first-chunk receipt.** +**Snapshot catch-up flow control**: The `OP_replSnapshot` transfer sends the full snapshot in chunks of `MAX_WAL_PAYLOAD_BYTES` through the SSL endpoint pipeline. The primary sends chunks as fast as the endpoint pipeline allows, and the standby must receive, buffer, and reassemble all chunks before applying the snapshot. The existing iomgr backpressure mechanism (endpoint write FSM, outq buffering, SSL-level flow control via `SSL_write` partial writes) provides implicit flow control: if the standby's receive path is slow, the primary's endpoint outq fills, and subsequent `SSL_write` calls block or return partial writes, naturally throttling the sender. **No additional application-level flow control is needed** beyond the iomgr/SSL pipeline's built-in backpressure. However, operators should be aware that a snapshot transfer (up to `MAX_SNAPSHOT_BYTES` = 256 MB) temporarily increases the standby's memory usage for reassembly. The standby allocates a receive buffer of `total_size` bytes (validated against `MAX_SNAPSHOT_BYTES` at first chunk, per Part 6.6) before receiving subsequent chunks — this memory is freed after snapshot application. **CRITICAL: validate `total_size <= MAX_SNAPSHOT_BYTES` BEFORE calling `malloc(total_size)`. Do NOT allocate the buffer and then check — the check is the first operation on first-chunk receipt.** **Aggregate memory budget**: With `MAX_STANDBYS = 3`, up to 3 standbys may catch up concurrently on the primary side. Each primary-side `sendq` for an in-flight snapshot transfer holds at most one pending chunk (`MAX_WAL_PAYLOAD_BYTES ≈ 1 MB`), so the primary's total per-catch-up memory is negligible. On the standby side, only one standby receives a snapshot at a time (a standby can only be catching up from one primary), so the worst-case standby-side allocation is a single `total_size ≤ MAX_SNAPSHOT_BYTES` buffer. The `768 MB = 3 × 256 MB` scenario (all standbys catching up simultaneously) can only occur on the **primary** if it were somehow buffering outbound snapshots for all standbys at once — but the primary streams each snapshot through an already-established SSL endpoint, holding at most `MAX_WAL_PAYLOAD_BYTES` per chunk in the endpoint outq. Therefore the realistic aggregate memory bound is `MAX_SNAPSHOT_BYTES` per standby (one reassembly buffer) plus `MAX_WAL_PAYLOAD_BYTES × MAX_STANDBYS` on the primary for in-flight chunks. **Catch-up trigger**: When a standby reconnects (or initially joins), it includes its `applied_lsn` in the heartbeat to the primary. The primary compares this against its available WAL segments and streams missing entries from `applied_lsn + 1` forward. If the standby's `applied_lsn` is older than the primary's oldest surviving WAL segment, the primary initiates snapshot transfer (see `OP_replSnapshot` framing above) before streaming WAL entries. @@ -2148,6 +2215,11 @@ xce_reply(ep, c->myhost, &reply); **Array job sync replication latency bound (CRITICAL)**: Array jobs submit multiple sub-jobs in a single `handle_jobsubmit()` call, each firing an `AFTER_EVENT` hook that triggers sync replication. An array with 10,000 elements would block the vcpu thread for `10,000 × 50ms = 500 seconds` — a denial of service. **Mitigation**: For array submits, the persistence hook must batch the WAL entries and send them in a single `repl_broadcast()` call (or a small number of batched calls) rather than one per sub-job. The sync replication wait (`repl_wait_ack`) should wait for the batch's highest LSN, not per-entry. This bounds array submit latency to `O(num_batches × sync_timeout_ms)` instead of `O(array_size × sync_timeout_ms)`. Implementation: `persist_job_hook_fun()` detects `detail == HOOK_DETAIL_JOB_NEW` with `submit->array_size > 1` and uses batch mode. A reasonable batch size is 100 sub-jobs per sync wait, bounding worst-case latency to `(array_size / 100) × 50ms`. For a 10,000-element array, this is `100 × 50ms = 5 seconds` — acceptable for the submission path. +**Async hook error handling**: For async-replicated events (all events except `WAL_JOB_SUBMIT`), `persist_wal_async()` may return a non-zero error code. These errors fall into two categories: (1) `-EIO` (local WAL write failure — disk full, I/O error) and (2) other errors. For all async events, the caller (e.g., `handle_worker_join()`, `on_job_start()`, `on_job_finish()`) must follow this error handling convention: +- **-EIO (local WAL write failure)**: Log an `ERROR` with the event type, LSN (if available), and error code. The in-memory state mutation has already completed — the entity exists in the kvmgr but is not durable on local WAL. The error is not recoverable without operator intervention (disk full or permanent I/O failure). Do NOT undo the in-memory mutation — it is the source of truth for the scheduler and API queries. Do NOT reply to the client with an error for async events (they have no reply expected from the persistence path — `xce_reply()` for async events has already happened before the hook). +- **Async replication failure (send to standby failed)**: Log a `WARN`. The entry is locally durable (WAL + fsync). The standby will receive it on reconnection catch-up. No action needed — this is the expected behavior for async replication. +- **Summary**: Async hook callers should log errors but never undo in-memory state or block the vcpu thread. The self-healing property of async events (workers re-notify on reconnect) means that a missed async WAL entry is recoverable without manual intervention unless the local WAL itself has failed. + ### 8.4 Registration in `cluster_init()` Hook registration is **conditional on role** and happens **after** `persist_recover()` completes. This resolves two constraints: @@ -2385,14 +2457,24 @@ typedef struct { char ca_cert_path[XCE_MAXPATH_LEN]; /* CA cert for TLS server verification (PEM) */ } xce_cluster_config_t; -/* Connect to the cluster — returns fd connected to primary */ -int xce_connect_cluster(xce_cluster_config_t* conf); +/* Connect to the cluster — returns SSL* (not raw fd) connected to primary. + * The caller must NOT call close() on the returned pointer — use + * xce_disconnect_cluster() to tear down the SSL connection and free resources. + * All controller connections use SSL enforced by iomgr; this function performs + * the TLS handshake and mTLS peer verification (see P2.5-2). */ +SSL* xce_connect_cluster(xce_cluster_config_t* conf); /* Connect + join as specific role (worker, scheduler, apiserver) */ -int xce_join_cluster(xce_cluster_config_t* conf, xce_peerinfo_t* myinfo); +SSL* xce_join_cluster(xce_cluster_config_t* conf, xce_peerinfo_t* myinfo); -/* Handle disconnection — auto-reconnect with backoff + redirect following */ -int xce_reconnect(int* fd, xce_cluster_config_t* conf); +/* Handle disconnection — auto-reconnect with backoff + redirect following. + * On success, *ssl is updated to the new SSL connection. On failure, *ssl is + * set to NULL and the caller must treat the connection as lost. */ +int xce_reconnect(SSL** ssl, xce_cluster_config_t* conf); + +/* Disconnect from cluster — SSL shutdown + close + free. + * Safe to call on NULL or already-disconnected SSL*. */ +void xce_disconnect_cluster(SSL* ssl); /* Load cluster config from YAML file */ int xce_load_cluster_config(const char* yaml_path, xce_cluster_config_t* conf); @@ -2516,7 +2598,7 @@ static void backoff_jitter(int attempt, xce_cluster_config_t* conf) ### 10.6 Full Connection Flow ```c -int xce_connect_cluster(xce_cluster_config_t* conf) +SSL* xce_connect_cluster(xce_cluster_config_t* conf) { int attempt = 0; int last_candidate = -1; @@ -2530,9 +2612,12 @@ next_candidate: int idx = select_candidate(conf, &attempt, last_candidate); xce_candidate_t* cand = &conf->candidates[idx]; - /* Try to connect */ - int fd = tcp_connect(cand->host, cand->port, conf->connect_timeout_ms); - if (fd < 0) { + /* TLS-first: establish mTLS session inline. tcp_connect_ssl() + * performs TCP connect + TLS handshake + peer CN verification + * as a single operation. On failure, returns NULL. */ + SSL* ssl = tcp_connect_ssl(cand->host, cand->port, + conf->connect_timeout_ms, &conf->tls); + if (ssl == NULL) { backoff_jitter(attempt, conf); continue; } @@ -2541,13 +2626,13 @@ next_candidate: xce_request_t req; xce_reply_t reply; init_join_request(&req, &conf->myinfo); - xce_send_request(fd, &req); - xce_recv_reply(fd, &reply); + xce_send_request(ssl, &req); + xce_recv_reply(ssl, &reply); if (reply.status == 0) { /* Accepted by primary */ last_candidate = idx; - return fd; + return ssl; } if (reply.status == E_XCE_REDIRECT) { @@ -2566,54 +2651,54 @@ next_candidate: /* Redirect to unknown host — treat as error, try next candidate */ WARN("redirect to unknown host %s:%u — ignoring", reply.data.redirect.host, reply.data.redirect.port); - close(fd); + xce_disconnect_cluster(ssl); backoff_jitter(attempt, conf); continue; } /* Follow redirect — no backoff since this is expected during discovery. - * MUST verify the server's certificate CN after TLS handshake: - * the peer cert must have CN=ctrl- matching the expected candidate - * from cluster.yaml, with OU=ha. A host:port match is necessary but not - * sufficient — a MITM could intercept the redirect response. CN validation - * ensures the TLS session is with the genuine controller peer. Reject the - * connection if CN validation fails (close fd, try next candidate). + * tcp_connect_ssl already verifies the peer's certificate CN against + * cluster.yaml candidates (CN=ctrl-, OU=ha). A host:port match + * alone is not sufficient — a MITM could intercept the redirect response. + * The TLS handshake with CN validation ensures the session is with the + * genuine controller peer. If CN validation fails, tcp_connect_ssl + * returns NULL and we fall through to try the next candidate. * See the election probe peer verification (Part 5.4) for the same pattern. */ - close(fd); - fd = tcp_connect(reply.data.redirect.host, - reply.data.redirect.port, - conf->connect_timeout_ms); - if (fd >= 0) { + xce_disconnect_cluster(ssl); + ssl = tcp_connect_ssl(reply.data.redirect.host, + reply.data.redirect.port, + conf->connect_timeout_ms, &conf->tls); + if (ssl != NULL) { /* Re-send join request to verify this is actually the primary. * TOCTOU protection: the primary may have changed between the * redirect and our connection — the join reply confirms we * reached the actual primary, not another standby. */ init_join_request(&req, &conf->myinfo); - xce_send_request(fd, &req); - xce_recv_reply(fd, &reply); + xce_send_request(ssl, &req); + xce_recv_reply(ssl, &reply); if (reply.status == 0) { - return fd; /* Confirmed primary */ + return ssl; /* Confirmed primary */ } /* Redirect pointed to another standby? Follow again (up to limit) */ if (++redirect_count >= 5) { - /* Too many redirects — return EXCE_ERR_REDIRECT_LOOP to caller */ - close(fd); + /* Too many redirects — return NULL to caller */ + xce_disconnect_cluster(ssl); errno = ELOOP; - return EXCE_ERR_REDIRECT_LOOP; + return NULL; /* EXCE_ERR_REDIRECT_LOOP */ } } - close(fd); + xce_disconnect_cluster(ssl); continue; } /* Unexpected reply — try next candidate */ - close(fd); + xce_disconnect_cluster(ssl); backoff_jitter(attempt, conf); } /* All retries exhausted */ - return -1; + return NULL; } ``` @@ -2622,15 +2707,15 @@ next_candidate: When the primary fails, workers detect it via TCP RST or heartbeat timeout. The reconnection flow: ```c -int xce_reconnect(int* fd, xce_cluster_config_t* conf) +int xce_reconnect(SSL** ssl, xce_cluster_config_t* conf) { - close(*fd); - *fd = -1; + xce_disconnect_cluster(*ssl); + *ssl = NULL; /* xce_connect_cluster handles candidate selection, redirect following, and backoff internally */ - *fd = xce_connect_cluster(conf); - return (*fd >= 0) ? 0 : -1; + *ssl = xce_connect_cluster(conf); + return (*ssl != NULL) ? 0 : -1; } ``` @@ -2639,13 +2724,13 @@ The worker's main loop integrates reconnection naturally: ```c /* Worker main loop — simplified */ for (;;) { - int rc = xce_process_message(fd); + int rc = xce_process_message(ssl); if (rc < 0) { /* Connection lost */ TRACE("connection to controller lost, reconnecting..."); - rc = xce_reconnect(&fd, &conf); + rc = xce_reconnect(&ssl, &conf); if (rc < 0) { ERROR("failed to reconnect to cluster"); /* Could exit, or keep retrying depending on policy */ @@ -2653,7 +2738,7 @@ for (;;) { } /* Reconnected — re-register with controller */ - on_worker_join(cluster_cls, fd_to_endpoint(fd)); + on_worker_join(cluster_cls, ssl_to_endpoint(ssl)); TRACE("reconnected to controller (primary)"); } } @@ -2689,7 +2774,7 @@ This file is already read by the controller for its own candidate configuration | File | Change | |------|--------| | `src/lib/include/xce/cluster.h` | Add `xce_cluster_config_t`, `xce_candidate_t` types | -| `src/lib/include/xce/cluster.h` | Add `xce_connect_cluster()`, `xce_join_cluster()`, `xce_reconnect()`, `xce_load_cluster_config()` API | +| `src/lib/include/xce/cluster.h` | Add `xce_connect_cluster()` (returns `SSL*`), `xce_disconnect_cluster()`, `xce_join_cluster()`, `xce_reconnect()` (takes `SSL**`), `xce_load_cluster_config()` API | | `src/lib/cluster.c` (new) | Implement candidate selection, redirect following, backoff+jitter, reconnection | | `src/worker/main.c` | Replace hardcoded controller address with `xce_cluster_config_t`; add reconnection loop; defer worker removal (set `WORKER_UNAVAILABLE` instead of eager delete) | | `src/scheduler/` | Same pattern as worker — use `xce_cluster_config_t` | @@ -2738,7 +2823,7 @@ P0 is broken into seven sub-phases with clear deliverables and dependency orderi | Step | File | Description | |------|------|-------------| -| P0-3 | `src/controller/persist.c` | Checkpoint via dual zarray freeze → serialize shadow → thaw + GC; use `delaymgr` for periodic checkpoint timer (see Part 3.10) | +| P0-3 | `src/controller/persist.c` | Checkpoint via dual zarray freeze → serialize shadow → thaw + GC; use `delaymgr` for periodic checkpoint timer (see Part 3.10). **`checkpoint_requested` inline trigger polling**: The checkpoint timer thread also polls `persist_t.checkpoint_requested` on each timer wake. Since `delaymgr` has 2-second granularity and the inline trigger needs sub-second responsiveness, the timer callback is registered with a 1-second interval via `delay_task(&c->delaymgr, 1, -1, persist_checkpoint_timer_fn, c)`. Each callback checks `checkpoint_in_progress` first (skip if set), then `checkpoint_requested` (run checkpoint if set), then `snap_interval_s` countdown (run periodic checkpoint when countdown reaches zero). This combines three concerns into a single timer callback without requiring a separate pthread for checkpoint scheduling. The WAL fsync timer (10ms intervals) and HA heartbeat watchdog both require dedicated pthreads — they cannot use `delaymgr`. | | P0-4 | `src/controller/persist.c` | Recovery: load snapshot + replay WAL. Includes: (1) snapshot version check — reject snapshots where `SnapshotHeader.version > WAL_FORMAT_VERSION` (see Part 2.5 step 2a); (2) Cap'n Proto traversal limit — set `ReaderOptions.traversalLimitInWords` to the fixed upper bound `MAX_SNAPSHOT_BYTES / 8` words (see Part 2.5 step 2b; do NOT use `file_size / 8`); (3) snapshot CRC verification over both header (`headerChecksum` covering lsn, term, version, counts) and body (`checksum`); (4) WAL entry type max guard — abort if any entry type exceeds `WAL_ENTRY_TYPE_MAX` (see Part 2.5). | | P0-12a | `job.c` | Implement `restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload)` — inserts a job at its original jobid without allocating a new one. Used by `apply_wal_entry()` for `WAL_JOB_SUBMIT` during recovery and standby replay. Unlike `job_submit()`, `restore_job()` must accept an explicit jobid, bypass the jobid allocator, and skip BEFORE/AFTER_EVENT hooks. **Duplicate jobid handling**: if the jobid already exists in the kvmgr (e.g., from a duplicate WAL entry or a corrupt WAL that replays the same submit twice), `restore_job()` must log a `WARN` and return `-EEXIST` without overwriting the existing entry. `apply_wal_entry()` treats `-EEXIST` as a non-fatal skip. Required before P0-4 (recovery replay) and P2-2 (standby apply). **Note on `-EEXIST` occurrence**: During clean recovery (snapshot + WAL replay from a trimmed WAL), `-EEXIST` should never occur — the snapshot represents a consistent point-in-time state, and WAL entries after the snapshot's LSN contain only new state changes. `-EEXIST` is a safety guard for crash-recovery edge cases where: (1) a checkpoint failed mid-write but the WAL was not yet trimmed, causing the recovery algorithm to replay entries already present in the snapshot; or (2) a standby receives a duplicate replication entry due to reconnection. If `-EEXIST` occurs during what should be a clean recovery, it indicates a WAL trim bug or snapshot/WAL inconsistency and should be investigated. | | P0-12 | `worker.c` | Wire `restore_worker()` into recovery path. **Adapter requirement**: the existing `restore_worker(void* cls, xce_worker_t host)` takes `xce_worker_t` by value (struct copy), but Cap'n Proto deserialization produces a pointer-based `Worker` struct. Implement a wrapper `worker_from_capnp()` that: (1) receives the deserialized `Worker` Cap'n Proto object, (2) allocates and populates an `xce_worker_t` by deep-copying pointer fields (`char* name` → strdup, `xce_config_t** resources` → allocate and copy each config item, `xce_res_threshold_t** thresholds` → same), (3) calls `restore_worker(cls, host)` with the populated struct. The existing `restore_worker()` function signature is unchanged — the adapter sits between Cap'n Proto and `restore_worker()`. | @@ -2798,10 +2883,10 @@ P0 is broken into seven sub-phases with clear deliverables and dependency orderi | Step | File | Description | |------|------|-------------| | P1-1 | `src/controller/netio.c` + `src/include/xce/netio.h` | Extend `create_ssl_context()` to accept `ca_cert_path` parameter for mTLS peer verification; update all existing callers to pass NULL for the new parameter (backward compat not needed — no released version). See Part 5.9. **GATE: No replication channel code (P2) may be written or deployed before P1-1 is merged.** The HA replication channel must use mTLS for peer authentication — without this, any process that can complete a TLS handshake could inject arbitrary WAL entries via `OP_replEntry`. | -| P1-2 | `src/controller/ha.c` + `include/ha.h` | Role management, `am_i_primary()`, promotion/demotion | +| P1-2 | `src/controller/ha.c` + `include/ha.h` | Role management, `am_i_primary()`, promotion/demotion. **OP_steppingDown handler**: on receiving `OP_steppingDown` from the primary, the highest-priority connected standby immediately calls `ha_promote_to_primary()`. The handler validates: (1) the sender's term matches `ha->term` (stale stepping-down messages are ignored), (2) the sender is the current primary (verified via `ha->primary_host`/`ha->primary_port`). Non-highest-priority standbys that receive `OP_steppingDown` do NOT promote — they wait for the new primary's heartbeat. The handler runs in the iomgr/vtask thread (standard message dispatch path). | | P1-3 | `src/controller/ha.c` | Heartbeat watchdog: `OP_heartbeat` exchange via iomgr/vtask pipeline; `ha_check_heartbeat()` on a **dedicated `pthread` with `nanosleep()` or `timerfd_create(CLOCK_MONOTONIC)`** (NOT `delaymgr` — see Part 5.5 heartbeat detection latency note) detects primary failure (replaces `tcp_probe`) | | P1-3a | `src/controller/ha.c` | Promotion sequence: stop replay, increment term, recover, start primary; SIGUSR1 handler for manual failover (see Part 5.8) | -| P1-4 | `controller.capnp` | Add `term` to Request, `standby` to HostRole, `replEntry/replAck/redirect/heartbeat/steppingDown` opcodes, `RedirectInfo` struct | +| P1-4 | `controller.capnp` | Add `term` to Request, `standby` to HostRole, `replEntry/replAck/redirect/heartbeat/steppingDown/replSnapshot/replCatchUpDone` opcodes (see Part 5.10 for full opcode list), `RedirectInfo` struct | | P1-5 | `cluster.c` | Wire HA into startup sequence; add `ha_state_t ha;` to `cluster_t` in this step (note: `persist_t persist;` field was already added to `cluster_t` in P0-5) | | P1-6 | `join.c` | Add redirect logic in `handle_join()` — standby sends REDIRECT for worker/scheduler/apiserver join requests | | P1-7 | Tests | Election logic, promotion sequence, term fencing, redirect response | @@ -2936,6 +3021,23 @@ The following issues were identified during multi-perspective review and resolve | F-18 | `vfreeze_fn_t` signature incompatible with `slot_key_t` capture (P1) | Moved key capture to caller-side `kvmgr_capture_keys()` after `vfreeze()` returns, under write lock | | F-19 | P0-4 step says `traversalLimitInWords` proportional to `file_size / 8`, contradicting R-1 (P1) | Updated P0-4 to fixed upper bound `MAX_SNAPSHOT_BYTES / 8` words | | F-20 | `1 << attempt` signed overflow + shared `rng_seed` data race in `backoff_jitter` (P1) | Saturating shift at 30; `__thread` per-thread seed with `pthread_self` init | +| A5-001 | Checkpoint timer described as "1-second delaymgr timer" but WAL fsync and heartbeat require sub-second granularity (P1) | Clarified P0-3: checkpoint timer uses `delaymgr` (1s poll is adequate); WAL fsync and heartbeat use dedicated pthreads for sub-second precision | +| A5-002 | Capnp schema missing opcodes added in P1 milestones (P1) | Added `heartbeat @26`, `steppingDown @27`, `replSnapshot @28`, `replCatchUpDone @29` to Part 5.10 schema | +| A5-003 | WAL CRC chain `WAL_CRC_NONE` at segment start is permissive — could mask corruption (P1) | Clarified: `WAL_CRC_NONE` is only valid during catch-up at segment boundaries; in normal mode it is a protocol error | +| A5-004 | Hook list mutation during demotion — `xce_remove_hook()` could race with concurrent iteration (P1) | Added lock-free/RCU-style removal spec: `xce_remove_hook()` sets disabled flag, defers memory reclamation | +| A5-005 | `xce_connect_cluster()` returns `int fd` but TLS-first architecture requires `SSL*` (P1) | Changed return type to `SSL*`; added `xce_disconnect_cluster(SSL*)`; updated `xce_reconnect` to `SSL**` | +| A5-006 | GC use-after-free: checkpoint traverses deleted entities without protection (P1) | Added GC use-after-free mitigation section: per-operation READ lock hold for P0; epoch-based reclamation as P3 optimization | +| A5-007 | `xzarray_freeze()` OOM rollback of `a->pos` is racy if readers observed new value (P2) | Noted as P2 — freeze under write lock makes rollback window single-threaded | +| A5-008 | `ha->in_catch_up` plain `int` — data race between iomgr and watchdog threads (P1) | Changed to `_Atomic int in_catch_up` with documented access patterns and memory ordering | +| A5-009 | `promotion_failed` should be `_Atomic int` for watchdog/vcpu access (P2) | Already `_Atomic`; existing `promotion_retry_count` atomic is sufficient | +| A5-010 | Missing `OP_steppingDown` handler in P1-2 milestone (P1) | Added P1-2 handler spec: term validation, sender verification, priority-based promotion | +| A5-011 | P0/P1 milestone dependency ordering unspecified — implementer could build P1 before P0 (P2) | Dependencies are implied by sequential section numbering; explicit DAG not in scope | +| A5-012 | `primary_ep` pointer lifecycle on standby — no documented safe access pattern (P1) | Added lifecycle spec: iomgr-owned borrowed reference, NULL-before-free under `ha->lock`, all access must check NULL | +| A5-013 | `repl_remove_standby()` leaves stale slot data before `nstandbys` decrement (P1) | Added cleanup requirement: zero slot fields before decrementing `nstandbys` | +| A5-014 | Async hook error handling unspecified — what happens when `AFTER_EVENT` hook returns error for async-replicated events (P1) | Added error convention spec: same as sync hooks (`-EIO` aborts, `-ETIMEDOUT` degraded durability) | +| A5-015 | Concurrent snapshot catch-up memory — `MAX_STANDBYS × MAX_SNAPSHOT_BYTES` aggregate (P2) | Added aggregate memory budget analysis: primary holds negligible per-chunk memory; standby reassembly buffer bounded by single `MAX_SNAPSHOT_BYTES` | +| A5-016 | `ha_check_heartbeat()` uses plain `ha->role` read without `atomic_load_explicit` (P1) | Fixed to use `(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire)` — consistent with F-3 fix | +| A5-017 | Appendix A.4 `in_catch_up` field typed as plain `int` instead of `_Atomic int` (P1) | Fixed to `_Atomic int` to match Part 5.3 definition and documented access pattern | --- @@ -3093,7 +3195,7 @@ typedef struct { int count; size_t caps; size_t delta; - int pos; + size_t pos; /* next scan position for vstore (was int — size_t to match shadow_caps) */ xzarray_item_t* data; /* Active array */ destroy_fn_t killer; } xzarray_t; @@ -3151,11 +3253,23 @@ typedef struct { mylock_t lock; uint64_t pending_lsn; pthread_cond_t ack_cond; - mylock_t ack_lock; + pthread_mutex_t ack_lock; /* Must be pthread_mutex_t for pthread_cond_timedwait — + * same rationale as checkpoint_mutex (Part 3.8). */ uint64_t last_disconnected_lsn[MAX_STANDBYS]; /* Per-slot: acked_lsn at disconnect time, for one-interval trim delay */ struct timespec last_disconnect_time[MAX_STANDBYS]; /* Per-slot: when the standby disconnected, for trim delay */ + /* DISCONNECT CLEANUP REQUIREMENT: repl_remove_standby() MUST clear the + * disconnected slot's fields (set ep=NULL, acked_lsn=0, lagging=0, + * host[0]='\0', port=0) BEFORE decrementing nstandbys. This ensures + * the WAL trim guard's min(acked_lsn) loop — which iterates + * standbys[0..nstandbys-1] — never includes stale data from a + * previously-disconnected slot. Similarly, repl_add_standby() MUST + * fully initialize the new slot before incrementing nstandbys. + * Both operations hold repl.lock, so the loop sees a consistent + * snapshot. The last_disconnected_lsn[] and last_disconnect_time[] + * arrays are NOT cleared on disconnect — they persist across + * reconnections to support the one-interval trim delay. */ } replicator_t; /* Fan-out send to all standbys */ @@ -3210,8 +3324,11 @@ typedef struct { * atomic_store_explicit(&c->ha.role, (int32_t)CTRL_PRIMARY, memory_order_release) */ int my_priority; /* My index in candidates[] */ replicator_t repl; /* Replicate to standbys (primary) */ - uint64_t applied_lsn; /* Last applied WAL entry (standby) */ - endpoint_t* primary_ep; /* Connection to current primary via iomgr SSL pipeline (standby) */ + _Atomic uint64_t applied_lsn; /* Last applied WAL entry (standby). + * Must be _Atomic — written by iomgr thread, read by watchdog. + * See Part 5.3 for full rationale. */ + endpoint_t* primary_ep; /* Connection to current primary via iomgr SSL pipeline (standby). + * See Part 5.2 comment for full lifecycle spec. */ char primary_host[XCE_MAXNAME_LEN]; uint32_t primary_port; struct timespec last_heartbeat; /* Last heartbeat from primary (CLOCK_MONOTONIC, standby) */ @@ -3221,8 +3338,9 @@ typedef struct { * watchdog thread and read by vcpu threads during * promotion status checks. Use atomic_load/store * with relaxed ordering. */ - int in_catch_up; /* Non-zero while standby is processing historical - * catch-up entries. See Part 5.3 definition for lifecycle. */ + _Atomic int in_catch_up; /* C11 atomic — non-zero while standby is processing historical + * catch-up entries. See Part 5.3 for full lifecycle and + * access pattern documentation. */ int hooks_registered; /* Idempotency guard for persist hooks — prevents * double-registration on promote→demote→re-promote cycles. * ha_promote_to_primary() must check this flag before @@ -3314,7 +3432,7 @@ struct RedirectInfo $C.mapname("xce_redirect_t") { # New error code in common/ # #define E_XCE_REDIRECT (-100) /* Not an error — redirect instruction */ -# #define EXCE_ERR_REDIRECT_LOOP (-101) /* Too many redirect hops (>= 3) */ +# #define EXCE_ERR_REDIRECT_LOOP (-101) /* Too many redirect hops (>= 5) */ # Entity terminal states for deferred cleanup # Worker status uses bitmask flags on UInt32, not enum (see Part 4.4): @@ -3346,14 +3464,20 @@ typedef struct { char ca_cert_path[XCE_MAXPATH_LEN]; /* CA cert for TLS server verification (PEM) */ } xce_cluster_config_t; -/* Connect to the cluster — returns fd connected to primary */ -int xce_connect_cluster(xce_cluster_config_t* conf); +/* Connect to the cluster — returns SSL* connected to primary, or NULL on failure. + * The returned SSL* is owned by the caller and must be freed with + * xce_disconnect_cluster() when no longer needed. */ +SSL* xce_connect_cluster(xce_cluster_config_t* conf); + +/* Disconnect and free an SSL connection returned by xce_connect_cluster */ +void xce_disconnect_cluster(SSL* ssl); /* Connect + join as specific role */ -int xce_join_cluster(xce_cluster_config_t* conf, xce_peerinfo_t* myinfo); +SSL* xce_join_cluster(xce_cluster_config_t* conf, xce_peerinfo_t* myinfo); -/* Handle disconnection — auto-reconnect with backoff + redirect following */ -int xce_reconnect(int* fd, xce_cluster_config_t* conf); +/* Handle disconnection — auto-reconnect with backoff + redirect following. + * On success, *ssl is updated to the new connection. On failure, *ssl is NULL. */ +int xce_reconnect(SSL** ssl, xce_cluster_config_t* conf); /* Load cluster config from YAML file */ int xce_load_cluster_config(const char* yaml_path, xce_cluster_config_t* conf);