From 9f4f5096198d0435ccfd8776b1b5e494eb6f7da6 Mon Sep 17 00:00:00 2001 From: anandh8x Date: Sun, 16 Aug 2026 20:21:57 +0530 Subject: [PATCH 01/12] feat: add push fetch and sync latency metrics --- crates/gitlawb-node/src/api/peers.rs | 41 +++++++++++- crates/gitlawb-node/src/api/repos.rs | 22 ++++++- crates/gitlawb-node/src/db/mod.rs | 89 +++++++++++++++++++++++-- crates/gitlawb-node/src/metrics.rs | 98 +++++++++++++++++++++++++++- crates/gitlawb-node/src/p2p/mod.rs | 1 + crates/gitlawb-node/src/sync.rs | 53 +++++++++++++-- 6 files changed, 291 insertions(+), 13 deletions(-) diff --git a/crates/gitlawb-node/src/api/peers.rs b/crates/gitlawb-node/src/api/peers.rs index 3934e7a36..5e224d8cf 100644 --- a/crates/gitlawb-node/src/api/peers.rs +++ b/crates/gitlawb-node/src/api/peers.rs @@ -365,6 +365,7 @@ pub async fn trigger_sync(State(state): State) -> Result 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. @@ -1500,6 +1501,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) @@ -1661,6 +1663,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, @@ -1669,7 +1672,7 @@ async fn notify_peer_of_ref( "node_did": node_did, "pusher_did": pusher_did, "old_sha": old_sha, - "timestamp": chrono::Utc::now().to_rfc3339(), + "timestamp": origin_timestamp, "owner_did": owner_did, }); let body_bytes = match serde_json::to_vec(&body) { @@ -1719,6 +1722,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( @@ -1733,6 +1737,7 @@ async fn notify_peer_of_refs( node_did, pusher_did, owner_did, + origin_timestamp, ) .await; } @@ -1747,6 +1752,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 @@ -2051,6 +2057,7 @@ pub async fn git_receive_pack( // Record the successful push for metrics. The body has already been // consumed by smart_http::receive_pack so we observe size up front. crate::metrics::record_push(&record.id); + crate::metrics::observe_push_duration(push_started.elapsed()); crate::metrics::observe_pack_size(body_len as f64); // Record push event for trust score and issue a signed ref certificate. @@ -2149,6 +2156,10 @@ async fn post_receive_replication_tail( disk_path: std::path::PathBuf, did: String, ) { + // Capture this before the asynchronous replication work below. The same + // origin time travels over gossip and HTTP so receiving peers can measure + // push-to-visible lag rather than notification-to-visible lag. + let origin_timestamp = Utc::now().to_rfc3339(); // Replication enforcement (Phase 2): decide once per push whether the public // may read this repo at all and, if so, which blob OIDs must not leave the // node. `withheld == None` means this push pins nothing (private / mode A / @@ -2362,6 +2373,7 @@ 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 @@ -2468,7 +2480,7 @@ async fn post_receive_replication_tail( ref_name: ref_name.clone(), old_sha: old_sha.clone(), new_sha: new_sha.clone(), - timestamp: chrono::Utc::now().to_rfc3339(), + timestamp: origin_timestamp_for_announce.clone(), cert_id: None, cid: cid.map(|s| s.to_string()), }) @@ -2570,6 +2582,7 @@ async fn post_receive_replication_tail( &node_did_str, &pusher_did_clone, &record.owner_did, + &origin_timestamp_for_announce, ) .await; } @@ -4254,6 +4267,9 @@ mod tests { mockito::Matcher::PartialJsonString( r#"{"owner_did":"did:key:zOwner"}"#.to_string(), ), + mockito::Matcher::PartialJsonString( + r#"{"timestamp":"2026-08-16T00:00:00Z"}"#.to_string(), + ), ])) .with_status(200) .expect(1) @@ -4290,6 +4306,7 @@ mod tests { "did:key:zNode", "did:key:zPusher", "did:key:zOwner", + "2026-08-16T00:00:00Z", ) .await; @@ -4338,6 +4355,7 @@ mod tests { "did:key:zNode", "did:key:zPusher", "did:key:zOwner", + "2026-08-16T00:00:00Z", ) .await; diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 50c3bddad..61a13b5fb 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -901,6 +901,16 @@ const MIGRATIONS: &[Migration] = &[ "ALTER TABLE sync_queue ADD COLUMN IF NOT EXISTS attempted_at TEXT", ], }, + Migration { + version: 18, + 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", + ], + }, ]; // ── Repos ───────────────────────────────────────────────────────────────────── @@ -1629,6 +1639,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, } @@ -1641,10 +1652,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()) @@ -1653,6 +1665,7 @@ impl Db { .bind(ref_name) .bind(new_sha) .bind(cid) + .bind(origin_timestamp) .bind(Utc::now().to_rfc3339()) .execute(&self.pool) .await?; @@ -1695,7 +1708,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()) @@ -1710,6 +1723,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"), }) @@ -3977,7 +3991,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( @@ -3986,6 +4000,7 @@ mod migration_tests { "refs/heads/main", &"0".repeat(40), None, + None, ) .await .unwrap(); @@ -4055,6 +4070,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_v18_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 = 18") + .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 dequeue_stamps_attempted_at_on_every_row_it_hands_out(pool: sqlx::PgPool) { // The stamp is what stops a deferred row from holding the window, and diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index c95ef1d18..403f84142 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? — @@ -13,6 +15,8 @@ //! `gitlawb_webhook_deliveries_total{result}` //! * how big are the packs we're sending and receiving? — //! `gitlawb_pack_size_bytes` +//! * how long until an origin ref update is 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` //! @@ -26,11 +30,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, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry, @@ -50,6 +53,9 @@ static AUTH_FAILURES: OnceLock = OnceLock::new(); static SYNC_PROCESSED: OnceLock = OnceLock::new(); static WEBHOOK_DELIVERIES: 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(); /// One-time initializer. Builds the registry, registers every metric, @@ -108,6 +114,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", + "Duration of successful git push (receive-pack) handlers in seconds", + ) + .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 an origin ref-update timestamp 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", @@ -223,6 +274,27 @@ 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_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. @@ -277,6 +349,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() { @@ -321,6 +399,9 @@ 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(Duration::from_secs(12)); let body = encode().expect("encode should succeed after init"); assert!( @@ -335,6 +416,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}" + ); + } } /// #192 F4: `init` is idempotent and safe to call repeatedly. The panic that @@ -366,6 +457,9 @@ mod tests { record_sync_processed("done"); record_webhook_delivery("ok"); observe_pack_size(1024.0); + observe_push_duration(Duration::from_secs(1)); + observe_fetch_duration(Duration::from_secs(1)); + observe_sync_lag(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 80e28a4aa..4e45ff1ce 100644 --- a/crates/gitlawb-node/src/p2p/mod.rs +++ b/crates/gitlawb-node/src/p2p/mod.rs @@ -329,6 +329,7 @@ pub async fn start( &event.ref_name, &event.new_sha, event.cid.as_deref(), + Some(event.timestamp.as_str()), ).await; } } diff --git a/crates/gitlawb-node/src/sync.rs b/crates/gitlawb-node/src/sync.rs index 0ed4a9f9a..97fb18dcb 100644 --- a/crates/gitlawb-node/src/sync.rs +++ b/crates/gitlawb-node/src/sync.rs @@ -15,7 +15,9 @@ use std::collections::HashMap; use std::path::Path; use std::sync::Arc; +use std::time::Duration; +use chrono::{DateTime, Utc}; use gitlawb_core::identity::Keypair; use tracing::{error, info, warn}; @@ -179,6 +181,20 @@ 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. +fn sync_lag_from_origin_timestamp( + origin_timestamp: Option<&str>, + visible_at: DateTime, +) -> Option { + let origin = DateTime::parse_from_rfc3339(origin_timestamp?).ok()?; + visible_at + .signed_duration_since(origin.with_timezone(&Utc)) + .to_std() + .ok() +} + async fn process_batch( db: &Db, config: &Config, @@ -395,7 +411,7 @@ async fn process_batch( false }; // Register in DB so git smart HTTP can serve the mirrored repo - let _ = db + let mirror_visible = db .upsert_mirror_repo( owner_short, repo_name, @@ -403,7 +419,8 @@ async fn process_batch( machine_id, quarantined, ) - .await; + .await + .is_ok(); // Option B2: carry the encrypted withheld-blob envelopes too, so an // authorized reader can recover private content from this mirror if // the origin dies. `item.repo` is the slug "{owner_short}/{name}", @@ -418,8 +435,15 @@ async fn process_batch( &config.ipfs_api, ) .await; - let _ = db.mark_sync_done(&item.id).await; + let marked_done = db.mark_sync_done(&item.id).await; crate::metrics::record_sync_processed("done"); + if mirror_visible && marked_done.is_ok() { + if let Some(lag) = + sync_lag_from_origin_timestamp(item.origin_timestamp.as_deref(), Utc::now()) + { + crate::metrics::observe_sync_lag(lag); + } + } // Tell the origin we now host a replica so its replica_count // reflects reality. Best-effort: idempotent on the origin and @@ -1194,11 +1218,32 @@ 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 dir_entries(dir: &Path) -> Vec { let mut names: Vec = std::fs::read_dir(dir) .unwrap() From 2f99ad85558f751aee55f78dd869872cc821b25d Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Tue, 10 Nov 2026 22:44:01 -0600 Subject: [PATCH 02/12] Merge remote-tracking branch 'origin/main' into pr-362 --- crates/gitlawb-node/src/db/mod.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index a5764037e..0ee6e63e7 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -2370,6 +2370,8 @@ const MIGRATIONS: &[Migration] = &[ stmts: &[ "ALTER TABLE webhook_deliveries ADD COLUMN IF NOT EXISTS sent_at TEXT", "CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_pending ON webhook_deliveries (sent_at, created_at)", + ], + }, Migration { version: 47, name: "sync_queue_origin_timestamp", @@ -2380,9 +2382,6 @@ const MIGRATIONS: &[Migration] = &[ "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). From 966b2b8e0abe48fb9817001e297c4e28957e6928 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Tue, 10 Nov 2026 22:52:34 -0600 Subject: [PATCH 03/12] fix(node): sample sync lag at mirror visibility, clamp untrusted origin ages - the lag observation moves to the upsert point, before the encrypted-blob tail and mark_sync_done, so the metric reads push-to-visible rather than push-to-fully-replicated - a quarantined mirror is admitted but withheld from readers and now reports no visibility lag - MAX_SYNC_LAG_OBSERVATION (24h) discards absurd sender-supplied origin ages that would park fabricated values in the tail buckets; the timestamp is unauthenticated peer input - the migration lands at v47 after main's slot moves; the test initializer gains origin_timestamp --- crates/gitlawb-node/src/db/mod.rs | 6 +++--- crates/gitlawb-node/src/sync.rs | 32 +++++++++++++++++++++---------- 2 files changed, 25 insertions(+), 13 deletions(-) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 0ee6e63e7..000c2b989 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -10550,7 +10550,7 @@ mod migration_tests { /// 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_v18_adds_sync_queue_origin_timestamp(pool: sqlx::PgPool) { + async fn migration_v47_adds_sync_queue_origin_timestamp(pool: sqlx::PgPool) { let db = super::Db::for_testing(pool); db.migrate().await.unwrap(); @@ -10558,7 +10558,7 @@ mod migration_tests { .execute(&db.pool) .await .unwrap(); - sqlx::query("DELETE FROM schema_migrations WHERE version = 18") + sqlx::query("DELETE FROM schema_migrations WHERE version = 47") .execute(&db.pool) .await .unwrap(); @@ -14153,7 +14153,7 @@ mod ref_certificate_tests { .execute(&pool) .await .unwrap(); - sqlx::query("DELETE FROM schema_migrations WHERE version = 18") + sqlx::query("DELETE FROM schema_migrations WHERE version = 47") .execute(&pool) .await .unwrap(); diff --git a/crates/gitlawb-node/src/sync.rs b/crates/gitlawb-node/src/sync.rs index 6d7510860..c5e29f6dc 100644 --- a/crates/gitlawb-node/src/sync.rs +++ b/crates/gitlawb-node/src/sync.rs @@ -185,15 +185,22 @@ fn resolve_origin_url(peers: &[crate::db::PeerRecord], node_did: &str) -> Option /// 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()?; - visible_at + let lag = visible_at .signed_duration_since(origin.with_timezone(&Utc)) .to_std() - .ok() + .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( @@ -422,6 +429,17 @@ async fn process_batch( ) .await .is_ok(); + // 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. + if mirror_visible && !quarantined { + if let Some(lag) = + sync_lag_from_origin_timestamp(item.origin_timestamp.as_deref(), Utc::now()) + { + crate::metrics::observe_sync_lag(lag); + } + } // Option B2: carry the encrypted withheld-blob envelopes too, so an // authorized reader can recover private content from this mirror if // the origin dies. `item.repo` is the slug "{owner_short}/{name}", @@ -436,15 +454,8 @@ async fn process_batch( &config.ipfs_api, ) .await; - let marked_done = db.mark_sync_done(&item.id).await; + let _ = db.mark_sync_done(&item.id).await; crate::metrics::record_sync_processed("done"); - if mirror_visible && marked_done.is_ok() { - if let Some(lag) = - sync_lag_from_origin_timestamp(item.origin_timestamp.as_deref(), Utc::now()) - { - crate::metrics::observe_sync_lag(lag); - } - } // Tell the origin we now host a replica so its replica_count // reflects reality. Best-effort: idempotent on the origin and @@ -2853,6 +2864,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") From d5d9bbb5b3d5705ea5a12d108e4c3016c478a098 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 11:58:11 -0600 Subject: [PATCH 04/12] fix(node): split sync-lag origin from gossip publish freshness The single origin_timestamp captured after the durability wait was reused as the signed RefUpdateEvent timestamp, so a tail that published more than 600s later minted an already-stale event that verified ingest drops in check_freshness, silently disabling gossip auto-sync on mirrors. - Capture the lag origin at push landing, before the durability wait, and carry it on a new optional signed field RefUpdateEvent::lag_origin (serde default, skipped when absent, so events from older peers still parse and the null-optional signing shape is byte-identical). - Sign the publish instant in event.timestamp again, matching merge-base, so check_freshness sees the time the event actually went on the wire. - Ingest now feeds enqueue_sync lag_origin with a timestamp fallback, so sync-lag still measures push-to-visible instead of notify-to-visible. - Re-pin the all-Some GOLDEN_SIGNING_BYTES for the new field; the all-None production golden is unchanged. Round-trip test asserts the field survives serde. Also restore the v18 pinned-cid upgrade oracle: the test's rollback delete targeted schema_migrations version 47 instead of 18, so v18 never re-applied and the index_exists assertion failed on stable and beta. --- crates/gitlawb-node/src/api/repos.rs | 15 +++++++++------ crates/gitlawb-node/src/db/mod.rs | 2 +- crates/gitlawb-node/src/p2p/mod.rs | 25 ++++++++++++++++++++++--- 3 files changed, 32 insertions(+), 10 deletions(-) diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index 2593a8e67..58f172e5f 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -3697,6 +3697,13 @@ 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 timestamp (HTTP) — never as the signed event's + // own `timestamp`, which must stay at publish time for check_freshness. + let origin_timestamp = Utc::now().to_rfc3339(); + if !publish_durability_confirmed(&publish_durability, durability_wait).await { tracing::warn!( repo = %record.id, @@ -3705,11 +3712,6 @@ pub(crate) async fn post_receive_replication_tail( return; } - // Capture this before the asynchronous replication work below. The same - // origin time travels over gossip and HTTP so receiving peers can measure - // push-to-visible lag rather than notification-to-visible lag. - let origin_timestamp = Utc::now().to_rfc3339(); - // Replication enforcement (Phase 2): decide once per push whether the public // may read this repo at all and, if so, which blob OIDs must not leave the // node. `withheld == None` means this push pins nothing (private / mode A / @@ -4049,7 +4051,8 @@ pub(crate) async fn post_receive_replication_tail( ref_name: ref_name.clone(), old_sha: old_sha.clone(), new_sha: new_sha.clone(), - timestamp: origin_timestamp_for_announce.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, diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 000c2b989..fd7fb59bc 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -14153,7 +14153,7 @@ mod ref_certificate_tests { .execute(&pool) .await .unwrap(); - sqlx::query("DELETE FROM schema_migrations WHERE version = 47") + sqlx::query("DELETE FROM schema_migrations WHERE version = 18") .execute(&pool) .await .unwrap(); diff --git a/crates/gitlawb-node/src/p2p/mod.rs b/crates/gitlawb-node/src/p2p/mod.rs index e132b0f2e..bdfcc0b3f 100644 --- a/crates/gitlawb-node/src/p2p/mod.rs +++ b/crates/gitlawb-node/src/p2p/mod.rs @@ -376,8 +376,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) @@ -1498,7 +1505,12 @@ pub(crate) async fn ingest_ref_update( &event.ref_name, &event.new_sha, event.cid.as_deref(), - Some(event.timestamp.as_str()), + Some( + event + .lag_origin + .as_deref() + .unwrap_or(event.timestamp.as_str()), + ), ) .await { @@ -7203,6 +7215,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, @@ -7214,6 +7227,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] @@ -7265,6 +7279,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, @@ -7404,7 +7419,8 @@ mod tests { 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] @@ -7452,6 +7468,9 @@ mod tests { 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() } } From bfb491e0d8330074231d6a4f69c5a082aaf6bea4 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:30:44 -0600 Subject: [PATCH 05/12] fix(node): version-gate lag_origin, read persisted quarantine, keep push-land instant - lag_origin ships under v=1: CURRENT_REF_UPDATE_VERSION bumped, ingest rejects a v0 event carrying the v1 field, goldens re-pinned for the v1 publish shape, and the legacy-parse test now pins the fleet rule (v0 still parses, v1 refuses loudly on the v key) - the sync-lag observation gate reads is_repo_quarantined instead of the attempt-local flag, so a re-sync of a quarantined mirror no longer samples visibility readers still deny; fail-closed on lookup - post_receive_replication_tail takes push_landed_at so a deferred drain reports the request row's landing instant rather than its own drain time; live callers pass the acceptance instant, tests pass None --- crates/gitlawb-node/src/api/repos.rs | 44 +++++++++++- crates/gitlawb-node/src/db/mod.rs | 18 +++++ crates/gitlawb-node/src/durable_outbox.rs | 5 ++ crates/gitlawb-node/src/p2p/mod.rs | 85 +++++++++++++++++------ crates/gitlawb-node/src/sync.rs | 8 ++- 5 files changed, 137 insertions(+), 23 deletions(-) diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index 58f172e5f..0ff533552 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -3321,6 +3321,7 @@ pub async fn git_receive_pack( landed_refs.clone(), disk_path.clone(), auth.0.to_string(), + Some(Utc::now().to_rfc3339()), None, )); } @@ -3349,6 +3350,7 @@ pub async fn git_receive_pack( landed_refs, disk_path.clone(), auth.0.to_string(), + Some(Utc::now().to_rfc3339()), Some(publish_durability.arc()), )); } @@ -3681,12 +3683,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>>, >, @@ -3702,7 +3716,7 @@ pub(crate) async fn post_receive_replication_tail( // measures. Travels to receiving peers as RefUpdateEvent::lag_origin // (gossip) and the notify timestamp (HTTP) — never as the signed event's // own `timestamp`, which must stay at publish time for check_freshness. - let origin_timestamp = Utc::now().to_rfc3339(); + let origin_timestamp = lag_origin_timestamp(push_landed_at); if !publish_durability_confirmed(&publish_durability, durability_wait).await { tracing::warn!( @@ -13984,6 +13998,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(), @@ -14016,6 +14048,7 @@ mod tests { f2a_update("refs/heads/main", &c1), repo.path().to_path_buf(), F2A_PUSHER.to_string(), + None, Some(slot), ) .await; @@ -14062,6 +14095,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; let after_first = f2a_walks(&log); @@ -14083,6 +14117,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -14159,6 +14194,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -14241,6 +14277,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; @@ -14358,6 +14395,7 @@ mod tests { repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -15509,6 +15547,7 @@ exit 0 repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -15565,6 +15604,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; @@ -15607,6 +15647,7 @@ exit 0 repo.path().to_path_buf(), F2A_PUSHER.to_string(), None, + None, ) .await; @@ -15706,6 +15747,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 fd7fb59bc..1c3989b25 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -13209,6 +13209,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) { diff --git a/crates/gitlawb-node/src/durable_outbox.rs b/crates/gitlawb-node/src/durable_outbox.rs index 0abb45064..17389094c 100644 --- a/crates/gitlawb-node/src/durable_outbox.rs +++ b/crates/gitlawb-node/src/durable_outbox.rs @@ -1711,6 +1711,10 @@ fn spawn_deferred_tail_after_proof( } let repo_id = req.repo_id.clone(); let pusher_did = req.pusher_did.clone(); + // The request row was created when the push was accepted, so its + // created_at IS the push-land instant the lag origin should carry, + // not this deferred drain's Utc::now(). + let push_landed_at = 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 +1755,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/p2p/mod.rs b/crates/gitlawb-node/src/p2p/mod.rs index bdfcc0b3f..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. @@ -1199,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 @@ -7206,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(), @@ -7270,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(), @@ -7334,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] @@ -7414,7 +7446,7 @@ 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","#, @@ -7465,6 +7497,9 @@ 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, @@ -7695,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(); @@ -8001,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 c5e29f6dc..77189dba0 100644 --- a/crates/gitlawb-node/src/sync.rs +++ b/crates/gitlawb-node/src/sync.rs @@ -432,8 +432,12 @@ async fn process_batch( // 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. - if mirror_visible && !quarantined { + // 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); + if mirror_visible && !withheld { if let Some(lag) = sync_lag_from_origin_timestamp(item.origin_timestamp.as_deref(), Utc::now()) { From a42aab77e7f3d3773ab9b82f5739ffee9068d8e0 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 16:19:56 -0600 Subject: [PATCH 06/12] fix(node): tighten the two new lag-observation edges - deferred tails source push_landed_at from the accepted transitions' applied_at (the instant refs actually moved), falling back to the request row's created_at rather than drain time - the visibility gate confirms the mirror's ref tip matches the queued outcome before sampling, so a row superseded by a newer origin update (or a deletion the mirror absorbed) never records lag --- crates/gitlawb-node/src/durable_outbox.rs | 13 +++++--- crates/gitlawb-node/src/sync.rs | 39 ++++++++++++++++++++++- 2 files changed, 47 insertions(+), 5 deletions(-) diff --git a/crates/gitlawb-node/src/durable_outbox.rs b/crates/gitlawb-node/src/durable_outbox.rs index 17389094c..2d6093573 100644 --- a/crates/gitlawb-node/src/durable_outbox.rs +++ b/crates/gitlawb-node/src/durable_outbox.rs @@ -1711,10 +1711,15 @@ fn spawn_deferred_tail_after_proof( } let repo_id = req.repo_id.clone(); let pusher_did = req.pusher_did.clone(); - // The request row was created when the push was accepted, so its - // created_at IS the push-land instant the lag origin should carry, - // not this deferred drain's Utc::now(). - let push_landed_at = req.created_at.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())) diff --git a/crates/gitlawb-node/src/sync.rs b/crates/gitlawb-node/src/sync.rs index 77189dba0..7f4b39bae 100644 --- a/crates/gitlawb-node/src/sync.rs +++ b/crates/gitlawb-node/src/sync.rs @@ -437,7 +437,18 @@ async fn process_batch( // `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); - if mirror_visible && !withheld { + // Only sample when the queued outcome is what the mirror now + // shows: 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. Deletions confirm when + // the ref is absent. + let tip = mirror_ref_tip(&local_path, &item.ref_name).await; + let outcome_visible = if item.new_sha.bytes().all(|b| b == b'0') { + tip.is_none() + } else { + tip.as_deref() == Some(item.new_sha.as_str()) + }; + if mirror_visible && !withheld && outcome_visible { if let Some(lag) = sync_lag_from_origin_timestamp(item.origin_timestamp.as_deref(), Utc::now()) { @@ -688,6 +699,32 @@ async fn git_run(args: &[&str]) -> anyhow::Result<()> { Ok(()) } +/// The sha `ref_name` resolves to inside `repo`, or `None` when the ref is +/// absent or the read fails. Used by the lag observation gate to confirm the +/// queued outcome is what the mirror actually carries before sampling. +async fn mirror_ref_tip(repo: &std::path::Path, ref_name: &str) -> Option { + // 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 = tokio::time::timeout( + std::time::Duration::from_secs(10), + cmd.args(["-C"]) + .arg(repo) + .args(["rev-parse", "--verify", ref_name]) + .output(), + ) + .await + .ok()? + .ok()?; + if !out.status.success() { + return None; + } + let sha = String::from_utf8_lossy(&out.stdout).trim().to_string(); + (!sha.is_empty()).then_some(sha) +} + /// 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]) { From ea0b0571cc7198c50a01eba1e737c80fc3e88236 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 16:55:56 -0600 Subject: [PATCH 07/12] fix(node): split notify publish/land clocks and unify live push-land instant - HTTP sync/notify now stamps timestamp with send time (publish semantics, matching what gossip ingest stores on the feed row) and carries the land instant in a new optional lag_origin field; the receiver plumbs lag_origin to sync_queue.origin_timestamp with a timestamp fallback for older senders - both live replication-tail spawns pass the request row's acceptance instant (the same clock the deferred drain's created_at fallback uses) instead of minting a fresh clock at spawn time --- crates/gitlawb-node/src/api/peers.rs | 26 ++++++++++++++++++++++++-- crates/gitlawb-node/src/api/repos.rs | 16 +++++++++++++--- 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/crates/gitlawb-node/src/api/peers.rs b/crates/gitlawb-node/src/api/peers.rs index 06ccc4800..1f0c73a90 100644 --- a/crates/gitlawb-node/src/api/peers.rs +++ b/crates/gitlawb-node/src/api/peers.rs @@ -475,6 +475,13 @@ pub struct NotifyRequest { pub old_sha: Option, #[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. @@ -527,7 +534,7 @@ pub async fn notify_sync( &req.ref_name, &req.new_sha, None, - req.timestamp.as_deref(), + req.lag_origin.as_deref().or(req.timestamp.as_deref()), ) .await?; @@ -1295,13 +1302,17 @@ mod tests { 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); @@ -1318,7 +1329,18 @@ mod tests { let queued = db.dequeue_pending_syncs(100).await.unwrap(); assert_eq!(queued.len(), 1); - assert_eq!(queued[0].origin_timestamp.as_deref(), Some(timestamp)); + 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] diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index 0ff533552..328c058d7 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -2201,7 +2201,13 @@ async fn notify_peer_of_ref( "node_did": node_did, "pusher_did": pusher_did, "old_sha": old_sha, - "timestamp": origin_timestamp, + // 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) { @@ -3321,7 +3327,10 @@ pub async fn git_receive_pack( landed_refs.clone(), disk_path.clone(), auth.0.to_string(), - Some(Utc::now().to_rfc3339()), + // The acceptance instant from the request row insert, matching the + // deferred drain's land definition — a fresh clock here would + // understate live push-to-mirror lag by the whole receive-pack run. + Some(now.clone()), None, )); } @@ -3350,7 +3359,8 @@ pub async fn git_receive_pack( landed_refs, disk_path.clone(), auth.0.to_string(), - Some(Utc::now().to_rfc3339()), + // Same acceptance instant as the carve-out spawn above. + Some(now.clone()), Some(publish_durability.arc()), )); } From 66d01f8a6faf2a51fdd5a65feee73f0810ecbc07 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 18:16:20 -0600 Subject: [PATCH 08/12] fix(node): use the deferred land-instant rule on live replication tails Both live post_receive_replication_tail spawns now resolve push_landed_at through db.request_push_landed_at: the latest applied_at across this request's applied transitions, falling back to the request row's created_at, the same rule the deferred drain uses. The pre-receive-pack acceptance stamp no longer flows into gossip lag_origin, the HTTP notify land field, or sync_queue.origin_timestamp on the live path, so live and deferred pushes report one lag origin. Regression: request_push_landed_at_returns_max_applied_or_none covers the max-applied arm and both fallback cases. --- crates/gitlawb-node/src/api/repos.rs | 29 ++++++++++--- crates/gitlawb-node/src/db/mod.rs | 63 ++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 6 deletions(-) diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index 328c058d7..3769d69cb 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -3327,10 +3327,18 @@ pub async fn git_receive_pack( landed_refs.clone(), disk_path.clone(), auth.0.to_string(), - // The acceptance instant from the request row insert, matching the - // deferred drain's land definition — a fresh clock here would - // understate live push-to-mirror lag by the whole receive-pack run. - Some(now.clone()), + // Same rule the deferred drain uses: the truest land instant is the + // latest applied_at on this request's landed transitions, with the + // request row's created_at (`now`) as the fallback. + Some( + state + .db + .request_push_landed_at(&request_id) + .await + .ok() + .flatten() + .unwrap_or_else(|| now.clone()), + ), None, )); } @@ -3359,8 +3367,17 @@ pub async fn git_receive_pack( landed_refs, disk_path.clone(), auth.0.to_string(), - // Same acceptance instant as the carve-out spawn above. - Some(now.clone()), + // Same max(applied_at) ?? created_at rule as the carve-out + // spawn and the deferred drain. + Some( + state + .db + .request_push_landed_at(&request_id) + .await + .ok() + .flatten() + .unwrap_or_else(|| now.clone()), + ), Some(publish_durability.arc()), )); } diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 1c3989b25..42358fca9 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -5406,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, @@ -16708,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); + } } From c564f8422b2bdf4b25f16e0d5b0f29d646c6863d Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 21:04:46 -0600 Subject: [PATCH 09/12] fix(node): align notify land surfaces and resolve land instant in-task - the multi-ref notify regression now matches lag_origin for the supplied land instant instead of timestamp, which is a live send instant after the publish/land split; the tail comment now names the notify lag_origin field rather than timestamp - both live replication-tail spawns resolve push_landed_at inside the spawned task so the handler no longer awaits the lookup while holding the reclaimed write lease or leaving the tail un-owned on disconnect --- crates/gitlawb-node/src/api/repos.rs | 108 ++++++++++++++++----------- 1 file changed, 66 insertions(+), 42 deletions(-) diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index 3769d69cb..f1ef71c18 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -3321,26 +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(), - // Same rule the deferred drain uses: the truest land instant is the - // latest applied_at on this request's landed transitions, with the - // request row's created_at (`now`) as the fallback. - Some( - state - .db - .request_push_landed_at(&request_id) - .await - .ok() - .flatten() - .unwrap_or_else(|| now.clone()), - ), - 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 @@ -3361,25 +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(), - // Same max(applied_at) ?? created_at rule as the carve-out - // spawn and the deferred drain. - Some( - state - .db - .request_push_landed_at(&request_id) - .await - .ok() - .flatten() - .unwrap_or_else(|| now.clone()), - ), - 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 @@ -3741,8 +3762,9 @@ pub(crate) async fn post_receive_replication_tail( // 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 timestamp (HTTP) — never as the signed event's - // own `timestamp`, which must stay at publish time for check_freshness. + // (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 { @@ -8120,8 +8142,10 @@ 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#"{"timestamp":"2026-08-16T00:00:00Z"}"#.to_string(), + r#"{"lag_origin":"2026-08-16T00:00:00Z"}"#.to_string(), ), ])) .with_status(200) From 2219c5d23f7cda7be549002793f29271ddb51a00 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Wed, 11 Nov 2026 23:59:46 -0600 Subject: [PATCH 10/12] test(node): realign the inv22 tail-spawn gate and metrics help text (#580) - inv22_replication_tail_spawns_at_the_durability_boundary matched the literal tokio::spawn(post_receive_replication_tail( shape, which went blind when production wrapped the tail in tokio::spawn(async move { ... }). The gate now matches the tail call itself (filtering the fn definition) while keeping the disposition.spawn_tail gate, brace depth, no-second-spawn, landed-set, and ordering assertions. - metrics help for gitlawb_push_duration_seconds described receive-pack subprocess time but observes the whole successful handler interval; gitlawb_sync_lag_seconds described the origin ref-update timestamp but samples lag_origin (the land instant). Both strings and the module bullet now name what is measured. --- crates/gitlawb-node/src/metrics.rs | 6 ++--- crates/gitlawb-node/tests/inv22_gates.rs | 30 ++++++++++++++---------- 2 files changed, 21 insertions(+), 15 deletions(-) diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index 12b34baad..63342cd18 100644 --- a/crates/gitlawb-node/src/metrics.rs +++ b/crates/gitlawb-node/src/metrics.rs @@ -20,7 +20,7 @@ //! `gitlawb_gossip_replay_guard_saturated_total` //! * how big are the packs we're sending and receiving? — //! `gitlawb_pack_size_bytes` -//! * how long until an origin ref update is visible on this mirror? — +//! * 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` @@ -134,7 +134,7 @@ fn init_inner(version: &str, node_did: &str) { let push_duration = Histogram::with_opts( HistogramOpts::new( "gitlawb_push_duration_seconds", - "Duration of successful git push (receive-pack) handlers in seconds", + "Wall time of a successful git push (receive-pack) request in seconds, from handler entry to the point its effects are recorded", ) .buckets(operation_duration_buckets()), ) @@ -164,7 +164,7 @@ fn init_inner(version: &str, node_did: &str) { let sync_lag = Histogram::with_opts( HistogramOpts::new( "gitlawb_sync_lag_seconds", - "Lag from an origin ref-update timestamp to successful local mirror visibility in 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, diff --git a/crates/gitlawb-node/tests/inv22_gates.rs b/crates/gitlawb-node/tests/inv22_gates.rs index af844a5ec..d5cf55b40 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,15 @@ fn inv22_replication_tail_spawns_at_the_durability_boundary() { ); let call_end = production[pos..].find(");").unwrap() + pos; let call = &production[pos..call_end]; + // `landed` is the in-scope alias for `landed_refs` in the carve-out + // spawn; either name is the proven-landed set, `ref_updates` is not. assert!( - call.contains("landed_refs"), + call.contains("landed"), "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)") From d826a05f488aeebb7d11501aedba33f2ba1b553e Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:21:56 -0600 Subject: [PATCH 11/12] fix(node): tri-state mirror tip probe, full-interval push duration (#580) - mirror_ref_tip returns a TipProbe (Tip/Absent/Unknown): a failed rev-parse is no longer a confirmed-absent ref, so a delete row only samples sync lag on proven absence (regression test included) - observe_push_duration now fires at the response paths (claim-loser and post-effects), so the sample covers the whole handler interval the client waits on rather than stopping before inline effects --- crates/gitlawb-node/src/api/repos.rs | 3 +- crates/gitlawb-node/src/metrics.rs | 17 +- crates/gitlawb-node/src/sync.rs | 226 +++++++++++++++++++++++---- 3 files changed, 212 insertions(+), 34 deletions(-) diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index f1ef71c18..0f09e8125 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -3455,7 +3455,6 @@ pub async fn git_receive_pack( // that case via `run_effects`, so this is defensive). let _ = state.db.touch_repo(&record.id).await; crate::metrics::record_push(&record.id); - crate::metrics::observe_push_duration(push_started.elapsed()); crate::metrics::observe_pack_size(body_len as f64); // Claim ownership before inline effects: a background worker may @@ -3468,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") @@ -3572,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") diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index 63342cd18..a94476774 100644 --- a/crates/gitlawb-node/src/metrics.rs +++ b/crates/gitlawb-node/src/metrics.rs @@ -374,6 +374,14 @@ fn observe_duration(histogram: &OnceLock, duration: Duration) { } } +/// 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. @@ -565,7 +573,9 @@ mod tests { .inc(); observe_push_duration(Duration::from_millis(42)); observe_fetch_duration(Duration::from_millis(24)); - observe_sync_lag(Duration::from_secs(12)); + // `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); @@ -634,12 +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)); - observe_sync_lag(Duration::from_secs(1)); set_peers_connected(0); } diff --git a/crates/gitlawb-node/src/sync.rs b/crates/gitlawb-node/src/sync.rs index 7f4b39bae..874ab936e 100644 --- a/crates/gitlawb-node/src/sync.rs +++ b/crates/gitlawb-node/src/sync.rs @@ -51,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. /// @@ -437,24 +454,8 @@ async fn process_batch( // `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); - // Only sample when the queued outcome is what the mirror now - // shows: 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. Deletions confirm when - // the ref is absent. - let tip = mirror_ref_tip(&local_path, &item.ref_name).await; - let outcome_visible = if item.new_sha.bytes().all(|b| b == b'0') { - tip.is_none() - } else { - tip.as_deref() == Some(item.new_sha.as_str()) - }; - if mirror_visible && !withheld && outcome_visible { - if let Some(lag) = - sync_lag_from_origin_timestamp(item.origin_timestamp.as_deref(), Utc::now()) - { - crate::metrics::observe_sync_lag(lag); - } - } + 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 // the origin dies. `item.repo` is the slug "{owner_short}/{name}", @@ -699,30 +700,75 @@ async fn git_run(args: &[&str]) -> anyhow::Result<()> { Ok(()) } -/// The sha `ref_name` resolves to inside `repo`, or `None` when the ref is -/// absent or the read fails. Used by the lag observation gate to confirm the -/// queued outcome is what the mirror actually carries before sampling. -async fn mirror_ref_tip(repo: &std::path::Path, ref_name: &str) -> Option { +/// 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 = tokio::time::timeout( + let out = match tokio::time::timeout( std::time::Duration::from_secs(10), cmd.args(["-C"]) .arg(repo) - .args(["rev-parse", "--verify", ref_name]) + .args(["rev-parse", "--verify", "--quiet", ref_name]) .output(), ) .await - .ok()? - .ok()?; - if !out.status.success() { - return None; + { + 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, } - let sha = String::from_utf8_lossy(&out.stdout).trim().to_string(); - (!sha.is_empty()).then_some(sha) } /// Run a git subprocess, ignoring a non-zero exit. Used for idempotent @@ -1684,6 +1730,126 @@ mod tests { ); } + 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() From f1b21ebb74bbe7223e485740ae414d2b6e635c8f Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:24:44 -0600 Subject: [PATCH 12/12] fix(node): tighten inv22 landed-ref gate, align push-duration help (#580) - inv22_gates U5: match the third call argument against landed/landed_refs instead of contains("landed"), which push_landed_at also satisfied and made the proven-landed check vacuous. - metrics: gitlawb_push_duration_seconds help now describes handler wall time through the response including claim-loser returns. - sync: log mirror-repo registration failures instead of silently discarding them. --- crates/gitlawb-node/src/metrics.rs | 2 +- crates/gitlawb-node/src/sync.rs | 10 ++++++++-- crates/gitlawb-node/tests/inv22_gates.rs | 15 ++++++++++++--- 3 files changed, 21 insertions(+), 6 deletions(-) diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index a94476774..cf9111714 100644 --- a/crates/gitlawb-node/src/metrics.rs +++ b/crates/gitlawb-node/src/metrics.rs @@ -134,7 +134,7 @@ fn init_inner(version: &str, node_did: &str) { 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 the point its effects are recorded", + "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()), ) diff --git a/crates/gitlawb-node/src/sync.rs b/crates/gitlawb-node/src/sync.rs index 874ab936e..69c9522a6 100644 --- a/crates/gitlawb-node/src/sync.rs +++ b/crates/gitlawb-node/src/sync.rs @@ -436,7 +436,7 @@ async fn process_batch( false }; // Register in DB so git smart HTTP can serve the mirrored repo - let mirror_visible = db + let mirror_visible = match db .upsert_mirror_repo( owner_short, repo_name, @@ -445,7 +445,13 @@ async fn process_batch( quarantined, ) .await - .is_ok(); + { + 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 diff --git a/crates/gitlawb-node/tests/inv22_gates.rs b/crates/gitlawb-node/tests/inv22_gates.rs index d5cf55b40..d7d5d847a 100644 --- a/crates/gitlawb-node/tests/inv22_gates.rs +++ b/crates/gitlawb-node/tests/inv22_gates.rs @@ -582,10 +582,19 @@ fn inv22_replication_tail_spawns_at_the_durability_boundary() { ); let call_end = production[pos..].find(");").unwrap() + pos; let call = &production[pos..call_end]; - // `landed` is the in-scope alias for `landed_refs` in the carve-out - // spawn; either name is the proven-landed set, `ref_updates` is not. + // 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"), + refset_arg == "landed" || refset_arg == "landed_refs", "U5 gate bypassed: tail #{i} must carry the proven-landed ref set, not unfiltered ref_updates" ); }