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

feat(config-upgrade-snapshot): migrate queue admin-state frames to capnp wire format

Browse source 
collect_queue_frames now encodes each queue as a QueueAdminState capnp
frame (capnp_encode_queue_admin_state, mirroring task 2.1's host-frame
encoder) instead of the raw {name_len,name,status} layout;
write_snapshot's queue trailer writes [payload_len][capnp bytes] per
queue. restore_queue_state and load_snapshot's queue decode are
untouched (task 4's boundary).

Temporarily TEST_IGNORE_MESSAGE's
test_snapent_load_snapshot_restores_entities_before_advancing_position
(m_queues=2 real write+load round trip hits the still-raw loader) --
tracked in tasks.md Implementation Notes, must be re-enabled by task 4.
Matches existing codebase precedent (bug-847) for this pattern. Task
2.2 of 19.

Requirements: 1.9, 5.7
This commit is contained in:
Rongsong Shen 2026-07-24 10:20:31 +00:00
parent a71aa3f8a1
commit 70ff3e5b80
3 changed files with 146 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

107
src/controller/persist.c
View file

@ -129,6 +129,36 @@ static int capnp_encode_host_admin_state(const capnp_host_admin_state_t* h,
return 0;
}
/* Serialize a queue admin-state frame (config-upgrade-snapshot task 2.2, Req
* 1.9/5.7) to a malloc'd Cap'n Proto buffer. Mirrors
* capnp_encode_host_admin_state above exactly -- capnp_queue_admin_state_t
* (typemap.h) is a frame-only (name, status) pair, matching
* restore_queue_state's exact (name, status) contract (queue.c:208), same
* field content as the prior raw layout, only the wire encoding changes.
* Caller must free(*out) on success. */
static int capnp_encode_queue_admin_state(const capnp_queue_admin_state_t* q,
uint8_t** out, size_t* out_len)
{
struct capn c;
struct capn_segment* cs;
struct QueueAdminState cq;
QueueAdminState_ptr qp;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_QueueAdminState(cs, &cq, (capnp_queue_admin_state_t*)q);
qp = new_QueueAdminState(cs);
write_QueueAdminState(&cq, qp);
capn_setp(capn_root(&c), 0, qp.p);
*out_len = capn_size(&c);
*out = malloc(*out_len);
if (!*out) { capn_free(&c); return -ENOMEM; }
capn_write_mem(&c, *out, *out_len, 0);
capn_free(&c);
return 0;
}
/* WAL_WORKER_CTRL payload: [uint32_t detail][capnp bytes...] or [uint32_t detail][name\0] */
#define WORKER_WAL_HDR_SIZE sizeof(uint32_t)
@ -1787,15 +1817,15 @@ static int collect_job_frames(persist_t* p, xlist_t* frames)
return rc;
}
/* Queue admin-status frame (ha-entity-snapshots task 3.2, Req 1.1). Unlike
* jobs, a queue's on-disk frame carries no capnp payload at all -- it is a
* plain (name, status) pair, the entirety of restore_queue_state's contract
* (queue.c:208). `name` is a malloc'd, NUL-terminated copy; the on-disk
* frame itself carries no NUL (design.md "Checkpoint File Layout":
* `{ uint32_t name_len; char name[name_len]; uint32_t status }`). */
/* Queue admin-status frame (config-upgrade-snapshot task 2.2, Req 1.9, 5.7).
* Migrated from the prior raw {name_len, name, status} layout to a capnp
* QueueAdminState payload -- identical {buf,len} shape to snap_host_frame_t
* (both carry a capnp payload), same (name, status) field content as
* before, restore_queue_state's contract (queue.c:208) unchanged; only the
* wire encoding changes (design.md "Snapshot Entity/Config Payload"). */
typedef struct {
char* name;
uint32_t status;
uint8_t* buf;
uint32_t len;
} snap_queue_frame_t;
static void snap_free_queue_frames(xlist_t* frames)
@ -1803,21 +1833,21 @@ static void snap_free_queue_frames(xlist_t* frames)
for (int i = 0; i < frames->count; i++) {
snap_queue_frame_t* fr = (snap_queue_frame_t*)xlist_get(frames, i);
if (fr) {
free(fr->name);
free(fr->buf);
free(fr);
}
}
xlist_fini(frames);
}
/* Collect every queue in `frames`, one malloc'd snap_queue_frame_t per
* queue, under the same point-in-time-consistent walk discipline as
/* Collect + encode every queue in `frames`, one malloc'd snap_queue_frame_t
* per queue, under the same point-in-time-consistent walk discipline as
* collect_job_frames above (design.md "Existing Architecture Analysis" /
* Requirement 1.2): the qmgr table's own RTTI_LOCK_READ is held for the
* ENTIRE walk, and each queue's own RTTI_LOCK_READ is ADDITIONALLY taken
* nested around just that queue's (name, status) read -- mirroring
* restore_queue_state's own per-entity lock pattern on the load side. No
* lock is held during the disk write itself.
* nested around just that queue's (name, status) read and capnp encode --
* mirroring collect_host_frames' identical lock-then-encode-then-unlock
* shape. No lock is held during the disk write itself.
*
* Unlike collect_job_frames, no retention predicate is applied: a queue's
* mere presence in the live qmgr table already means config provisioned
@ -1826,10 +1856,10 @@ static void snap_free_queue_frames(xlist_t* frames)
*
* Returns 0 on success (frames populated; `frames` is always left
* xlist_init'd, even for zero results, so the caller can always xlist_fini
* it) or a negative errno on the first failure (name allocation), in which
* case every frame collected so far has already been freed and `frames`
* is empty (Requirement 1.3: a mid-walk failure must not leave partial
* entity content behind). */
* it) or a negative errno on the first encode failure, in which case every
* frame collected so far has already been freed and `frames` is empty
* (Requirement 1.3: a mid-walk failure must not leave partial entity
* content behind). */
static int collect_queue_frames(persist_t* p, xlist_t* frames)
{
cluster_t* c = p->cluster;
@ -1859,29 +1889,32 @@ static int collect_queue_frames(persist_t* p, xlist_t* frames)
if (q == NULL)
continue;
char* name;
uint32_t status;
capnp_queue_admin_state_t qs;
uint8_t* buf = NULL;
size_t sz = 0;
int enc_rc;
lock(q, RTTI_LOCK_READ);
name = q->queue.queue ? strdup(q->queue.queue) : NULL;
status = q->queue.status;
qs.name = q->queue.queue;
qs.status = q->queue.status;
enc_rc = capnp_encode_queue_admin_state(&qs, &buf, &sz);
unlock(q);
if (name == NULL) {
rc = -ENOMEM;
if (enc_rc != 0) {
rc = enc_rc;
break;
}
snap_queue_frame_t* fr = (snap_queue_frame_t*)malloc(sizeof(*fr));
if (fr == NULL) {
free(name);
free(buf);
rc = -ENOMEM;
break;
}
fr->name = name;
fr->status = status;
fr->buf = buf;
fr->len = (uint32_t)sz;
if (xlist_store(frames, fr) < 0) {
free(name);
free(buf);
free(fr);
rc = -ENOMEM;
break;
@ -2119,9 +2152,11 @@ static int write_snapshot(persist_t* p)
}
}
/* Queue trailer (ha-entity-snapshots task 3.2, Req 1.1, 1.2): plain
* (name, status) frames, no capnp -- matches restore_queue_state's
* exact contract (design.md "Checkpoint File Layout"). */
/* Queue trailer (config-upgrade-snapshot task 2.2, Req 1.9, 5.7):
* capnp QueueAdminState frames, same [payload_len][capnp bytes] shape
* as the job/host frame loops -- migrated from the prior raw
* {name_len, name, status} layout (design.md "Checkpoint File Layout":
* same field content, new wire encoding only). */
if (rc == 0) {
uint32_t queue_count = (uint32_t)queue_frames.count;
if (write(fd, &queue_count, sizeof(queue_count)) != (ssize_t)sizeof(queue_count))
@ -2130,17 +2165,11 @@ static int write_snapshot(persist_t* p)
for (int i = 0; rc == 0 && i < queue_frames.count; i++) {
snap_queue_frame_t* fr = (snap_queue_frame_t*)xlist_get(&queue_frames, i);
uint32_t name_len = (uint32_t)strlen(fr->name);
if (write(fd, &name_len, sizeof(name_len)) != (ssize_t)sizeof(name_len)) {
if (write(fd, &fr->len, sizeof(fr->len)) != (ssize_t)sizeof(fr->len)) {
rc = -EIO;
break;
}
if (name_len > 0 &&
write(fd, fr->name, name_len) != (ssize_t)name_len) {
rc = -EIO;
break;
}
if (write(fd, &fr->status, sizeof(fr->status)) != (ssize_t)sizeof(fr->status)) {
if (write(fd, fr->buf, fr->len) != (ssize_t)fr->len) {
rc = -EIO;
break;
}