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

rustev_runtime/
capture.rs
