feat: add push, fetch, and sync latency metrics

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

@@ -365,6 +365,7 @@ pub async fn trigger_sync(State(state): State<AppState>) -> Result<Json<serde_js
                     "refs/heads/main",
                     "0000000000000000000000000000000000000000",
                     None,
+                    None,
                 )
                 .await;
             enqueued += 1;
@@ -446,7 +447,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
@@ -1162,6 +1170,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

@@ -1298,6 +1298,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.
@@ -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<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
@@ -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;
 

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<String>,
+    pub origin_timestamp: Option<String>,
     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