feat(ha-entity-snapshots): consume restoration indicator in the catch-up receive handler (task 6.1)

ha_handle_snapshot_chunk now logs whether a received checkpoint was
fully entity-restored or was a legacy position-only file (INFO level,
framed as an expected rollout state -- never an error, per design.md).
The rc-gates-applied_lsn control flow is byte-for-byte unchanged from
task 5.2 (confirmed by review) -- this task is purely additive
observability, matching design.md's own note that this call site
"needs no change" once persist_load_snapshot's gate is correct.

New test drives the legacy path end-to-end through the real production
handler (rc==0, file renamed, applied_lsn advances to the legacy LSN).

This completes the integration layer -- write path (jobs+queues),
restore-before-advance read path, and the production wiring are all in
place and independently reviewed. Only validation tasks remain (7.1
chaotic soak, 8.1 rollback compatibility).

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

@@ -92,8 +92,8 @@
   - _Requirements: 2.1, 2.2, 2.3, 2.4, 5.1, 5.2_
   - _Boundary: Persistence Layer (load_snapshot)_
 
-- [ ] 6. Integration: wire the catch-up receive path to the new restore contract
-- [ ] 6.1 Consume the new restoration indicator in the snapshot-chunk receive handler
+- [x] 6. Integration: wire the catch-up receive path to the new restore contract
+- [x] 6.1 Consume the new restoration indicator in the snapshot-chunk receive handler
   - The handler advances the member's applied position only after the load succeeds (unchanged
     ordering); on a legacy (position-only) file it logs the fallback rather than treating it as
     a failure

src/controller/ha.c

@@ -2594,9 +2594,10 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
     }
 
     /* Load the snapshot — sets p->checkpoint_lsn and p->current_lsn.
-     * ha-entity-snapshots task 5.1: call-site signature compat only for the
-     * new out-param — entities_restored itself is not yet consumed here
-     * (that is task 6.1's job, once 5.2 makes the flag meaningful). */
+     * ha-entity-snapshots task 6.1: entities_restored is consumed below,
+     * purely for observability — it does not change the rc-gates-applied_lsn
+     * ordering here (persist_load_snapshot/load_snapshot already gate their
+     * own position advance on entity-restore success, task 5.2). */
     bool entities_restored = false;
     int rc = persist_load_snapshot(p, final_path, chunk->snapshot_lsn,
                                    &entities_restored);
@@ -2608,8 +2609,19 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
     /* Advance applied_lsn to checkpoint_lsn (in-memory only — Req 4.2) */
     atomic_store_explicit(&ha->applied_lsn, p->checkpoint_lsn,
                           memory_order_release);
-    INFO("HA: snapshot loaded lsn=%" PRIu64 " applied_lsn set",
-         p->checkpoint_lsn);
+
+    /* Observability only (Requirements 2.4, 4.1, 4.2, 5.1, 5.2) — a legacy
+     * (position-only) checkpoint is an expected rollout state, not an error:
+     * log it as informational and let the member continue through the
+     * existing incremental catch-up path for the remaining history. */
+    if (entities_restored) {
+        INFO("HA: snapshot loaded lsn=%" PRIu64 " with full job/queue entity "
+             "restoration — applied_lsn set", p->checkpoint_lsn);
+    } else {
+        INFO("HA: snapshot lsn=%" PRIu64 " is a legacy position-only "
+             "checkpoint (no entity content) — applied_lsn set, continuing "
+             "via normal incremental catch-up", p->checkpoint_lsn);
+    }
     return 0;
 }
 

src/controller/testsuite/test_config_repl.c

@@ -3057,6 +3057,83 @@ static void test_snapshot_recv_two_chunks_assembles_and_renames(void)
     printf("test_snapshot_recv_two_chunks_assembles_and_renames: PASS\n");
 }
 
