player_plugin/
benchmark.rs

1use std::collections::BTreeMap;
2
3use serde::{Deserialize, Serialize};
4use thiserror::Error;
5
6use crate::{
7    MAX_PLUGIN_DIAGNOSTICS, MAX_PLUGIN_EVENT_ID_BYTES, MAX_PLUGIN_EVENT_NAME_BYTES,
8    MAX_PLUGIN_MEASUREMENTS, MAX_PLUGIN_PLATFORM_BYTES, MAX_PLUGIN_PROTOCOL_BYTES,
9    MAX_PLUGIN_THREAD_BYTES, PluginDiagnostic, PluginMeasurement, PluginProtocolViolation,
10    protocol::{validate_attributes, validate_optional_text, validate_text},
11};
12
13pub const MAX_BENCHMARK_BATCH_EVENTS: usize = 512;
14pub const MAX_BENCHMARK_THRESHOLD_VIOLATIONS: usize = 128;
15
16/// One high-resolution benchmark event emitted by a host playback session.
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(rename_all = "camelCase")]
19pub struct BenchmarkEvent {
20    pub run_id: String,
21    pub session_id: String,
22    pub platform: String,
23    pub source_protocol: Option<String>,
24    pub event_name: String,
25    pub timestamp_ns: u64,
26    pub elapsed_ns: u64,
27    pub thread: Option<String>,
28    #[serde(default)]
29    pub attributes: BTreeMap<String, String>,
30}
31
32impl BenchmarkEvent {
33    pub fn validate(&self) -> Result<(), BenchmarkSinkError> {
34        validate_text("benchmark.run_id", &self.run_id, MAX_PLUGIN_EVENT_ID_BYTES)?;
35        validate_text(
36            "benchmark.session_id",
37            &self.session_id,
38            MAX_PLUGIN_EVENT_ID_BYTES,
39        )?;
40        validate_text(
41            "benchmark.platform",
42            &self.platform,
43            MAX_PLUGIN_PLATFORM_BYTES,
44        )?;
45        validate_optional_text(
46            "benchmark.source_protocol",
47            self.source_protocol.as_deref(),
48            MAX_PLUGIN_PROTOCOL_BYTES,
49        )?;
50        validate_text(
51            "benchmark.event_name",
52            &self.event_name,
53            MAX_PLUGIN_EVENT_NAME_BYTES,
54        )?;
55        validate_optional_text(
56            "benchmark.thread",
57            self.thread.as_deref(),
58            MAX_PLUGIN_THREAD_BYTES,
59        )?;
60        validate_attributes(&self.attributes)?;
61        Ok(())
62    }
63}
64
65/// Batch payload sent from the host to a benchmark sink plugin.
66#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
67#[serde(rename_all = "camelCase")]
68pub struct BenchmarkEventBatch {
69    pub events: Vec<BenchmarkEvent>,
70}
71
72impl BenchmarkEventBatch {
73    pub fn validate(&self) -> Result<(), BenchmarkSinkError> {
74        if self.events.len() > MAX_BENCHMARK_BATCH_EVENTS {
75            return Err(BenchmarkSinkError::ProtocolViolation(format!(
76                "benchmark batch exceeds the {MAX_BENCHMARK_BATCH_EVENTS}-event protocol limit"
77            )));
78        }
79        for event in &self.events {
80            event.validate()?;
81        }
82        Ok(())
83    }
84}
85
86/// Lightweight acknowledgement returned after a sink receives one event batch.
87#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
88#[serde(rename_all = "camelCase")]
89pub struct BenchmarkSinkStatus {
90    pub accepted_events: u64,
91}
92
93impl BenchmarkSinkStatus {
94    pub fn validate_for_batch(&self, batch_event_count: usize) -> Result<(), BenchmarkSinkError> {
95        let batch_event_count = u64::try_from(batch_event_count).map_err(|_| {
96            BenchmarkSinkError::ProtocolViolation(
97                "benchmark batch size cannot be represented by the protocol".to_owned(),
98            )
99        })?;
100        if self.accepted_events > batch_event_count {
101            return Err(BenchmarkSinkError::ProtocolViolation(format!(
102                "benchmark sink accepted {} events from a {batch_event_count}-event batch",
103                self.accepted_events
104            )));
105        }
106        Ok(())
107    }
108}
109
110#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
111#[serde(rename_all = "camelCase")]
112pub struct BenchmarkThresholdViolation {
113    pub measurement: String,
114    pub actual: f64,
115    pub threshold: f64,
116    pub comparison: String,
117}
118
119/// Final report returned by a benchmark sink when the host flushes a run.
120#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
121#[serde(rename_all = "camelCase")]
122pub struct BenchmarkSinkReport {
123    pub accepted_events: u64,
124    pub dropped_events: u64,
125    #[serde(default)]
126    pub measurements: Vec<PluginMeasurement>,
127    #[serde(default)]
128    pub threshold_violations: Vec<BenchmarkThresholdViolation>,
129    #[serde(default)]
130    pub diagnostics: Vec<PluginDiagnostic>,
131}
132
133impl BenchmarkSinkReport {
134    pub fn validate(&self) -> Result<(), BenchmarkSinkError> {
135        if self.measurements.len() > MAX_PLUGIN_MEASUREMENTS {
136            return Err(BenchmarkSinkError::ProtocolViolation(format!(
137                "benchmark report exceeds the {MAX_PLUGIN_MEASUREMENTS}-measurement protocol limit"
138            )));
139        }
140        if self.diagnostics.len() > MAX_PLUGIN_DIAGNOSTICS {
141            return Err(BenchmarkSinkError::ProtocolViolation(format!(
142                "benchmark report exceeds the {MAX_PLUGIN_DIAGNOSTICS}-diagnostic protocol limit"
143            )));
144        }
145        if self.threshold_violations.len() > MAX_BENCHMARK_THRESHOLD_VIOLATIONS {
146            return Err(BenchmarkSinkError::ProtocolViolation(format!(
147                "benchmark report exceeds the {MAX_BENCHMARK_THRESHOLD_VIOLATIONS}-threshold-violation protocol limit"
148            )));
149        }
150        for measurement in &self.measurements {
151            measurement.validate()?;
152        }
153        for diagnostic in &self.diagnostics {
154            diagnostic.validate()?;
155        }
156        for violation in &self.threshold_violations {
157            validate_text(
158                "threshold_violation.measurement",
159                &violation.measurement,
160                MAX_PLUGIN_EVENT_NAME_BYTES,
161            )?;
162            validate_text(
163                "threshold_violation.comparison",
164                &violation.comparison,
165                MAX_PLUGIN_EVENT_NAME_BYTES,
166            )?;
167            if !violation.actual.is_finite() || !violation.threshold.is_finite() {
168                return Err(BenchmarkSinkError::ProtocolViolation(format!(
169                    "threshold violation `{}` contains a non-finite value",
170                    violation.measurement
171                )));
172            }
173        }
174        Ok(())
175    }
176}
177
178/// Error payload shared by benchmark sink plugins and host-side adapters.
179#[derive(Debug, Error, Clone, PartialEq, Eq, Serialize, Deserialize)]
180#[serde(rename_all = "camelCase", tag = "code", content = "message")]
181pub enum BenchmarkSinkError {
182    #[error("payload codec error: {0}")]
183    PayloadCodec(String),
184    #[error("plugin ABI violation: {0}")]
185    AbiViolation(String),
186    #[error("sink failed: {0}")]
187    SinkFailed(String),
188    #[error("sink protocol violation: {0}")]
189    ProtocolViolation(String),
190}
191
192impl From<PluginProtocolViolation> for BenchmarkSinkError {
193    fn from(value: PluginProtocolViolation) -> Self {
194        Self::ProtocolViolation(value.to_string())
195    }
196}
197
198pub trait BenchmarkSink: Send + Sync {
199    fn name(&self) -> &str;
200
201    fn on_event_batch(
202        &self,
203        batch: &BenchmarkEventBatch,
204    ) -> Result<BenchmarkSinkStatus, BenchmarkSinkError>;
205
206    fn flush(&self) -> Result<BenchmarkSinkReport, BenchmarkSinkError> {
207        Ok(BenchmarkSinkReport::default())
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214
215    #[test]
216    fn batch_limit_is_part_of_the_protocol() {
217        let event = BenchmarkEvent {
218            run_id: "run".to_owned(),
219            session_id: "session".to_owned(),
220            platform: "test".to_owned(),
221            source_protocol: None,
222            event_name: "tick".to_owned(),
223            timestamp_ns: 0,
224            elapsed_ns: 0,
225            thread: None,
226            attributes: BTreeMap::new(),
227        };
228        let batch = BenchmarkEventBatch {
229            events: vec![event; MAX_BENCHMARK_BATCH_EVENTS + 1],
230        };
231        assert!(matches!(
232            batch.validate(),
233            Err(BenchmarkSinkError::ProtocolViolation(_))
234        ));
235    }
236
237    #[test]
238    fn sink_status_cannot_accept_more_events_than_the_input_batch() {
239        let status = BenchmarkSinkStatus { accepted_events: 2 };
240        assert!(matches!(
241            status.validate_for_batch(1),
242            Err(BenchmarkSinkError::ProtocolViolation(_))
243        ));
244    }
245
246    #[test]
247    fn benchmark_events_validate_all_transport_text_fields() {
248        let event = BenchmarkEvent {
249            run_id: String::new(),
250            session_id: "session".to_owned(),
251            platform: "test".to_owned(),
252            source_protocol: None,
253            event_name: "tick".to_owned(),
254            timestamp_ns: 0,
255            elapsed_ns: 0,
256            thread: None,
257            attributes: BTreeMap::new(),
258        };
259
260        assert!(matches!(
261            event.validate(),
262            Err(BenchmarkSinkError::ProtocolViolation(message))
263                if message.contains("benchmark.run_id")
264        ));
265    }
266
267    #[test]
268    fn benchmark_reports_bound_threshold_violations() {
269        let violation = BenchmarkThresholdViolation {
270            measurement: "latency".to_owned(),
271            actual: 2.0,
272            threshold: 1.0,
273            comparison: "greater-than".to_owned(),
274        };
275        let report = BenchmarkSinkReport {
276            threshold_violations: vec![violation; MAX_BENCHMARK_THRESHOLD_VIOLATIONS + 1],
277            ..BenchmarkSinkReport::default()
278        };
279
280        assert!(matches!(
281            report.validate(),
282            Err(BenchmarkSinkError::ProtocolViolation(message))
283                if message.contains("threshold-violation")
284        ));
285    }
286}
player_plugin/benchmark.rs

player_plugin/
benchmark.rs
