diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index b817318d..35c49dd8 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/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> { + static TABLE: OnceLock>> = 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, + 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> { + #[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> { + #[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(), diff --git a/crates/gitlawb-node/src/test_support.rs b/crates/gitlawb-node/src/test_support.rs index 0d0ae2e8..f97981b4 100644 --- a/crates/gitlawb-node/src/test_support.rs +++ b/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>>; + + fn lines() -> &'static Lines { + static LINES: OnceLock = OnceLock::new(); + LINES.get_or_init(|| Arc::new(Mutex::new(Vec::new()))) + } + + struct Capture; + impl Layer 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 { + 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" + ); + } + } } }