fix: address fourth-round adversarial review findings for persistence-and-ha-design
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
parent
454b4aa355
commit
c623e37b84
1 changed files with 67 additions and 29 deletions
|
|
@ -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. */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue