From a3a247f8069a14eb829f0ff324318bb0989c25c8 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Sun, 12 Oct 2025 23:34:46 -0400 Subject: [PATCH] feat: more verbose seq tracking --- consumer/src/indexer/mod.rs | 115 ++++++++++++++++++++++++++++-------- 1 file changed, 91 insertions(+), 24 deletions(-) diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index e8e2c99c..1ae455bd 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -48,6 +48,8 @@ struct RelayIndexerState { do_backfill: bool, do_handle_res: bool, req_backfill: bool, + /// Stores the last sequence number and timestamp for rate calculation (seq, timestamp) + last_seq_time: Option<(u64, chrono::DateTime)>, } /// Extract time_us sequence number from a JSON preview string @@ -354,6 +356,7 @@ impl RelayIndexer { req_backfill: opts.request_backfill, do_handle_res: !opts.skip_handle_validation, idxc_tx, + last_seq_time: None, }, hasher: RandomState::default(), resume, @@ -456,32 +459,96 @@ impl RelayIndexer { crate::consumer::UnifiedConsumer::Jetstream(_) => ("jetstream", "Jetstream"), }; - // Log status every 10 seconds (not on every tick for better readability) - use std::sync::atomic::{AtomicU64, Ordering}; - static LAST_LOGGED_SEQ: AtomicU64 = AtomicU64::new(0); - let last_seq = LAST_LOGGED_SEQ.load(Ordering::Relaxed); - let should_log = if seq > last_seq + 10000000 || seq < last_seq { // Log if we've advanced 10M or if we've reset - LAST_LOGGED_SEQ.store(seq, Ordering::Relaxed); - true - } else { - false - }; - - if should_log { - tracing::info!("Indexer running with {} consumer at sequence {}", consumer_type, seq); - } - - tracing::info!( - "Saved sequence {}, consumer={}, resume_key={}", - seq, - consumer_type, - resume_key - ); if let Err(e) = self.resume.insert(resume_key, &seq.to_le_bytes()) { tracing::error!("Failed to update seq ({seq}) in resume: {e}"); } counter!("consumer_seq").absolute(seq); + // If this is Jetstream, convert the sequence (which is microseconds since epoch) to a timestamp + // and calculate how far behind we are from current time + let log_message = if consumer_type == "Jetstream" { + let now = chrono::Utc::now(); + // Convert microseconds to milliseconds for chrono + let event_time = chrono::DateTime::::from_timestamp_micros(seq as i64) + .unwrap_or(now); + + // Calculate how far behind we are + let behind_duration = now.signed_duration_since(event_time); + + // Format lag as days:hours:minutes:seconds + let total_seconds = behind_duration.num_seconds(); + let days = total_seconds / 86400; + let hours = (total_seconds % 86400) / 3600; + let minutes = (total_seconds % 3600) / 60; + let seconds = total_seconds % 60; + + let behind_formatted = if days > 0 { + format!("{}d {}h {}m {}s", days, hours, minutes, seconds) + } else if hours > 0 { + format!("{}h {}m {}s", hours, minutes, seconds) + } else if minutes > 0 { + format!("{}m {}s", minutes, seconds) + } else { + format!("{}s", seconds) + }; + + // Calculate processing rate relative to real-time + let rate_info = if let Some((prev_seq, prev_time)) = self.state.last_seq_time { + // Calculate time difference to see if we can compute a meaningful rate + let wall_time_diff = (now - prev_time).num_milliseconds() as f64 / 1000.0; + + // Only calculate rate if sequence has changed + if prev_seq != seq && wall_time_diff > 0.0 { + // Calculate event time difference in seconds + let event_time_diff_us = seq.saturating_sub(prev_seq) as f64; + let event_time_diff_sec = event_time_diff_us / 1_000_000.0; + + if event_time_diff_sec > 0.0 { + // Calculate rate (how many times faster than real-time) + let rate = event_time_diff_sec / wall_time_diff; + + // If rate > 1.0, we're catching up, otherwise falling behind + if rate > 1.1 { + format!(", rate=x{:.1} (catching up)", rate) + } else if rate < 0.9 { + format!(", rate=x{:.1} (falling behind)", rate) + } else { + format!(", rate=x{:.1} (keeping pace)", rate) + } + } else { + String::new() + } + } else { + String::new() + } + } else { + String::new() + }; + + // Always update last sequence and time for next calculation + self.state.last_seq_time = Some((seq, now)); + + format!( + "Saved sequence {} ({}, {}), consumer={}, resume_key={}, behind={}{}", + seq, + event_time.to_rfc3339(), + event_time.format("%Y-%m-%d %H:%M:%S UTC"), + consumer_type, + resume_key, + behind_formatted, + rate_info + ) + } else { + format!( + "Saved sequence {}, consumer={}, resume_key={}", + seq, + consumer_type, + resume_key + ) + }; + + tracing::info!("{}", log_message); + if stop.has_changed().unwrap_or(true) { break; } @@ -1148,7 +1215,7 @@ async fn process_op( } let Some((cid, decoded)) = decode_op(op, blocks) else { - tracing::warn!("failed to decode op for {} {}", op.action, full_path); + tracing::debug!("failed to decode op for {} {}", op.action, full_path); return Ok(()); }; @@ -1175,8 +1242,8 @@ fn decode_op(op: &CommitOp, blocks: &HashMap>) -> Option<(Cid, Reco return Some((cid, record)); } Err(err) => { - tracing::warn!("Failed to deserialize Jetstream record: {}", err); - // Fall through to CBOR decoding + tracing::debug!("Failed to deserialize Jetstream record: {}", err); + return None; } } } -- 2.51.2