test(ha-entity-snapshots): RED test for queue status checkpoint content (task 3.1)

Queue's (name, status) round-trip through write_snapshot's raw on-disk
bytes -- a plain fixed record (no capnp; queue definitions are
config-owned, only runtime admin status is WAL/snapshot-tracked, per
design.md's corrected queue model). Fails today (Expected 4 Was 0) since
write_snapshot still hardcodes queue_count=0 from task 2.2's placeholder.
Does not call persist_load_snapshot, following task 2.2's established
raw-byte-parsing pattern.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JDYFbrgyDHbvNYcxApKhrX

.kiro/specs/ha-entity-snapshots/tasks.md

@@ -32,7 +32,7 @@
   - _Boundary: Persistence Layer (write_snapshot)_
 
 - [ ] 3. Core: entity-bearing checkpoint write path — queues
-- [ ] 3.1 RED: queue admin-status round-trip test
+- [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
   - Observable: new test compiles and fails (RED)
@@ -129,3 +129,4 @@
 - (2.1) The write-failure-discard test (`test_snapent_checkpoint_write_failure_leaves_no_final_file`) only exercises the pre-existing `open()`-failure branch (via directory removal) and passes today without any entity-encode loop existing yet. Task 2.2 should add a genuine mid-encode-loop fault-injection test (failing to encode e.g. the Nth of N jobs) once the per-job encode loop exists — the current test cannot validate that path.
 - (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.

src/controller/testsuite/test_persist_snapshot_entities.c

@@ -753,6 +753,194 @@ static void test_snapent_checkpoint_mid_encode_failure_discards_whole_file(void)
     printf("test_snapent_checkpoint_mid_encode_failure_discards_whole_file: PASS\n");
 }
 
+/* ------------------------------------------------------------------ */
+/* Test-local raw checkpoint reader for the queue section (task 3.1).   */
+/*                                                                       */
+/* This task's boundary is write_snapshot ONLY -- load_snapshot's entity */
+/* restore is task 5.x's job (see design.md Requirement 2), mirroring    */
+/* the job-section reader above (task 2.2 remediation note). Reads the   */
+/* checkpoint's on-disk bytes directly per design.md's exact "Checkpoint */
+/* File Layout (Physical)":                                              */
+/*   [57-byte legacy sentinel][uint32_t format_version]                  */
+/*   [uint32_t job_count][job_count * {payload_len; capnp Job bytes}]    */
+/*   [uint32_t queue_count]                                              */
+/*   [queue_count * {uint32_t name_len; char name[name_len]; status}]    */
+/* Skips past the job section (whatever it contains) to reach the queue  */
+/* section, so this reader works regardless of how many jobs a fixture   */
+/* also seeded. */
+/* ------------------------------------------------------------------ */
+
+#define SNAPENT_MAX_QNAME_LEN 63
+
+typedef struct {
+    char     name[SNAPENT_MAX_QNAME_LEN + 1];
+    uint32_t status;
+} snapent_decoded_queue_t;
+
+/* Reads every queue frame in the checkpoint at `path` into `out`
+ * (caller-allocated array of `max_out` slots). Returns the number of
+ * frames decoded (<= queue_count on disk, capped at max_out), or -1 on
+ * any structural read error (bad magic, bad/absent format_version, short
+ * read, oversized job payload_len, oversized queue name_len). */
+static int snapent_read_checkpoint_queues(const char* path,
+                                          snapent_decoded_queue_t* out,
+                                          int max_out)
+{
+    int fd = open(path, O_RDONLY);
+    if (fd < 0)
+        return -1;
+
+    char header[SNAPENT_LEGACY_HEADER_BYTES];
+    if (read(fd, header, sizeof(header)) != (ssize_t)sizeof(header) ||
+        memcmp(header, "XCESNAP\0", 8) != 0) {
+        close(fd);
+        return -1;
+    }
+
+    uint32_t format_version = 0;
+    if (read(fd, &format_version, sizeof(format_version)) !=
+            (ssize_t)sizeof(format_version) ||
+        format_version != SNAPENT_TEST_FORMAT_VERSION) {
+        close(fd);
+        return -1;
+    }
+
+    /* Skip the entire job section -- this reader only cares about what
+     * follows it, but the on-disk layout is sequential so every job frame
+     * must be walked past (by its own length prefix) to reach the queue
+     * section. */
+    uint32_t job_count = 0;
+    if (read(fd, &job_count, sizeof(job_count)) != (ssize_t)sizeof(job_count)) {
+        close(fd);
+        return -1;
+    }
+    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)) {
+            close(fd);
+            return -1;
+        }
+        if (payload_len == 0 || payload_len > SNAPENT_MAX_DECODE_PAYLOAD) {
+            close(fd);
+            return -1;
+        }
+        uint8_t* discard = (uint8_t*)malloc(payload_len);
+        if (discard == NULL) {
+            close(fd);
+            return -1;
+        }
+        ssize_t got = read(fd, discard, payload_len);
+        free(discard);
+        if (got != (ssize_t)payload_len) {
+            close(fd);
+            return -1;
+        }
+    }
+
+    uint32_t queue_count = 0;
+    if (read(fd, &queue_count, sizeof(queue_count)) !=
+            (ssize_t)sizeof(queue_count)) {
+        close(fd);
+        return -1;
+    }
+
+    int n = 0;
+    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)) {
+            close(fd);
+            return -1;
+        }
+        if (name_len == 0 || name_len > SNAPENT_MAX_QNAME_LEN) {
+            close(fd);
+            return -1;
+        }
+
+        char namebuf[SNAPENT_MAX_QNAME_LEN + 1];
+        if (read(fd, namebuf, name_len) != (ssize_t)name_len) {
+            close(fd);
+            return -1;
+        }
+        namebuf[name_len] = '\0';
+
+        uint32_t status = 0;
+        if (read(fd, &status, sizeof(status)) != (ssize_t)sizeof(status)) {
+            close(fd);
+            return -1;
+        }
+
+        if (n < max_out) {
+            snprintf(out[n].name, sizeof(out[n].name), "%s", namebuf);
+            out[n].status = status;
+            n++;
+        }
+    }
+
+    close(fd);
+    return n;
+}
+
+static snapent_decoded_queue_t* snapent_find_decoded_queue(
+    snapent_decoded_queue_t* queues, int count, const char* name)
+{
+    for (int i = 0; i < count; i++) {
+        if (strcmp(queues[i].name, name) == 0)
+            return &queues[i];
+    }
+    return NULL;
+}
+
+/* ------------------------------------------------------------------ */
+/* Test 6 (task 3.1, RED): a queue's (name, status) pair written into    */
+/* the checkpoint by write_snapshot is read back byte-identical          */
+/* (Requirement 1.1).                                                    */
+/*                                                                       */
+/* write_snapshot currently hardcodes queue_count=0 (persist.c's task    */
+/* 3.2 placeholder -- see design.md "write_snapshot (extended)" and the   */
+/* Implementation Notes at the bottom of tasks.md) -- so this MUST fail  */
+/* today: the checkpoint's queue section is always empty regardless of  */
+/* how many queues are actually seeded in the fixture's qmgr. Task 3.2   */
+/* makes this pass by collecting the live queue table into that section.*/
+/* ------------------------------------------------------------------ */
+
+static void test_snapent_checkpoint_queue_status_roundtrip(void)
+{
+    const int m_queues = 4;
+
+    snapent_fixture_t* src = snapent_make(0, m_queues);
+    TEST_ASSERT_NOT_NULL(src);
+
+    TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&src->c->persist));
+
+    char path[4096];
+    snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp",
+             src->c->persist.snap_path, src->c->persist.checkpoint_lsn);
+
+    snapent_decoded_queue_t decoded[8];
+    int count = snapent_read_checkpoint_queues(path, decoded, 8);
+    TEST_ASSERT_TRUE_MESSAGE(count >= 0,
+        "checkpoint file must parse as a well-formed entity-bearing "
+        "format_version=1 payload through the queue section");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(m_queues, count,
+        "checkpoint must contain exactly the seeded queues -- write_snapshot "
+        "currently hardcodes queue_count=0 (task 3.2 not yet implemented)");
+
+    for (int i = 0; i < m_queues; i++) {
+        char expect_name[32];
+        snapent_queue_name(i, expect_name, sizeof(expect_name));
+
+        snapent_decoded_queue_t* dq =
+            snapent_find_decoded_queue(decoded, count, expect_name);
+        TEST_ASSERT_NOT_NULL_MESSAGE(dq,
+            "checkpoint's raw queue frames must contain the seeded queue");
+        TEST_ASSERT_EQUAL_UINT32(snapent_queue_status(i), dq->status);
+    }
+
+    snapent_teardown(src);
+    printf("test_snapent_checkpoint_queue_status_roundtrip: PASS\n");
+}
+
 /* ------------------------------------------------------------------ */
 /* Suite entry point                                                    */
 /* ------------------------------------------------------------------ */
@@ -777,4 +965,5 @@ void test_persist_snapshot_entities(void)
     RUN_TEST(test_snapent_checkpoint_concurrent_mutation_not_torn);
     RUN_TEST(test_snapent_checkpoint_write_failure_leaves_no_final_file);
     RUN_TEST(test_snapent_checkpoint_mid_encode_failure_discards_whole_file);
+    RUN_TEST(test_snapent_checkpoint_queue_status_roundtrip);
 }