diff --git a/crates/gitlawb-node/src/api/peers.rs b/crates/gitlawb-node/src/api/peers.rs index 3934e7a3..5e224d8c 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 50c3bdda..61a13b5f 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 c95ef1d1..403f8414 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 80e28a4a..4e45ff1c 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 0ed4a9f9..97fb18dc 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()