feat(ha-entity-snapshots): rotation guard for in-flight checkpoint transfers (task 4.3)

rotate_snapshots now parses each candidate file's LSN and skips deletion
via a new snapshot_lsn_in_flight() check against
p->cluster->ha.snapshot_inflight_lsn[] (task 4.1) before unlinking --
protected files are retried on the next rotation pass once the transfer
completes and the LSN clears.

Checks LSN directly (nonzero + matches) rather than gating on
snapshot_inflight[]==1 first, per task 4.1's own note: the LSN clears
before the flag in every completion/failure path, so a stale flag=1 can
momentarily coexist with an already-cleared LSN=0. A checkpoint LSN is
never 0, so this is race-free without needing the flag at all.
p->cluster==NULL (standalone lib-level tests) means no HA layer exists to
protect anything.

Task 4.2's RED rotation-guard test now passes; full regression sweep
(test_persist_checkpoint, test_persist_hooks, test_commit_barrier_reply,
test_commit_durability, test_e2e_hooks, test_hooks, test_config_repl,
test_ha_liveness) all clean. Both controller_unit_test and production
xce_controller build clean; nm confirms only the real production helper
(snapshot_lsn_in_flight) appears, no test-only symbols leaked in.

Executed directly (no subagent dispatch) per the hybrid execution-mode
decision.

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

@@ -49,7 +49,7 @@
   - _Requirements: 1.1, 1.2_
   - _Boundary: Persistence Layer (write_snapshot)_
 
-- [ ] 4. Core: protect an in-flight checkpoint transfer from rotation
+- [x] 4. Core: protect an in-flight checkpoint transfer from rotation
 - [x] 4.1 (P) Track which checkpoint position each in-flight peer transfer references
   - Add a per-peer tracking array alongside the existing in-flight-transfer flag; set it when a
     transfer begins, clear it when the transfer completes
@@ -63,7 +63,7 @@
   - Observable: new test compiles and fails (RED) against the current rotation logic
   - _Requirements: 3.1, 3.2, 3.3_
   - _Depends: 4.1_
-- [ ] 4.3 Skip rotating out a checkpoint file that is currently mid-transfer
+- [x] 4.3 Skip rotating out a checkpoint file that is currently mid-transfer
   - Extend the existing file-rotation pass to check the in-flight tracking before deleting a
     candidate file
   - Observable: the RED test from 4.2 passes

src/controller/persist.c

@@ -1996,6 +1996,29 @@ static int snap_name_cmp(const void* a, const void* b)
 }
 
 /* Keep only the snap_keep_n most recent snapshot files; delete the rest. */
+/* True when `lsn` is the checkpoint position an in-flight peer snapshot
+ * transfer currently references (ha-entity-snapshots task 4.1/4.3, Req
+ * 3.1-3.3) — checked directly against snapshot_inflight_lsn[], not gated on
+ * snapshot_inflight[] first: the LSN clears before the flag in every
+ * ha_snapshot_send_dispatch/snapshot_lane_fn path, so a stale flag=1 can
+ * momentarily coexist with an already-cleared LSN=0 (see task 4.1's
+ * Implementation Note). A checkpoint's LSN is never 0, so "nonzero and
+ * matches" is the correct, race-free protection signal on its own. No
+ * cluster (p->cluster == NULL, e.g. standalone lib-level tests) means no HA
+ * layer exists to be mid-transfer — nothing is protected. */
+static bool snapshot_lsn_in_flight(persist_t* p, uint64_t lsn)
+{
+    if (p->cluster == NULL || lsn == 0)
+        return false;
+
+    for (int i = 0; i < MAX_HA_CANDIDATES; i++) {
+        if (atomic_load_explicit(&p->cluster->ha.snapshot_inflight_lsn[i],
+                                 memory_order_acquire) == lsn)
+            return true;
+    }
+    return false;
+}
+
 static void rotate_snapshots(persist_t* p)
 {
     DIR* d = opendir(p->snap_path);
@@ -2018,11 +2041,18 @@ static void rotate_snapshots(persist_t* p)
     /* Sort ascending — oldest first */
     qsort(names, (size_t)n, sizeof(names[0]), snap_name_cmp);
 
-    /* Delete all but the snap_keep_n newest */
+    /* Delete all but the snap_keep_n newest, skipping any file an in-flight
+     * peer transfer is still reading from (task 4.3) — it is retried on the
+     * next rotation pass once that transfer completes. */
     int keep = (int)p->config.snap_keep_n;
     int delete_up_to = n - keep; /* first index to keep */
 
     for (int i = 0; i < delete_up_to && i < n; i++) {
+        uint64_t lsn = 0;
+        if (sscanf(names[i], "snap-%" SCNx64 ".capnp", &lsn) == 1
+                && snapshot_lsn_in_flight(p, lsn))
+            continue;
+
         char path[4096];
         snprintf(path, sizeof(path), "%s/%s", p->snap_path, names[i]);
         unlink(path);