diff --git a/src/docs/persistence-and-ha-design.md b/src/docs/persistence-and-ha-design.md index 480dac2..846686c 100644 --- a/src/docs/persistence-and-ha-design.md +++ b/src/docs/persistence-and-ha-design.md @@ -735,13 +735,25 @@ persist_checkpoint() fires on timer thread } slot_key_t; ``` - `xzarray_freeze()` receives an output `slot_key_t**` and `size_t* n_slots`. - The caller (per-kvmgr checkpoint code) allocates the array under the write - lock, copies key pointers and lengths from kvtab entries for live indices, - and returns it. The serialization phase reads this array (no locks held) - to map shadow slots to their keys in the snapshot. After serialization - completes and thaw runs, the array is freed. This avoids any kvtab access - during the no-locks-held serialization phase. + `xzarray_freeze()` returns after the pointer swap. While still holding the + kvmgr write lock (before releasing it), the caller captures key→index + mappings by walking the kvtab and building a `slot_key_t[]` array: + + ```c + /* Caller-side step: after vfreeze(obj) returns, still under kvmgr write lock */ + vfreeze(obj); /* xzarray_freeze — O(1) pointer swap */ + slot_key_t* keys = NULL; + size_t n_keys = 0; + kvmgr_capture_keys(kvmgr, &keys, &n_keys); /* walks kvtab under write lock */ + /* Now release write lock. Serialization reads keys[] (no locks held). */ + ``` + + This approach avoids threading output parameters through the RTTI interface. + The `vfreeze_fn_t` signature remains `int (*vfreeze_fn_t)(meta_obj_t* obj)`. + Key capture is a separate caller-side function that walks the kvtab hash table + while the write lock is still held. The serialization phase reads the keys[] + array (no locks held) to map shadow slots to their keys in the snapshot. + After serialization completes and thaw runs, the keys[] array is freed. │ ▼ After serialization confirmed on disk: @@ -1221,7 +1233,8 @@ The controller reads only the binary Cap'n Proto config file — it has no YAML typedef enum { CTRL_PRIMARY = 0, CTRL_STANDBY = 1 } ctrl_role_t; typedef struct { - ctrl_role_t role; + _Atomic uint64_t term; /* C11 atomic election term (fencing token) — see Part 5.7 */ + _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) */ @@ -1235,6 +1248,14 @@ 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 + * OP_replCatchUpDone). Gates ACK suppression: during + * catch-up, the standby does NOT send OP_replAck + * for historical entries (see Part 6.4 line 1778). */ ha_config_t config; /* HA configuration (timeouts, intervals) */ } ha_state_t; @@ -1431,7 +1452,7 @@ When ctrl-1 recovers: 3. ctrl-1 stays standby and connects to ctrl-2 as primary 4. Admin can trigger manual failover to restore optimal controller placement: -**Manual failover protocol**: To move primary back to a higher-priority candidate (e.g., ctrl-1 after recovery), the operator sends `SIGUSR1` to the current primary (ctrl-2). The signal handler calls `ha_demote_to_standby()`, which: (a) unregisters persistence hooks (see Part 5.7 hook unregistration), (b) sets `ha.role = CTRL_STANDBY` using `atomic_store_explicit` with `memory_order_release`, (c) stops accepting new writes, (d) flushes and sync-replicates any pending WAL entries, (e) sends a `OP_steppingDown` message to all connected standbys with its current term and the election deadline. On receiving `OP_steppingDown`, the highest-priority connected standby (ctrl-1) immediately begins its election (calls `ha_promote_to_primary()`). The stepping-down controller then reconnects to the newly promoted primary as a standby. **During the transfer window (~1-2 election_timeout_ms), no controller is primary — job submits receive `EAGAIN` and must retry.** This is the same brief unavailability as an automatic failover. The `SIGUSR1` approach avoids needing a dedicated admin RPC or CLI command in the initial implementation; a future `xce hostcontrol --failover` CLI can wrap this by sending the signal remotely via SSH. +**Manual failover protocol**: To move primary back to a higher-priority candidate (e.g., ctrl-1 after recovery), the operator sends `SIGUSR1` to the current primary (ctrl-2). The SIGUSR1 signal handler **must only set an atomic flag**: `atomic_store(&ha->failover_requested, 1)`. It does NOT call `ha_demote_to_standby()` directly — that function performs operations (mutex locks, fsync, SSL writes, memory allocation) that are not async-signal-safe per POSIX. The failover is executed by the dedicated HA watchdog thread (the same pthread running `ha_check_heartbeat()` — P1-3), which polls `ha->failover_requested` on each tick. When the flag is observed, the watchdog thread calls `ha_demote_to_standby()` from its safe thread context, which: (a) unregisters persistence hooks (see Part 5.7 hook unregistration), (b) sets `ha.role = CTRL_STANDBY` using `atomic_store_explicit` with `memory_order_release`, (c) stops accepting new writes, (d) flushes and sync-replicates any pending WAL entries, (e) sends a `OP_steppingDown` message to all connected standbys with its current term and the election deadline. On receiving `OP_steppingDown`, the highest-priority connected standby (ctrl-1) immediately begins its election (calls `ha_promote_to_primary()`). The stepping-down controller then reconnects to the newly promoted primary as a standby. **During the transfer window (~1-2 election_timeout_ms), no controller is primary — job submits receive `EAGAIN` and must retry.** This is the same brief unavailability as an automatic failover. The `SIGUSR1` approach avoids needing a dedicated admin RPC or CLI command in the initial implementation; a future `xce hostcontrol --failover` CLI can wrap this by sending the signal remotely via SSH. Add `atomic_int failover_requested;` to `ha_state_t` for this purpose. **Demotion failure handling**: If `persist_flush()` fails during `ha_demote_to_standby()` (step d above), the controller has already set `ha.role = CTRL_STANDBY` and incremented term. The safest recovery is to **not rollback** the role change — instead, log a `FATAL("persist_flush failed during demotion — WAL may be incomplete. Manual investigation required.")`. The demoted controller will not accept new writes (role is STANDBY), and the WAL will be flushed again on the next checkpoint. A role rollback (setting `ha.role = CTRL_PRIMARY` on flush failure) would leave the cluster in a split-brain-eligible state where the original primary continues accepting writes while a new primary may already be promoting. Failing forward as STANDBY is safer: the cluster temporarily has no primary until another standby wins election, which is the same state as a normal failover. @@ -1510,7 +1531,9 @@ typedef struct { mylock_t lock; /* Protects standbys array */ uint64_t pending_lsn; /* LSN we're waiting ACK for (sync mode) */ pthread_cond_t ack_cond; /* Signaled when ACK arrives */ - mylock_t ack_lock; /* Protects ack_cond + pending_lsn */ + pthread_mutex_t ack_lock; /* Protects ack_cond + pending_lsn. MUST be pthread_mutex_t + * (not mylock_t) because pthread_cond_timedwait requires a + * pthread_mutex_t* — same rationale as checkpoint_mutex. */ } replicator_t; ``` @@ -2455,33 +2478,36 @@ static int select_candidate(xce_cluster_config_t* conf, int* attempt, ### 10.5 Backoff with Jitter ```c -/* Thread-safe RNG initialization using pthread_once. - * Uses getrandom() for an unpredictable seed (avoids the correlated seeds - * from getpid() ^ time(NULL) on HPC clusters where many nodes boot simultaneously). - * The static seeded flag + double-checked locking in the original implementation - * was a data race — use pthread_once for correct one-time initialization. */ -static unsigned int rng_seed; /* per-process seed for rand_r() */ -static pthread_once_t rng_once = PTHREAD_ONCE_INIT; +/* Thread-local RNG seed — each thread initializes its own seed once. + * Using __thread avoids the data race on a shared global rng_seed when + * multiple threads call backoff_jitter() concurrently (rand_r with the + * same shared seed is concurrent read-modify-write, which is a race). */ +static __thread unsigned int rng_seed; +static __thread int rng_init = 0; -static void rng_do_init(void) { - if (getrandom(&rng_seed, sizeof(rng_seed), 0) != sizeof(rng_seed)) - rng_seed = (unsigned int)time(NULL) ^ (unsigned int)getpid(); +static void rng_init_once(void) { + if (rng_init) return; + /* getrandom is not available per-thread in one syscall, so seed from + * combination of thread ID and time — good enough for jitter purposes. */ + unsigned long tid = (unsigned long)pthread_self(); + rng_seed = (unsigned int)(tid ^ time(NULL)); + rng_init = 1; } /* Exponential backoff with random jitter to spread reconnect storm. - * Uses rand_r() (thread-safe) instead of rand() (per-process, not thread-safe). */ + * Uses per-thread rand_r() to avoid data races. */ static void backoff_jitter(int attempt, xce_cluster_config_t* conf) { - int delay_ms = conf->retry_base_ms * (1 << attempt); + int64_t delay = (int64_t)conf->retry_base_ms * ((int64_t)1 << (attempt > 30 ? 30 : attempt)); - if (delay_ms > conf->retry_max_ms) - delay_ms = conf->retry_max_ms; + if (delay > conf->retry_max_ms) + delay = conf->retry_max_ms; - pthread_once(&rng_once, rng_do_init); - int jitter = rand_r(&rng_seed) % (delay_ms / 2 + 1); - delay_ms += jitter; + rng_init_once(); + int jitter = rand_r(&rng_seed) % ((int)delay / 2 + 1); + delay += jitter; - usleep(delay_ms * 1000); + usleep((useconds_t)(delay * 1000)); } ``` @@ -2713,7 +2739,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-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` proportional to `file_size / 8` and reject snapshots larger than `MAX_SNAPSHOT_BYTES` (see Part 2.5 step 2b); (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-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()`. | @@ -2904,6 +2930,12 @@ The following issues were identified during multi-perspective review and resolve | F-12 | Async WAL write failure silently swallowed — no logging (P1) | Added `WARN()` log on `persist_wal_replicated_entry()` failure | | F-13 | Step 5.5 references `persist_recover()` but promotion doesn't call it (P1) | Changed wording to reference `cluster_start()` instead | | F-14 | `apply_wal_entry()` return value ignored — corrupted entry silently accepted (P0) | Added return value check; break replication stream on failure | +| F-15 | `replicator_t.ack_lock` is `mylock_t` but `pthread_cond_timedwait` requires `pthread_mutex_t*` (P0) | Changed to `pthread_mutex_t` — same rationale as `checkpoint_mutex` (R-6) | +| F-16 | `ha->in_catch_up` referenced in ACK gate but absent from `ha_state_t` (P0) | Added `in_catch_up` field to `ha_state_t` in Part 5.3 and Appendix A.4 with lifecycle spec | +| F-17 | SIGUSR1 handler calls non-async-signal-safe `ha_demote_to_standby()` (P0) | SIGUSR1 sets `atomic_int failover_requested` only; watchdog thread executes actual demotion | +| 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 | --- @@ -3189,10 +3221,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. See Part 5.3 definition for lifecycle. */ 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 * calling xce_add_hook(). Reset to 0 on demote. */ + atomic_int failover_requested; /* Async-signal-safe flag set by SIGUSR1 handler. + * Polled by HA watchdog thread which executes the + * actual failover (Part 5.8). Never read/written + * outside of SIGUSR1 handler and watchdog thread. */ uint32_t last_received_crc; /* CRC of the last received WAL entry on the replication * stream — used for chained CRC validation (P0-C fix). * Mirrors persist_t.last_crc but on the standby side. */