Fix: add explicit monotonic <-> realtime clock sync to trace format (#210)

* fix: emit monotonic/realtime anchors for absolute-time rendering

* Use nearest clock-sync anchor for wall time

* Clean up

Author: yulnr

dial9-tokio-telemetry/examples/trace_to_fat_jsonl.rs

@@ -165,6 +165,7 @@ fn to_fat_event(event: &TelemetryEvent, reader: &TraceReader) -> Option<FatEvent
         TelemetryEvent::TaskSpawn { .. }
         | TelemetryEvent::TaskTerminate { .. }
         | TelemetryEvent::ThreadNameDef { .. }
-        | TelemetryEvent::SegmentMetadata { .. } => None,
+        | TelemetryEvent::SegmentMetadata { .. }
+        | TelemetryEvent::ClockSync { .. } => None,
     }
 }

dial9-tokio-telemetry/src/background_task/sealed.rs

@@ -12,9 +12,9 @@ pub(crate) struct SealedSegment {
     pub(crate) index: u32,
 }
 
-/// Segment creation time as epoch seconds, parsed from SegmentMetadata header.
-/// Returns `(epoch_secs, true)` if the header was valid, or falls back to
-/// file mtime / current time with `(epoch_secs, false)`.
+/// Segment creation time as epoch seconds, parsed from the first clock
+/// anchor in the trace. Returns `(secs, true)` on a successful parse, or
+/// falls back to file mtime / current time with `(secs, false)`.
 pub(crate) fn creation_epoch_secs(data: &[u8], path: &Path) -> (u64, bool) {
     match parse_segment_timestamp(data) {
         Ok(ts) => return (ts / 1_000_000_000, true),
@@ -42,17 +42,32 @@ pub(crate) fn creation_epoch_secs(data: &[u8], path: &Path) -> (u64, bool) {
     (secs, false)
 }
 
-/// Parse the timestamp (nanos) from the first SegmentMetadata event in a trace segment.
+/// Legacy epoch-vs-monotonic discriminator for ambiguous
+/// `SegmentMetadataEvent.timestamp_ns` values.
+///
+/// Pre-clock-sync traces may store epoch nanoseconds there, while new traces
+/// use monotonic nanoseconds. Epoch timestamps are ~1e18, while monotonic uptime is much smaller.
+/// So we use `2020-01-01` ns as a floor, monotonic would need decades
+/// of continuous runtime to reach that range.
+///
+/// Keep in sync with `LEGACY_EPOCH_NS_FLOOR` in `dial9-viewer/ui/trace_parser.js`.
+pub(crate) const LEGACY_EPOCH_NS_FLOOR: u64 = 1_577_836_800_000_000_000;
+
+/// Parse wall-clock creation time from the first `ClockSyncEvent`,
+/// or from a legacy `SegmentMetadataEvent.timestamp_ns` that predates clock-sync support.
 fn parse_segment_timestamp(data: &[u8]) -> Result<u64, ParseTimestampError> {
     use dial9_trace_format::decoder::{DecodedFrameRef, Decoder};
+    use dial9_trace_format::types::FieldValueRef;
 
     let mut dec = Decoder::new(data).ok_or(ParseTimestampError::InvalidHeader)?;
     let mut events_seen = 0;
+    let mut legacy_fallback: Option<u64> = None;
     loop {
         match dec.next_frame_ref() {
             Ok(Some(DecodedFrameRef::Event {
                 type_id,
                 timestamp_ns,
+                values,
                 ..
             })) => {
                 events_seen += 1;
@@ -61,16 +76,25 @@ fn parse_segment_timestamp(data: &[u8]) -> Result<u64, ParseTimestampError> {
                     .get(type_id)
                     .map(|s| s.name.as_str())
                     .ok_or(ParseTimestampError::UnknownTypeId(type_id.0))?;
-                if name == "SegmentMetadataEvent" {
-                    return timestamp_ns.ok_or(ParseTimestampError::MissingTimestamp);
+                if name == "ClockSyncEvent" {
+                    return match values.first() {
+                        Some(FieldValueRef::Varint(v)) => Ok(*v),
+                        _ => Err(ParseTimestampError::MissingRealtimeField),
+                    };
+                }
+                if name == "SegmentMetadataEvent"
+                    && let Some(ts) = timestamp_ns
+                    && ts >= LEGACY_EPOCH_NS_FLOOR
+                {
+                    legacy_fallback = Some(ts);
                 }
                 if events_seen >= 10 {
-                    return Err(ParseTimestampError::NotFoundInFirst10Events);
+                    return legacy_fallback.ok_or(ParseTimestampError::NoAnchorInFirst10Events);
                 }
             }
             Ok(Some(_)) => {} // schema/pool frame, keep going
             Ok(None) => {
-                return Err(ParseTimestampError::EndOfStream { events_seen });
+                return legacy_fallback.ok_or(ParseTimestampError::EndOfStream { events_seen });
             }
             Err(e) => {
                 return Err(ParseTimestampError::DecodeError(e.to_string()));
@@ -83,8 +107,8 @@ fn parse_segment_timestamp(data: &[u8]) -> Result<u64, ParseTimestampError> {
 enum ParseTimestampError {
     InvalidHeader,
     UnknownTypeId(u16),
-    MissingTimestamp,
-    NotFoundInFirst10Events,
+    MissingRealtimeField,
+    NoAnchorInFirst10Events,
     EndOfStream { events_seen: u32 },
     DecodeError(String),
 }
@@ -94,13 +118,14 @@ impl std::fmt::Display for ParseTimestampError {
         match self {
             Self::InvalidHeader => write!(f, "invalid trace header"),
             Self::UnknownTypeId(id) => write!(f, "unknown type_id {id} not in registry"),
-            Self::MissingTimestamp => write!(f, "SegmentMetadataEvent had no timestamp"),
-            Self::NotFoundInFirst10Events => {
-                write!(f, "SegmentMetadataEvent not found in first 10 events")
-            }
+            Self::MissingRealtimeField => write!(f, "ClockSyncEvent had no realtime_ns field"),
+            Self::NoAnchorInFirst10Events => write!(
+                f,
+                "no ClockSyncEvent or legacy wall-clock SegmentMetadataEvent in first 10 events"
+            ),
             Self::EndOfStream { events_seen } => write!(
                 f,
-                "end of stream after {events_seen} events without SegmentMetadataEvent"
+                "end of stream after {events_seen} events without a clock anchor"
             ),
             Self::DecodeError(e) => write!(f, "decode error: {e}"),
         }
@@ -330,6 +355,63 @@ mod tests {
         check!(now_nanos.abs_diff(ts) < 60_000_000_000);
     }
 
+    fn legacy_segment_metadata_trace(timestamp_ns: u64) -> Vec<u8> {
+        use crate::telemetry::format::{SegmentMetadataEvent, WorkerParkEvent};
+        use dial9_trace_format::encoder::Encoder;
+
+        let mut enc = Encoder::new_to(Vec::new()).unwrap();
+        enc.write(&SegmentMetadataEvent {
+            timestamp_ns,
+            entries: vec![("k".into(), "v".into())],
+        })
+        .unwrap();
+        enc.write_infallible(&WorkerParkEvent {
+            timestamp_ns: 1_000_000_000,
+            worker_id: crate::telemetry::format::WorkerId::from(0usize),
+            local_queue: 0,
+            cpu_time_ns: 0,
+        });
+        enc.into_inner()
+    }
+
+    /// Legacy pre-clock-sync files stored wall clock in `SegmentMetadataEvent.timestamp_ns`,
+    /// parser fallback should recover it.
+    #[test]
+    fn test_parse_segment_timestamp_legacy_segment_metadata_fallback() {
+        // 2024-06-01T00:00:00Z: comfortably epoch-scale.
+        const LEGACY_WALL_NS: u64 = 1_717_200_000_000_000_000;
+        let data = legacy_segment_metadata_trace(LEGACY_WALL_NS);
+
+        let parsed = parse_segment_timestamp(&data).unwrap();
+        check!(parsed == LEGACY_WALL_NS);
+    }
+
+    #[test]
+    fn test_creation_epoch_secs_uses_legacy_segment_metadata_fallback() {
+        const LEGACY_WALL_NS: u64 = 1_717_200_000_000_000_000;
+        let data = legacy_segment_metadata_trace(LEGACY_WALL_NS);
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("legacy.0.bin");
+        std::fs::write(&path, &data).unwrap();
+        let (epoch_secs, header_valid) = creation_epoch_secs(&data, &path);
+        check!(header_valid);
+        check!(epoch_secs == LEGACY_WALL_NS / 1_000_000_000);
+    }
+
+    /// A small (monotonic-looking) SegmentMeta.timestamp_ns must NOT be
+    /// mistaken for a legacy wall-clock value.
+    #[test]
+    fn test_parse_segment_timestamp_monotonic_segment_metadata_is_not_legacy() {
+        let data = legacy_segment_metadata_trace(12_345);
+
+        let result = parse_segment_timestamp(&data);
+        check!(matches!(
+            result,
+            Err(ParseTimestampError::EndOfStream { .. })
+                | Err(ParseTimestampError::NoAnchorInFirst10Events)
+        ));
+    }
+
     #[test]
     fn parse_segment_index_valid() {
         check!(parse_segment_index("trace.0.bin", "trace") == Some(0));

dial9-tokio-telemetry/src/telemetry/analysis.rs

@@ -88,6 +88,7 @@ impl TraceReader {
                     TelemetryEvent::TaskSpawn { .. }
                         | TelemetryEvent::ThreadNameDef { .. }
                         | TelemetryEvent::SegmentMetadata { .. }
+                        | TelemetryEvent::ClockSync { .. }
                 )
             })
             .cloned()
@@ -297,7 +298,8 @@ pub fn analyze_trace(events: &[TelemetryEvent]) -> TraceAnalysis {
             | TelemetryEvent::CpuSample { .. }
             | TelemetryEvent::ThreadNameDef { .. }
             | TelemetryEvent::WakeEvent { .. }
-            | TelemetryEvent::SegmentMetadata { .. } => {}
+            | TelemetryEvent::SegmentMetadata { .. }
+            | TelemetryEvent::ClockSync { .. } => {}
         }
     }
 

dial9-tokio-telemetry/src/telemetry/events.rs

@@ -188,6 +188,16 @@ pub enum TelemetryEvent {
         /// Key-value metadata pairs.
         entries: Vec<(String, String)>,
     },
+    /// Clock-correlation anchor pairing a monotonic timestamp with the
+    /// wall-clock value captured at the same instant.
+    ClockSync {
+        /// Monotonic nanoseconds.
+        #[serde(rename = "timestamp_ns")]
+        timestamp_nanos: u64,
+        /// Nanoseconds since the Unix epoch.
+        #[serde(rename = "realtime_ns")]
+        realtime_nanos: u64,
+    },
 }
 
 impl TelemetryEvent {
@@ -224,6 +234,9 @@ impl TelemetryEvent {
             TelemetryEvent::ThreadNameDef { .. } => None,
             TelemetryEvent::SegmentMetadata {
                 timestamp_nanos, ..
+            }
+            | TelemetryEvent::ClockSync {
+                timestamp_nanos, ..
             } => Some(*timestamp_nanos),
         }
     }
@@ -241,7 +254,8 @@ impl TelemetryEvent {
             | TelemetryEvent::TaskTerminate { .. }
             | TelemetryEvent::ThreadNameDef { .. }
             | TelemetryEvent::WakeEvent { .. }
-            | TelemetryEvent::SegmentMetadata { .. } => None,
+            | TelemetryEvent::SegmentMetadata { .. }
+            | TelemetryEvent::ClockSync { .. } => None,
         }
     }
 
@@ -367,15 +381,49 @@ pub fn clock_monotonic_ns() -> u64 {
     ts.tv_sec as u64 * 1_000_000_000 + ts.tv_nsec as u64
 }
 
+/// `CLOCK_MONOTONIC` in nanoseconds. Non-Linux fallback: elapsed time
+/// since the first call on this process via `Instant`.
 #[cfg(not(target_os = "linux"))]
 pub fn clock_monotonic_ns() -> u64 {
-    // Fallback: use Instant. This is fine for non-Linux where perf isn't available.
     use std::sync::OnceLock;
     use std::time::Instant;
     static EPOCH: OnceLock<Instant> = OnceLock::new();
     EPOCH.get_or_init(Instant::now).elapsed().as_nanos() as u64
 }
 
+/// `CLOCK_REALTIME` in nanoseconds since the Unix epoch.
+#[cfg(target_os = "linux")]
+pub(crate) fn clock_realtime_ns() -> u64 {
+    let mut ts = libc::timespec {
+        tv_sec: 0,
+        tv_nsec: 0,
+    };
+    unsafe {
+        libc::clock_gettime(libc::CLOCK_REALTIME, &mut ts);
+    }
+    ts.tv_sec as u64 * 1_000_000_000 + ts.tv_nsec as u64
+}
+
+#[cfg(not(target_os = "linux"))]
+pub(crate) fn clock_realtime_ns() -> u64 {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_nanos() as u64)
+        .unwrap_or(0)
+}
+
+/// Snapshot `(monotonic_ns, realtime_ns)` as close together as possible.
+/// Reads M₁ -> R -> M₂ and pairs `R` with the midpoint of M₁ and M₂ so
+/// the correlation error is half the `clock_gettime` interval.
+pub(crate) fn clock_pair() -> (u64, u64) {
+    let m1 = clock_monotonic_ns();
+    let r = clock_realtime_ns();
+    let m2 = clock_monotonic_ns();
+    let mono = m1 + m2.saturating_sub(m1) / 2;
+    (mono, r)
+}
+
 /// Per-thread scheduler stats from `/proc/<pid>/task/<tid>/schedstat`.
 /// Fields: run_time_ns wait_time_ns timeslices
 #[derive(Debug, Clone, Copy, Default)]

dial9-tokio-telemetry/src/telemetry/format.rs

@@ -215,6 +215,19 @@ pub(crate) struct SegmentMetadataEvent {
     pub entries: Vec<(String, String)>,
 }
 
+/// Clock-correlation anchor. `timestamp_ns` (monotonic) and `realtime_ns`
+/// (nanoseconds since Unix epoch) are captured at the same instant via
+/// [`clock_pair`], so offline consumers can recover wall clock from the
+/// monotonic event stream.
+///
+/// [`clock_pair`]: crate::telemetry::events::clock_pair
+#[derive(TraceEvent)]
+pub(crate) struct ClockSyncEvent {
+    #[traceevent(timestamp)]
+    pub timestamp_ns: u64,
+    pub realtime_ns: u64,
+}
+
 // ── dial9-trace-format: decode ──────────────────────────────────────────────
 
 /// Decode all events from a `dial9-trace-format` byte slice into `TelemetryEvent`s.
@@ -254,6 +267,7 @@ pub(crate) enum TelemetryEventRef<'a> {
     CpuSample(CpuSampleEventRef<'a>),
     WakeEvent(WakeEventEventRef<'a>),
     SegmentMetadata(SegmentMetadataEventRef<'a>),
+    ClockSync(ClockSyncEventRef<'a>),
 }
 
 #[cfg(any(feature = "analysis", test))]
@@ -272,6 +286,7 @@ impl<'a> TelemetryEventRef<'a> {
             Self::CpuSample(e) => Some(e.timestamp_ns),
             Self::WakeEvent(e) => Some(e.timestamp_ns),
             Self::SegmentMetadata(e) => Some(e.timestamp_ns),
+            Self::ClockSync(e) => Some(e.timestamp_ns),
         }
     }
 }
@@ -314,6 +329,9 @@ pub(crate) fn decode_ref<'a>(
         "SegmentMetadataEvent" => {
             TelemetryEventRef::SegmentMetadata(SegmentMetadataEvent::decode(timestamp_ns, fields)?)
         }
+        "ClockSyncEvent" => {
+            TelemetryEventRef::ClockSync(ClockSyncEvent::decode(timestamp_ns, fields)?)
+        }
         _ => return None,
     })
 }
@@ -392,5 +410,9 @@ pub(crate) fn to_owned_event(r: TelemetryEventRef<'_>, pool: &StringPool) -> Tel
                 .map(|(k, v)| (k.to_owned(), v.to_owned()))
                 .collect(),
         },
+        TelemetryEventRef::ClockSync(e) => TelemetryEvent::ClockSync {
+            timestamp_nanos: e.timestamp_ns,
+            realtime_nanos: e.realtime_ns,
+        },
     }
 }