1use 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
27pub const MAX_CAPTURE_BYTES: u64 = 16 * 1024 * 1024;
29
30#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct CaptureConfig {
35 pub scope: Scope,
36 pub max_bytes: u64,
38}
39
40#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum CaptureConfigError {
43 ZeroLimit,
44 LimitTooLarge {
45 max_bytes: u64,
46 },
47 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
87pub(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 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 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 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 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 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 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 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}