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/api/peers.rs

@@ -439,6 +439,7 @@ pub async fn trigger_sync(State(state): State<AppState>) -> Result<Json<serde_js
                     "refs/heads/main",
                     "0000000000000000000000000000000000000000",
                     None,
+                    None,
                 )
                 .await;
             enqueued += 1;
@@ -520,7 +521,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.timestamp.as_deref(),
+        )
         .await?;
 
     // Mirror the gossipsub-receive handler: insert the same record we'd
@@ -1282,6 +1290,37 @@ 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;
+        let timestamp = "2026-08-16T00:00:00Z";
+        let body = serde_json::json!({
+            "repo": "z6Mkfoo/timestamped",
+            "ref_name": "refs/heads/main",
+            "new_sha": "0000000000000000000000000000000000000000",
+            "node_did": peer_did,
+            "timestamp": timestamp,
+        })
+        .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(timestamp));
+    }
+
     #[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

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

@@ -1777,6 +1777,7 @@ pub async fn git_upload_pack(
     headers: axum::http::HeaderMap,
     body: Bytes,
 ) -> Result<Response> {
+    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,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) {
@@ -2248,6 +2251,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 +2266,7 @@ async fn notify_peer_of_refs(
             node_did,
             pusher_did,
             owner_did,
+            origin_timestamp,
         )
         .await;
     }
@@ -2280,6 +2285,7 @@ pub async fn git_receive_pack(
     headers: axum::http::HeaderMap,
     body: Bytes,
 ) -> Result<Response> {
+    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
@@ -3399,6 +3405,7 @@ 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
@@ -3698,6 +3705,11 @@ 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 /
@@ -3912,6 +3924,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
@@ -4036,7 +4049,7 @@ pub(crate) 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()),
                             sig: None,
@@ -4139,6 +4152,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 +8076,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)
@@ -8098,6 +8115,7 @@ mod tests {
             "did:key:zNode",
             "did:key:zPusher",
             "did:key:zOwner",
+            "2026-08-16T00:00:00Z",
         )
         .await;
 
@@ -8146,6 +8164,7 @@ mod tests {
             "did:key:zNode",
             "did:key:zPusher",
             "did:key:zOwner",
+            "2026-08-16T00:00:00Z",
         )
         .await;
 

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<String>,
+    pub origin_timestamp: Option<String>,
     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"),
             })
@@ -10454,7 +10468,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 +10477,7 @@ mod migration_tests {
             "refs/heads/main",
             &"0".repeat(40),
             None,
+            None,
         )
         .await
         .unwrap();
@@ -10532,6 +10547,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);
@@ -14072,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();

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 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`
 //!   * 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<IntCounterVec> = OnceLock::new();
 static GOSSIP_INGEST: OnceLock<IntCounterVec> = OnceLock::new();
 static GOSSIP_REPLAY_GUARD_SATURATED: OnceLock<IntCounter> = OnceLock::new();
 static PACK_SIZE: OnceLock<Histogram> = OnceLock::new();
+static PUSH_DURATION: OnceLock<Histogram> = OnceLock::new();
+static FETCH_DURATION: OnceLock<Histogram> = OnceLock::new();
+static SYNC_LAG: OnceLock<Histogram> = OnceLock::new();
 static PEERS_CONNECTED: OnceLock<IntGauge> = OnceLock::new();
 static RECONCILIATION_GAPS_FOUND: OnceLock<IntCounter> = OnceLock::new();
 static RECONCILIATION_GAPS_FILLED: OnceLock<IntCounter> = 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",
+            "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",
@@ -302,6 +353,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<Histogram>, 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.
@@ -420,6 +492,12 @@ pub fn observe_pack_size(bytes: f64) {
     }
 }
 
+fn operation_duration_buckets() -> Vec<f64> {
+    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 +563,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));
         record_reconciliation_gaps_found(7);
         record_reconciliation_gaps_filled(3);
 
@@ -501,6 +582,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}"
@@ -546,6 +637,9 @@ mod tests {
         // no diagnostic value.
         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);
     }
 

crates/gitlawb-node/src/p2p/mod.rs

@@ -1498,6 +1498,7 @@ pub(crate) async fn ingest_ref_update(
                 &event.ref_name,
                 &event.new_sha,
                 event.cid.as_deref(),
+                Some(event.timestamp.as_str()),
             )
             .await
         {

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};
 
@@ -180,6 +182,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<Utc>,
+) -> Option<Duration> {
+    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 +419,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,
@@ -404,7 +427,19 @@ async fn process_batch(
                         machine_id,
                         quarantined,
                     )
-                    .await;
+                    .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}",
@@ -1582,11 +1617,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<String> {
         let mut names: Vec<String> = std::fs::read_dir(dir)
             .unwrap()
@@ -2808,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")