diff --git a/.kiro/specs/ha-entity-snapshots/tasks.md b/.kiro/specs/ha-entity-snapshots/tasks.md index 1f2c990a..48335eaf 100644 --- a/.kiro/specs/ha-entity-snapshots/tasks.md +++ b/.kiro/specs/ha-entity-snapshots/tasks.md @@ -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. diff --git a/src/controller/persist.c b/src/controller/persist.c index 8c222317..1247fdd7 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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);