diff --git a/crates/gitlawb-node/src/api/peers.rs b/crates/gitlawb-node/src/api/peers.rs index 2ae2cc183..1f0c73a90 100644 --- a/crates/gitlawb-node/src/api/peers.rs +++ b/crates/gitlawb-node/src/api/peers.rs @@ -439,6 +439,7 @@ pub async fn trigger_sync(State(state): State) -> Result, #[serde(default)] pub timestamp: Option, + /// The instant the origin push landed, for sync-lag measurement only — + /// mirrors RefUpdateEvent::lag_origin on the gossip transport. `timestamp` + /// stays publish semantics to match what gossip ingest stores on the feed + /// row; when an older sender omits this field the enqueue falls back to + /// `timestamp`, which was the pre-split approximation of land time. + #[serde(default)] + pub lag_origin: Option, #[serde(default)] pub cert_id: Option, /// Full owner DID — added in #144 for DID-aware feed gating. @@ -520,7 +528,14 @@ pub async fn notify_sync( state .db - .enqueue_sync(&req.repo, &req.node_did, &req.ref_name, &req.new_sha, None) + .enqueue_sync( + &req.repo, + &req.node_did, + &req.ref_name, + &req.new_sha, + None, + req.lag_origin.as_deref().or(req.timestamp.as_deref()), + ) .await?; // Mirror the gossipsub-receive handler: insert the same record we'd @@ -1282,6 +1297,52 @@ mod tests { ); } + #[sqlx::test] + async fn sync_notify_preserves_origin_timestamp_for_sync_lag(pool: PgPool) { + let state = test_state(pool).await; + let db = state.db.clone(); + let peer_did = seed_peer(&state).await; + // Distinct publish and land instants: the feed row must carry publish + // semantics while only the queue row sees lag_origin. + let timestamp = "2026-08-16T00:00:00Z"; + let lag_origin = "2026-08-15T23:59:00Z"; + let body = serde_json::json!({ + "repo": "z6Mkfoo/timestamped", + "ref_name": "refs/heads/main", + "new_sha": "0000000000000000000000000000000000000000", + "node_did": peer_did, + "timestamp": timestamp, + "lag_origin": lag_origin, + }) + .to_string(); + let router = crate::server::build_router(state); + + let resp = router + .oneshot(unsigned_post( + "/api/v1/sync/notify", + &body, + "198.51.100.23:5000", + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let queued = db.dequeue_pending_syncs(100).await.unwrap(); + assert_eq!(queued.len(), 1); + assert_eq!(queued[0].origin_timestamp.as_deref(), Some(lag_origin)); + let feed = db + .list_ref_updates_keyset(Some("z6Mkfoo/timestamped"), 10, None) + .await + .unwrap(); + assert_eq!( + feed.iter() + .find(|u| u.repo == "z6Mkfoo/timestamped") + .map(|u| u.timestamp.as_str()), + Some(timestamp), + "the feed row must keep publish semantics, matching gossip ingest" + ); + } + #[sqlx::test] async fn sync_notify_unknown_peer_wins_over_malformed_slug(pool: PgPool) { // Ordering: the slug check sits after the peer gate, so an unknown peer diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index 0e3d4930c..5ade5b6f9 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -1777,6 +1777,7 @@ pub async fn git_upload_pack( headers: axum::http::HeaderMap, body: Bytes, ) -> Result { + let fetch_started = std::time::Instant::now(); // #62 cheap load shed. Permit-less snapshot, not admission; see git_info_refs for // what it does and does not bound. The authoritative hold is `git_permit` below, // after the per-source cap. @@ -1979,6 +1980,7 @@ pub async fn git_upload_pack( // that mislabel predates this change and is tracked as a follow-up. if should_count_fetch(finalizes_fetch, served_filtered_pack, served_pack) { crate::metrics::record_fetch(&format!("{owner}/{name}")); + crate::metrics::observe_fetch_duration(fetch_started.elapsed()); crate::metrics::observe_pack_size(body_len as f64); } Ok(resp) @@ -2190,6 +2192,7 @@ async fn notify_peer_of_ref( node_did: &str, pusher_did: &str, owner_did: &str, + origin_timestamp: &str, ) { let body = serde_json::json!({ "repo": repo_slug, @@ -2198,7 +2201,13 @@ async fn notify_peer_of_ref( "node_did": node_did, "pusher_did": pusher_did, "old_sha": old_sha, - "timestamp": chrono::Utc::now().to_rfc3339(), + // Publish semantics, matching what gossip ingest stores on the feed + // row: the send instant, freshness-window meaningful. Land time travels + // separately in lag_origin (also an Option on the receiver); + // stamping land here would put a different clock on HTTP rows than on + // gossip rows for the same logical update. + "timestamp": Utc::now().to_rfc3339(), + "lag_origin": origin_timestamp, "owner_did": owner_did, }); let body_bytes = match serde_json::to_vec(&body) { @@ -2248,6 +2257,7 @@ async fn notify_peer_of_refs( node_did: &str, pusher_did: &str, owner_did: &str, + origin_timestamp: &str, ) { for (ref_name, old_sha, new_sha) in ref_updates { notify_peer_of_ref( @@ -2262,6 +2272,7 @@ async fn notify_peer_of_refs( node_did, pusher_did, owner_did, + origin_timestamp, ) .await; } @@ -2280,6 +2291,7 @@ pub async fn git_receive_pack( headers: axum::http::HeaderMap, body: Bytes, ) -> Result { + let push_started = std::time::Instant::now(); let name = smart_http_repo_name(&repo)?; // Fast-path shed before the DB lookup when the write pool is ALREADY saturated, so a // push flood against a full pool does not hit Postgres per request. Best-effort @@ -3309,14 +3321,37 @@ pub async fn git_receive_pack( Vec::new() }; if disposition.spawn_tail && !push_succeeded && !landed_refs.is_empty() { - tokio::spawn(post_receive_replication_tail( - state.clone(), - record.clone(), - landed_refs.clone(), - disk_path.clone(), - auth.0.to_string(), - None, - )); + let state = state.clone(); + let record = record.clone(); + let landed = landed_refs.clone(); + let disk_path = disk_path.clone(); + let did = auth.0.to_string(); + let db = state.db.clone(); + let request_id = request_id.clone(); + let fallback = now.clone(); + tokio::spawn(async move { + // Resolve inside the owned task: awaiting the lookup in the handler + // would hold the reclaimed write lease during a DB wait and leave + // the tail un-owned on a disconnect. Same land rule as the deferred + // drain — max applied_at across this request's landed transitions, + // else the request row's created_at. + let push_landed_at = db + .request_push_landed_at(&request_id) + .await + .ok() + .flatten() + .unwrap_or(fallback); + post_receive_replication_tail( + state, + record, + landed, + disk_path, + did, + Some(push_landed_at), + None, + ) + .await; + }); } // Always release the advisory lock — even on error — to prevent stale locks @@ -3337,14 +3372,35 @@ pub async fn git_receive_pack( // than by the mere fact that release was entered. publish_durability.arm_for_release(reclaimed.publish_stage()); if disposition.spawn_tail && !landed_refs.is_empty() { - tokio::spawn(post_receive_replication_tail( - state.clone(), - record.clone(), - landed_refs, - disk_path.clone(), - auth.0.to_string(), - Some(publish_durability.arc()), - )); + let state = state.clone(); + let record = record.clone(); + let disk_path = disk_path.clone(); + let did = auth.0.to_string(); + let db = state.db.clone(); + let request_id = request_id.clone(); + let fallback = now.clone(); + let durability_slot = publish_durability.arc(); + tokio::spawn(async move { + // Same in-task resolution and land rule as the carve-out spawn + // above — the handler must not await the lookup while it still + // holds the reclaimed write lease. + let push_landed_at = db + .request_push_landed_at(&request_id) + .await + .ok() + .flatten() + .unwrap_or(fallback); + post_receive_replication_tail( + state, + record, + landed_refs, + disk_path, + did, + Some(push_landed_at), + Some(durability_slot), + ) + .await; + }); } } // Short-circuit on a refused publish BEFORE anything downstream observes @@ -3411,6 +3467,7 @@ pub async fn git_receive_pack( { Ok(true) => {} _ => { + crate::metrics::observe_push_duration(push_started.elapsed()); return axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header("Content-Type", "application/x-git-receive-pack-result") @@ -3515,6 +3572,7 @@ pub async fn git_receive_pack( } } + crate::metrics::observe_push_duration(push_started.elapsed()); axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header("Content-Type", "application/x-git-receive-pack-result") @@ -3674,12 +3732,24 @@ async fn publish_durability_confirmed( /// complete without ever having a live tail (capability-free pushes whose /// landing reconcile proved). `pub(crate)` so the drain can call it; the /// live handler is the other caller. +/// The lag origin the tail reports: the caller-supplied push-land instant when +/// one exists (a deferred drain passes the request row's created_at), else the +/// tail's own entry time for callers that never knew the landing instant. +fn lag_origin_timestamp(push_landed_at: Option) -> String { + push_landed_at.unwrap_or_else(|| Utc::now().to_rfc3339()) +} + pub(crate) async fn post_receive_replication_tail( state: AppState, record: RepoRecord, ref_updates: Vec, disk_path: std::path::PathBuf, did: String, + // The instant the push landed, captured by the caller at acceptance so a + // deferred tail reports push-land lag rather than drain-time lag. Callers + // that cannot recover the landing instant pass None and the tail falls + // back to its own entry time, the pre-param behavior. + push_landed_at: Option, publish_durability: Option< Arc>>, >, @@ -3690,6 +3760,14 @@ pub(crate) async fn post_receive_replication_tail( .lock_held_transfer_timeout_secs .saturating_add(5), ); + // The sync-lag origin: the instant this push landed, captured BEFORE the + // durability wait so a slow publish does not shrink the lag a mirror + // measures. Travels to receiving peers as RefUpdateEvent::lag_origin + // (gossip) and the notify lag_origin field (HTTP) — never as the signed + // event's own `timestamp` or the notify `timestamp`, which both stay at + // publish time for check_freshness and feed-row parity. + let origin_timestamp = lag_origin_timestamp(push_landed_at); + if !publish_durability_confirmed(&publish_durability, durability_wait).await { tracing::warn!( repo = %record.id, @@ -3912,6 +3990,7 @@ pub(crate) async fn post_receive_replication_tail( let owner_did_for_arweave = record.owner_did.clone(); let self_public_url = state.config.public_url.clone(); let node_keypair = Arc::clone(&state.node_keypair); + let origin_timestamp_for_announce = origin_timestamp.clone(); // #174 F2a: gated on the cheap announce predicate, not on `withheld`. // `withheld` is None for a push that coalesced (it never walked), and // this task's work is per-push and non-idempotent, so keying it on the @@ -4037,6 +4116,7 @@ pub(crate) async fn post_receive_replication_tail( old_sha: old_sha.clone(), new_sha: new_sha.clone(), timestamp: chrono::Utc::now().to_rfc3339(), + lag_origin: Some(origin_timestamp_for_announce.clone()), cert_id: None, cid: cid.map(|s| s.to_string()), sig: None, @@ -4139,6 +4219,7 @@ pub(crate) async fn post_receive_replication_tail( &node_did_str, &pusher_did_clone, &record.owner_did, + &origin_timestamp_for_announce, ) .await; } @@ -8062,6 +8143,11 @@ mod tests { mockito::Matcher::PartialJsonString( r#"{"owner_did":"did:key:zOwner"}"#.to_string(), ), + // Land time rides lag_origin after the publish/land split; + // timestamp is a live send instant and cannot be pinned. + mockito::Matcher::PartialJsonString( + r#"{"lag_origin":"2026-08-16T00:00:00Z"}"#.to_string(), + ), ])) .with_status(200) .expect(1) @@ -8098,6 +8184,7 @@ mod tests { "did:key:zNode", "did:key:zPusher", "did:key:zOwner", + "2026-08-16T00:00:00Z", ) .await; @@ -8146,6 +8233,7 @@ mod tests { "did:key:zNode", "did:key:zPusher", "did:key:zOwner", + "2026-08-16T00:00:00Z", ) .await; @@ -13962,6 +14050,24 @@ mod tests { } #[cfg(unix)] + #[test] + fn lag_origin_timestamp_prefers_the_caller_supplied_push_land() { + let landed = "2019-01-01T00:00:00Z".to_string(); + assert_eq!( + lag_origin_timestamp(Some(landed.clone())), + landed, + "a deferred tail must report the request row's landing instant, not drain time" + ); + let fallback = lag_origin_timestamp(None); + let parsed = chrono::DateTime::parse_from_rfc3339(&fallback) + .expect("the fallback must still be a valid RFC-3339 instant"); + let skew = (chrono::Utc::now() - parsed.with_timezone(&chrono::Utc)).num_seconds(); + assert!( + skew.abs() < 60, + "a caller that never knew the landing instant falls back to tail entry time" + ); + } + fn f2a_update(ref_name: &str, new_sha: &str) -> Vec { vec![RefUpdate { old_sha: ZERO_SHA.to_string(), @@ -13994,6 +14100,7 @@ mod tests { f2a_update("refs/heads/main", &c1), repo.path().to_path_buf(), F2A_PUSHER.to_string(), + None, Some(slot), ) .await; @@ -14040,6 +14147,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; let after_first = f2a_walks(&log); @@ -14061,6 +14169,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -14137,6 +14246,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -14219,6 +14329,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, )); f2a_wait_for(|| started.exists(), "the admitted push's walk to start").await; @@ -14336,6 +14447,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -15487,6 +15599,7 @@ exit 0 repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -15543,6 +15656,7 @@ exit 0 repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; tokio::time::sleep(std::time::Duration::from_millis(300)).await; @@ -15585,6 +15699,7 @@ exit 0 repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -15684,6 +15799,7 @@ exit 0 repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index d1fa22e25..8c42987ae 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -2372,6 +2372,16 @@ const MIGRATIONS: &[Migration] = &[ "CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_pending ON webhook_deliveries (sent_at, created_at)", ], }, + Migration { + version: 47, + name: "sync_queue_origin_timestamp", + stmts: &[ + // The source timestamp accompanies a ref update across either + // transport. It remains nullable for rows from older peers and + // manually-triggered syncs, which must not report a fabricated lag. + "ALTER TABLE sync_queue ADD COLUMN IF NOT EXISTS origin_timestamp TEXT", + ], + }, ]; /// Max distinct source repos recorded per pinned object (F1, #173 jatmn round 8). @@ -3532,6 +3542,7 @@ pub struct SyncQueueItem { pub ref_name: String, pub new_sha: String, pub cid: Option, + pub origin_timestamp: Option, pub status: String, pub enqueued_at: String, } @@ -3544,10 +3555,11 @@ impl Db { ref_name: &str, new_sha: &str, cid: Option<&str>, + origin_timestamp: Option<&str>, ) -> Result<()> { sqlx::query( - "INSERT INTO sync_queue (id, repo, node_did, ref_name, new_sha, cid, status, enqueued_at) - VALUES ($1, $2, $3, $4, $5, $6, 'pending', $7) + "INSERT INTO sync_queue (id, repo, node_did, ref_name, new_sha, cid, origin_timestamp, status, enqueued_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, 'pending', $8) ON CONFLICT DO NOTHING", ) .bind(Uuid::new_v4().to_string()) @@ -3556,6 +3568,7 @@ impl Db { .bind(ref_name) .bind(new_sha) .bind(cid) + .bind(origin_timestamp) .bind(Utc::now().to_rfc3339()) .execute(&self.pool) .await?; @@ -3598,7 +3611,7 @@ impl Db { SELECT id FROM sync_queue WHERE status = 'pending' ORDER BY COALESCE(attempted_at, enqueued_at) ASC LIMIT $1 ) - RETURNING id, repo, node_did, ref_name, new_sha, cid, status, enqueued_at", + RETURNING id, repo, node_did, ref_name, new_sha, cid, origin_timestamp, status, enqueued_at", ) .bind(limit) .bind(Utc::now().to_rfc3339()) @@ -3613,6 +3626,7 @@ impl Db { ref_name: r.get("ref_name"), new_sha: r.get("new_sha"), cid: r.get("cid"), + origin_timestamp: r.get("origin_timestamp"), status: r.get("status"), enqueued_at: r.get("enqueued_at"), }) @@ -5392,6 +5406,25 @@ impl Db { /// this once and processes each row by re-deriving the push event, /// the per-ref cert, and the anchor handoff. #[allow(dead_code)] // wired by the handler refactor in the next slice + /// The instant this request's refs actually moved: the latest `applied_at` + /// stamp across its applied transitions, or `None` when none are applied + /// yet. Shared definition of "push landed" for the live replication tail, + /// which must match the deferred drain's `max(applied_at) ?? created_at` + /// rule so live and deferred pushes report one lag origin. + pub async fn request_push_landed_at(&self, request_id: &str) -> Result> { + let row = sqlx::query_scalar::<_, Option>( + "SELECT MAX(applied_at) FROM pending_ref_transitions WHERE request_id = $1 AND state = 'applied'", + ) + .bind(request_id) + .fetch_one(&self.pool) + .await?; + Ok(row) + } + + /// Test-only: the drain callers that used this moved to the request-scoped + /// `request_push_landed_at` and count helpers; kept for the transition-state + /// tests below. + #[cfg(test)] pub async fn list_pending_ref_transitions_applied( &self, limit: i64, @@ -10454,7 +10487,7 @@ mod migration_tests { db.migrate().await.unwrap(); } - // ── sync_queue scheduling (attempted_at, v17) ──────────────────────────── + // ── sync_queue scheduling and origin timestamps (v17/v18) ─────────────── async fn enqueue_one(db: &super::Db, repo: &str) { db.enqueue_sync( @@ -10463,6 +10496,7 @@ mod migration_tests { "refs/heads/main", &"0".repeat(40), None, + None, ) .await .unwrap(); @@ -10532,6 +10566,72 @@ mod migration_tests { db.migrate().await.unwrap(); } + /// Upgrade-path test for the nullable source timestamp. The worker must + /// preserve it when it exists, while rows written before v18 remain valid. + #[sqlx::test] + async fn migration_v47_adds_sync_queue_origin_timestamp(pool: sqlx::PgPool) { + let db = super::Db::for_testing(pool); + db.migrate().await.unwrap(); + + sqlx::query("ALTER TABLE sync_queue DROP COLUMN origin_timestamp") + .execute(&db.pool) + .await + .unwrap(); + sqlx::query("DELETE FROM schema_migrations WHERE version = 47") + .execute(&db.pool) + .await + .unwrap(); + + sqlx::query( + "INSERT INTO sync_queue (id, repo, node_did, ref_name, new_sha, status, enqueued_at) + VALUES ('legacy-sync', 'z6Mkfoo/legacy', 'did:key:zPEER', 'refs/heads/main', $1, 'pending', $2)", + ) + .bind("0".repeat(40)) + .bind("2026-08-16T00:00:00Z") + .execute(&db.pool) + .await + .unwrap(); + + db.migrate().await.unwrap(); + + let col: (String, String) = sqlx::query_as( + "SELECT data_type, is_nullable + FROM information_schema.columns + WHERE table_name = 'sync_queue' AND column_name = 'origin_timestamp'", + ) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(col.0, "text"); + assert_eq!(col.1, "YES", "origin_timestamp must be nullable"); + + let items = db.dequeue_pending_syncs(10).await.unwrap(); + assert_eq!(items.len(), 1); + assert_eq!(items[0].origin_timestamp, None); + db.migrate().await.unwrap(); + } + + #[sqlx::test] + async fn dequeue_preserves_sync_origin_timestamp(pool: sqlx::PgPool) { + let db = super::Db::for_testing(pool); + db.migrate().await.unwrap(); + let timestamp = "2026-08-16T00:00:00Z"; + db.enqueue_sync( + "z6Mkfoo/timestamped", + "did:key:zPEER", + "refs/heads/main", + &"0".repeat(40), + None, + Some(timestamp), + ) + .await + .unwrap(); + + let items = db.dequeue_pending_syncs(10).await.unwrap(); + assert_eq!(items.len(), 1); + assert_eq!(items[0].origin_timestamp.as_deref(), Some(timestamp)); + } + #[sqlx::test] async fn migration_v27_drops_raw_assignee_index(pool: sqlx::PgPool) { let db = super::Db::for_testing(pool); @@ -13128,6 +13228,24 @@ mod icaptcha_quarantine_tests { ); } + /// The sync-lag gate in `sync.rs` reads this flag rather than the + /// attempt-local one, so the preserved flag is the exact contract the + /// gate's "withheld mirrors never sample" rule depends on. + #[sqlx::test] + async fn resync_preserved_quarantine_is_visible_to_the_lag_gate(pool: PgPool) { + let db = db(pool).await; + db.upsert_mirror_repo("z6owner", "garbage", "/srv/g", None, true) + .await + .unwrap(); + db.upsert_mirror_repo("z6owner", "garbage", "/srv/g", None, false) + .await + .unwrap(); + assert!( + db.is_repo_quarantined("z6owner/garbage").await.unwrap(), + "the lag gate reads is_repo_quarantined: a re-sync that passes quarantined=false must still read withheld here" + ); + } + /// Quarantined repos are withheld from the deduped listing surfaces. #[sqlx::test] async fn listings_exclude_quarantined(pool: PgPool) { @@ -16609,4 +16727,48 @@ mod policy_fence_tests { "the unfenced sentinel skips the epoch comparison" ); } + /// The live replication tail reads its land instant through the same rule + /// the deferred drain uses: max applied_at across the request's applied + /// transitions, or None so the caller falls back to created_at. + #[sqlx::test] + async fn request_push_landed_at_returns_max_applied_or_none(pool: PgPool) { + let db = db(pool).await; + let repo_id = seed_repo(&db).await; + db.insert_pending_ref_transitions( + "req-live", + &repo_id, + "did:key:zNode", + "did:key:zPusher", + &[ + crate::api::repos::RefUpdate { + ref_name: "refs/heads/main".into(), + old_sha: "0".repeat(40), + new_sha: "a".repeat(40), + }, + crate::api::repos::RefUpdate { + ref_name: "refs/heads/dev".into(), + old_sha: "0".repeat(40), + new_sha: "b".repeat(40), + }, + ], + "sig", + "sigin", + "digest", + ) + .await + .unwrap(); + // Nothing applied yet: the caller must fall back to created_at. + assert_eq!(db.request_push_landed_at("req-live").await.unwrap(), None); + + db.mark_pending_ref_transitions_applied("req-live") + .await + .unwrap(); + let landed = db.request_push_landed_at("req-live").await.unwrap(); + assert!( + landed.is_some(), + "an applied request must report a land instant" + ); + // A different request's rows never leak into this request's instant. + assert_eq!(db.request_push_landed_at("req-other").await.unwrap(), None); + } } diff --git a/crates/gitlawb-node/src/durable_outbox.rs b/crates/gitlawb-node/src/durable_outbox.rs index 0abb45064..2d6093573 100644 --- a/crates/gitlawb-node/src/durable_outbox.rs +++ b/crates/gitlawb-node/src/durable_outbox.rs @@ -1711,6 +1711,15 @@ fn spawn_deferred_tail_after_proof( } let repo_id = req.repo_id.clone(); let pusher_did = req.pusher_did.clone(); + // The truest landing instant available: the applied_at stamp on the + // accepted transitions (written when the refs actually moved), with the + // request row's created_at (push acceptance) as the fallback. Either is + // truer than this deferred drain's Utc::now(). + let push_landed_at = accepted + .iter() + .filter_map(|t| t.applied_at.clone()) + .max() + .unwrap_or_else(|| req.created_at.clone()); let landed: Vec<(String, String, String)> = accepted .iter() .map(|c| (c.ref_name.clone(), c.old_sha.clone(), c.new_sha.clone())) @@ -1751,6 +1760,7 @@ fn spawn_deferred_tail_after_proof( ref_updates, std::path::PathBuf::from(repo.disk_path.clone()), pusher_did, + Some(push_landed_at), None, ) .await; diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index 34468d04b..cf9111714 100644 --- a/crates/gitlawb-node/src/metrics.rs +++ b/crates/gitlawb-node/src/metrics.rs @@ -5,6 +5,8 @@ //! //! * is push traffic flowing? — `gitlawb_pushes_total` //! * is fetch traffic flowing? — `gitlawb_fetches_total` +//! * how long do successful push and fetch operations take? — +//! `gitlawb_push_duration_seconds` / `gitlawb_fetch_duration_seconds` //! * are signature checks passing or failing? — //! `gitlawb_auth_successes_total` / `gitlawb_auth_failures_total` //! * is the sync worker making progress? — @@ -18,6 +20,8 @@ //! `gitlawb_gossip_replay_guard_saturated_total` //! * how big are the packs we're sending and receiving? — //! `gitlawb_pack_size_bytes` +//! * how long after an origin lands a push is it visible on this mirror? — +//! `gitlawb_sync_lag_seconds` //! * a single `gitlawb_info{version, did}` gauge = 1, for joins/dashboards //! * currently-connected peer count — `gitlawb_peers_connected` //! * reconciliation sweep gaps found and filled — @@ -34,11 +38,10 @@ //! free. //! //! Follow-ups (not in this module): -//! * per-route latency histograms (TraceLayer already gives us spans) //! * per-peer p2p counters //! * ipfs / pinata counters -use std::sync::OnceLock; +use std::{sync::OnceLock, time::Duration}; use prometheus::{ Encoder, Histogram, HistogramOpts, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts, @@ -60,6 +63,9 @@ static WEBHOOK_DELIVERIES: OnceLock = OnceLock::new(); static GOSSIP_INGEST: OnceLock = OnceLock::new(); static GOSSIP_REPLAY_GUARD_SATURATED: OnceLock = OnceLock::new(); static PACK_SIZE: OnceLock = OnceLock::new(); +static PUSH_DURATION: OnceLock = OnceLock::new(); +static FETCH_DURATION: OnceLock = OnceLock::new(); +static SYNC_LAG: OnceLock = OnceLock::new(); static PEERS_CONNECTED: OnceLock = OnceLock::new(); static RECONCILIATION_GAPS_FOUND: OnceLock = OnceLock::new(); static RECONCILIATION_GAPS_FILLED: OnceLock = OnceLock::new(); @@ -125,6 +131,51 @@ fn init_inner(version: &str, node_did: &str) { .expect("register gitlawb_fetches_total"); FETCHES.set(fetches).expect("set FETCHES once"); + let push_duration = Histogram::with_opts( + HistogramOpts::new( + "gitlawb_push_duration_seconds", + "Wall time of a successful git push (receive-pack) request in seconds, from handler entry to its response, including claim-loser returns where another worker owns the effects", + ) + .buckets(operation_duration_buckets()), + ) + .expect("gitlawb_push_duration_seconds definition"); + registry + .register(Box::new(push_duration.clone())) + .expect("register gitlawb_push_duration_seconds"); + PUSH_DURATION + .set(push_duration) + .expect("set PUSH_DURATION once"); + + let fetch_duration = Histogram::with_opts( + HistogramOpts::new( + "gitlawb_fetch_duration_seconds", + "Duration of successful git fetch (upload-pack) completion requests in seconds", + ) + .buckets(operation_duration_buckets()), + ) + .expect("gitlawb_fetch_duration_seconds definition"); + registry + .register(Box::new(fetch_duration.clone())) + .expect("register gitlawb_fetch_duration_seconds"); + FETCH_DURATION + .set(fetch_duration) + .expect("set FETCH_DURATION once"); + + let sync_lag = Histogram::with_opts( + HistogramOpts::new( + "gitlawb_sync_lag_seconds", + "Lag from the origin push-land timestamp (lag_origin) to successful local mirror visibility in seconds", + ) + .buckets(vec![ + 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1_800.0, 3_600.0, + ]), + ) + .expect("gitlawb_sync_lag_seconds definition"); + registry + .register(Box::new(sync_lag.clone())) + .expect("register gitlawb_sync_lag_seconds"); + SYNC_LAG.set(sync_lag).expect("set SYNC_LAG once"); + let auth_successes = IntCounterVec::new( Opts::new( "gitlawb_auth_successes_total", @@ -302,6 +353,35 @@ pub fn record_fetch(repo: &str) { } } +/// Observe the elapsed time of a successful git push handler. +pub fn observe_push_duration(duration: Duration) { + observe_duration(&PUSH_DURATION, duration); +} + +/// Observe the elapsed time of a successful git fetch completion request. +pub fn observe_fetch_duration(duration: Duration) { + observe_duration(&FETCH_DURATION, duration); +} + +/// Observe end-to-end lag until an origin ref update is visible on this mirror. +pub fn observe_sync_lag(duration: Duration) { + observe_duration(&SYNC_LAG, duration); +} + +fn observe_duration(histogram: &OnceLock, duration: Duration) { + if let Some(h) = histogram.get() { + h.observe(duration.as_secs_f64()); + } +} + +/// Test-only: current `gitlawb_sync_lag_seconds` sample count (0 if the +/// registry is not initialized). Lets the sync lag-gate tests assert that a +/// skipped observation really never reached the histogram. +#[cfg(test)] +pub fn sync_lag_sample_count_for_test() -> u64 { + SYNC_LAG.get().map(|h| h.get_sample_count()).unwrap_or(0) +} + /// Test-only: current `gitlawb_fetches_total` value for `repo` (0 if the registry /// is not initialized). Lets api-layer tests assert the completed-fetch count with /// a unique label instead of scraping the encoded text. @@ -420,6 +500,12 @@ pub fn observe_pack_size(bytes: f64) { } } +fn operation_duration_buckets() -> Vec { + vec![ + 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, + ] +} + /// Update the currently-connected peer count gauge. pub fn set_peers_connected(count: i64) { if let Some(g) = PEERS_CONNECTED.get() { @@ -485,6 +571,11 @@ mod tests { .expect("PUSHES set after init") .with_label_values(&["alice/repo"]) .inc(); + observe_push_duration(Duration::from_millis(42)); + observe_fetch_duration(Duration::from_millis(24)); + // `observe_sync_lag` is deliberately NOT called. The sync lag-gate + // test asserts a before/after delta on the process-wide histogram, so + // a sample from this test running concurrently would flake it. record_reconciliation_gaps_found(7); record_reconciliation_gaps_filled(3); @@ -501,6 +592,16 @@ mod tests { body.contains("gitlawb_pushes_total{repo=\"alice/repo\"} 1"), "expected the incremented counter to be visible in: {body}" ); + for metric in [ + "gitlawb_push_duration_seconds", + "gitlawb_fetch_duration_seconds", + "gitlawb_sync_lag_seconds", + ] { + assert!( + body.contains(&format!("# TYPE {metric} histogram")), + "expected {metric} histogram in: {body}" + ); + } assert!( body.contains("gitlawb_reconciliation_gaps_found_total 7"), "expected the reconciliation gaps-found counter to be visible in: {body}" @@ -543,9 +644,13 @@ mod tests { // here. It is a label-less process-wide counter and the seen-set test // that asserts it asserts a before/after delta, so an increment from a // second test running concurrently in this binary would be a flake with - // no diagnostic value. + // no diagnostic value. `observe_sync_lag` is withheld for the same + // reason: the sync lag-gate test asserts a sample-count delta on the + // process-wide histogram. record_gossip_ingest("accepted"); observe_pack_size(1024.0); + observe_push_duration(Duration::from_secs(1)); + observe_fetch_duration(Duration::from_secs(1)); set_peers_connected(0); } diff --git a/crates/gitlawb-node/src/p2p/mod.rs b/crates/gitlawb-node/src/p2p/mod.rs index cba91970e..f1753ba75 100644 --- a/crates/gitlawb-node/src/p2p/mod.rs +++ b/crates/gitlawb-node/src/p2p/mod.rs @@ -306,8 +306,9 @@ impl IngestLimiters { /// /// 0 is the versionless form: the field set this struct shipped with, before /// `v` existed. It is not a placeholder for "unset", it is a real version whose -/// wire encoding happens to omit the key. -pub(crate) const CURRENT_REF_UPDATE_VERSION: u32 = 0; +/// wire encoding happens to omit the key. 1 adds `lag_origin`, the first +/// version-gated field: it is emitted and verified only under v1. +pub(crate) const CURRENT_REF_UPDATE_VERSION: u32 = 1; /// True for the version whose key is omitted from the wire form. Free function /// because `skip_serializing_if` takes a path, not a closure. @@ -376,8 +377,15 @@ pub struct RefUpdateEvent { pub old_sha: String, /// SHA after the push pub new_sha: String, - /// RFC-3339 timestamp + /// RFC-3339 timestamp of the publish, freshness-checked at ingest pub timestamp: String, + /// RFC-3339 instant the origin push landed, for sync-lag measurement only. + /// Kept out of `timestamp` so `check_freshness` still sees the publish + /// instant: a push whose durability wait ran long would otherwise sign an + /// already-stale event. Optional for older peers; ingest falls back to + /// `timestamp` when it is absent. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lag_origin: Option, /// Certificate ID (from the ref certificate, if issued) pub cert_id: Option, /// IPFS CID of the latest commit object (set after pinning completes) @@ -1192,6 +1200,14 @@ pub(crate) async fn ingest_ref_update( event.v, CURRENT_REF_UPDATE_VERSION )); } + // `lag_origin` is a v1 field: a v0 event carrying it is a protocol + // violation, not a newer peer's wire form, and refusing it keeps the field + // set inside each version's signing bytes exactly what that version means. + if event.v < 1 && event.lag_origin.is_some() { + return IngestOutcome::Rejected( + "v0 ref-update event must not carry the v1 field lag_origin".to_string(), + ); + } // The cheapest denial under enforcement, ahead of every gate that reads a // field: an unsigned event is refused before its `node_did` is resolved or @@ -1498,6 +1514,12 @@ pub(crate) async fn ingest_ref_update( &event.ref_name, &event.new_sha, event.cid.as_deref(), + Some( + event + .lag_origin + .as_deref() + .unwrap_or(event.timestamp.as_str()), + ), ) .await { @@ -7193,7 +7215,7 @@ mod tests { #[test] fn ref_update_event_round_trip_with_owner_did() { let event = RefUpdateEvent { - v: 0, + v: CURRENT_REF_UPDATE_VERSION, node_did: "did:key:zNode".into(), pusher_did: "did:key:zPusher".into(), repo: "zOwner/myrepo".into(), @@ -7202,6 +7224,7 @@ mod tests { old_sha: "0000000000000000000000000000000000000000".into(), new_sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(), timestamp: "2026-07-02T12:00:00Z".into(), + lag_origin: Some("2026-07-02T11:59:30Z".into()), cert_id: None, cid: None, sig: None, @@ -7213,6 +7236,7 @@ mod tests { let deserialized: RefUpdateEvent = serde_json::from_value(json).unwrap(); assert_eq!(deserialized.owner_did, Some("did:key:zOwner".into())); + assert_eq!(deserialized.lag_origin, Some("2026-07-02T11:59:30Z".into())); } #[test] @@ -7255,7 +7279,7 @@ mod tests { /// field is Some so the serialized form exercises the widest field set. fn populated_event() -> RefUpdateEvent { RefUpdateEvent { - v: 0, + v: CURRENT_REF_UPDATE_VERSION, node_did: "did:key:zNode".into(), pusher_did: "did:key:zPusher".into(), repo: "zOwner/myrepo".into(), @@ -7264,6 +7288,7 @@ mod tests { old_sha: "0000000000000000000000000000000000000000".into(), new_sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(), timestamp: "2026-07-02T12:00:00Z".into(), + lag_origin: Some("2026-07-02T11:59:30Z".into()), cert_id: Some("cert-1".into()), cid: Some("bafycid".into()), sig: None, @@ -7318,30 +7343,53 @@ mod tests { cid: Option, } - let mut event = populated_event(); - event.sig = Some("c2lnbmF0dXJl".into()); - let bytes = serde_json::to_vec(&event).unwrap(); + // The fleet rule this version split is built around: + // + // * A v0 event is the pre-change wire form: no `v` key, no `lag_origin`, + // and it must still parse under the OLD struct verbatim. + // * A v1 event is for nodes that bumped with us. An un-upgraded peer + // rejects it on the `v` key itself, which is the LOUD version drop the + // ingest ceiling was designed to name, not a silent signature + // mismatch against bytes it could never compute. + let mut v0_event = populated_event(); + v0_event.v = 0; + v0_event.lag_origin = None; + v0_event.sig = Some("c2lnbmF0dXJl".into()); + let bytes = serde_json::to_vec(&v0_event).unwrap(); let legacy: LegacyRefUpdateEvent = serde_json::from_slice(&bytes) - .expect("new-code bytes must deserialize under the old field set"); + .expect("v0 bytes must deserialize under the old field set"); assert_eq!(legacy.repo, "zOwner/myrepo"); assert_eq!(legacy.owner_did, Some("did:key:zOwner".into())); - // An UNSIGNED event carries no `sig` key at all, so it is byte-identical + // An UNSIGNED v0 event carries no `sig` key at all, so it is byte-identical // in shape to the pre-change wire form and parses even under the strict // reader. This is what `skip_serializing_if` buys; drop it and a // `"sig": null` key appears here and this goes red. - let unsigned_bytes = serde_json::to_vec(&populated_event()).unwrap(); + v0_event.sig = None; + let unsigned_bytes = serde_json::to_vec(&v0_event).unwrap(); let strict: StrictLegacyRefUpdateEvent = serde_json::from_slice(&unsigned_bytes) - .expect("an unsigned event must carry no field the pre-change reader did not know"); + .expect("an unsigned v0 event must carry no field the pre-change reader did not know"); assert_eq!(strict.repo, "zOwner/myrepo"); - // A SIGNED event does carry the new key, and that is intentional, not a - // compatibility bug: the permissive reader above is what makes it - // harmless. Pinning the refusal here documents `sig` as the one added - // field, so a SECOND addition cannot slip in unnoticed. - serde_json::from_slice::(&bytes) - .expect_err("a signed event must be visibly carrying the added `sig` key"); + // A SIGNED v0 event does carry the added `sig` key, and that is + // intentional, not a compatibility bug: the permissive reader above is + // what makes it harmless. Pinning the refusal here documents `sig` as + // an added field, so a SECOND unsigned-field addition cannot slip in + // unnoticed. + v0_event.sig = Some("c2lnbmF0dXJl".into()); + serde_json::from_slice::( + &serde_json::to_vec(&v0_event).unwrap(), + ) + .expect_err("a signed v0 event must be visibly carrying the added `sig` key"); + + // A v1 event is refused by the pre-change reader on the `v` key: + // refusing to parse IS the version drop old nodes perform, the same + // failure the ingest ceiling produces in the other direction. + serde_json::from_slice::( + &serde_json::to_vec(&populated_event()).unwrap(), + ) + .expect_err("a v1 event must fail loudly on the `v` key for a pre-change reader"); } #[test] @@ -7398,12 +7446,13 @@ mod tests { /// this failure, and the emit-to-ingest round trip is structurally blind to /// it because both sides re-serialize the same new declaration order. const GOLDEN_SIGNING_BYTES: &str = concat!( - r#"{"node_did":"did:key:zNode","pusher_did":"did:key:zPusher","#, + r#"{"v":1,"node_did":"did:key:zNode","pusher_did":"did:key:zPusher","#, r#""repo":"zOwner/myrepo","owner_did":"did:key:zOwner","#, r#""ref_name":"refs/heads/main","#, r#""old_sha":"0000000000000000000000000000000000000000","#, r#""new_sha":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","#, - r#""timestamp":"2026-07-02T12:00:00Z","cert_id":"cert-1","cid":"bafycid"}"#, + r#""timestamp":"2026-07-02T12:00:00Z","lag_origin":"2026-07-02T11:59:30Z","#, + r#""cert_id":"cert-1","cid":"bafycid"}"#, ); #[test] @@ -7448,9 +7497,15 @@ mod tests { /// The all-`None` optional shape: what the production publish site emits. fn all_none_optionals_event() -> RefUpdateEvent { RefUpdateEvent { + // v0 pins the versionless form: the key stays off the wire and the + // signing bytes are byte-identical to what a pre-version build emits. + v: 0, owner_did: None, cert_id: None, cid: None, + // None skips the key entirely, so the production-shape golden stays + // byte-identical even though lag_origin exists. + lag_origin: None, ..populated_event() } } @@ -7675,8 +7730,11 @@ mod tests { let keypair = Keypair::generate(); // Signed at v0, where the key is absent from the signing bytes. - // Raising it on the received copy makes the key appear. + // Raising it on the received copy makes the key appear. The v0 shape + // drops lag_origin too, since v0 may not carry the v1 field. let mut at_zero = event_for(&keypair); + at_zero.v = 0; + at_zero.lag_origin = None; sign_ref_update(&keypair, &mut at_zero).unwrap(); verify_ref_update(&at_zero).expect("baseline must verify before tampering"); let mut raised = at_zero.clone(); @@ -7981,7 +8039,14 @@ mod tests { &bytes_of(&event), "v1 event", Some(( - "unsupported ref-update event version 1; this build understands version 0", + Box::leak( + format!( + "unsupported ref-update event version {}; this build understands version {}", + CURRENT_REF_UPDATE_VERSION + 1, + CURRENT_REF_UPDATE_VERSION + ) + .into_boxed_str(), + ), "an unknown version must be named as one, never reported as a signature failure", )), ) diff --git a/crates/gitlawb-node/src/sync.rs b/crates/gitlawb-node/src/sync.rs index 20a8f0b59..69c9522a6 100644 --- a/crates/gitlawb-node/src/sync.rs +++ b/crates/gitlawb-node/src/sync.rs @@ -16,7 +16,9 @@ use std::collections::HashMap; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; +use chrono::{DateTime, Utc}; use gitlawb_core::identity::Keypair; use tracing::{error, info, warn}; @@ -49,6 +51,23 @@ enum PromisorProbe { Unknown, } +/// The result of reading a ref's tip from a local mirror, three-valued so a +/// failed read is not mistaken for a confirmed-absent ref. `rev-parse` exits +/// non-zero for both "ref missing" and "read failed", and classifying only on +/// success/failure lets a transient error satisfy the deletion arm of the lag +/// observation gate, and the mirror would then report a deletion it never +/// proved. +#[derive(Debug, Clone, PartialEq, Eq)] +enum TipProbe { + /// `rev-parse --verify` resolved the ref to this sha. + Tip(String), + /// The read completed and the ref definitively does not resolve. + Absent, + /// The read itself failed (spawn error, timeout, or an exit status that is + /// neither "resolved" nor "absent"); nothing about the ref was proven. + Unknown, +} + /// Decide the mirror mode from the origin's `withheld-paths` response plus, when /// the response is unknown, the existing mirror's on-disk promisor state. /// @@ -180,6 +199,27 @@ fn resolve_origin_url(peers: &[crate::db::PeerRecord], node_did: &str) -> Option .map(|p| p.http_url.trim_end_matches('/').to_string()) } +/// Convert a sender-supplied RFC-3339 ref-update timestamp into an observable +/// lag. Missing, malformed, and future timestamps are deliberately ignored: +/// older peers and clock-skewed peers must not create fabricated observations. +const MAX_SYNC_LAG_OBSERVATION: Duration = Duration::from_secs(86_400); + +fn sync_lag_from_origin_timestamp( + origin_timestamp: Option<&str>, + visible_at: DateTime, +) -> Option { + let origin = DateTime::parse_from_rfc3339(origin_timestamp?).ok()?; + let lag = visible_at + .signed_duration_since(origin.with_timezone(&Utc)) + .to_std() + .ok()?; + // The origin timestamp is unauthenticated input from a known-but-not-proofed + // peer: an absurd age (a 1970 date, a stale clock) would park a fabricated + // observation in the histogram's tail buckets, so past a day the value is + // ignored rather than sampled. + (lag <= MAX_SYNC_LAG_OBSERVATION).then_some(lag) +} + async fn process_batch( db: &Db, config: &Config, @@ -396,7 +436,7 @@ async fn process_batch( false }; // Register in DB so git smart HTTP can serve the mirrored repo - let _ = db + let mirror_visible = match db .upsert_mirror_repo( owner_short, repo_name, @@ -404,6 +444,23 @@ async fn process_batch( machine_id, quarantined, ) + .await + { + Ok(_) => true, + Err(e) => { + warn!(repo = %item.repo, err = %e, "failed to register mirror repo"); + false + } + }; + // The lag is push-to-visible: sample at registration, before + // the encrypted-blob tail and mark_sync_done. A quarantined + // mirror is admitted but withheld from readers, so it reports + // no visibility lag. The gate reads the PERSISTED flag, not the + // attempt-local one: on a re-sync `quarantined` is false while + // `upsert_mirror_repo` preserves an existing quarantine, and + // sampling then would record visibility the readers still deny. + let withheld = db.is_repo_quarantined(&item.repo).await.unwrap_or(true); + observe_sync_lag_if_outcome_visible(&local_path, &item, mirror_visible, withheld) .await; // Option B2: carry the encrypted withheld-blob envelopes too, so an // authorized reader can recover private content from this mirror if @@ -649,6 +706,77 @@ async fn git_run(args: &[&str]) -> anyhow::Result<()> { Ok(()) } +/// Observe `gitlawb_sync_lag_seconds` for one dequeued row, but only when the +/// mirror provably shows the queued outcome. If the origin advanced the same +/// ref past this row before the fetch ran, the mirror carries the newer tip +/// and this row's visibility was never real; a deletion confirms only when the +/// ref is proven absent. A failed tip read proves nothing in either direction, +/// so it never samples: on a delete-sentinel row a read error must not satisfy +/// the absence arm the way a bare `Option` once allowed. +async fn observe_sync_lag_if_outcome_visible( + local_path: &Path, + item: &crate::db::SyncQueueItem, + mirror_visible: bool, + withheld: bool, +) { + if !mirror_visible || withheld { + return; + } + let outcome_visible = match mirror_ref_tip(local_path, &item.ref_name).await { + TipProbe::Tip(sha) => sha == item.new_sha, + // `new_sha` is the all-zero sentinel on a delete row: the queued + // outcome is visible only once the mirror proves the ref is gone. + TipProbe::Absent => item.new_sha.bytes().all(|b| b == b'0'), + TipProbe::Unknown => false, + }; + if !outcome_visible { + return; + } + if let Some(lag) = sync_lag_from_origin_timestamp(item.origin_timestamp.as_deref(), Utc::now()) + { + crate::metrics::observe_sync_lag(lag); + } +} + +/// Resolve `ref_name` inside `repo` as a three-valued [`TipProbe`] for the lag +/// observation gate: the sha it points to, a confirmed-absent ref, or a read +/// that proved nothing. `rev-parse --verify --quiet` exits 0 with the sha on +/// stdout when the ref resolves, 1 when it definitively does not, and any +/// other status (128 on a path that is not a repo, a signal, a usage error) +/// means the read failed. A spawn error or the 10s timeout firing is also +/// [`TipProbe::Unknown`]. +async fn mirror_ref_tip(repo: &std::path::Path, ref_name: &str) -> TipProbe { + // allow-unbounded-git: bounded by the tokio::time::timeout wrapping this + // read below, plus kill_on_drop; rev-parse is a metadata lookup, not a + // transfer, so the 10s cap is generous. + let mut cmd = tokio::process::Command::new("git"); + cmd.kill_on_drop(true); + let out = match tokio::time::timeout( + std::time::Duration::from_secs(10), + cmd.args(["-C"]) + .arg(repo) + .args(["rev-parse", "--verify", "--quiet", ref_name]) + .output(), + ) + .await + { + Ok(Ok(out)) => out, + _ => return TipProbe::Unknown, + }; + match out.status.code() { + Some(0) => { + let sha = String::from_utf8_lossy(&out.stdout).trim().to_string(); + if sha.is_empty() { + TipProbe::Unknown + } else { + TipProbe::Tip(sha) + } + } + Some(1) => TipProbe::Absent, + _ => TipProbe::Unknown, + } +} + /// Run a git subprocess, ignoring a non-zero exit. Used for idempotent /// `config --unset`, which exits non-zero when the key is already absent. async fn git_run_lenient(args: &[&str]) { @@ -1582,11 +1710,152 @@ mod tests { } async fn enqueue(db: &Db, repo: &str, did: &str) { - db.enqueue_sync(repo, did, "refs/heads/main", &"0".repeat(40), None) + db.enqueue_sync(repo, did, "refs/heads/main", &"0".repeat(40), None, None) .await .unwrap(); } + #[test] + fn sync_lag_uses_origin_timestamp_only_when_it_is_in_the_past() { + let visible_at = chrono::DateTime::parse_from_rfc3339("2026-08-16T00:00:12Z") + .unwrap() + .with_timezone(&chrono::Utc); + + assert_eq!( + sync_lag_from_origin_timestamp(Some("2026-08-16T00:00:00Z"), visible_at), + Some(Duration::from_secs(12)) + ); + assert_eq!(sync_lag_from_origin_timestamp(None, visible_at), None); + assert_eq!( + sync_lag_from_origin_timestamp(Some("not-a-timestamp"), visible_at), + None + ); + assert_eq!( + sync_lag_from_origin_timestamp(Some("2026-08-16T00:00:13Z"), visible_at), + None + ); + } + + fn lag_item( + ref_name: &str, + new_sha: &str, + origin_timestamp: Option<&str>, + ) -> crate::db::SyncQueueItem { + crate::db::SyncQueueItem { + id: "row".to_string(), + repo: "o/r".to_string(), + node_did: "did:key:x".to_string(), + ref_name: ref_name.to_string(), + new_sha: new_sha.to_string(), + cid: None, + origin_timestamp: origin_timestamp.map(str::to_string), + status: "pending".to_string(), + enqueued_at: String::new(), + } + } + + /// The tip probe is three-valued: a resolved ref, a confirmed-absent ref + /// (git exits 1 under `--verify --quiet`), and a failed read (exit 128 on a + /// path that is not a repo) must not collapse into each other. + #[tokio::test] + async fn tip_probe_reports_tip_absent_and_unknown_distinctly() { + let (td, _url) = bare_remote(&[("a.txt", b"hi\n")]); + let repo = td.path().join("bare.git"); + let expected = { + let out = Command::new("git") + .args(["-C"]) + .arg(&repo) + .args(["rev-parse", "HEAD"]) + .output() + .unwrap(); + String::from_utf8_lossy(&out.stdout).trim().to_string() + }; + + assert_eq!(mirror_ref_tip(&repo, "HEAD").await, TipProbe::Tip(expected)); + assert_eq!( + mirror_ref_tip(&repo, "refs/heads/never-pushed").await, + TipProbe::Absent + ); + // A failed read proves nothing about the ref; it is not absent. + assert_eq!( + mirror_ref_tip( + Path::new("/nonexistent/gitlawb-tip-probe-xyz"), + "refs/heads/main" + ) + .await, + TipProbe::Unknown + ); + } + + /// Regression for the #580 review finding: a zero `new_sha` is the delete + /// sentinel, and a FAILED tip read must not satisfy the deletion arm. The + /// mirror never proved the ref is gone, so no `gitlawb_sync_lag_seconds` + /// sample may be emitted. `git -C` on a path that is not a repo exits 128 + /// (Unknown); under the old `tip.is_none()` gate that read satisfied the + /// arm and sampled a deletion nobody proved. + #[tokio::test] + async fn lag_gate_skips_the_sample_when_the_tip_read_fails() { + crate::metrics::init("0.0.0-test", "did:key:test"); + let ts = (Utc::now() - Duration::from_secs(60)).to_rfc3339(); + let item = lag_item("refs/heads/main", &"0".repeat(40), Some(&ts)); + + let before = crate::metrics::sync_lag_sample_count_for_test(); + observe_sync_lag_if_outcome_visible( + Path::new("/nonexistent/gitlawb-tip-probe-xyz"), + &item, + true, + false, + ) + .await; + assert_eq!( + crate::metrics::sync_lag_sample_count_for_test(), + before, + "a failed tip read must not emit a lag sample for a delete row" + ); + } + + /// The live controls around the skip: a confirmed-absent ref on a delete + /// row does sample (the skip above is not vacuously green), while a + /// still-resolving ref and a hidden or unregistered mirror do not. + #[tokio::test] + async fn lag_gate_samples_only_when_the_queued_outcome_is_proven() { + crate::metrics::init("0.0.0-test", "did:key:test"); + let (td, _url) = bare_remote(&[("a.txt", b"hi\n")]); + let repo = td.path().join("bare.git"); + let ts = (Utc::now() - Duration::from_secs(60)).to_rfc3339(); + let mut count = crate::metrics::sync_lag_sample_count_for_test(); + + // Confirmed absent on a delete-sentinel row: samples. + let gone = lag_item("refs/heads/never-pushed", &"0".repeat(40), Some(&ts)); + observe_sync_lag_if_outcome_visible(&repo, &gone, true, false).await; + count += 1; + assert_eq!(crate::metrics::sync_lag_sample_count_for_test(), count); + + // The ref still resolves, so the mirror carries a different state than + // the queued deletion: no sample. + let still_there = lag_item("HEAD", &"0".repeat(40), Some(&ts)); + observe_sync_lag_if_outcome_visible(&repo, &still_there, true, false).await; + assert_eq!(crate::metrics::sync_lag_sample_count_for_test(), count); + + // A non-delete row samples only when the tip equals the queued sha. + let tip = match mirror_ref_tip(&repo, "HEAD").await { + TipProbe::Tip(sha) => sha, + other => panic!("expected a resolved tip, got {other:?}"), + }; + let landed = lag_item("HEAD", &tip, Some(&ts)); + observe_sync_lag_if_outcome_visible(&repo, &landed, true, false).await; + count += 1; + assert_eq!(crate::metrics::sync_lag_sample_count_for_test(), count); + let wrong_sha = lag_item("HEAD", &"f".repeat(40), Some(&ts)); + observe_sync_lag_if_outcome_visible(&repo, &wrong_sha, true, false).await; + assert_eq!(crate::metrics::sync_lag_sample_count_for_test(), count); + + // A quarantined or unregistered mirror never samples. + observe_sync_lag_if_outcome_visible(&repo, &gone, false, false).await; + observe_sync_lag_if_outcome_visible(&repo, &gone, true, true).await; + assert_eq!(crate::metrics::sync_lag_sample_count_for_test(), count); + } + fn dir_entries(dir: &Path) -> Vec { let mut names: Vec = std::fs::read_dir(dir) .unwrap() @@ -2808,6 +3077,7 @@ mod tests { cid: None, status: "pending".to_string(), enqueued_at: chrono::Utc::now().to_rfc3339(), + origin_timestamp: None, }; async fn status_of(pool: &sqlx::PgPool, id: &str) -> String { let row: (String,) = sqlx::query_as("SELECT status FROM sync_queue WHERE id = $1") diff --git a/crates/gitlawb-node/tests/inv22_gates.rs b/crates/gitlawb-node/tests/inv22_gates.rs index 8fd6e48ef..a8bd889d3 100644 --- a/crates/gitlawb-node/tests/inv22_gates.rs +++ b/crates/gitlawb-node/tests/inv22_gates.rs @@ -517,7 +517,7 @@ fn inv22_ipfs_walk_admission_reaches_every_blocking_site() { /// drives the actual disconnect through a parked `release`. Same instrument the F3 /// gate above uses. /// -/// MUTATION (RED): move the `tokio::spawn(post_receive_replication_tail` call below +/// MUTATION (RED): move the `post_receive_replication_tail(` call below /// `guard.release(` and the ordering assertion fails; take it out of the /// `if disposition.spawn_tail` block and the failed-push assertion fails; gate /// `spawn_tail`/`release_ok` on the outcome commit and @@ -542,17 +542,21 @@ fn inv22_replication_tail_spawns_at_the_durability_boundary() { // position (carve-out early, or durability-bound at release) must sit // inside a `disposition.spawn_tail` gate and feed `landed_refs` — // never the unfiltered `ref_updates`. - let spawn_count = production - .matches("tokio::spawn(post_receive_replication_tail(") - .count(); + // Match the tail CALL, not the `tokio::spawn(...)` wrapper: production + // currently wraps it in `tokio::spawn(async move { ... })`, and keying on + // the wrapper's exact shape is what let this gate go blind on the refactor. + // The `pub(crate) async fn` definition is the only other text occurrence, + // filtered by the `fn ` that must precede it. + let spawn_positions: Vec = production + .match_indices("post_receive_replication_tail(") + .map(|(p, _)| p) + .filter(|&p| !production[..p].trim_end().ends_with("fn")) + .collect(); assert!( - spawn_count >= 1, + !spawn_positions.is_empty(), "U5 gate missing: the replication tail must be spawned by git_receive_pack" ); - for (i, (pos, _)) in production - .match_indices("tokio::spawn(post_receive_replication_tail(") - .enumerate() - { + for (i, pos) in spawn_positions.iter().copied().enumerate() { let gate = production[..pos].rfind("if disposition.spawn_tail").expect( "U5 gate missing: every tail spawn must be gated on `if disposition.spawn_tail`", ); @@ -578,13 +582,24 @@ fn inv22_replication_tail_spawns_at_the_durability_boundary() { ); let call_end = production[pos..].find(");").unwrap() + pos; let call = &production[pos..call_end]; + // Match the ref-set ARGUMENT (third positional), not a substring of the + // call: `landed` is the in-scope alias for `landed_refs` in the + // carve-out spawn and either name is the proven-landed set, but a bare + // `call.contains("landed")` is also satisfied by the `push_landed_at` + // timestamp argument, which would let a call feeding unfiltered + // `ref_updates` pass. + let refset_arg = call + .split(',') + .nth(2) + .map(str::trim) + .expect("U5 gate stale: tail call no longer takes a third argument"); assert!( - call.contains("landed_refs"), + refset_arg == "landed" || refset_arg == "landed_refs", "U5 gate bypassed: tail #{i} must carry the proven-landed ref set, not unfiltered ref_updates" ); } - let spawn = production - .find("tokio::spawn(post_receive_replication_tail(") + let spawn = *spawn_positions + .first() .expect("U5 gate missing: the replication tail must be spawned by git_receive_pack"); let release = production .find(".release(disposition.release_ok)")