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

feat(config-upgrade-snapshot): decode host/queue admin-state frames on checkpoint load

Browse source 
load_snapshot_entities' queue-frame decode migrates from the old raw
{name_len,name,status} layout to {payload_len, capnp QueueAdminState
bytes}, still calling the unmodified restore_queue_state. Adds a new
host-frame section decode after queues, dispatching to
restore_host_state (worker.c, task 3) with the same bounds-checking
and restore-before-advance discipline as the existing job/queue loops.

Un-ignores test_snapent_load_snapshot_restores_entities_before_advancing_position
(blocked since task 2.2) -- now passes for real. Also fixes a genuine
regression in test_snapshot_recv_out_of_order_chunks_assemble_correctly
(test_config_repl.c): its synthetic checkpoint-body filler landed on
the new host_count field's offset, tripping the cluster==NULL guard --
pinned to a well-formed host_count=0, matching that test's existing
fmt/job_count/queue_count precedent. Task 4 of 19.

Requirements: 1.10
This commit is contained in:
Rongsong Shen 2026-07-24 14:08:12 +00:00
parent 5b1600a8b2
commit 00cd26a927
4 changed files with 347 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

176
src/controller/persist.c
View file

@ -37,6 +37,11 @@ extern int worker_from_capnp(cluster_t* c,
const void* capnp_payload, size_t len);
extern int restore_queue_state(cluster_t* c,
const char* name, uint32_t status);
/* restore_host_state defined in worker.c (config-upgrade-snapshot task 3,
* Req 1.10/1.11/5.8) -- create-if-missing host admin-state restore, called
* from load_snapshot_entities' host-frame loop below (task 4). */
extern int restore_host_state(cluster_t* c, const char* name,
uint32_t admin_status);
/* raft-apply-decoupling task 3.2: pending command-apply hand-off list node
* (producer side; the not-yet-built ordered apply pump, task 6.1/6.2, is
@ -2742,10 +2747,17 @@ done:
/* Recovery internals */
/* ------------------------------------------------------------------ */
/* 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}].
/* Parse and restore every job frame, then every queue frame, then every host
* frame from the entity payload beginning at the current `fd` offset
* (design.md "Checkpoint File Layout", SNAPSHOT_FORMAT_VERSION=2):
* [job_count][job_count * {payload_len; capnp Job bytes}]
* [queue_count][queue_count * {payload_len; capnp QueueAdminState bytes}]
* [host_count][host_count * {payload_len; capnp HostAdminState bytes}]
* A trailing config-bytes section follows the host section in the file
* (config-upgrade-snapshot task 2.3) but is deliberately NOT read here --
* out of this function's scope; any bytes after the host section are simply
* left unread, exactly as host-frame bytes were already tolerated as
* present-but-unread before this function grew a host section of its own.
*
* `file_size` is the checkpoint file's total size (from fstat) and
* `consumed` is the number of bytes already read from it (legacy header +
@ -2755,8 +2767,13 @@ done:
* 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
* restore_queue_state, then every host via restore_host_state
* (config-upgrade-snapshot task 4, Req 1.10 — design.md sequence diagram
* and "Referential integrity" note: restore order does not matter
* functionally, since re-homing decisions are already baked into each
* entity's encoded content at materialization time, never re-derived at
* restore time, Design Invariant 2). 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).
@ -2765,9 +2782,12 @@ done:
* 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. */
* rather than aborting on them again. restore_queue_state and
* restore_host_state have no such idempotent-duplicate case of their own to
* special-case here (re-applying the same status is already a harmless
* overwrite/OR, and restore_host_state's create-if-missing placeholder path
* is independently idempotent under retry — Requirement 5.8, worker.c), so
* any non-zero return from either 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)
{
@ -2814,45 +2834,123 @@ static int load_snapshot_entities(int fd, off_t file_size, off_t consumed,
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))
uint32_t payload_len = 0;
if (read(fd, &payload_len, sizeof(payload_len)) !=
(ssize_t)sizeof(payload_len))
return -EIO;
consumed += (off_t)sizeof(name_len);
consumed += (off_t)sizeof(payload_len);
if (name_len == 0 || (off_t)name_len > file_size - consumed)
if (payload_len == 0 || (off_t)payload_len > file_size - consumed)
return -EIO;
char* name = (char*)malloc((size_t)name_len + 1);
if (name == NULL)
uint8_t* buf = (uint8_t*)malloc(payload_len);
if (buf == NULL)
return -ENOMEM;
if (read(fd, name, name_len) != (ssize_t)name_len) {
free(name);
if (read(fd, buf, payload_len) != (ssize_t)payload_len) {
free(buf);
return -EIO;
}
name[name_len] = '\0';
consumed += (off_t)name_len;
consumed += (off_t)payload_len;
uint32_t status = 0;
if (read(fd, &status, sizeof(status)) != (ssize_t)sizeof(status)) {
free(name);
return -EIO;
}
consumed += (off_t)sizeof(status);
/* Decode the capnp QueueAdminState frame (config-upgrade-snapshot
* task 4, Req 1.9/5.7 decode side — migrated off the prior raw
* {name_len, name, status} layout collect_queue_frames stopped
* writing in task 2.2). Mirrors the WAL_JOB_SUBMIT re-decode block
* above in this file (capn_init_mem/capn_getp/read_*) and
* test_persist_snapshot_entities.c's own
* snapent_read_checkpoint_queues reader exactly. */
struct capn c_ctx;
capn_init_mem(&c_ctx, buf, payload_len, 0);
QueueAdminState_ptr qp;
qp.p = capn_getp(capn_root(&c_ctx), 0, 1);
struct QueueAdminState cq;
memset(&cq, 0, sizeof(cq));
read_QueueAdminState(&cq, qp);
int qrc = restore_queue_state(p->cluster, name, status);
free(name);
capnp_queue_admin_state_t decoded;
memset(&decoded, 0, sizeof(decoded));
decode_QueueAdminState(&decoded, &cq);
capn_free(&c_ctx);
free(buf);
int qrc = restore_queue_state(p->cluster, decoded.name, decoded.status);
free_QueueAdminState(&decoded);
if (qrc != 0)
return qrc;
}
/* Host section (config-upgrade-snapshot task 4, Req 1.10): new in
* SNAPSHOT_FORMAT_VERSION=2, added after the queue section (design.md
* "Checkpoint File Layout"). Same [payload_len][capnp bytes] per-frame
* shape as the job/queue sections above. Every host frame dispatches
* to restore_host_state (worker.c, task 3), which is create-if-missing
* (a worker_t is join-provisioned, not config-provisioned like a queue
* — Design Invariant 1) and idempotent under retry on its own
* (Requirement 5.8), so this loop applies the exact same "any failure
* returns immediately" discipline as the job/queue loops above without
* needing any extra idempotency handling here. */
uint32_t host_count = 0;
if (read(fd, &host_count, sizeof(host_count)) !=
(ssize_t)sizeof(host_count))
return -EIO;
consumed += (off_t)sizeof(host_count);
if (host_count > 0 &&
(p->cluster == NULL || p->cluster->workermgr == NULL))
return -EINVAL; /* entities present but nowhere to restore them into */
for (uint32_t i = 0; i < host_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;
/* Decode the capnp HostAdminState frame -- mirrors the queue decode
* block immediately above and
* test_persist_snapshot_entities.c's snapent_read_checkpoint_hosts
* reader exactly. */
struct capn c_ctx;
capn_init_mem(&c_ctx, buf, payload_len, 0);
HostAdminState_ptr hp;
hp.p = capn_getp(capn_root(&c_ctx), 0, 1);
struct HostAdminState ch;
memset(&ch, 0, sizeof(ch));
read_HostAdminState(&ch, hp);
capnp_host_admin_state_t decoded;
memset(&decoded, 0, sizeof(decoded));
decode_HostAdminState(&decoded, &ch);
capn_free(&c_ctx);
free(buf);
int hrc = restore_host_state(p->cluster, decoded.name, decoded.status);
free_HostAdminState(&decoded);
if (hrc != 0)
return hrc;
}
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.
* config-ref tail), then — for an entity-bearing checkpoint — every job,
* queue, and host frame that follows it (ha-entity-snapshots task 5.2,
* config-upgrade-snapshot task 4; Requirements 2.1-2.4, 5.1, 5.2, 1.10).
* 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
@ -2861,16 +2959,18 @@ static int load_snapshot_entities(int fd, off_t file_size, off_t consumed,
* 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.
* job, then every queue, then every host 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. */
* open/fstat failure, or a restore_job/restore_queue_state/restore_host_state
* error code. */
static int load_snapshot(persist_t* p, const char* path, uint64_t lsn,
bool* out_entities_restored)
{