diff --git a/.kiro/specs/ha-entity-snapshots/tasks.md b/.kiro/specs/ha-entity-snapshots/tasks.md index 74a8c66..1f2c990 100644 --- a/.kiro/specs/ha-entity-snapshots/tasks.md +++ b/.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. diff --git a/src/controller/testsuite/test_persist_snapshot_entities.c b/src/controller/testsuite/test_persist_snapshot_entities.c index 6dacf61..fb4887f 100644 --- a/src/controller/testsuite/test_persist_snapshot_entities.c +++ b/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); }