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

feat(ha-entity-snapshots): restore-before-advance load_snapshot implementation (task 5.2)

Browse source 
Replaces the naive task-5.1 stopgap with the real restore contract:
detects legacy vs entity-bearing checkpoints by remaining file size;
restores every job then every queue status via restore_job/
restore_queue_state, in file order, with bounds-checked length-prefixed
reads; only after every entity restores successfully does
current_lsn/checkpoint_lsn commit to the checkpoint's position. Every
failure path (bounds check, decode, restore error, unrecognized
format_version) returns before that single commit point -- position is
provably untouched on failure, never partially advanced.

Also fixes the pre-existing restore_job decode leak (job.c): job_init
deep-copies submit (never nulled) and MOVE_PTRs allocs/pidinfo/
pend_reasons (nulled on the one path that runs) -- free_Job(&j) is safe
unconditionally on every reachable return path (EEXIST early-return,
create_job success, create_job failure), verified against the exact
generated free_*_ptr null-check guards.

Two pre-existing test_config_repl.c wire-protocol tests needed their
synthetic garbage snapshot bodies updated to a well-formed empty-entity
tail, since load_snapshot now genuinely parses past the legacy header
(previously ignored); their actual assertions (chunk-reassembly
correctness) are unchanged.

This is the exact function/contract that caused task 2.2's rejection
(position advancing before restore validated). Independent review
traced the memory-safety fix through the generated capnp code field-by-
field, confirmed the test_config_repl.c change doesn't weaken any
assertion, and confirmed restore_queue_state's ENOENT-aborts-the-whole-
checkpoint behavior is faithful to design.md's unconditional contract
(flagged as a watch-item for task 7.1's soak, not a defect).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JDYFbrgyDHbvNYcxApKhrX
This commit is contained in:
Rongsong Shen 2026-07-01 08:30:10 +00:00
parent a8e300d96a
commit 9fdb6778d4
4 changed files with 297 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

268
src/controller/persist.c
View file

@ -2162,10 +2162,137 @@ done:
/* Recovery internals */
/* ------------------------------------------------------------------ */
/* P0 sentinel format: "XCESNAP\0" (8 bytes) + checkpoint_lsn (8 bytes LE).
* Returns 0 on success and sets p->checkpoint_lsn / p->current_lsn.
* Returns -EINVAL on bad magic, -EIO on short read, -errno on open failure. */
static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
/* Parse and restore every job frame then every queue frame from the entity
* payload beginning at the current `fd` offset (design.md "Checkpoint File
* Layout"): [job_count][job_count * {payload_len; capnp Job bytes}]
* [queue_count][queue_count * {name_len; name; status}].
*
* `file_size` is the checkpoint file's total size (from fstat) and
* `consumed` is the number of bytes already read from it (legacy header +
* format_version) — every length-prefixed frame's length is bounds-checked
* against the ACTUAL remaining bytes in the file before being trusted for a
* read or allocation, never blindly followed (a corrupt/attacker-controlled
* length must fail fast, not drive an oversized read).
*
* Restores in file order: every job via restore_job, then every queue via
* restore_queue_state (design.md sequence diagram). Returns 0 only if every
* frame restored; on the FIRST failure of any kind (short read, oversized
* length, decode error, or restore error) returns immediately with a
* negative errno — the caller MUST NOT commit any position change on a
* non-zero return (Requirement 2.3).
*
* restore_job's -EEXIST is treated as success, not a failure: it is job.c's
* own documented idempotent-replay guard (job already present), needed so a
* retried load — after an earlier partial attempt already created some jobs
* before failing on a later one — can continue past already-restored entries
* rather than aborting on them again. restore_queue_state has no such
* idempotent-duplicate case (re-applying the same status is already a
* harmless overwrite), so any non-zero return from it is a genuine failure. */
static int load_snapshot_entities(int fd, off_t file_size, off_t consumed,
persist_t* p, uint64_t stored_lsn)
{
uint32_t job_count = 0;
if (read(fd, &job_count, sizeof(job_count)) != (ssize_t)sizeof(job_count))
return -EIO;
consumed += (off_t)sizeof(job_count);
if (job_count > 0 && (p->cluster == NULL || p->cluster->jobmgr == NULL))
return -EINVAL; /* entities present but nowhere to restore them into */
for (uint32_t i = 0; i < job_count; i++) {
uint32_t payload_len = 0;
if (read(fd, &payload_len, sizeof(payload_len)) !=
(ssize_t)sizeof(payload_len))
return -EIO;
consumed += (off_t)sizeof(payload_len);
if (payload_len == 0 || (off_t)payload_len > file_size - consumed)
return -EIO;
uint8_t* buf = (uint8_t*)malloc(payload_len);
if (buf == NULL)
return -ENOMEM;
if (read(fd, buf, payload_len) != (ssize_t)payload_len) {
free(buf);
return -EIO;
}
consumed += (off_t)payload_len;
int rrc = restore_job(p->cluster, 0, buf, payload_len, stored_lsn);
free(buf);
if (rrc != 0 && rrc != -EEXIST)
return rrc;
}
uint32_t queue_count = 0;
if (read(fd, &queue_count, sizeof(queue_count)) !=
(ssize_t)sizeof(queue_count))
return -EIO;
consumed += (off_t)sizeof(queue_count);
if (queue_count > 0 && (p->cluster == NULL || p->cluster->qmgr == NULL))
return -EINVAL; /* entities present but nowhere to restore them into */
for (uint32_t i = 0; i < queue_count; i++) {
uint32_t name_len = 0;
if (read(fd, &name_len, sizeof(name_len)) != (ssize_t)sizeof(name_len))
return -EIO;
consumed += (off_t)sizeof(name_len);
if (name_len == 0 || (off_t)name_len > file_size - consumed)
return -EIO;
char* name = (char*)malloc((size_t)name_len + 1);
if (name == NULL)
return -ENOMEM;
if (read(fd, name, name_len) != (ssize_t)name_len) {
free(name);
return -EIO;
}
name[name_len] = '\0';
consumed += (off_t)name_len;
uint32_t status = 0;
if (read(fd, &status, sizeof(status)) != (ssize_t)sizeof(status)) {
free(name);
return -EIO;
}
consumed += (off_t)sizeof(status);
int qrc = restore_queue_state(p->cluster, name, status);
free(name);
if (qrc != 0)
return qrc;
}
return 0;
}
/* Load a snapshot: legacy header (magic + checkpoint_lsn + optional
* config-ref tail), then — for an entity-bearing checkpoint — every job and
* queue frame that follows it (ha-entity-snapshots task 5.2, Requirements
* 2.1-2.4, 5.1, 5.2). See design.md "load_snapshot (extended)" and its
* sequence diagram for the exact contract this implements.
*
* Ordering (Requirement 2.1, 2.2, 2.3 — restore FIRST, validate FULLY, THEN
* advance position, never the reverse, never partial): the header and any
* config-ref tail are parsed into LOCAL variables only. Format is then
* detected by whether any bytes remain past the header (design.md's stated
* discriminator — not a flag bit). A short file (nothing remains) is a
* legacy, pre-feature checkpoint: not an error, so its recorded position is
* committed immediately and *out_entities_restored is false. Otherwise every
* job then every queue frame is restored via load_snapshot_entities() before
* ANYTHING is written to p->checkpoint_lsn/p->current_lsn/p->cfgref_*; on
* any restore failure this function returns non-zero with p's fields
* PROVABLY untouched (the single commit point below is only ever reached
* after load_snapshot_entities() returns 0). *out_entities_restored is only
* set on a rc==0 return, matching persist.h's documented contract.
*
* Returns 0 on success, -EINVAL on bad magic / unrecognized entity format,
* -EIO on a short/corrupt read, -ENOMEM on allocation failure, -errno on
* open/fstat failure, or a restore_job/restore_queue_state error code. */
static int load_snapshot(persist_t* p, const char* path, uint64_t lsn,
bool* out_entities_restored)
{
(void)lsn; /* lsn is encoded in the file itself — we re-read it */
@ -2173,66 +2300,116 @@ static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
if (fd < 0)
return -errno;
struct stat st;
if (fstat(fd, &st) != 0) {
int saved = errno;
close(fd);
return -saved;
}
/* 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);
if (n < 16)
if (n < 16) {
close(fd);
return -EIO;
}
if (memcmp(header, "XCESNAP\0", 8) != 0)
if (memcmp(header, "XCESNAP\0", 8) != 0) {
close(fd);
return -EINVAL;
}
/* Extract checkpoint_lsn from bytes 8-15 (little-endian on this arch) */
/* Extract checkpoint_lsn from bytes 8-15 (little-endian on this arch).
* Held LOCALLY -- not yet written to p-> -- until the single commit
* point below (legacy short-circuit, or after every entity restores). */
uint64_t stored_lsn;
memcpy(&stored_lsn, header + 8, 8);
/* Optional committed-config reference (ha-term-safety task 4.2 r1):
* present only when the sentinel carries the 41-byte tail. Also held
* locally until the same commit point. */
int cfgref_present = 0;
uint64_t cfgref_tag = 0;
uint8_t cfgref_sha[32];
memset(cfgref_sha, 0, sizeof(cfgref_sha));
if (n >= 16 + 1 + 8 + 32 && header[16] != 0) {
cfgref_present = 1;
memcpy(&cfgref_tag, header + 17, 8);
memcpy(cfgref_sha, header + 25, 32);
}
off_t consumed = (off_t)n;
/* Legacy-format detection (Requirement 2.4, 5.1, 5.2): "does more data
* exist" is the discriminator (design.md), not a flag bit -- a file
* ending exactly at the legacy header is unambiguously a pre-feature
* checkpoint. This is an expected rollout shape, not corruption: commit
* the position it records (existing, preserved legacy behavior) and
* report entities_restored=false so the caller falls back to normal
* catch-up rather than treating this checkpoint as fully complete. */
if (st.st_size <= consumed) {
close(fd);
p->checkpoint_lsn = stored_lsn;
p->current_lsn = stored_lsn;
if (cfgref_present) {
p->cfgref_present = 1;
p->cfgref_tag = cfgref_tag;
memcpy(p->cfgref_sha, cfgref_sha, sizeof(p->cfgref_sha));
}
if (out_entities_restored)
*out_entities_restored = false;
return 0;
}
/* Entity-bearing checkpoint: format_version, then job frames, then queue
* frames (design.md "Checkpoint File Layout"). */
uint32_t fmt = 0;
if (read(fd, &fmt, sizeof(fmt)) != (ssize_t)sizeof(fmt)) {
close(fd);
return -EIO;
}
consumed += (off_t)sizeof(fmt);
/* An unrecognized format_version is corruption or a future layout this
* binary cannot read -- fail loud rather than silently treating it as
* legacy (task 5.1 review note: a version mismatch must never be masked
* as an expected legacy shape, or real corruption goes undetected). */
if (fmt != SNAPSHOT_FORMAT_VERSION) {
close(fd);
return -EINVAL;
}
int rc = load_snapshot_entities(fd, st.st_size, consumed, p, stored_lsn);
close(fd);
if (rc != 0)
return rc; /* Requirement 2.3: p->*_lsn / p->cfgref_* untouched */
/* Single commit point (Requirement 2.1, 2.2): reached ONLY after every
* job and queue frame restored successfully above. */
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) {
if (cfgref_present) {
p->cfgref_present = 1;
memcpy(&p->cfgref_tag, header + 17, 8);
memcpy(p->cfgref_sha, header + 25, 32);
p->cfgref_tag = cfgref_tag;
memcpy(p->cfgref_sha, cfgref_sha, sizeof(p->cfgref_sha));
}
if (out_entities_restored)
*out_entities_restored = true;
return 0;
}
/* Public wrapper for load_snapshot — exposed via persist.h for ha.c (Task 3.5c).
*
* ha-entity-snapshots task 5.1 (RED, signature-compat stopgap ONLY — task 5.2
* owns the real restore-before-advance implementation): `load_snapshot` above
* is UNCHANGED, still today's watermark-only behavior (never inspects/restores
* any entity payload). This wrapper's *out_entities_restored mapping is
* deliberately the most naive one possible — "the underlying call reported
* success" — precisely so it does NOT mask the two real gaps task 5.2 must
* close: (1) a legacy (position-only) file will report entities_restored=true
* here even though it never carried an entity, since load_snapshot() returns
* 0 for ANY well-formed header regardless of what (if anything) follows it —
* a live instance of the exact false-completeness hazard Requirement 2.4
* exists to prevent; (2) a corrupted entity frame is never even inspected —
* load_snapshot() returns 0 and already advances p->checkpoint_lsn/
* current_lsn before this wrapper is reached, so the "position unchanged on
* failure" contract (Requirement 2.3) does not hold yet either. Both are
* exactly the behaviors the task 5.1 RED tests assert against. */
/* Public wrapper for load_snapshot — exposed via persist.h for ha.c (Task
* 3.5c). load_snapshot() above now owns the full restore-before-advance
* implementation (ha-entity-snapshots task 5.2): this wrapper is a thin
* pass-through, since load_snapshot() already gates *out_entities_restored
* and the p->checkpoint_lsn/current_lsn commit on entity-restore success
* (Requirements 2.1-2.4, 5.1, 5.2). */
int persist_load_snapshot(persist_t *p, const char *path, uint64_t lsn,
bool *out_entities_restored)
{
int rc = load_snapshot(p, path, lsn);
/* Contract: only set on rc==0 (design.md "load_snapshot (extended)" —
* "*out_entities_restored is only set when rc==0"); left untouched
* otherwise so a failed call never overwrites a caller's own state. */
if (rc == 0 && out_entities_restored != NULL)
*out_entities_restored = true;
return rc;
return load_snapshot(p, path, lsn, out_entities_restored);
}
/* P2-2: Standby WAL replay — apply a single replicated entry to in-memory state.
@ -2612,7 +2789,10 @@ int persist_recover(persist_t* p)
/* --- Step 2: load snapshot (if any) --- */
if (snap_lsn > 0 && snap_file[0] != '\0') {
int rc = load_snapshot(p, snap_file, snap_lsn);
/* Local recovery doesn't need the entities-restored indicator: the
* WAL replay in Step 3 applies everything above checkpoint_lsn
* regardless, unchanged by ha-entity-snapshots. */
int rc = load_snapshot(p, snap_file, snap_lsn, NULL);
if (rc != 0) {
ERROR("persist_recover: failed to load snapshot %s: %s",
snap_file, strerror(-rc));