fix(node): retry the requeue re-read so a coalesced push is not silently dropped

`requeue_or_release` clears the dirty flag atomically with the decision to loop,
which is correct and stays as it is. The problem was what the pass did next: a
failed `get_repo_by_id` fell into an Err arm that yielded no rules, and
`list_visibility_rules(...).ok()` collapsed its own error the same way. With no
rules, `replication_withheld_set` returns `(false, None)`, the pass skipped its
work entirely, and the loop exited having consumed the dirty bit that stood for
the coalesced push. There is no reconciliation sweep to re-derive it, so that
push's objects were never pinned and its withheld blobs never sealed.

`requeue_refresh_state` now separates the three cases the old code collapsed.
`Ok(Some)` proceeds. `Ok(None)` means the repo really is gone and releases
without touching the retry budget. `Err` from either read is transient and is
retried with doubling backoff up to three attempts. A rules-read error is no
longer indistinguishable from a repo that has no rules.

Exhausting the retries still drops the pass, which is the pre-existing residual,
but it now logs at error level with the repo id and attempt count instead of
disappearing behind a warning. Six tests cover the retried-then-lands path, the
bounded-exhaustion path, repo-gone, the rules-read arm, freshness of the applied
rules, and the coalescing property that had to survive. Reverting the Err arm,
restoring the `.ok()`, raising the bound, or downgrading the error log each turns
a test red.

crates/gitlawb-node/src/api/repos.rs

@@ -849,6 +849,177 @@ async fn requeue_full_scan_object_list(
     .await
 }
 
