From dc5def98c02076ac18e5bd09d7661546410fb7e1 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Tue, 20 Oct 2026 09:16:51 -0500 Subject: [PATCH] feat(metrics): export commit-time lag gauges Carry the commit record's xact_time with the emitter ack frontier and export it as walshadow_emitter_ack_commit_timestamp_seconds, plus the newest drained commit as walshadow_source_last_commit_timestamp_seconds. Both are Unix seconds, NaN until the first commit, so replication lag is a plain subtraction with no PromQL rate window. --- src/backfill/wal_replay.rs | 6 +-- src/bin/stream/metrics_publish.rs | 4 ++ src/bin/stream/session.rs | 2 + src/emit/pipeline/ack.rs | 75 ++++++++++++++++++++++++++----- src/emit/pipeline/bootstrap.rs | 4 +- src/emit/pipeline/reorder.rs | 6 +-- src/filter/engine.rs | 25 +++++++++++ src/ops/metrics.rs | 39 ++++++++++++++++ tests/emitter_budget_flush.rs | 2 +- tests/emitter_native_types.rs | 2 +- tests/emitter_tls.rs | 2 +- 11 files changed, 144 insertions(+), 23 deletions(-) diff --git a/src/backfill/wal_replay.rs b/src/backfill/wal_replay.rs index cea3e08a..b639f60a 100644 --- a/src/backfill/wal_replay.rs +++ b/src/backfill/wal_replay.rs @@ -453,9 +453,9 @@ impl WalReplaySink { let value_permit = detoast_heap(&mut heap, spool, &ref_maps, &self.resolver) .await .map_err(SinkError::from)?; - let (open, rows) = self - .open - .get_or_insert_with(|| (self.seqs.open(commit_lsn, Publish::Commit), 0)); + let (open, rows) = self.open.get_or_insert_with(|| { + (self.seqs.open(commit_lsn, None, Publish::Commit), 0) + }); *rows += 1; let seq = open.seq(); if !route.drops_deletes() diff --git a/src/bin/stream/metrics_publish.rs b/src/bin/stream/metrics_publish.rs index eb26a944..61fac480 100644 --- a/src/bin/stream/metrics_publish.rs +++ b/src/bin/stream/metrics_publish.rs @@ -260,6 +260,8 @@ pub(crate) async fn populate_metrics( shadow_replay_lsn: Pos, decoder_commit_lsn: Pos, emitter_ack_lsn: Pos, + ack_commit_ts: Option, + source_commit_ts: Option, rec_metrics: &MetricsRecordSink, pump_queue_depth: u64, queue_records_out_total: u64, @@ -284,6 +286,8 @@ pub(crate) async fn populate_metrics( shadow_replay_lsn, decoder_commit_lsn, emitter_ack_lsn, + emitter_ack_commit_timestamp_seconds: ack_commit_ts, + source_last_commit_timestamp_seconds: source_commit_ts, source_endpoint_swaps_total: source_swap.swaps, source_endpoint_swap_failures_total: source_swap.failures, source_endpoint_swap_pending: u64::from(source_swap.pending), diff --git a/src/bin/stream/session.rs b/src/bin/stream/session.rs index 94314485..695a75a7 100644 --- a/src/bin/stream/session.rs +++ b/src/bin/stream/session.rs @@ -2239,6 +2239,8 @@ async fn session_body( shadow_replay, drain_for_metric, emitter_ack_for_metric, + ack_probe.borrow().ack_commit_ts, + stream.filter().stats().last_commit_ts, &record_sink.metrics, record_sink.decoder_xact.in_flight(), record_sink.decoder_xact.processed(), diff --git a/src/emit/pipeline/ack.rs b/src/emit/pipeline/ack.rs index 03e52871..1c7c15de 100644 --- a/src/emit/pipeline/ack.rs +++ b/src/emit/pipeline/ack.rs @@ -39,6 +39,9 @@ pub enum AckEvent { Register { seq: u64, commit_lsn: u64, + /// Commit record's `xact_time` (PG epoch µs); `None` for aborts and + /// bootstrap seqs, which carry no source commit time + commit_ts: Option, publish: bool, }, Placed { @@ -99,6 +102,7 @@ impl Progress { struct Seq { commit_lsn: u64, + commit_ts: Option, publish: bool, progress: Progress, } @@ -119,6 +123,9 @@ pub struct AckSnapshot { pub frontier: u64, pub oldest_incomplete: Option, pub trailing_held: Pos, + /// `xact_time` (PG epoch µs) of the newest commit the watermark has + /// retired; `None` until the first one + pub ack_commit_ts: Option, /// Protocol faults; the first trips pipeline fatal pub wedged: u64, /// Events for retired seqs, benign: inserter batches straddle retirement @@ -146,6 +153,8 @@ pub struct AckState { registered: u64, /// Highest publishing `commit_lsn` retired from map watermark: u64, + /// Newest published commit time, advances with `watermark` + watermark_ts: Option, /// Retained across events because quiescent source cannot resend it trailing: u64, emitter_ack: Arc>, @@ -169,6 +178,7 @@ impl AckState { frontier: 0, registered: 0, watermark: 0, + watermark_ts: None, trailing: 0, emitter_ack, frontier_cell, @@ -189,8 +199,9 @@ impl AckState { AckEvent::Register { seq, commit_lsn, + commit_ts, publish, - } => self.open_seq(seq, commit_lsn, publish), + } => self.open_seq(seq, commit_lsn, commit_ts, publish), AckEvent::Placed { seq, rows } => self.step(seq, |p| p.place(rows)), AckEvent::Acked { counts } => { for (seq, n) in counts { @@ -206,7 +217,7 @@ impl AckState { } } - fn open_seq(&mut self, seq: u64, commit_lsn: u64, publish: bool) { + fn open_seq(&mut self, seq: u64, commit_lsn: u64, commit_ts: Option, publish: bool) { // Preserve live counts, never resurrect retired seqs if seq < self.frontier || self.map.contains_key(&seq) { self.wedge(seq, "seq registered twice"); @@ -216,6 +227,7 @@ impl AckState { seq, Seq { commit_lsn, + commit_ts, publish, progress: Progress::default(), }, @@ -264,10 +276,11 @@ impl AckState { if !s.progress.is_done() { break; } - let (publish, commit_lsn) = (s.publish, s.commit_lsn); + let (publish, commit_lsn, commit_ts) = (s.publish, s.commit_lsn, s.commit_ts); self.map.remove(&self.frontier); if publish { self.watermark = self.watermark.max(commit_lsn); + self.watermark_ts = self.watermark_ts.max(commit_ts); } self.frontier += 1; } @@ -291,6 +304,7 @@ impl AckState { frontier: self.frontier, oldest_incomplete: self.oldest_incomplete(), trailing_held: Pos::new(self.trailing), + ack_commit_ts: self.watermark_ts, wedged: self.wedged, late: self.late, } @@ -406,12 +420,15 @@ impl SeqAlloc { self.next } - pub fn open(&mut self, commit_lsn: u64, publish: Publish) -> OpenSeq { + /// `commit_ts` is the commit record's `xact_time`, which the ack frontier + /// carries once the seq publishes + pub fn open(&mut self, commit_lsn: u64, commit_ts: Option, publish: Publish) -> OpenSeq { let seq = self.next; self.next += 1; let ev = AckEvent::Register { seq, commit_lsn, + commit_ts, publish: publish == Publish::Commit, }; if let Some(held) = &mut self.held { @@ -423,8 +440,8 @@ impl SeqAlloc { } /// Ordering marker carrying no rows - pub fn mark(&mut self, commit_lsn: u64) { - let seq = self.open(commit_lsn, Publish::Commit); + pub fn mark(&mut self, commit_lsn: u64, commit_ts: Option) { + let seq = self.open(commit_lsn, commit_ts, Publish::Commit); self.place(seq, 0); } @@ -499,10 +516,11 @@ pub fn spawn(emitter_ack: Arc>, fatal: Fatal) -> (AckHandle #[cfg(test)] impl AckState { - fn register(&mut self, seq: u64, commit_lsn: u64, publish: bool) { + fn register(&mut self, seq: u64, commit_lsn: u64, commit_ts: Option, publish: bool) { self.apply(AckEvent::Register { seq, commit_lsn, + commit_ts, publish, }); } @@ -541,7 +559,7 @@ mod tests { let (mut s, ack) = state(); let n = 50_000u64; for seq in 0..n { - s.register(seq, (seq + 1) * 10, true); + s.register(seq, (seq + 1) * 10, None, true); s.placed(seq, 1); } assert_eq!(ack.get(), 0); @@ -556,7 +574,7 @@ mod tests { #[test] fn probe_names_what_holds_the_watermark() { let (mut s, _ack) = state(); - s.register(0, 100, true); + s.register(0, 100, None, true); let snap = s.snapshot(); assert_eq!(snap.stall_reason(), Some("rows not yet placed")); assert_eq!( @@ -580,10 +598,42 @@ mod tests { assert_eq!((snap.wedged, snap.late), (0, 0)); } + #[test] + fn commit_ts_advances_with_the_contiguous_frontier() { + let (mut s, ack) = state(); + assert_eq!(s.snapshot().ack_commit_ts, None); + s.register(0, 100, Some(1_000), true); + s.register(1, 200, Some(2_000), true); + s.placed(0, 1); + s.placed(1, 1); + // seq 1 done first: seq 0 gates both the LSN and its timestamp + s.acked(1, 1); + assert_eq!((ack.get().get(), s.snapshot().ack_commit_ts), (0, None)); + s.acked(0, 1); + assert_eq!( + (ack.get().get(), s.snapshot().ack_commit_ts), + (200, Some(2_000)) + ); + } + + #[test] + fn partial_slices_and_untimed_seqs_do_not_move_commit_ts() { + let (mut s, _ack) = state(); + s.register(0, 100, Some(1_000), true); + // non-final slice of a later commit, then an abort-style untimed seq + s.register(1, 200, Some(5_000), false); + s.register(2, 300, None, true); + for seq in 0..3 { + s.placed(seq, 1); + s.acked(seq, 1); + } + assert_eq!(s.snapshot().ack_commit_ts, Some(1_000)); + } + #[test] fn events_for_retired_seqs_are_benign() { let (mut s, _ack) = state(); - s.register(0, 100, true); + s.register(0, 100, None, true); s.placed(0, 1); s.acked(0, 1); s.acked(0, 1); @@ -595,14 +645,14 @@ mod tests { #[test] fn illegal_transitions_pin_the_seq_instead_of_being_absorbed() { let (mut s, ack) = state(); - s.register(0, 100, true); + s.register(0, 100, None, true); s.placed(0, 2); assert!(!s.fatal.is_set()); s.placed(0, 5); let first = s.fatal.message().expect("first fault trips fatal"); assert!(first.contains("seq placed twice"), "{first}"); s.acked(0, 3); - s.register(0, 100, true); + s.register(0, 100, None, true); assert_eq!(s.snapshot().wedged, 3); assert!(!s.all_done(), "a faulted seq stays incomplete"); assert_eq!(ack.get(), 0); @@ -755,6 +805,7 @@ mod interleavings { Ev::Reg { seq, lsn, publish } => AckEvent::Register { seq, commit_lsn: lsn, + commit_ts: None, publish, }, Ev::Place { seq, rows } => AckEvent::Placed { seq, rows }, diff --git a/src/emit/pipeline/bootstrap.rs b/src/emit/pipeline/bootstrap.rs index 466a84be..e69dd1ed 100644 --- a/src/emit/pipeline/bootstrap.rs +++ b/src/emit/pipeline/bootstrap.rs @@ -210,7 +210,7 @@ pub async fn drain( out.flush(&msg_tx).await?; seqs.place(prev, prev_rows); } - let seq = seqs.open(source_lsn, Publish::Commit); + let seq = seqs.open(source_lsn, None, Publish::Commit); let s = seq.seq(); open = Some((rfn, seq, 0)); s @@ -618,7 +618,7 @@ async fn route_batch( for row in rows { let mut tuple = row.tuple; let at = seq - .get_or_insert_with(|| seqs.open(tuple.source_lsn, Publish::Commit)) + .get_or_insert_with(|| seqs.open(tuple.source_lsn, None, Publish::Commit)) .seq(); render_ext_columns(&row.rel.attributes, &mut tuple.columns); out.push(msg_tx, at, row.rel, row.route, tuple, permit.clone()) diff --git a/src/emit/pipeline/reorder.rs b/src/emit/pipeline/reorder.rs index fd868fca..ae99121d 100644 --- a/src/emit/pipeline/reorder.rs +++ b/src/emit/pipeline/reorder.rs @@ -915,7 +915,7 @@ impl ReorderSink { let heaps = std::mem::take(pending); let bytes = std::mem::take(pending_bytes); let permit = self.out.admit(bytes).await?.map(Arc::new); - let seq = self.out.seqs.open(commit_lsn, publish); + let seq = self.out.seqs.open(commit_lsn, Some(commit_ts), publish); self.stats.queue_jobs_out.fetch_add(1, Ordering::Relaxed); let chunk_rows = self.dest.current().decode_chunk_rows; let rows = decode::place_rows( @@ -1258,7 +1258,7 @@ impl ReorderSink { // rows=0 marker: publishes commit_lsn once every earlier partial // segment is durable. Covers empty / read-only commits and plans // whose tail is a control or truncate. - self.out.seqs.mark(commit_lsn); + self.out.seqs.mark(commit_lsn, Some(commit_ts)); } Ok(()) } @@ -1271,7 +1271,7 @@ impl ReorderSink { // ABORT PREPARED: buffered state keys off the prepared xid let xid = payload.twophase_xid.unwrap_or(xid); self.note_pending(xid, &payload.subxacts, false).await?; - let seq = self.out.seqs.open(record.source_lsn, Publish::Commit); + let seq = self.out.seqs.open(record.source_lsn, None, Publish::Commit); { let mut buf = self.buffer.lock().await; buf.abort(xid, Pos::new(record.source_lsn), &payload.subxacts) diff --git a/src/filter/engine.rs b/src/filter/engine.rs index 4bb4db7e..bdfb02ba 100644 --- a/src/filter/engine.rs +++ b/src/filter/engine.rs @@ -48,6 +48,8 @@ pub struct FilterStats { pub kept_user: u64, pub kept_special: u64, pub kept_empty: u64, + /// Newest commit record `xact_time` (PG epoch µs) routed + pub last_commit_ts: Option, } impl FilterStats { @@ -539,6 +541,9 @@ impl Filter { return Ok(XactEnd::default()); } let payload = parse_xact_payload(info, &record.main_data, page_magic)?; + if is_commit { + self.stats.last_commit_ts = self.stats.last_commit_ts.max(Some(payload.xact_time)); + } let header_xid = record.header.xact_id; self.creating_dbs.remove(&header_xid); let (merged, members) = @@ -1283,6 +1288,26 @@ mod tests { assert!(v.boundary.is_none(), "dirty mark consumed once"); } + #[test] + fn last_commit_ts_tracks_newest_commit_only() { + let at = |op, xid, ts: i64| { + let mut r = xact_end(op, xid, &[], None); + r.main_data.to_mut()[..8].copy_from_slice(&ts.to_le_bytes()); + r + }; + let mut f = target_filter(); + assert_eq!(f.stats().last_commit_ts, None); + f.decide_record(&at(XLOG_XACT_ABORT, 6, 100), 0, 0xD116) + .unwrap(); + assert_eq!(f.stats().last_commit_ts, None); + f.decide_record(&at(XLOG_XACT_COMMIT, 7, 300), 0, 0xD116) + .unwrap(); + // Backends stamp xact_time before inserting WAL, so commits can land out of order + f.decide_record(&at(XLOG_XACT_COMMIT, 8, 200), 0, 0xD116) + .unwrap(); + assert_eq!(f.stats().last_commit_ts, Some(300)); + } + #[test] fn abort_clears_dirty_without_boundary() { let mut f = target_filter(); diff --git a/src/ops/metrics.rs b/src/ops/metrics.rs index 542aca7b..ed52d929 100644 --- a/src/ops/metrics.rs +++ b/src/ops/metrics.rs @@ -63,6 +63,12 @@ macro_rules! encode_metric { gauge($enc, $name, $help, $value)? } }; + // Same for a commit time not seen yet: a 0 sample reads as 56 years of lag + ($enc:expr, commit_ts, $name:expr, $help:expr, $value:expr) => { + if let Some(pg_us) = $value { + gauge($enc, $name, $help, unix_seconds(pg_us))? + } + }; ($enc:expr, counter, $name:expr, $help:expr, $value:expr, $key:literal, $labels:expr) => { counter_series($enc, $name, $help, $key, $labels.into_iter().zip($value))? }; @@ -236,6 +242,13 @@ snapshot! { lsn decoder_commit_lsn: Pos = "Highest commit LSN drained out of the xact buffer.", /// Live insert watermark, unlike manifest's resume-safe `emitter_ack` lsn emitter_ack_lsn: Pos = "Contiguous-done watermark from the insert pipeline.", + /// `xact_time` in PG epoch µs. `time() - ack_ts` over-reads on an idle + /// source; `source_last_commit_ts - ack_ts` does not + commit_ts emitter_ack_commit_timestamp_seconds: Option = + "Commit time (Unix seconds) of the newest transaction ClickHouse has acked.", + /// `xact_time` in PG epoch µs, read where the pump routes the commit record + commit_ts source_last_commit_timestamp_seconds: Option = + "Commit time (Unix seconds) of the newest source transaction the pump has read.", lsn floor_lsn: Pos = "Durable resume floor: restart resumes here and pruners cut to it.", gauge xact_active: u64 = "Active transactions buffered in memory or on spill.", gauge xact_bytes_in_memory: u64 = "Bytes held in memory across all buffered xacts.", @@ -503,6 +516,11 @@ snapshot! { "Cumulative seconds the pump parked in released catalog-boundary holds.", } +/// `TimestampTz` (microseconds since 2000-01-01) as Unix seconds +fn unix_seconds(pg_us: i64) -> f64 { + pg_us.saturating_add(walrus::pg::replication::PG_EPOCH_USEC) as f64 / 1e6 +} + #[derive(Debug, Default, Clone, PartialEq, Eq)] pub struct SkippedRel { pub database: String, @@ -1498,6 +1516,27 @@ mod tests { ); } + #[test] + fn render_omits_commit_timestamps_until_the_first() { + let body = render(MetricsSnapshot { + // 2000-01-01T00:00:00Z plus 1.5 s + emitter_ack_commit_timestamp_seconds: Some(1_500_000), + ..MetricsSnapshot::default() + }); + assert!( + body.contains("# TYPE walshadow_emitter_ack_commit_timestamp_seconds gauge"), + "{body}" + ); + assert!( + body.contains("walshadow_emitter_ack_commit_timestamp_seconds 946684801.5"), + "{body}" + ); + assert!( + !body.contains("walshadow_source_last_commit_timestamp_seconds"), + "{body}" + ); + } + #[test] fn rate_estimator_rate_across_window() { let mut e = RateEstimator::new(Duration::from_secs(30)); diff --git a/tests/emitter_budget_flush.rs b/tests/emitter_budget_flush.rs index 28da142f..6999cfc6 100644 --- a/tests/emitter_budget_flush.rs +++ b/tests/emitter_budget_flush.rs @@ -183,7 +183,7 @@ async fn budget_trips_seal_complete_inserts() { const N: i32 = 5; let commit_lsn = 0xC0FFEE; let mut seqs = ack.seqs(0); - let seq = seqs.open(commit_lsn, Publish::Commit); + let seq = seqs.open(commit_lsn, None, Publish::Commit); for i in 0..N { msg_tx .send(BatcherMsg::Row(RoutedRow { diff --git a/tests/emitter_native_types.rs b/tests/emitter_native_types.rs index 8e730003..ff78a7ec 100644 --- a/tests/emitter_native_types.rs +++ b/tests/emitter_native_types.rs @@ -175,7 +175,7 @@ async fn native_numeric_time_timetz_round_trip() { // Drive the single row through the tail: register the seq, route the // row, seal with FlushAll, wait for it durable, then drain. let mut seqs = ack.seqs(0); - let seq = seqs.open(tuple.commit_lsn, Publish::Commit); + let seq = seqs.open(tuple.commit_lsn, None, Publish::Commit); msg_tx .send(BatcherMsg::Row(RoutedRow { seq: seq.seq(), diff --git a/tests/emitter_tls.rs b/tests/emitter_tls.rs index cc28fcd6..5eb7b6a0 100644 --- a/tests/emitter_tls.rs +++ b/tests/emitter_tls.rs @@ -376,7 +376,7 @@ async fn emitter_tls_round_trip() { commit_lsn: 0xABCD, }; let mut seqs = ack.seqs(0); - let seq = seqs.open(tuple.commit_lsn, Publish::Commit); + let seq = seqs.open(tuple.commit_lsn, None, Publish::Commit); msg_tx .send(BatcherMsg::Row(RoutedRow { seq: seq.seq(),