rustev_runtime/
capture.rs

1//! Opt-in bounded capture of supplied values (spec 004, 5.3, 5.4 and 5.7).
2//!
3//! Capture records each value the runtime supplies to the core, in actual
4//! supply order, with its request identity for the actual target and its
5//! producing attempt. It is an in-memory handoff returned beside the normal
6//! result: never persisted, never sent to the evidence sink, never a
7//! callback. Exceeding its byte or entry cap discards every captured payload
8//! and reports `limit_exceeded`; the decision itself is unaffected.
9
10use std::fmt;
11
12use rustev_contract::Document;
13use rustev_contract::Identified;
14use rustev_contract::output::BackendOutputDoc;
15use rustev_contract::reason::RuntimeReasonDoc;
16use rustev_contract::replay::{
17    Capture, CaptureStatus, CapturedAcquisition, CapturedPage, CapturedSupply, CapturedValue,
18    MAX_BUNDLE_ITEMS,
19};
20use rustev_contract::run::RequestRecord;
21use rustev_contract::schema;
22use rustev_contract::scope::Scope;
23use rustev_core::evaluate::{Evaluation, Supplied};
24
25use crate::PreparedPlan;
26
27/// The largest byte limit a capture may be configured with.
28pub const MAX_CAPTURE_BYTES: u64 = 16 * 1024 * 1024;
29
30/// What to capture under. The scope is the host's isolation choice for
31/// retained evidence; it is separate from, and never changes, the principal
32/// handle passed to backends.
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct CaptureConfig {
35    pub scope: Scope,
36    /// Positive, at most [`MAX_CAPTURE_BYTES`].
37    pub max_bytes: u64,
38}
39
40/// A capture configuration refused before admission.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum CaptureConfigError {
43    ZeroLimit,
44    LimitTooLarge {
45        max_bytes: u64,
46    },
47    /// A plan whose items no replay bundle admits yet (spec 032, 3.11.3).
48    CaptureUnsupported {
49        plan_schema: &'static str,
50    },
51}
52
53impl fmt::Display for CaptureConfigError {
54    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
55        match self {
56            CaptureConfigError::ZeroLimit => f.write_str("a capture byte limit is positive"),
57            CaptureConfigError::LimitTooLarge { max_bytes } => write!(
58                f,
59                "a capture byte limit is at most {MAX_CAPTURE_BYTES}, not {max_bytes}"
60            ),
61            CaptureConfigError::CaptureUnsupported { plan_schema } => {
62                write!(
63                    f,
64                    "capture_unsupported: no replay bundle admits a {plan_schema}"
65                )
66            }
67        }
68    }
69}
70
71impl std::error::Error for CaptureConfigError {}
72
73impl CaptureConfig {
74    pub fn check(&self) -> Result<(), CaptureConfigError> {
75        if self.max_bytes == 0 {
76            return Err(CaptureConfigError::ZeroLimit);
77        }
78        if self.max_bytes > MAX_CAPTURE_BYTES {
79            return Err(CaptureConfigError::LimitTooLarge {
80                max_bytes: self.max_bytes,
81            });
82        }
83        Ok(())
84    }
85}
86
87/// A capture in progress. `bytes` counts up to and including the entry
88/// that overflowed a cap; nothing is counted after a discard.
89pub(crate) struct Capturing<'a> {
90    config: &'a CaptureConfig,
91    bytes: u64,
92    exceeded: bool,
93    supplies: Vec<CapturedSupply>,
94    pages: Vec<CapturedAcquisition>,
95}
96
97impl<'a> Capturing<'a> {
98    pub fn new(config: &'a CaptureConfig) -> Self {
99        Capturing {
100            config,
101            bytes: 0,
102            exceeded: false,
103            supplies: vec![],
104            pages: vec![],
105        }
106    }
107
108    /// Record `supplied` for `record`'s request, which must still be pending
109    /// in `ev`: call right before supplying it.
110    pub fn record(
111        &mut self,
112        ev: &Evaluation<'_>,
113        plan: &PreparedPlan,
114        record: &RequestRecord,
115        supplied: &Supplied,
116        failure_context: Option<rustev_core::FailureContext>,
117    ) {
118        if self.exceeded {
119            return;
120        }
121        let request = record.supplied_from().and_then(|from| {
122            ev.request_document(
123                &record.step,
124                &record.instance,
125                from.target,
126                &self.config.scope,
127            )
128            .ok()?
129            .id()
130            .ok()
131        });
132        let mut entry = request.and_then(|request| self.entry(request, plan, record, supplied));
133        if let Some((entry, _)) = &mut entry
134            && plan.compiled.plan.schema == schema::PLAN_V3
135        {
136            entry.failure_context =
137                failure_context.map(|c| rustev_contract::replay::CapturedFailureContext {
138                    origin: c.origin,
139                    determinism: c.determinism,
140                });
141        }
142        self.push(entry);
143    }
144
145    /// Capture generation before supplying it, including a cache hit's origin.
146    pub fn record_generation(
147        &mut self,
148        stage: &rustev_core::generation_eval::GenerationEvaluation<'_>,
149        plan: &PreparedPlan,
150        record: &RequestRecord,
151        supplied: &Supplied,
152    ) {
153        if self.exceeded {
154            return;
155        }
156        let request = record.supplied_from().and_then(|from| {
157            stage
158                .request_document(
159                    &record.step,
160                    &record.instance,
161                    from.target,
162                    &self.config.scope,
163                )
164                .ok()?
165                .id()
166                .ok()
167        });
168        let entry = request.and_then(|request| self.entry(request, plan, record, supplied));
169        self.push(entry);
170    }
171
172    /// Keep an entry within both caps, or discard every payload. An entry
173    /// that could not be built is a value that cannot be retained, so it
174    /// cannot be within any limit: the capture is never `complete` without
175    /// it.
176    fn push(&mut self, entry: Option<(CapturedSupply, u64)>) {
177        if self.exceeded {
178            return;
179        }
180        let Some((entry, bytes)) = entry else {
181            self.discard();
182            return;
183        };
184        self.bytes = self.bytes.saturating_add(bytes);
185        if self.bytes > self.config.max_bytes || self.supplies.len() >= MAX_BUNDLE_ITEMS {
186            self.discard();
187        } else {
188            self.supplies.push(entry);
189        }
190    }
191
192    fn discard(&mut self) {
193        self.exceeded = true;
194        self.supplies = vec![];
195        self.pages = vec![];
196    }
197
198    fn entry(
199        &self,
200        request: rustev_contract::ids::RequestId,
201        plan: &PreparedPlan,
202        record: &RequestRecord,
203        supplied: &Supplied,
204    ) -> Option<(CapturedSupply, u64)> {
205        let from = record.supplied_from()?;
206        let (value, len) = match supplied {
207            Supplied::Output(raw) => {
208                let artifact = plan
209                    .steps
210                    .get(&record.step)?
211                    .targets
212                    .get(from.target as usize)?
213                    .artifact
214                    .clone();
215                let out = BackendOutputDoc {
216                    schema: schema::BACKEND_OUTPUT.into(),
217                    step: record.step.clone(),
218                    instance: record.instance.clone(),
219                    artifact,
220                    output: raw.clone(),
221                };
222                let len = out.record_canonical().ok()?.len();
223                (CapturedValue::Output(out), len)
224            }
225            Supplied::Failed(u) => {
226                let r = RuntimeReasonDoc::new(u.clone()).ok()?;
227                let len = r.record_canonical().ok()?.len();
228                (CapturedValue::Reason(r), len)
229            }
230        };
231        let bytes = (len + request.as_str().len()) as u64;
232        Some((
233            CapturedSupply {
234                failure_context: None,
235                sequence: (self.supplies.len() + self.pages.len()) as u64,
236                step: record.step.clone(),
237                instance: record.instance.clone(),
238                request,
239                attempt_id: from
240                    .attempt
241                    .map(|a| a.attempt_id.clone())
242                    .or_else(|| from.reused_from.map(str::to_string)),
243                target: from.target,
244                value,
245            },
246            bytes,
247        ))
248    }
249
250    /// Capture a page before the driver supplies it, using the shared sequence.
251    pub fn record_page(
252        &mut self,
253        record: &rustev_contract::acquisition_record::AcquisitionRecord,
254        bytes: Option<&[u8]>,
255    ) {
256        use rustev_contract::acquisition_record::AcquisitionResult;
257        if self.exceeded || matches!(record.result, AcquisitionResult::NotSupplied) {
258            return;
259        }
260        let value = match (&record.result, bytes) {
261            (AcquisitionResult::Page, Some(bytes)) => {
262                rustev_contract::acquisition::AcquisitionPage::parse(bytes)
263                    .ok()
264                    .map(CapturedPage::Page)
265            }
266            (AcquisitionResult::Failed(reason), _) => {
267                Some(CapturedPage::Reason(RuntimeReasonDoc {
268                    schema: schema::RUNTIME_REASON.into(),
269                    reason: reason.clone(),
270                }))
271            }
272            _ => None,
273        };
274        let Some(value) = value else {
275            self.discard();
276            return;
277        };
278        let canonical = match &value {
279            CapturedPage::Page(p) => p.record_canonical(),
280            CapturedPage::Reason(r) => r.record_canonical(),
281        };
282        let Ok(bytes) = canonical else {
283            self.discard();
284            return;
285        };
286        self.bytes = self
287            .bytes
288            .saturating_add((bytes.len() + record.request.as_str().len()) as u64);
289        // The bundle bounds each member on its own (`ReplayBundle::check`):
290        // `acquisition_pages` and `supplies` each hold at most MAX_BUNDLE_ITEMS,
291        // and the shared sequence is only required to increase.
292        if self.bytes > self.config.max_bytes || self.pages.len() >= MAX_BUNDLE_ITEMS {
293            self.discard();
294            return;
295        }
296        self.pages.push(CapturedAcquisition {
297            sequence: (self.supplies.len() + self.pages.len()) as u64,
298            step: record.step.clone(),
299            tier: record.tier,
300            page: record.page,
301            request: record.request.clone(),
302            attempt_id: record.attempts.last().map(|a| a.attempt_id.clone()),
303            value,
304        });
305    }
306
307    /// The capture of a decision that ran to `cancelled` or a judgment.
308    pub fn finish(self, decision_id: &str, cancelled: bool) -> Capture {
309        let status = if self.exceeded {
310            CaptureStatus::LimitExceeded
311        } else if cancelled {
312            CaptureStatus::Cancelled
313        } else {
314            CaptureStatus::Complete
315        };
316        Capture {
317            decision_id: decision_id.to_string(),
318            scope: self.config.scope.clone(),
319            status,
320            supplies: self.supplies,
321            acquisition_pages: self.pages,
322            bytes: self.bytes,
323        }
324    }
325}
326
327#[cfg(test)]
328mod tests {
329    use super::*;
330    use rustev_contract::ids::RequestId;
331    use rustev_contract::scope::Handle;
332
333    fn config(max_bytes: u64) -> CaptureConfig {
334        CaptureConfig {
335            scope: Scope::tenant_only(Handle::new("t").unwrap(), Handle::new("r").unwrap()),
336            max_bytes,
337        }
338    }
339
340    fn entry(i: u64) -> Option<(CapturedSupply, u64)> {
341        let reason =
342            RuntimeReasonDoc::new(rustev_contract::judgment::Unresolved::DeadlineExceeded).unwrap();
343        Some((
344            CapturedSupply {
345                failure_context: None,
346                sequence: i,
347                step: format!("s{i}"),
348                instance: vec![],
349                request: RequestId::parse(&format!("sha256:{}", "a".repeat(64))).unwrap(),
350                attempt_id: None,
351                target: 0,
352                value: CapturedValue::Reason(reason),
353            },
354            10,
355        ))
356    }
357
358    #[test]
359    fn config_bounds_are_inclusive() {
360        assert_eq!(config(0).check(), Err(CaptureConfigError::ZeroLimit));
361        assert!(config(1).check().is_ok());
362        assert!(config(MAX_CAPTURE_BYTES).check().is_ok());
363        assert!(config(MAX_CAPTURE_BYTES + 1).check().is_err());
364    }
365
366    #[test]
367    fn an_entry_that_cannot_be_built_discards_everything() {
368        let c = config(1_000);
369        let mut cap = Capturing::new(&c);
370        cap.push(entry(0));
371        cap.push(None);
372        cap.push(entry(2));
373        let done = cap.finish("d", false);
374        assert_eq!(done.status, CaptureStatus::LimitExceeded);
375        assert!(done.supplies.is_empty());
376    }
377
378    #[test]
379    fn overflowing_after_a_kept_entry_discards_it_too() {
380        let c = config(15);
381        let mut cap = Capturing::new(&c);
382        cap.push(entry(0));
383        assert_eq!(cap.supplies.len(), 1);
384        cap.push(entry(1));
385        let done = cap.finish("d", false);
386        assert_eq!(done.status, CaptureStatus::LimitExceeded);
387        assert!(done.supplies.is_empty());
388        assert_eq!(done.bytes, 20);
389        // Negative control: exactly at the limit is complete.
390        let c = config(20);
391        let mut cap = Capturing::new(&c);
392        cap.push(entry(0));
393        cap.push(entry(1));
394        assert_eq!(cap.finish("d", false).status, CaptureStatus::Complete);
395    }
396
397    #[test]
398    fn at_most_4096_entries_are_kept() {
399        let c = config(MAX_CAPTURE_BYTES);
400        let mut cap = Capturing::new(&c);
401        for i in 0..MAX_BUNDLE_ITEMS as u64 {
402            cap.push(entry(i));
403        }
404        assert_eq!(cap.supplies.len(), MAX_BUNDLE_ITEMS);
405        assert!(!cap.exceeded);
406        cap.push(entry(MAX_BUNDLE_ITEMS as u64));
407        let done = cap.finish("d", false);
408        assert_eq!(done.status, CaptureStatus::LimitExceeded);
409        assert!(done.supplies.is_empty());
410    }
411
412    #[test]
413    fn a_cap_overrun_outranks_cancellation() {
414        let c = config(5);
415        let mut cap = Capturing::new(&c);
416        cap.push(entry(0));
417        let done = cap.finish("d", true);
418        assert_eq!(done.status, CaptureStatus::LimitExceeded);
419        assert!(done.supplies.is_empty());
420        // Negative control: a cancelled capture within its cap keeps entries.
421        let c = config(100);
422        let mut cap = Capturing::new(&c);
423        cap.push(entry(0));
424        let done = cap.finish("d", true);
425        assert_eq!(done.status, CaptureStatus::Cancelled);
426        assert_eq!(done.supplies.len(), 1);
427    }
428}
rustev_runtime/capture.rs

rustev_runtime/
capture.rs
