keyhog_profile/
schema.rs

1use crate::collector::CollectorCapability;
2use crate::metrics::{MacroStageId, MetricId, METRICS};
3use serde::{Deserialize, Serialize};
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::time::{SystemTime, UNIX_EPOCH};
6
7/// Stable wire schema for persisted profiling records.
8pub const PROFILE_SCHEMA: &str = "keyhog-profile-v1";
9
10const fn legacy_component_version() -> u16 {
11    1
12}
13
14pub const STAGE_MEASUREMENT_VERSION: u16 = 1;
15pub const RUN_IDENTITY_VERSION: u16 = 1;
16pub const STATE_TRANSITION_VERSION: u16 = 1;
17pub const RESOURCE_SAMPLE_VERSION: u16 = 1;
18pub const RESOURCE_SNAPSHOT_VERSION: u16 = 2;
19pub const STATE_MEASUREMENT_VERSION: u16 = 1;
20pub const RESOURCE_USAGE_VERSION: u16 = 1;
21pub const RUN_PROFILE_VERSION: u16 = 3;
22pub const WORKLOAD_MEASUREMENTS_VERSION: u16 = 1;
23
24static RUN_SEQUENCE: AtomicU64 = AtomicU64::new(0);
25
26/// Stable micro-function identifier shared by scanner, source, verifier, and reporter paths.
27#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
28#[serde(rename_all = "kebab-case")]
29#[repr(usize)]
30pub enum Stage {
31    SourceAcquire = 0,
32    SourceWalk,
33    SourceRead,
34    Preprocess,
35    Phase1Triggers,
36    BackendDispatch,
37    HotPatterns,
38    ConfirmedPatterns,
39    Phase2Prefilter,
40    Phase2KeywordAc,
41    Phase2SharedAc,
42    Phase2AnchoredVerify,
43    Phase2WholeChunk,
44    GenericDetection,
45    Entropy,
46    MachineLearning,
47    Decode,
48    Suppression,
49    LiveVerification,
50    Reporting,
51    SourceQueueWait,
52    IncrementalLookup,
53    BackendSelect,
54    ResultMerge,
55    ScannerQueueWait,
56    /// Timing a probe scan to choose a backend, not scanning for credentials.
57    AutorouteCalibration,
58    /// Rescanning the seam between adjacent chunks so a match spanning the
59    /// boundary is not lost. Separated from the phase-2 leaves it sits inside
60    /// because seam work grows with chunk count, not with input size.
61    BoundaryScan,
62    /// Loading detector bytes, cache entries, and parsed specifications.
63    DetectorLoad,
64    /// Validating detector selection, policy, and effective corpus identity.
65    DetectorValidate,
66    /// Selecting the backend- and policy-specific execution plan generation.
67    ExecutionPackSelect,
68    /// Materializing the selected execution plan into the scanner runtime.
69    ExecutionPackMap,
70    /// Discovering backend hardware and runtime availability.
71    BackendAcquire,
72    /// Initializing the selected backend runtime and compiled databases.
73    BackendInit,
74    /// Releasing scanner plans, backend resources, and retained buffers.
75    Teardown,
76}
77
78impl Stage {
79    /// Every stage in stable wire order.
80    pub const ALL: [Self; 34] = [
81        Self::SourceAcquire,
82        Self::SourceWalk,
83        Self::SourceRead,
84        Self::Preprocess,
85        Self::Phase1Triggers,
86        Self::BackendDispatch,
87        Self::HotPatterns,
88        Self::ConfirmedPatterns,
89        Self::Phase2Prefilter,
90        Self::Phase2KeywordAc,
91        Self::Phase2SharedAc,
92        Self::Phase2AnchoredVerify,
93        Self::Phase2WholeChunk,
94        Self::GenericDetection,
95        Self::Entropy,
96        Self::MachineLearning,
97        Self::Decode,
98        Self::Suppression,
99        Self::LiveVerification,
100        Self::Reporting,
101        Self::SourceQueueWait,
102        Self::IncrementalLookup,
103        Self::BackendSelect,
104        Self::ResultMerge,
105        Self::ScannerQueueWait,
106        Self::AutorouteCalibration,
107        Self::BoundaryScan,
108        Self::DetectorLoad,
109        Self::DetectorValidate,
110        Self::ExecutionPackSelect,
111        Self::ExecutionPackMap,
112        Self::BackendAcquire,
113        Self::BackendInit,
114        Self::Teardown,
115    ];
116
117    #[inline]
118    pub(crate) const fn index(self) -> usize {
119        self as usize
120    }
121
122    /// Stable metric identifier shared by wire records and runtime storage.
123    ///
124    /// Named explicitly rather than derived from position, so a metric can be
125    /// appended to the registry without silently re-pointing a stage.
126    pub const fn metric_id(self) -> MetricId {
127        match self {
128            Self::SourceAcquire => MetricId::SourceAcquire,
129            Self::SourceWalk => MetricId::SourceWalk,
130            Self::SourceRead => MetricId::SourceRead,
131            Self::Preprocess => MetricId::Preprocess,
132            Self::Phase1Triggers => MetricId::Phase1Triggers,
133            Self::BackendDispatch => MetricId::BackendDispatch,
134            Self::HotPatterns => MetricId::HotPatterns,
135            Self::ConfirmedPatterns => MetricId::ConfirmedPatterns,
136            Self::Phase2Prefilter => MetricId::Phase2Prefilter,
137            Self::Phase2KeywordAc => MetricId::Phase2KeywordAc,
138            Self::Phase2SharedAc => MetricId::Phase2SharedAc,
139            Self::Phase2AnchoredVerify => MetricId::Phase2AnchoredVerify,
140            Self::Phase2WholeChunk => MetricId::Phase2WholeChunk,
141            Self::GenericDetection => MetricId::GenericDetection,
142            Self::Entropy => MetricId::Entropy,
143            Self::MachineLearning => MetricId::MachineLearning,
144            Self::Decode => MetricId::Decode,
145            Self::Suppression => MetricId::Suppression,
146            Self::LiveVerification => MetricId::LiveVerification,
147            Self::Reporting => MetricId::Reporting,
148            Self::SourceQueueWait => MetricId::SourceQueueWait,
149            Self::IncrementalLookup => MetricId::IncrementalLookup,
150            Self::BackendSelect => MetricId::BackendSelect,
151            Self::ResultMerge => MetricId::ResultMerge,
152            Self::ScannerQueueWait => MetricId::ScannerQueueWait,
153            Self::AutorouteCalibration => MetricId::AutorouteCalibration,
154            Self::BoundaryScan => MetricId::BoundaryScan,
155            Self::DetectorLoad => MetricId::DetectorLoad,
156            Self::DetectorValidate => MetricId::DetectorValidate,
157            Self::ExecutionPackSelect => MetricId::ExecutionPackSelect,
158            Self::ExecutionPackMap => MetricId::ExecutionPackMap,
159            Self::BackendAcquire => MetricId::BackendAcquire,
160            Self::BackendInit => MetricId::BackendInit,
161            Self::Teardown => MetricId::Teardown,
162        }
163    }
164
165    /// Stable text label used by human reports.
166    pub const fn as_str(self) -> &'static str {
167        METRICS[self.metric_id() as usize].name
168    }
169
170    /// Stable macro-stage identifier that owns this micro-function.
171    pub const fn macro_stage_id(self) -> MacroStageId {
172        match self {
173            Self::SourceAcquire | Self::SourceWalk | Self::SourceRead | Self::SourceQueueWait => {
174                MacroStageId::Acquire
175            }
176            Self::Preprocess
177            | Self::Phase1Triggers
178            | Self::BackendDispatch
179            | Self::HotPatterns
180            | Self::ConfirmedPatterns
181            | Self::Phase2Prefilter
182            | Self::Phase2KeywordAc
183            | Self::Phase2SharedAc
184            | Self::Phase2AnchoredVerify
185            | Self::Phase2WholeChunk
186            | Self::GenericDetection
187            | Self::Entropy
188            | Self::MachineLearning
189            | Self::Decode
190            | Self::IncrementalLookup
191            | Self::BackendSelect
192            | Self::ScannerQueueWait
193            | Self::AutorouteCalibration
194            | Self::BoundaryScan
195            | Self::DetectorLoad
196            | Self::DetectorValidate
197            | Self::ExecutionPackSelect
198            | Self::ExecutionPackMap
199            | Self::BackendAcquire
200            | Self::BackendInit
201            | Self::Teardown => MacroStageId::Scan,
202            Self::Suppression | Self::ResultMerge => MacroStageId::Resolve,
203            Self::LiveVerification => MacroStageId::Verify,
204            Self::Reporting => MacroStageId::Report,
205        }
206    }
207}
208
209/// One aggregate fixed-stage measurement.
210#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
211pub struct StageMeasurement {
212    #[serde(default = "legacy_component_version")]
213    pub version: u16,
214    pub stage: Stage,
215    pub elapsed_ns: u64,
216    pub calls: u64,
217    pub attributed_ns: u64,
218}
219
220/// Cache state that materially changes run cost.
221#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
222#[serde(rename_all = "kebab-case")]
223pub enum CacheState {
224    #[default]
225    Unknown,
226    Disabled,
227    Cold,
228    Warm,
229}
230
231/// Daemon state that materially changes startup and resident work.
232#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
233#[serde(rename_all = "kebab-case")]
234pub enum DaemonState {
235    #[default]
236    Off,
237    Client,
238    Worker,
239    Mass,
240}
241
242/// Coarse causal state of a profiling run.
243#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
244#[serde(rename_all = "kebab-case")]
245pub enum RunState {
246    Created,
247    Acquiring,
248    Scanning,
249    Resolving,
250    Verifying,
251    Reporting,
252    Completed,
253    Failed,
254}
255
256/// Identity and execution choices required to compare two run records honestly.
257#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
258pub struct RunIdentity {
259    #[serde(default = "legacy_component_version")]
260    pub version: u16,
261    pub run_id: String,
262    pub binary_version: String,
263    pub detector_digest: String,
264    pub config_digest: String,
265    pub source_kind: String,
266    pub workload_class: String,
267    pub backend_requested: String,
268    pub backend_selected: Option<String>,
269    pub cache_state: CacheState,
270    pub daemon_state: DaemonState,
271    pub scanner_threads: usize,
272    pub reader_threads: Option<usize>,
273    pub logical_cpus: usize,
274}
275
276impl RunIdentity {
277    /// Construct a run identity with a process-unique identifier and explicit state.
278    pub fn new(
279        binary_version: impl Into<String>,
280        detector_digest: impl Into<String>,
281        config_digest: impl Into<String>,
282        source_kind: impl Into<String>,
283        workload_class: impl Into<String>,
284        backend_requested: impl Into<String>,
285    ) -> Self {
286        let sequence = RUN_SEQUENCE.fetch_add(1, Ordering::Relaxed);
287        let unix_ns = SystemTime::now()
288            .duration_since(UNIX_EPOCH)
289            .unwrap_or_default()
290            .as_nanos();
291        Self {
292            version: RUN_IDENTITY_VERSION,
293            run_id: format!("{}-{unix_ns}-{sequence}", std::process::id()),
294            binary_version: binary_version.into(),
295            detector_digest: detector_digest.into(),
296            config_digest: config_digest.into(),
297            source_kind: source_kind.into(),
298            workload_class: workload_class.into(),
299            backend_requested: backend_requested.into(),
300            backend_selected: None,
301            cache_state: CacheState::Unknown,
302            daemon_state: DaemonState::Off,
303            scanner_threads: 0,
304            reader_threads: None,
305            logical_cpus: std::thread::available_parallelism().map_or(1, usize::from),
306        }
307    }
308}
309
310/// One run-state transition relative to session start.
311#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
312pub struct StateTransition {
313    #[serde(default = "legacy_component_version")]
314    pub version: u16,
315    pub state: RunState,
316    pub elapsed_ns: u64,
317}
318
319/// Process resource observation associated with a run-state boundary.
320#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
321pub struct ResourceSample {
322    #[serde(default = "legacy_component_version")]
323    pub version: u16,
324    pub state: RunState,
325    pub elapsed_ns: u64,
326    pub snapshot: ResourceSnapshot,
327}
328
329/// Process resource observation at a macro boundary.
330#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
331pub struct ResourceSnapshot {
332    #[serde(default = "legacy_component_version")]
333    pub version: u16,
334    pub cpu_time_ms: Option<u64>,
335    pub resident_bytes: Option<u64>,
336    pub virtual_bytes: Option<u64>,
337    pub thread_count: Option<u64>,
338    /// Kernel-maintained exact resident high water (VmHWM); version 2 field.
339    #[serde(default)]
340    pub resident_high_water_bytes: Option<u64>,
341    /// Bytes swapped out at sample time (VmSwap); version 2 field.
342    #[serde(default)]
343    pub swap_bytes: Option<u64>,
344}
345
346impl Default for ResourceSnapshot {
347    fn default() -> Self {
348        Self {
349            version: RESOURCE_SNAPSHOT_VERSION,
350            cpu_time_ms: None,
351            resident_bytes: None,
352            virtual_bytes: None,
353            thread_count: None,
354            resident_high_water_bytes: None,
355            swap_bytes: None,
356        }
357    }
358}
359
360/// One completed macro state with its wall time and boundary resource deltas.
361#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
362pub struct StateMeasurement {
363    #[serde(default = "legacy_component_version")]
364    pub version: u16,
365    pub state: RunState,
366    pub elapsed_ns: u64,
367    pub cpu_time_ms: Option<u64>,
368    pub aggregate_cpu_milli_percent: Option<u64>,
369    pub resident_start_bytes: Option<u64>,
370    pub resident_end_bytes: Option<u64>,
371    pub threads_start: Option<u64>,
372    pub threads_end: Option<u64>,
373}
374
375/// Resource change across a completed profile session.
376#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
377pub struct ResourceUsage {
378    #[serde(default = "legacy_component_version")]
379    pub version: u16,
380    pub start: ResourceSnapshot,
381    pub finish: ResourceSnapshot,
382    pub max_observed_resident_bytes: Option<u64>,
383    pub max_observed_threads: Option<u64>,
384    pub aggregate_cpu_percent: Option<f64>,
385}
386
387impl Default for ResourceUsage {
388    fn default() -> Self {
389        Self {
390            version: RESOURCE_USAGE_VERSION,
391            start: ResourceSnapshot::default(),
392            finish: ResourceSnapshot::default(),
393            max_observed_resident_bytes: None,
394            max_observed_threads: None,
395            aggregate_cpu_percent: None,
396        }
397    }
398}
399
400/// Optional byte domains whose totals distinguish source, expansion, decode, and dispatch work.
401#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
402pub struct WorkloadMeasurements {
403    #[serde(default = "legacy_component_version")]
404    pub version: u16,
405    pub container_bytes: Option<u64>,
406    pub expanded_payload_bytes: Option<u64>,
407    pub derived_decoder_bytes: Option<u64>,
408    pub backend_dispatched_bytes: Option<u64>,
409}
410
411impl Default for WorkloadMeasurements {
412    fn default() -> Self {
413        Self {
414            version: WORKLOAD_MEASUREMENTS_VERSION,
415            container_bytes: None,
416            expanded_payload_bytes: None,
417            derived_decoder_bytes: None,
418            backend_dispatched_bytes: None,
419        }
420    }
421}
422
423impl WorkloadMeasurements {
424    pub(crate) fn measured(derived_decoder_bytes: u64, backend_dispatched_bytes: u64) -> Self {
425        Self {
426            version: WORKLOAD_MEASUREMENTS_VERSION,
427            container_bytes: None,
428            expanded_payload_bytes: None,
429            derived_decoder_bytes: Some(derived_decoder_bytes),
430            backend_dispatched_bytes: Some(backend_dispatched_bytes),
431        }
432    }
433}
434
435/// Complete replayable profile record.
436#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
437pub struct RunProfile {
438    #[serde(default = "legacy_component_version")]
439    pub version: u16,
440    pub schema: String,
441    pub identity: RunIdentity,
442    pub status: RunState,
443    pub wall_time_ns: u64,
444    pub input_bytes: u64,
445    pub input_units: u64,
446    #[serde(default)]
447    pub workload: WorkloadMeasurements,
448    pub stages: Vec<StageMeasurement>,
449    pub transitions: Vec<StateTransition>,
450    #[serde(default)]
451    pub states: Vec<StateMeasurement>,
452    #[serde(default)]
453    pub collectors: Vec<CollectorCapability>,
454    pub resource_samples: Vec<ResourceSample>,
455    pub resources: ResourceUsage,
456    /// CPU hardware evidence; absent in profiles recorded before version 2.
457    #[serde(default = "legacy_hardware_gap")]
458    pub hardware: crate::Evidence<crate::hardware::HardwareRunEvidenceV2>,
459    /// Memory, IO, and system evidence; absent in profiles before version 3.
460    #[serde(default = "legacy_system_gap")]
461    pub system: crate::Evidence<crate::system::SystemRunEvidenceV2>,
462}
463
464fn legacy_system_gap() -> crate::Evidence<crate::system::SystemRunEvidenceV2> {
465    crate::Evidence::unavailable(crate::EvidenceGap::LegacyV1NotRecorded)
466}
467
468fn legacy_hardware_gap() -> crate::Evidence<crate::hardware::HardwareRunEvidenceV2> {
469    crate::Evidence::unavailable(crate::EvidenceGap::LegacyV1NotRecorded)
470}
471
472impl RunProfile {
473    /// Serialize the stable record as pretty JSON.
474    pub fn to_json_pretty(&self) -> serde_json::Result<String> {
475        serde_json::to_string_pretty(self)
476    }
477
478    /// Render a compact operator report without secrets or source content.
479    pub fn render_text(&self) -> String {
480        let reader_threads = self
481            .identity
482            .reader_threads
483            .map_or_else(|| "auto".to_owned(), |threads| threads.to_string());
484        let throughput_mib_s = if self.wall_time_ns == 0 {
485            0.0
486        } else {
487            self.input_bytes as f64 * 1_000_000_000.0 / self.wall_time_ns as f64 / (1024.0 * 1024.0)
488        };
489        let mut output = format!(
490            "KeyHog profile {}\n\
491             state={} source={} workload={} backend_requested={} backend_selected={} cache={} daemon={} wall_ms={:.3}\n\
492             version={} detector_digest={} config_digest={}\n\
493             input_bytes={} input_units={} throughput_mib_s={throughput_mib_s:.3} scanner_threads={} reader_threads={} logical_cpus={}\n",
494            self.identity.run_id,
495            state_name(self.status),
496            self.identity.source_kind,
497            self.identity.workload_class,
498            self.identity.backend_requested,
499            self.identity
500                .backend_selected
501                .as_deref()
502                .unwrap_or("unselected"),
503            cache_name(self.identity.cache_state),
504            daemon_name(self.identity.daemon_state),
505            self.wall_time_ns as f64 / 1_000_000.0,
506            self.identity.binary_version,
507            self.identity.detector_digest,
508            self.identity.config_digest,
509            self.input_bytes,
510            self.input_units,
511            self.identity.scanner_threads,
512            reader_threads,
513            self.identity.logical_cpus,
514        );
515        for state in &self.states {
516            output.push_str(&format!(
517                "macro {:<12} wall_ms={:.3}",
518                state_name(state.state),
519                state.elapsed_ns as f64 / 1_000_000.0,
520            ));
521            if let Some(cpu) = state.aggregate_cpu_milli_percent {
522                output.push_str(&format!(" cpu={:.1}%", cpu as f64 / 1_000.0));
523            } else {
524                output.push_str(" cpu=unavailable");
525            }
526            if let (Some(start), Some(finish)) =
527                (state.resident_start_bytes, state.resident_end_bytes)
528            {
529                output.push_str(&format!(" rss_bytes={start}->{finish}"));
530            }
531            if let (Some(start), Some(finish)) = (state.threads_start, state.threads_end) {
532                output.push_str(&format!(" threads={start}->{finish}"));
533            }
534            output.push('\n');
535        }
536        for stage in &self.stages {
537            let per_call_us = if stage.calls == 0 {
538                0.0
539            } else {
540                stage.elapsed_ns as f64 / stage.calls as f64 / 1_000.0
541            };
542            output.push_str(&format!(
543                "  {:<24} {:>10.3} ms calls={} per_call_us={per_call_us:.3} attributed_ms={:.3}\n",
544                stage.stage.as_str(),
545                stage.elapsed_ns as f64 / 1_000_000.0,
546                stage.calls,
547                stage.attributed_ns as f64 / 1_000_000.0,
548            ));
549        }
550        if let Some(state) = self.states.iter().max_by_key(|state| state.elapsed_ns) {
551            output.push_str(&format!(
552                "bottleneck macro={} wall_ms={:.3}",
553                state_name(state.state),
554                state.elapsed_ns as f64 / 1_000_000.0,
555            ));
556        }
557        if let Some(stage) = self
558            .stages
559            .iter()
560            .filter(|stage| stage.stage != Stage::BackendDispatch)
561            .max_by_key(|stage| stage.elapsed_ns)
562        {
563            output.push_str(&format!(
564                " summed_stage={} summed_ms={:.3}",
565                stage.stage.as_str(),
566                stage.elapsed_ns as f64 / 1_000_000.0,
567            ));
568        }
569        if !self.states.is_empty() || !self.stages.is_empty() {
570            output.push('\n');
571        }
572        if let Some(cpu) = self.resources.aggregate_cpu_percent {
573            output.push_str(&format!("resources aggregate_cpu={cpu:.1}%"));
574        } else {
575            output.push_str("resources aggregate_cpu=unavailable");
576        }
577        if let Some(rss) = self.resources.max_observed_resident_bytes {
578            output.push_str(&format!(" max_observed_rss_bytes={rss}"));
579        }
580        if let Some(threads) = self.resources.max_observed_threads {
581            output.push_str(&format!(" max_observed_threads={threads}"));
582        }
583        output.push('\n');
584        for capability in &self.collectors {
585            output.push_str(&format!(
586                "collector {} availability={}",
587                capability.collector.as_str(),
588                capability.availability.as_str(),
589            ));
590            if let Some(detail) = &capability.detail {
591                output.push_str(&format!(" detail={detail}"));
592            }
593            output.push('\n');
594        }
595        output
596    }
597}
598
599fn state_name(state: RunState) -> &'static str {
600    match state {
601        RunState::Created => "created",
602        RunState::Acquiring => "acquiring",
603        RunState::Scanning => "scanning",
604        RunState::Resolving => "resolving",
605        RunState::Verifying => "verifying",
606        RunState::Reporting => "reporting",
607        RunState::Completed => "completed",
608        RunState::Failed => "failed",
609    }
610}
611
612fn cache_name(state: CacheState) -> &'static str {
613    match state {
614        CacheState::Unknown => "unknown",
615        CacheState::Disabled => "disabled",
616        CacheState::Cold => "cold",
617        CacheState::Warm => "warm",
618    }
619}
620
621fn daemon_name(state: DaemonState) -> &'static str {
622    match state {
623        DaemonState::Off => "off",
624        DaemonState::Client => "client",
625        DaemonState::Worker => "worker",
626        DaemonState::Mass => "mass",
627    }
628}
keyhog_profile/schema.rs

keyhog_profile/
schema.rs
