test(v3): complete reader owners before committing a prune backlog (#104)

The native 64-owner backlog regression failed at its first bulk ACTIVE-owner commit, before any explicit prune assertion, because the shared deferred ownership guard rejected the final state. Source inspection supports the 60-second reader deadline as the most likely explanation; the runtime log does not identify which guard predicate failed. Complete each of the 65 admitted readers immediately in the same transaction, preserving the terminal transaction fence and deferred validation. After its first successful commit, assert 65 FINISHED owners, no ACTIVE owners or REQUEST/READER anchors, and an issuance high-water advance of 65. Preserve the same-transaction prune=0 and later prune=64 then1, with retained-row counts1 then0 and all existing tests. Production SQL, deadlines, ownership guards, implementation fingerprints and the focused24-test workflow remain unchanged. Nightly formatting, scoped source checks, candidate whitespace and bidirectional patch checks pass; native execution of this new fixture remains pending.

Author: Ivanbeethoven

src/api/router/snapshot_reader_retention_tests.rs

@@ -189,18 +189,45 @@ async fn terminal_owner_survives_its_deferred_completion_and_prunes_in_a_later_t
 #[tokio::test]
 async fn reader_pruning_respects_the_64_owner_budget_with_a_committed_backlog() {
     let fixture = Fixture::new_with_pg_config(true).await;
+    let initial = q_count(
+        &fixture,
+        "SELECT high_water FROM {q}.mst2_metadata_reader_issuance",
+    )
+    .await;
     let txn = transaction(&fixture).await;
-    let mut tickets = Vec::new();
+    // Complete each owner before admitting the next. Their terminal transaction
+    // fence retains the entire backlog until its deferred checks have committed.
     for _ in 0..65 {
-        tickets.push(admission(&txn, &fixture).await);
-    }
-    txn.commit().await.unwrap();
-    let txn = transaction(&fixture).await;
-    for ticket in &tickets {
-        finish(&txn, ticket).await;
+        let ticket = admission(&txn, &fixture).await;
+        finish(&txn, &ticket).await;
     }
     assert_eq!(prune(&txn, 64).await, 0);
     txn.commit().await.unwrap();
+    assert_eq!(
+        q_count(
+            &fixture,
+            "SELECT count(*) FROM {q}.mst2_metadata_reader_operation WHERE state='FINISHED'"
+        )
+        .await,
+        65
+    );
+    assert_eq!(
+        q_count(
+            &fixture,
+            "SELECT count(*) FROM {q}.mst2_metadata_reader_operation WHERE state='ACTIVE'"
+        )
+        .await,
+        0
+    );
+    assert_eq!(q_count(&fixture,"SELECT count(*) FROM {q}.mst2_metadata_root_anchor WHERE anchor_kind IN ('REQUEST','READER')").await,0);
+    assert_eq!(
+        q_count(
+            &fixture,
+            "SELECT high_water FROM {q}.mst2_metadata_reader_issuance"
+        )
+        .await,
+        initial + 65
+    );
     let txn = transaction(&fixture).await;
     assert_eq!(prune(&txn, 64).await, 64);
     txn.commit().await.unwrap();