+/// Test-only fault-injection seam for the requeue re-read (#173 U2). The defect this
+/// unit fixes lives entirely on the `Err` arm of the two re-reads, which a real Postgres
+/// pool will not produce on demand, so the two reads go through the wrappers below and
+/// consult this table first. Keyed by `repo_id` (a fresh uuid per test) so tests running
+/// in parallel in one process cannot see each other's injections, and it also records
+/// the ATTEMPT counts the retry-bound assertions key on.
+#[cfg(test)]
+pub(crate) mod requeue_faults {
+    use std::collections::HashMap;
+    use std::sync::{Mutex, OnceLock};
+
+    #[derive(Default, Clone, Copy, Debug)]
+    pub(crate) struct Counters {
+        pub(crate) repo_read_failures_left: usize,
+        pub(crate) rules_read_failures_left: usize,
+        pub(crate) repo_read_attempts: usize,
+        pub(crate) rules_read_attempts: usize,
+    }
+
+    fn table() -> &'static Mutex<HashMap<String, Counters>> {
+        static TABLE: OnceLock<Mutex<HashMap<String, Counters>>> = OnceLock::new();
+        TABLE.get_or_init(|| Mutex::new(HashMap::new()))
+    }
+
+    /// Make the next `repo_read_failures` repo re-reads and the next
+    /// `rules_read_failures` rule re-reads for `repo_id` return `Err`, then succeed.
+    pub(crate) fn inject(repo_id: &str, repo_read_failures: usize, rules_read_failures: usize) {
+        table().lock().unwrap().insert(
+            repo_id.to_string(),
+            Counters {
+                repo_read_failures_left: repo_read_failures,
+                rules_read_failures_left: rules_read_failures,
+                ..Default::default()
+            },
+        );
+    }
+
+    /// Observed attempt counts (and remaining injections) for `repo_id`.
+    pub(crate) fn counters(repo_id: &str) -> Counters {
+        table()
+            .lock()
+            .unwrap()
+            .get(repo_id)
+            .copied()
+            .unwrap_or_default()
+    }
+
+    /// Production-path hook: count one repo re-read attempt, return whether it must fail.
+    pub(crate) fn take_repo_read(repo_id: &str) -> bool {
+        let mut map = table().lock().unwrap();
+        let c = map.entry(repo_id.to_string()).or_default();
+        c.repo_read_attempts += 1;
+        if c.repo_read_failures_left > 0 {
+            c.repo_read_failures_left -= 1;
+            return true;
+        }
+        false
+    }
+
+    /// Production-path hook: count one rules re-read attempt, return whether it must fail.
+    pub(crate) fn take_rules_read(repo_id: &str) -> bool {
+        let mut map = table().lock().unwrap();
+        let c = map.entry(repo_id.to_string()).or_default();
+        c.rules_read_attempts += 1;
+        if c.rules_read_failures_left > 0 {
+            c.rules_read_failures_left -= 1;
+            return true;
+        }
+        false
+    }
+}
+
+/// Attempts allowed for the requeue re-read before the task gives up (#173 U2). The
+/// dirty flag that represented the coalesced push is already consumed by the atomic
+/// check-and-clear at the tail and `EncryptInflightGuard::drop` removes the key, so the
+/// flag cannot outlive the task and there is no reconciliation sweep to re-derive the
+/// work: a transient read error must be RETRIED here or the push's pin/encrypt pass is
+/// gone. The bound keeps a sustained outage from spinning forever; on exhaustion the
+/// work is still lost (the pre-existing residual), but the give-up is logged at ERROR
+/// so it is observable instead of silent.
+const REQUEUE_REREAD_MAX_ATTEMPTS: usize = 3;
+
+/// Backoff before the next re-read attempt. Doubles per attempt.
+const REQUEUE_REREAD_BACKOFF: std::time::Duration = std::time::Duration::from_millis(50);
+
+/// The outcome of the requeue's fresh state re-read, keeping the three cases the old
+/// code collapsed into one distinct: a usable refresh, a repo that genuinely no longer
+/// exists (terminal, and NOT a retry), and a transient read failure (retryable).
+enum RequeueRefresh {
+    State {
+        rules: Vec<crate::db::VisibilityRule>,
+        is_public: bool,
+        owner_did: String,
+    },
+    Gone,
+    Failed,
+}
+
+/// Re-read repo state for a requeue pass, retrying transient read errors.
+///
+/// Both reads are retryable and neither may be read as an absence: an `Err` from the
+/// repo row is not "the repo is gone", and an `Err` from the rule list is not "this repo
+/// has no rules" (the old `.ok()` made those indistinguishable, and a `None` rule set
+/// makes `replication_withheld_set` return `None`, which skips the entire pass). Only
+/// `Ok(None)` on the repo row is a terminal absence, and it consumes no retry budget.
+async fn requeue_refresh_state(ctx: &PostPushReplication) -> RequeueRefresh {
+    let mut backoff = REQUEUE_REREAD_BACKOFF;
+    for attempt in 1..=REQUEUE_REREAD_MAX_ATTEMPTS {
+        let record = match requeue_get_repo(ctx).await {
+            Ok(Some(rec)) => rec,
+            Ok(None) => {
+                tracing::debug!(repo = %ctx.repo_id, "repo gone before requeue pass; releasing");
+                return RequeueRefresh::Gone;
+            }
+            Err(e) => {
+                tracing::warn!(
+                    repo = %ctx.repo_id, err = %e, attempt,
+                    "requeue repo re-read failed; retrying"
+                );
+                tokio::time::sleep(backoff).await;
+                backoff *= 2;
+                continue;
+            }
+        };
+        match requeue_list_rules(ctx).await {
+            Ok(rules) => {
+                return RequeueRefresh::State {
+                    rules,
+                    is_public: record.is_public,
+                    owner_did: record.owner_did,
+                }
+            }
+            Err(e) => {
+                tracing::warn!(
+                    repo = %ctx.repo_id, err = %e, attempt,
+                    "requeue visibility-rule re-read failed; retrying"
+                );
+                tokio::time::sleep(backoff).await;
+                backoff *= 2;
+            }
+        }
+    }
+    tracing::error!(
+        repo = %ctx.repo_id,
+        attempts = REQUEUE_REREAD_MAX_ATTEMPTS,
+        "requeue re-read failed on every attempt; the coalesced push's pin/encrypt pass is \
+         dropped (no reconciliation sweep re-derives it)"
+    );
+    RequeueRefresh::Failed
+}
+
+/// The requeue's repo re-read, behind the test-only fault seam above.
+async fn requeue_get_repo(ctx: &PostPushReplication) -> anyhow::Result<Option<RepoRecord>> {
+    #[cfg(test)]
+    if requeue_faults::take_repo_read(&ctx.repo_id) {
+        return Err(anyhow::anyhow!("injected repo re-read failure"));
+    }
+    ctx.db.get_repo_by_id(&ctx.repo_id).await
+}
+
+/// The requeue's visibility-rule re-read, behind the test-only fault seam above.
+async fn requeue_list_rules(
+    ctx: &PostPushReplication,
+) -> anyhow::Result<Vec<crate::db::VisibilityRule>> {
+    #[cfg(test)]
+    if requeue_faults::take_rules_read(&ctx.repo_id) {
+        return Err(anyhow::anyhow!("injected visibility-rule re-read failure"));
+    }
+    ctx.db.list_visibility_rules(&ctx.repo_id).await
+}
+
 /// The detached post-push encryption + local-IPFS pin task, as a REQUEUE LOOP.
 ///
 /// Pass one uses the spawn-time captures (`first_*`) — the delta the push handler
