forked from lei.chen/myxce
feat(ha-entity-snapshots): entity-bearing checkpoint write path for queues (task 3.2)
write_snapshot replaces the queue_count=0 placeholder with a real collect_queue_frames helper: enumerates every queue in the live table under nested table+per-entity RTTI_LOCK_READ (no retention filter -- table presence already implies config-provisioned existence, unlike jobs), writes each queue's (name, status) as a plain fixed record (no capnp -- matches restore_queue_state's exact contract). Any failure anywhere still discards the whole checkpoint, including any already-collected job frames, no partial file at the final path. Entirely contained to persist.c -- load_snapshot and persist.h remain byte-identical to before this task, honoring the boundary lesson task 2.2 established (queue restoration stays task 5.2's scope). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JDYFbrgyDHbvNYcxApKhrX
This commit is contained in:
parent
86b12b8323
commit
eac7bb8ad8
2 changed files with 146 additions and 11 deletions
|
|
@ -31,7 +31,7 @@
|
|||
- _Requirements: 1.1, 1.2, 1.3, 1.4_
|
||||
- _Boundary: Persistence Layer (write_snapshot)_
|
||||
|
||||
- [ ] 3. Core: entity-bearing checkpoint write path — queues
|
||||
- [x] 3. Core: entity-bearing checkpoint write path — queues
|
||||
- [x] 3.1 RED: queue admin-status round-trip test
|
||||
- Assert a queue's (name, status) pair written into the checkpoint and read back matches
|
||||
exactly
|
||||
|
|
@ -39,7 +39,7 @@
|
|||
- _Requirements: 1.1_
|
||||
- _Boundary: Persistence Layer (write_snapshot)_
|
||||
- _Depends: 2.2_
|
||||
- [ ] 3.2 Collect queue admin-status into the checkpoint
|
||||
- [x] 3.2 Collect queue admin-status into the checkpoint
|
||||
- Enumerate every queue present in the live table (its table presence already implies
|
||||
config-provisioned existence), take each queue's own read lock nested inside the table walk,
|
||||
append its name+status as a plain record (no serialization schema needed — matches the
|
||||
|
|
@ -130,3 +130,4 @@
|
|||
- (2.2) `restore_job` (job.c) decodes via `decode_Job` → `decode_JobSubmitData_ptr`/`decode_JobStartInfo_ptr` (STRING_DUP'd fields) into a local `xce_job_t j`; that decoded struct is never released via `free_Job(&j)` after `create_job` takes its own copy — a real, pre-existing leak on `restore_job`'s decode path. Not touched by task 2.2 (its `load_snapshot` excursion was fully reverted, see next note) — task 5.2 will reintroduce this call site and must fix the leak there.
|
||||
- (2.2) First attempt ALSO touched `load_snapshot`/`persist.h` to make task 2.1's round-trip tests pass end-to-end, but review found it advanced `current_lsn`/`checkpoint_lsn` BEFORE the entity-decode loop ran -- a live silent-divergence hazard reachable through the unmodified `ha_handle_snapshot_chunk` production path (reachable today, not hypothetical, since `write_snapshot` now always emits entity-bearing checkpoints). REMEDIATED by fully reverting `load_snapshot`/`persist.h` to pre-2.2 state (byte-identical to commit `0f344da`) and rewriting the round-trip tests to parse `write_snapshot`'s raw on-disk bytes directly (new `snapent_read_checkpoint_jobs` helper using `read_Job`/`decode_Job`) instead of round-tripping through `persist_load_snapshot`. Task 5.2 must implement `load_snapshot`'s entity-decode from scratch per design.md's restore-before-advance-position contract -- do not assume any decode scaffolding exists yet.
|
||||
- (3.1) `snapent_read_checkpoint_queues`'s job-section skip loop duplicates `snapent_read_checkpoint_jobs`'s (task 2.2) skip logic verbatim rather than sharing a helper -- minor, non-blocking; consider factoring a shared "skip past job section" reader if task 5.x's test coverage grows the file further.
|
||||
- (3.2) Pre-existing ASan leak in the test fixture itself (`snapent_make` -> `makeInstance`/`makeJob`/`makeQueue`/`xkvmgr_add`, ~18KB/113 allocations), unrelated to any production code -- present since task 1.1, masked by the project's standing `ASAN_OPTIONS=detect_leaks=0` convention for this suite. Not blocking any task; worth a dedicated fixture-teardown cleanup pass if this test file keeps growing.
|
||||
|
|
|
|||
|
|
@ -1479,15 +1479,14 @@ static int persist_write_checkpoint_marker(persist_t* p)
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* Entity-bearing checkpoint payload (ha-entity-snapshots task 2.2, Req
|
||||
/* Entity-bearing checkpoint payload (ha-entity-snapshots task 2.2/3.2, Req
|
||||
* 1.1-1.4). Physical layout appended after the existing legacy header
|
||||
* (design.md "Checkpoint File Layout"):
|
||||
* [uint32_t format_version]
|
||||
* [uint32_t job_count]
|
||||
* [job_count * { uint32_t payload_len; uint8_t capnp_Job_bytes[payload_len] }]
|
||||
* [uint32_t queue_count]
|
||||
* [queue_count * { ... }] -- written as 0 for now; task 3.2 fills this in
|
||||
* without needing to touch the job path above. */
|
||||
* [queue_count * { uint32_t name_len; char name[name_len]; uint32_t status }] */
|
||||
#define SNAPSHOT_FORMAT_VERSION 1u
|
||||
|
||||
typedef struct {
|
||||
|
|
@ -1609,6 +1608,116 @@ 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 }`). */
|
||||
typedef struct {
|
||||
char* name;
|
||||
uint32_t status;
|
||||
} snap_queue_frame_t;
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
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_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.
|
||||
*
|
||||
* Unlike collect_job_frames, no retention predicate is applied: a queue's
|
||||
* mere presence in the live qmgr table already means config provisioned
|
||||
* it (design.md "write_snapshot (extended)" Responsibilities), so every
|
||||
* queue found is included.
|
||||
*
|
||||
* 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). */
|
||||
static int collect_queue_frames(persist_t* p, xlist_t* frames)
|
||||
{
|
||||
cluster_t* c = p->cluster;
|
||||
int rc = 0;
|
||||
|
||||
xlist_init(frames, 64);
|
||||
|
||||
/* No live cluster back-pointer (e.g. a bare unit-test persist_t) --
|
||||
* zero queues is not a failure. */
|
||||
if (c == NULL || c->qmgr == NULL)
|
||||
return 0;
|
||||
|
||||
lock(c->qmgr, RTTI_LOCK_READ);
|
||||
|
||||
iterator_t it;
|
||||
int irc = iter_make(&it, c->qmgr);
|
||||
if (irc < 0) {
|
||||
unlock(c->qmgr);
|
||||
return 0; /* nothing to walk -- mirrors collect_queues' empty-table handling */
|
||||
}
|
||||
|
||||
for (;;) {
|
||||
queue_t* q = NULL;
|
||||
int nrc = iter_next(&it, (void**)&q);
|
||||
if (nrc < 0 || nrc == RTTI_ITER_NO_DATA)
|
||||
break;
|
||||
if (q == NULL)
|
||||
continue;
|
||||
|
||||
char* name;
|
||||
uint32_t status;
|
||||
|
||||
lock(q, RTTI_LOCK_READ);
|
||||
name = q->queue.queue ? strdup(q->queue.queue) : NULL;
|
||||
status = q->queue.status;
|
||||
unlock(q);
|
||||
|
||||
if (name == NULL) {
|
||||
rc = -ENOMEM;
|
||||
break;
|
||||
}
|
||||
|
||||
snap_queue_frame_t* fr = (snap_queue_frame_t*)malloc(sizeof(*fr));
|
||||
if (fr == NULL) {
|
||||
free(name);
|
||||
rc = -ENOMEM;
|
||||
break;
|
||||
}
|
||||
fr->name = name;
|
||||
fr->status = status;
|
||||
if (xlist_store(frames, fr) < 0) {
|
||||
free(name);
|
||||
free(fr);
|
||||
rc = -ENOMEM;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
iter_destroy(&it);
|
||||
unlock(c->qmgr);
|
||||
|
||||
if (rc != 0)
|
||||
snap_free_queue_frames(frames); /* Requirement 1.3: discard, not partial */
|
||||
|
||||
return rc;
|
||||
}
|
||||
|
||||
/* Write an entity-bearing checkpoint: legacy header (unchanged) + entity
|
||||
* payload (ha-entity-snapshots task 2.2/3.2, Req 1.1-1.4, 3.1-3.3). Any
|
||||
* failure at any step -- collect, encode, write, sync -- leaves NO file at
|
||||
|
|
@ -1632,9 +1741,17 @@ static int write_snapshot(persist_t* p)
|
|||
if (rc != 0)
|
||||
return rc;
|
||||
|
||||
xlist_t queue_frames;
|
||||
rc = collect_queue_frames(p, &queue_frames);
|
||||
if (rc != 0) {
|
||||
snap_free_job_frames(&job_frames);
|
||||
return rc;
|
||||
}
|
||||
|
||||
int fd = open(tmp_path, O_CREAT | O_WRONLY | O_TRUNC, 0600);
|
||||
if (fd < 0) {
|
||||
snap_free_job_frames(&job_frames);
|
||||
snap_free_queue_frames(&queue_frames);
|
||||
return -errno;
|
||||
}
|
||||
|
||||
|
|
@ -1681,22 +1798,39 @@ static int write_snapshot(persist_t* p)
|
|||
}
|
||||
}
|
||||
|
||||
/* Queue trailer: task 3.2 populates this with the real collected count
|
||||
* and frames without needing to touch anything above. Writing an
|
||||
* explicit 0 now (rather than omitting the section) keeps the on-disk
|
||||
* layout complete per design.md, so a reader can always expect a
|
||||
* queue_count word after the job frames. */
|
||||
/* 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"). */
|
||||
if (rc == 0) {
|
||||
uint32_t queue_count = 0;
|
||||
uint32_t queue_count = (uint32_t)queue_frames.count;
|
||||
if (write(fd, &queue_count, sizeof(queue_count)) != (ssize_t)sizeof(queue_count))
|
||||
rc = -EIO;
|
||||
}
|
||||
|
||||
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)) {
|
||||
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)) {
|
||||
rc = -EIO;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (rc == 0 && fdatasync(fd) != 0)
|
||||
rc = -errno;
|
||||
|
||||
close(fd);
|
||||
snap_free_job_frames(&job_frames);
|
||||
snap_free_queue_frames(&queue_frames);
|
||||
|
||||
if (rc != 0) {
|
||||
unlink(tmp_path);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue