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    /// Start over for the rerun of a decision abandoned under `recompute`:
99    /// the capture holds the rerun's supplies alone (spec 044, 3.6.5 item 3).
100    pub(crate) fn restart(&mut self) {
101        self.bytes = 0;
102        self.exceeded = false;
103        self.supplies.clear();
104        self.pages.clear();
105    }
106
107    pub fn new(config: &'a CaptureConfig) -> Self {
108        Capturing {
109            config,
110            bytes: 0,
111            exceeded: false,
112            supplies: vec![],
113            pages: vec![],
114        }
115    }
116
117    /// Record `supplied` for `record`'s request, which must still be pending
118    /// in `ev`: call right before supplying it.
119    pub fn record(
120        &mut self,
121        ev: &Evaluation<'_>,
122        plan: &PreparedPlan,
123        record: &RequestRecord,
124        supplied: &Supplied,
125        failure_context: Option<rustev_core::FailureContext>,
126    ) {
127        if self.exceeded {
128            return;
129        }
130        let request = record.supplied_from().and_then(|from| {
131            ev.request_document(
132                &record.step,
133                &record.instance,
134                from.target,
135                &self.config.scope,
136            )
137            .ok()?
138            .id()
139            .ok()
140        });
141        let mut entry = request.and_then(|request| self.entry(request, plan, record, supplied));
142        if let Some((entry, _)) = &mut entry
143            && plan.compiled.plan.schema == schema::PLAN_V3
144        {
145            entry.failure_context =
146                failure_context.map(|c| rustev_contract::replay::CapturedFailureContext {
147                    origin: c.origin,
148                    determinism: c.determinism,
149                });
150        }
151        self.push(entry);
152    }
153
154    /// Capture generation before supplying it, including a cache hit's origin.
155    pub fn record_generation(
156        &mut self,
157        stage: &rustev_core::generation_eval::GenerationEvaluation<'_>,
158        plan: &PreparedPlan,
159        record: &RequestRecord,
160        supplied: &Supplied,
161    ) {
162        if self.exceeded {
163            return;
164        }
165        let request = record.supplied_from().and_then(|from| {
166            stage
167                .request_document(
168                    &record.step,
169                    &record.instance,
170                    from.target,
171                    &self.config.scope,
172                )
173                .ok()?
174                .id()
175                .ok()
176        });
177        let entry = request.and_then(|request| self.entry(request, plan, record, supplied));
178        self.push(entry);
179    }
180
181    /// Keep an entry within both caps, or discard every payload. An entry
182    /// that could not be built is a value that cannot be retained, so it
183    /// cannot be within any limit: the capture is never `complete` without
184    /// it.
185    fn push(&mut self, entry: Option<(CapturedSupply, u64)>) {
186        if self.exceeded {
187            return;
188        }
189        let Some((entry, bytes)) = entry else {
190            self.discard();
191            return;
192        };
193        self.bytes = self.bytes.saturating_add(bytes);
194        if self.bytes > self.config.max_bytes || self.supplies.len() >= MAX_BUNDLE_ITEMS {
195            self.discard();
196        } else {
197            self.supplies.push(entry);
198        }
199    }
200
201    fn discard(&mut self) {
202        self.exceeded = true;
203        self.supplies = vec![];
204        self.pages = vec![];
205    }
206
207    fn entry(
208        &self,
209        request: rustev_contract::ids::RequestId,
210        plan: &PreparedPlan,
211        record: &RequestRecord,
212        supplied: &Supplied,
213    ) -> Option<(CapturedSupply, u64)> {
214        let from = record.supplied_from()?;
215        let (value, len) = match supplied {
216            Supplied::Output(raw) => {
217                let artifact = plan
218                    .steps
219                    .get(&record.step)?
220                    .targets
221                    .get(from.target as usize)?
222                    .artifact
223                    .clone();
224                let out = BackendOutputDoc {
225                    schema: schema::BACKEND_OUTPUT.into(),
226                    step: record.step.clone(),
227                    instance: record.instance.clone(),
228                    artifact,
229                    output: raw.clone(),
230                };
231                let len = out.record_canonical().ok()?.len();
232                (CapturedValue::Output(out), len)
233            }
234            Supplied::Failed(u) => {
235                let r = RuntimeReasonDoc::new(u.clone()).ok()?;
236                let len = r.record_canonical().ok()?.len();
237                (CapturedValue::Reason(r), len)
238            }
239        };
240        let bytes = (len + request.as_str().len()) as u64;
241        Some((
242            CapturedSupply {
243                failure_context: None,
244                sequence: (self.supplies.len() + self.pages.len()) as u64,
245                step: record.step.clone(),
246                instance: record.instance.clone(),
247                request,
248                attempt_id: from
249                    .attempt
250                    .map(|a| a.attempt_id.clone())
251                    .or_else(|| from.reused_from.map(str::to_string)),
252                target: from.target,
253                value,
254            },
255            bytes,
256        ))
257    }
258
259    /// Capture a page before the driver supplies it, using the shared sequence.
260    pub fn record_page(
261        &mut self,
262        record: &rustev_contract::acquisition_record::AcquisitionRecord,
263        bytes: Option<&[u8]>,
264    ) {
265        use rustev_contract::acquisition_record::AcquisitionResult;
266        if self.exceeded || matches!(record.result, AcquisitionResult::NotSupplied) {
267            return;
268        }
269        let value = match (&record.result, bytes) {
270            (AcquisitionResult::Page, Some(bytes)) => {
271                rustev_contract::acquisition::AcquisitionPage::parse(bytes)
272                    .ok()
273                    .map(CapturedPage::Page)
274            }
275            (AcquisitionResult::Failed(reason), _) => {
276                Some(CapturedPage::Reason(RuntimeReasonDoc {
277                    schema: schema::RUNTIME_REASON.into(),
278                    reason: reason.clone(),
279                }))
280            }
281            _ => None,
282        };
283        let Some(value) = value else {
284            self.discard();
285            return;
286        };
287        let canonical = match &value {
288            CapturedPage::Page(p) => p.record_canonical(),
289            CapturedPage::Reason(r) => r.record_canonical(),
290        };
291        let Ok(bytes) = canonical else {
292            self.discard();
293            return;
294        };
295        self.bytes = self
296            .bytes
297            .saturating_add((bytes.len() + record.request.as_str().len()) as u64);
298        // The bundle bounds each member on its own (`ReplayBundle::check`):
299        // `acquisition_pages` and `supplies` each hold at most MAX_BUNDLE_ITEMS,
300        // and the shared sequence is only required to increase.
301        if self.bytes > self.config.max_bytes || self.pages.len() >= MAX_BUNDLE_ITEMS {
302            self.discard();
303            return;
304        }
305        self.pages.push(CapturedAcquisition {
306            sequence: (self.supplies.len() + self.pages.len()) as u64,
307            step: record.step.clone(),
308            tier: record.tier,
309            page: record.page,
310            request: record.request.clone(),
311            attempt_id: record.attempts.last().map(|a| a.attempt_id.clone()),
312            value,
313        });
314    }
315
316    /// The capture of a decision that ran to `cancelled` or a judgment.
317    pub fn finish(self, decision_id: &str, cancelled: bool) -> Capture {
318        let status = if self.exceeded {
319            CaptureStatus::LimitExceeded
320        } else if cancelled {
321            CaptureStatus::Cancelled
322        } else {
323            CaptureStatus::Complete
324        };
325        Capture {
326            decision_id: decision_id.to_string(),
327            scope: self.config.scope.clone(),
328            status,
329            supplies: self.supplies,
330            acquisition_pages: self.pages,
331            bytes: self.bytes,
332        }
333    }
334}
335
336#[cfg(test)]
337mod tests {
338    use super::*;
339    use rustev_contract::ids::RequestId;
340    use rustev_contract::scope::Handle;
341
342    fn config(max_bytes: u64) -> CaptureConfig {
343        CaptureConfig {
344            scope: Scope::tenant_only(Handle::new("t").unwrap(), Handle::new("r").unwrap()),
345            max_bytes,
346        }
347    }
348
349    fn entry(i: u64) -> Option<(CapturedSupply, u64)> {
350        let reason =
351            RuntimeReasonDoc::new(rustev_contract::judgment::Unresolved::DeadlineExceeded).unwrap();
352        Some((
353            CapturedSupply {
354                failure_context: None,
355                sequence: i,
356                step: format!("s{i}"),
357                instance: vec![],
358                request: RequestId::parse(&format!("sha256:{}", "a".repeat(64))).unwrap(),
359                attempt_id: None,
360                target: 0,
361                value: CapturedValue::Reason(reason),
362            },
363            10,
364        ))
365    }
366
367    #[test]
368    fn config_bounds_are_inclusive() {
369        assert_eq!(config(0).check(), Err(CaptureConfigError::ZeroLimit));
370        assert!(config(1).check().is_ok());
371        assert!(config(MAX_CAPTURE_BYTES).check().is_ok());
372        assert!(config(MAX_CAPTURE_BYTES + 1).check().is_err());
373    }
374
375    #[test]
376    fn an_entry_that_cannot_be_built_discards_everything() {
377        let c = config(1_000);
378        let mut cap = Capturing::new(&c);
379        cap.push(entry(0));
380        cap.push(None);
381        cap.push(entry(2));
382        let done = cap.finish("d", false);
383        assert_eq!(done.status, CaptureStatus::LimitExceeded);
384        assert!(done.supplies.is_empty());
385    }
386
387    #[test]
388    fn overflowing_after_a_kept_entry_discards_it_too() {
389        let c = config(15);
390        let mut cap = Capturing::new(&c);
391        cap.push(entry(0));
392        assert_eq!(cap.supplies.len(), 1);
393        cap.push(entry(1));
394        let done = cap.finish("d", false);
395        assert_eq!(done.status, CaptureStatus::LimitExceeded);
396        assert!(done.supplies.is_empty());
397        assert_eq!(done.bytes, 20);
398        // Negative control: exactly at the limit is complete.
399        let c = config(20);
400        let mut cap = Capturing::new(&c);
401        cap.push(entry(0));
402        cap.push(entry(1));
403        assert_eq!(cap.finish("d", false).status, CaptureStatus::Complete);
404    }
405
406    #[test]
407    fn at_most_4096_entries_are_kept() {
408        let c = config(MAX_CAPTURE_BYTES);
409        let mut cap = Capturing::new(&c);
410        for i in 0..MAX_BUNDLE_ITEMS as u64 {
411            cap.push(entry(i));
412        }
413        assert_eq!(cap.supplies.len(), MAX_BUNDLE_ITEMS);
414        assert!(!cap.exceeded);
415        cap.push(entry(MAX_BUNDLE_ITEMS as u64));
416        let done = cap.finish("d", false);
417        assert_eq!(done.status, CaptureStatus::LimitExceeded);
418        assert!(done.supplies.is_empty());
419    }
420
421    #[test]
422    fn a_cap_overrun_outranks_cancellation() {
423        let c = config(5);
424        let mut cap = Capturing::new(&c);
425        cap.push(entry(0));
426        let done = cap.finish("d", true);
427        assert_eq!(done.status, CaptureStatus::LimitExceeded);
428        assert!(done.supplies.is_empty());
429        // Negative control: a cancelled capture within its cap keeps entries.
430        let c = config(100);
431        let mut cap = Capturing::new(&c);
432        cap.push(entry(0));
433        let done = cap.finish("d", true);
434        assert_eq!(done.status, CaptureStatus::Cancelled);
435        assert_eq!(done.supplies.len(), 1);
436    }
437}
rustev_runtime/capture.rs

rustev_runtime/
capture.rs