+/* Test 6.1: a legacy (pre-ha-entity-snapshots) position-only snapshot —
+ * exactly the 16-byte XCESNAP magic+lsn header, no entity payload — driven
+ * through the real receive handler end-to-end. persist_load_snapshot's
+ * entities_restored out-param comes back false (design.md "Checkpoint File
+ * Layout": "does more data exist" is the legacy discriminator), and
+ * ha_handle_snapshot_chunk must still: (a) return rc==0 (a legacy file is
+ * not a failure, Requirement 2.4/5.1/5.2), and (b) advance ha->applied_lsn
+ * exactly as it does for an entity-bearing file (Requirement 4.1/4.2 —
+ * the applied_lsn gating is unchanged by task 6.1; only a new informational
+ * log line is added on this path).
+ * Requirements: 2.4, 4.1, 4.2, 5.1, 5.2 */
+static void test_snapshot_recv_legacy_file_advances_without_entities(void)
+{
+    char tmpdir[] = "/tmp/xcectest_snaprecv_legacy_XXXXXX";
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
+
+    char snap_dir[512];
+    snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir);
+    TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700));
+
+    /* Legacy body: magic(8) + checkpoint_lsn(8), nothing else — the file
+     * ends exactly at the legacy header, so load_snapshot's "does more data
+     * exist" discriminator reports this as legacy (entities_restored=false)
+     * rather than corrupt or entity-bearing. */
+    uint64_t snap_lsn = 300;
+    uint8_t body[16];
+    memcpy(body, "XCESNAP\0", 8);
+    memcpy(body + 8, &snap_lsn, 8);
+
+    persist_t *p = calloc(1, sizeof(persist_t));
+    TEST_ASSERT_NOT_NULL(p);
+    persist_config_t pcfg = persist_config_defaults();
+    persist_set_workdir(p, tmpdir, "ctrl-snaprecv-legacy");
+    persist_init(p, &pcfg, NULL);
+    persist_start(p);
+
+    ha_state_t ha;
+    snap_recv_ha_init(&ha);
+
+    snapshot_transfer_t c0;
+    memset(&c0, 0, sizeof(c0));
+    c0.snapshot_lsn = snap_lsn;
+    c0.chunk_index  = 0;
+    c0.chunk_offset = 0;
+    c0.total_size   = sizeof(body);
+    c0.is_last      = 1;
+
+    int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0, body, sizeof(body));
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
+        "a legacy position-only checkpoint must not be treated as a failure");
+
+    /* Final file must exist (rename still happens on the legacy path) */
+    char final_path[600];
+    snprintf(final_path, sizeof(final_path),
+             "%s/snap-%016" PRIx64 ".bin", snap_dir, snap_lsn);
+    struct stat st;
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(final_path, &st),
+                                  "snap-{lsn}.bin should exist after isLast");
+
+    /* applied_lsn advances to the legacy file's recorded position — same
+     * rc-gates-applied_lsn ordering as the entity-bearing path, unchanged by
+     * task 6.1 (Requirement 4.1, 4.2). */
+    uint64_t got_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn,
+                                                       memory_order_acquire);
+    TEST_ASSERT_EQUAL_UINT64(snap_lsn, got_lsn);
+
+    ha_fini(&ha);
+    persist_stop(p);
+    persist_fini(p);
+    free(p);
+
+    char cmd[512];
+    snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
+    system(cmd);
+    printf("test_snapshot_recv_legacy_file_advances_without_entities: PASS\n");
+}
+
 /* Test 3.5c-B: size mismatch on isLast returns error (verify size == total_size).
  * Requirements: 4.3 */
 static void test_snapshot_recv_size_mismatch_fails(void)
@@ -3531,6 +3608,9 @@ void test_config_repl(void)
     RUN_TEST(test_snapshot_transfer_small_less_than_one_chunk);
     /* Task 3.5c: standby snapshot receive handler */
     RUN_TEST(test_snapshot_recv_two_chunks_assembles_and_renames);
+    /* Task 6.1: legacy position-only file consumed end-to-end via the
+     * receive handler — entities_restored=false path */
+    RUN_TEST(test_snapshot_recv_legacy_file_advances_without_entities);
     RUN_TEST(test_snapshot_recv_size_mismatch_fails);
     RUN_TEST(test_snapshot_recv_null_args_fail);
     /* Task 6.4-A: 128KB snapshot in 2 chunks — full-size receive path */