@@ -966,20 +1137,19 @@ async fn run_post_push_replication(
         // A push coalesced during this pass. Re-read repo state FRESH (never the stale
         // spawn-time captures) so a coalesced push that changed `.gitlawb` withholding
         // is walked under the new policy, then re-enumerate the pin set fail-closed.
-        let (r_rules, r_is_public, r_owner) = match ctx.db.get_repo_by_id(&ctx.repo_id).await {
-            Ok(Some(rec)) => (
-                ctx.db.list_visibility_rules(&ctx.repo_id).await.ok(),
-                rec.is_public,
-                rec.owner_did,
-            ),
-            Ok(None) => {
-                tracing::debug!(repo = %ctx.repo_id, "repo gone before requeue pass; releasing");
-                (None, false, String::new())
-            }
-            Err(e) => {
-                tracing::warn!(repo = %ctx.repo_id, err = %e, "requeue repo re-read failed; skipping this pass's work");
-                (None, false, String::new())
-            }
+        // A read error here is retried rather than treated as "no state" (#173 U2): the
+        // dirty flag is already consumed, so skipping the pass would silently discard
+        // exactly the push this requeue exists to cover.
+        let (r_rules, r_is_public, r_owner) = match requeue_refresh_state(&ctx).await {
+            RequeueRefresh::State {
+                rules,
+                is_public,
+                owner_did,
+            } => (Some(rules), is_public, owner_did),
+            // The repo is gone (terminal) or the re-read never succeeded (already logged
+            // at ERROR). Either way there is no fresh state to act on, so exit; the guard
+            // Drop removes the key so the repo is never locked out of a future task.
+            RequeueRefresh::Gone | RequeueRefresh::Failed => break,
         };
         let (_announce, r_withheld) = replication_withheld_set(
             r_rules.clone(),

crates/gitlawb-node/src/test_support.rs

@@ -8323,5 +8323,506 @@ mod tests {
                 "the key is released after one pass"
             );
         }
+
+        // ---- U2 (#173): a transient re-read failure must not discard the coalesced
+        // push's work ----
+        //
+        // The requeue pass re-reads repo state fresh. Before this unit, the `Err` arm of
+        // either read (repo row, visibility rules) collapsed into "no rules", which made
+        // `replication_withheld_set` return `None`, which skipped the whole pass. The
+        // dirty flag was already consumed by the atomic check-and-clear at the tail and
+        // there is no reconciliation sweep, so the coalesced push's pin/encrypt work was
+        // lost silently. These tests drive that `Err` arm through the fault seam in
+        // `api::repos::requeue_faults` (a real pool will not fail on demand) and assert
+        // on the WORK PERFORMED, not on control flow.
+        mod u2_reread_retry {
+            use super::*;
+            use crate::api::repos::requeue_faults;
+
+            /// Process-wide tracing capture so a test can assert the give-up is logged at
+            /// ERROR. A global default subscriber can only be installed once per process,
+            /// so it is shared by every test here and assertions filter on the repo id,
+            /// which is a fresh uuid per test.
+            mod logcap {
+                use std::sync::{Arc, Mutex, OnceLock};
+                use tracing::{Event, Level, Subscriber};
+                use tracing_subscriber::layer::{Context, Layer};
+                use tracing_subscriber::prelude::*;
+
+                type Lines = Arc<Mutex<Vec<(Level, String)>>>;
+
+                fn lines() -> &'static Lines {
+                    static LINES: OnceLock<Lines> = OnceLock::new();
+                    LINES.get_or_init(|| Arc::new(Mutex::new(Vec::new())))
+                }
+
+                struct Capture;
+                impl<S: Subscriber> Layer<S> for Capture {
+                    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
+                        struct V(String);
+                        impl tracing::field::Visit for V {
+                            fn record_debug(
+                                &mut self,
+                                field: &tracing::field::Field,
+                                value: &dyn std::fmt::Debug,
+                            ) {
+                                self.0.push_str(&format!(" {}={:?}", field.name(), value));
+                            }
+                        }
+                        let mut v = V(String::new());
+                        event.record(&mut v);
+                        lines()
+                            .lock()
+                            .unwrap()
+                            .push((*event.metadata().level(), v.0));
+                    }
+                }
+
+                pub(super) fn install() {
+                    static ONCE: OnceLock<()> = OnceLock::new();
+                    ONCE.get_or_init(|| {
+                        let _ = tracing::subscriber::set_global_default(
+                            tracing_subscriber::registry().with(Capture),
+                        );
+                    });
+                }
+
+                pub(super) fn errors_containing(needle: &str) -> Vec<String> {
+                    lines()
+                        .lock()
+                        .unwrap()
+                        .iter()
+                        .filter(|(lvl, msg)| *lvl == Level::ERROR && msg.contains(needle))
+                        .map(|(_, msg)| msg.clone())
+                        .collect()
+                }
+            }
+
+            /// SCENARIO 1. The repo re-read fails once, then succeeds: the requeue pass
+            /// must still RUN, under the refreshed state, and pin the coalesced push's
+            /// object. RED before the fix (the single `Err` yielded `(None, false, "")`,
+            /// the pass was skipped, and the already-consumed dirty flag meant the work
+            /// was gone for good).
+            #[sqlx::test]
+            async fn u2_transient_repo_reread_failure_is_retried_and_work_lands(pool: PgPool) {
+                let state = test_state(pool).await;
+                let owner = new_did();
+                let repo = seed_repo(&owner, "u2-retry");
+                state.db.create_repo(&repo).await.expect("seed repo");
+                let git_repo = init_repo();
+                let obj1 = commit(&git_repo.path, "a.txt", "one\n");
+                // The coalesced push B adds obj2, absent from push A's spawn captures.
+                let obj2 = commit(&git_repo.path, "b.txt", "two\n");
+
+                let mut server = mockito::Server::new_async().await;
+                let _m = server
+                    .mock("POST", mockito::Matcher::Regex(r"^/api/v0/add".to_string()))
+                    .with_status(200)
+                    .with_body(r#"{"Hash":"bafyprovider"}"#)
+                    .expect_at_least(1)
+                    .create_async()
+                    .await;
+
+                // One transient repo re-read failure, then the real DB answers.
+                requeue_faults::inject(&repo.id, 1, 0);
+
+                let guard = state
+                    .encrypt_inflight
+                    .try_begin(&repo.id)
+                    .expect("push A admits");
+                assert!(
+                    state.encrypt_inflight.try_begin(&repo.id).is_none(),
+                    "push B coalesces while A is in flight"
+                );
+
+                crate::api::repos::run_post_push_replication_for_test(
+                    &state,
+                    guard,
+                    git_repo.path.clone(),
+                    repo.id.clone(),
+                    server.url(),
+                    true,
+                    owner.clone(),
+                    vec![obj1.clone()],
+                    Some(vec![]),
+                    HashSet::new(),
+                )
+                .await;
+
+                assert!(
+                    state.db.is_pinned(&obj2).await.unwrap(),
+                    "the coalesced push's object is pinned after the retried re-read (RED \
+                     before this unit: the Err arm dropped the pass and the work with it)"
+                );
+                let c = requeue_faults::counters(&repo.id);
+                assert_eq!(
+                    c.repo_read_attempts, 2,
+                    "the failed re-read is retried exactly once before it succeeds"
+                );
+                assert!(
+                    state.encrypt_inflight.is_empty(),
+                    "the guard key is released once the task exits"
+                );
+            }
+
+            /// SCENARIO 2. Every re-read attempt fails: the loop must give up on a BOUND
+            /// (asserted as a literal, so raising or removing the bound goes RED) and log
+            /// the give-up at ERROR so the residual loss is observable rather than silent.
+            #[sqlx::test]
+            async fn u2_sustained_repo_reread_failure_is_bounded_and_logged(pool: PgPool) {
+                logcap::install();
+                let state = test_state(pool).await;
+                let owner = new_did();
+                let repo = seed_repo(&owner, "u2-bounded");
+                state.db.create_repo(&repo).await.expect("seed repo");
+                let git_repo = init_repo();
+                let obj1 = commit(&git_repo.path, "a.txt", "one\n");
+                let obj2 = commit(&git_repo.path, "b.txt", "two\n");
+
+                let mut server = mockito::Server::new_async().await;
+                let _m = server
+                    .mock("POST", mockito::Matcher::Regex(r"^/api/v0/add".to_string()))
+                    .with_status(200)
+                    .with_body(r#"{"Hash":"bafyprovider"}"#)
+                    .expect_at_least(1)
+                    .create_async()
+                    .await;
+
+                // Far more failures than the bound allows: the outage never clears.
+                requeue_faults::inject(&repo.id, 10_000, 0);
+
+                let guard = state
+                    .encrypt_inflight
+                    .try_begin(&repo.id)
+                    .expect("push A admits");
+                assert!(
+                    state.encrypt_inflight.try_begin(&repo.id).is_none(),
+                    "push B coalesces"
+                );
+
+                crate::api::repos::run_post_push_replication_for_test(
+                    &state,
+                    guard,
+                    git_repo.path.clone(),
+                    repo.id.clone(),
+                    server.url(),
+                    true,
+                    owner.clone(),
+                    vec![obj1.clone()],
+                    Some(vec![]),
+                    HashSet::new(),
+                )
+                .await;
+
+                let c = requeue_faults::counters(&repo.id);
+                assert_eq!(
+                    c.repo_read_attempts, 3,
+                    "the re-read is bounded at 3 attempts; unbounded retry or a raised \
+                     bound must fail here"
+                );
+                assert!(
+                    !state.db.is_pinned(&obj2).await.unwrap(),
+                    "with the read never succeeding there is nothing fresh to act on"
+                );
+                let errs = logcap::errors_containing(&repo.id);
+                assert!(
+                    !errs.is_empty(),
+                    "the exhausted requeue re-read is logged at ERROR with the repo id, so \
+                     the residual work loss is observable; captured: {errs:?}"
+                );
+                assert!(
+                    state.encrypt_inflight.is_empty(),
+                    "the guard key is still released on the give-up path"
+                );
+            }
+
+            /// SCENARIO 3. `Ok(None)` (the repo was deleted during the in-flight window)
+            /// is NOT a transient failure: it must release immediately without burning the
+            /// retry budget. The repo id is never inserted, so the re-read legitimately
+            /// returns `Ok(None)`.
+            #[sqlx::test]
+            async fn u2_repo_gone_releases_without_consuming_retries(pool: PgPool) {
+                let state = test_state(pool).await;
+                let owner = new_did();
+                let missing_id = uuid::Uuid::new_v4().to_string();
+                let git_repo = init_repo();
+                let _obj1 = commit(&git_repo.path, "a.txt", "one\n");
+
+                let server = mockito::Server::new_async().await;
+
+                requeue_faults::inject(&missing_id, 0, 0);
+
+                let guard = state
+                    .encrypt_inflight
+                    .try_begin(&missing_id)
+                    .expect("push A admits");
+                assert!(
+                    state.encrypt_inflight.try_begin(&missing_id).is_none(),
+                    "push B coalesces"
+                );
+
+                // Empty object list: pass one touches no pin rows for a repo that is gone.
+                crate::api::repos::run_post_push_replication_for_test(
+                    &state,
+                    guard,
+                    git_repo.path.clone(),
+                    missing_id.clone(),
+                    server.url(),
+                    true,
+                    owner.clone(),
+                    vec![],
+                    Some(vec![]),
+                    HashSet::new(),
+                )
+                .await;
+
+                let c = requeue_faults::counters(&missing_id);
+                assert_eq!(
+                    c.repo_read_attempts, 1,
+                    "a deleted repo is a terminal answer, never retried"
+                );
+                assert_eq!(
+                    c.rules_read_attempts, 0,
+                    "no rules read is attempted once the repo row is gone"
+                );
+                assert!(
+                    state.encrypt_inflight.is_empty(),
+                    "the guard key is released cleanly"
+                );
+            }
+
+            /// SCENARIO 4. A failed visibility-rule read is transient, never an empty
+            /// policy. RED before the fix, where `.ok()` made "the rules read failed" and
+            /// "this repo has no rules" the same value: the withheld blob was then neither
+            /// sealed nor covered, because a `None` rule set skips the pass entirely.
+            #[sqlx::test]
+            async fn u2_transient_rules_read_failure_is_retried_not_read_as_empty(pool: PgPool) {
+                let state = test_state(pool).await;
+                let owner = new_did();
+                let reader = new_did();
+                let repo = seed_repo(&owner, "u2-rules");
+                state.db.create_repo(&repo).await.expect("seed repo");
+                let git_repo = init_repo();
+                let pub_oid = commit(&git_repo.path, "public/a.txt", "public\n");
+                let secret_oid = commit(&git_repo.path, "secret/b.txt", "TOP SECRET\n");
+
+                // The coalesced push B is what added the path-scoped rule.
+                state
+                    .db
+                    .set_visibility_rule(
+                        &repo.id,
+                        "/secret/**",
+                        VisibilityMode::B,
+                        &[reader],
+                        &owner,
+                    )
+                    .await
+                    .expect("set rule");
+
+                let mut server = mockito::Server::new_async().await;
+                let _m = server
+                    .mock("POST", mockito::Matcher::Regex(r"^/api/v0/add".to_string()))
+                    .with_status(200)
+                    .with_body(r#"{"Hash":"bafyprovider"}"#)
+                    .expect_at_least(1)
+                    .create_async()
+                    .await;
+
+                // The repo row reads fine; the RULES read is the one that blips.
+                requeue_faults::inject(&repo.id, 0, 1);
+
+                let guard = state
+                    .encrypt_inflight
+                    .try_begin(&repo.id)
+                    .expect("push A admits");
+                assert!(
+                    state.encrypt_inflight.try_begin(&repo.id).is_none(),
+                    "push B coalesces"
+                );
+
+                // Push A's captures are stale: no rule, nothing withheld.
+                crate::api::repos::run_post_push_replication_for_test(
+                    &state,
+                    guard,
+                    git_repo.path.clone(),
+                    repo.id.clone(),
+                    server.url(),
+                    true,
+                    owner.clone(),
+                    vec![pub_oid.clone()],
+                    Some(vec![]),
+                    HashSet::new(),
+                )
+                .await;
+
+                assert!(
+                    state
+                        .db
+                        .encrypted_blob_recipients_tag(&repo.id, &secret_oid)
+                        .await
+                        .unwrap()
+                        .is_some(),
+                    "the withheld blob is sealed under the RETRIED rule set (RED with \
+                     list_visibility_rules(..).ok(): an empty policy seals nothing)"
+                );
+                let c = requeue_faults::counters(&repo.id);
+                assert_eq!(
+                    c.rules_read_attempts, 2,
+                    "the failed rules read is retried, not collapsed into an empty rule set"
+                );
+                assert!(
+                    !state.db.is_pinned(&secret_oid).await.unwrap(),
+                    "the withheld blob is never pinned in the clear by the requeue"
+                );
+            }
+
+            /// SCENARIO 5. The fault-free control for scenario 4: the rules applied by the
+            /// requeue are the COALESCED push's fresh ones, never the spawn-time capture,
+            /// and the retry path does not perturb that (exactly one read of each).
+            #[sqlx::test]
+            async fn u2_requeue_applies_fresh_rules_not_spawn_captures(pool: PgPool) {
+                let state = test_state(pool).await;
+                let owner = new_did();
+                let reader = new_did();
+                let repo = seed_repo(&owner, "u2-fresh");
+                state.db.create_repo(&repo).await.expect("seed repo");
+                let git_repo = init_repo();
+                let pub_oid = commit(&git_repo.path, "public/a.txt", "public\n");
+                let secret_oid = commit(&git_repo.path, "secret/b.txt", "TOP SECRET\n");
+                state
+                    .db
+                    .set_visibility_rule(
+                        &repo.id,
+                        "/secret/**",
+                        VisibilityMode::B,
+                        &[reader],
+                        &owner,
+                    )
+                    .await
+                    .expect("set rule");
+
+                let mut server = mockito::Server::new_async().await;
+                let _m = server
+                    .mock("POST", mockito::Matcher::Regex(r"^/api/v0/add".to_string()))
+                    .with_status(200)
+                    .with_body(r#"{"Hash":"bafyprovider"}"#)
+                    .expect_at_least(1)
+                    .create_async()
+                    .await;
+
+                requeue_faults::inject(&repo.id, 0, 0);
+
+                let guard = state
+                    .encrypt_inflight
+                    .try_begin(&repo.id)
+                    .expect("push A admits");
+                assert!(
+                    state.encrypt_inflight.try_begin(&repo.id).is_none(),
+                    "push B coalesces"
+                );
+
+                crate::api::repos::run_post_push_replication_for_test(
+                    &state,
+                    guard,
+                    git_repo.path.clone(),
+                    repo.id.clone(),
+                    server.url(),
+                    true,
+                    owner.clone(),
+                    vec![pub_oid.clone()],
+                    Some(vec![]),
+                    HashSet::new(),
+                )
+                .await;
+
+                let c = requeue_faults::counters(&repo.id);
+                assert_eq!(
+                    (c.repo_read_attempts, c.rules_read_attempts),
+                    (1, 1),
+                    "a healthy DB is read exactly once per requeue pass"
+                );
+                assert!(
+                    state.db.is_pinned(&pub_oid).await.unwrap(),
+                    "the visible object is pinned under the fresh rules"
+                );
+                assert!(
+                    !state.db.is_pinned(&secret_oid).await.unwrap(),
+                    "the freshly-read rule withholds the secret blob (the spawn-time \
+                     capture had no rules at all)"
+                );
+            }
+
+            /// SCENARIO 6. Regression guard on the property the fix must not disturb: the
+            /// tail check-and-clear is atomic, so a push coalescing during it is still
+            /// covered by exactly one more pass, and the key is released after.
+            #[sqlx::test]
+            async fn u2_coalesced_push_still_covered_by_exactly_one_requeue_pass(pool: PgPool) {
+                let state = test_state(pool).await;
+                let owner = new_did();
+                let repo = seed_repo(&owner, "u2-coalesce");
+                state.db.create_repo(&repo).await.expect("seed repo");
+                let git_repo = init_repo();
+                let obj1 = commit(&git_repo.path, "a.txt", "one\n");
+                let obj2 = commit(&git_repo.path, "b.txt", "two\n");
+
+                let mut server = mockito::Server::new_async().await;
+                let _m = server
+                    .mock("POST", mockito::Matcher::Regex(r"^/api/v0/add".to_string()))
+                    .with_status(200)
+                    .with_body(r#"{"Hash":"bafyprovider"}"#)
+                    .expect_at_least(1)
+                    .create_async()
+                    .await;
+
+                requeue_faults::inject(&repo.id, 0, 0);
+
+                let guard = state
+                    .encrypt_inflight
+                    .try_begin(&repo.id)
+                    .expect("push A admits");
+                // Push B lands during the in-flight window: dirty flag set.
+                assert!(
+                    state.encrypt_inflight.try_begin(&repo.id).is_none(),
+                    "push B coalesces"
+                );
+                assert_eq!(
+                    state.encrypt_inflight.dirty(&repo.id),
+                    Some(true),
+                    "the coalesced push marked the repo dirty"
+                );
+
+                crate::api::repos::run_post_push_replication_for_test(
+                    &state,
+                    guard,
+                    git_repo.path.clone(),
+                    repo.id.clone(),
+                    server.url(),
+                    true,
+                    owner.clone(),
+                    vec![obj1.clone()],
+                    Some(vec![]),
+                    HashSet::new(),
+                )
+                .await;
+
+                assert_eq!(
+                    requeue_faults::counters(&repo.id).repo_read_attempts,
+                    1,
+                    "one coalesced push means exactly one requeue pass, no re-spin"
+                );
+                assert!(
+                    state.db.is_pinned(&obj1).await.unwrap(),
+                    "push A's object is pinned"
+                );
+                assert!(
+                    state.db.is_pinned(&obj2).await.unwrap(),
+                    "the coalesced push's object is covered by the requeue pass"
+                );
+                assert!(
+                    state.encrypt_inflight.is_empty(),
+                    "the key is released once the task is clean"
+                );
+            }
+        }
     }
 }