katra-trace 0.1.0

//! Trace verification: checks the invariants a well-formed trace must obey.

use std::collections::HashSet;

use katra_core::{EventKind, Phase};

use crate::reader::TraceReader;
use crate::record::TraceRecord;

/// A single verification finding.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct VerifyIssue {
    /// Whether this is an error (invariant violation) or a warning.
    pub is_error: bool,
    /// Human-readable description.
    pub message: String,
}

/// The result of verifying a trace.
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct VerifyReport {
    /// Issues found.
    pub issues: Vec<VerifyIssue>,
    /// Events inspected.
    pub event_count: u64,
    /// Highest sequence number.
    pub max_seq: u64,
    /// Whether sequence numbers were strictly monotonic.
    pub monotonic: bool,
    /// Whether all causal links referenced earlier, existing events.
    pub causal_ok: bool,
    /// Number of begin spans without a matching end (by thread+span).
    pub unclosed_spans: u64,
}

impl VerifyReport {
    /// True if no error-level issues were found.
    pub fn is_clean(&self) -> bool {
        !self.issues.iter().any(|i| i.is_error)
    }
}

/// Verify every invariant of a trace stream, consuming it.
pub fn verify(reader: &mut TraceReader) -> VerifyReport {
    let mut report = VerifyReport::default();
    let mut seen_seqs: HashSet<u64> = HashSet::new();
    let mut last_seq: Option<u64> = None;
    // (thread, span) -> count of open begins
    let mut open_spans: std::collections::HashMap<(u64, u64), u64> =
        std::collections::HashMap::new();

    loop {
        match reader.next_record() {
            Ok(Some(TraceRecord::Event(ev))) => {
                report.event_count += 1;
                if ev.seq.0 > report.max_seq {
                    report.max_seq = ev.seq.0;
                }
                if let Some(last) = last_seq {
                    if ev.seq.0 <= last {
                        report.monotonic = false;
                        report.issues.push(VerifyIssue {
                            is_error: true,
                            message: format!("sequence not monotonic: {} after {}", ev.seq.0, last),
                        });
                    }
                }
                last_seq = Some(ev.seq.0);
                if !seen_seqs.insert(ev.seq.0) {
                    report.issues.push(VerifyIssue {
                        is_error: true,
                        message: format!("duplicate sequence {}", ev.seq.0),
                    });
                }
                // Causal links must point at earlier events.
                for cause in &ev.causes {
                    if !seen_seqs.contains(cause) {
                        report.causal_ok = false;
                        report.issues.push(VerifyIssue {
                            is_error: true,
                            message: format!(
                                "event {} references unknown/future cause {}",
                                ev.seq.0, cause
                            ),
                        });
                    }
                }
                // Span balance.
                match ev.phase {
                    Phase::Begin => {
                        if let Some(span) = ev.span_id {
                            let key = (ev.thread_id, span.0);
                            *open_spans.entry(key).or_insert(0) += 1;
                        } else {
                            report.issues.push(VerifyIssue {
                                is_error: true,
                                message: format!("event {} is Begin without span_id", ev.seq.0),
                            });
                        }
                    }
                    Phase::End => {
                        if let Some(span) = ev.span_id {
                            let key = (ev.thread_id, span.0);
                            match open_spans.get_mut(&key) {
                                Some(n) if *n > 0 => {
                                    *n -= 1;
                                    if *n == 0 {
                                        open_spans.remove(&key);
                                    }
                                }
                                _ => {
                                    report.issues.push(VerifyIssue {
                                        is_error: true,
                                        message: format!(
                                            "event {} is End without matching Begin \
                                             (thread {}, span {})",
                                            ev.seq.0, ev.thread_id, span.0
                                        ),
                                    });
                                }
                            }
                        }
                    }
                    _ => {}
                }
            }
            Ok(Some(TraceRecord::SessionSummary(_))) => {}
            Ok(Some(_)) => {}
            Ok(None) => break,
            Err(e) => {
                report
                    .issues
                    .push(VerifyIssue { is_error: true, message: format!("read error: {e}") });
                break;
            }
        }
    }

    report.unclosed_spans = open_spans.values().sum();
    for ((thread, span), count) in open_spans {
        if count > 0 {
            report.issues.push(VerifyIssue {
                is_error: true,
                message: format!("unclosed span: thread {thread}, span {span}, {count} begins"),
            });
        }
    }
    if report.monotonic {
        report.issues.push(VerifyIssue {
            is_error: false,
            message: "sequence numbers strictly monotonic".into(),
        });
    }
    if report.causal_ok && report.event_count > 0 {
        report.issues.push(VerifyIssue {
            is_error: false,
            message: "all causal links reference existing earlier events".into(),
        });
    }
    report
}

/// Whether a kind participates in the semantic graph rebuild (nodes).
pub fn is_graph_kind(kind: EventKind) -> bool {
    matches!(
        kind,
        EventKind::FileRead
            | EventKind::NtReadFile
            | EventKind::IoRead
            | EventKind::KatraDecompress
            | EventKind::KatraStaging
            | EventKind::ResourceUpload
            | EventKind::Barrier
            | EventKind::FenceWait
            | EventKind::FenceSignal
            | EventKind::Present
            | EventKind::GpuDraw
            | EventKind::GpuComputeDispatch
            | EventKind::KatraPrefetch
            | EventKind::KatraAllocate
            | EventKind::CacheHit
            | EventKind::CacheMiss
            | EventKind::PipelineCreate
            | EventKind::ShaderTranslate
            | EventKind::KatraIoWait
    )
}