rustev_contract/
live.rs

1//! Live re-execution evidence, `rustev.live-reexecution/1` (spec 026).
2//!
3//! A live re-execution is a new decision: a new decision id, a newly
4//! identified snapshot, a fresh evaluation time, fresh admission and budgets
5//! and new runtime observations. Its run record (`rustev.run/1`) remains the
6//! evidence of execution. This document is a separate, bounded companion
7//! that binds what the run record cannot: the link to a historical case (a
8//! digest plus identities, never historical content), the host's
9//! authorization answer for every proposed external attempt, every effect
10//! with its charge and remote state, the evidence delivery observation, the
11//! capture and retention posture, and an optional comparison.
12//!
13//! Rustev defines no authorization token. What the runtime presents to the
14//! host before an attempt is an [`AuthorizationRequest`]; what this document
15//! keeps of the answer is the host's opaque, non-secret
16//! [`AuthorizationReference`] or the refusal. Neither is permission, neither
17//! authorizes any downstream action, and this document is not proof that
18//! the host's authorization decision was correct (spec 026, 3.3 and 3.6).
19
20use std::fmt;
21
22use serde::{Deserialize, Deserializer, Serialize, Serializer};
23
24use crate::canonical::{CanonicalError, canonical_bytes, tagged_digest};
25use crate::definition::Operation;
26use crate::ids::{ArtifactId, ContentDigest, PlanId, RequestId, SnapshotId};
27use crate::limits::REPLAY_V1;
28use crate::replay::{CaptureStatus, ReplayBundle};
29use crate::retention::Availability;
30use crate::run::{
31    AttemptEnd, Cancellation, Charge, CostBound, CostSummary, RemoteState, RunRecord, Termination,
32};
33use crate::scope::{Handle, Scope};
34use crate::time::Timestamp;
35use crate::{Document, schema};
36
37/// The most bytes of an authorization decision reference.
38pub const MAX_REFERENCE_BYTES: usize = 256;
39/// The most effects one live document lists: one per proposed attempt, so
40/// at most the plan's attempt bound, and never more than this.
41pub const MAX_EFFECTS: usize = 4096;
42/// The most named fields, exclusions or limitations of one comparison.
43pub const MAX_NOTES: usize = 64;
44/// The most bytes of one free-text member (a detail, a receipt, a note).
45pub const MAX_TEXT_BYTES: usize = 256;
46/// The most historical item observations (a bundle's items plus itself).
47pub const MAX_HISTORICAL_ITEMS: usize = 3 * crate::replay::MAX_BUNDLE_ITEMS + 4;
48
49/// The tag of an authorization request's digest. The request is presented
50/// in process and is not a wire document; its digest binds an answer to
51/// exactly one proposed attempt.
52pub const AUTHORIZATION_REQUEST_TAG: &str = "rustev.live-authorization/1";
53
54// ---------------------------------------------------------------------------
55// Authorization at dispatch (spec 026, 3.3)
56// ---------------------------------------------------------------------------
57
58/// Why a string is not an authorization reference.
59#[derive(Debug, Clone, PartialEq, Eq)]
60pub struct ReferenceError {
61    pub len: usize,
62}
63
64impl fmt::Display for ReferenceError {
65    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
66        write!(
67            f,
68            "an authorization reference is 1 to {MAX_REFERENCE_BYTES} bytes, not {}",
69            self.len
70        )
71    }
72}
73
74impl std::error::Error for ReferenceError {}
75
76/// The host's opaque, non-secret reference to its own authorization
77/// decision, kept for evidence. Rustev compares and records it, and never
78/// interprets it: it is not a token, a credential or a permission, and it
79/// authorizes nothing.
80#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
81pub struct AuthorizationReference(String);
82
83impl AuthorizationReference {
84    /// Refuse an empty or oversized reference before keeping it.
85    pub fn new(s: impl Into<String>) -> Result<Self, ReferenceError> {
86        let s = s.into();
87        if s.is_empty() || s.len() > MAX_REFERENCE_BYTES {
88            return Err(ReferenceError { len: s.len() });
89        }
90        Ok(Self(s))
91    }
92
93    pub fn as_str(&self) -> &str {
94        &self.0
95    }
96}
97
98impl Serialize for AuthorizationReference {
99    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
100        s.serialize_str(&self.0)
101    }
102}
103
104impl<'de> Deserialize<'de> for AuthorizationReference {
105    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
106        AuthorizationReference::new(String::deserialize(d)?).map_err(serde::de::Error::custom)
107    }
108}
109
110impl schemars::JsonSchema for AuthorizationReference {
111    fn schema_name() -> std::borrow::Cow<'static, str> {
112        "AuthorizationReference".into()
113    }
114
115    fn json_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
116        schemars::json_schema!({
117            "type": "string",
118            "minLength": 1,
119            "maxLength": MAX_REFERENCE_BYTES
120        })
121    }
122}
123
124/// Which kind of external attempt is proposed.
125#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
126#[serde(rename_all = "snake_case")]
127pub enum AttemptRole {
128    /// The first attempt on the bound backend.
129    Primary,
130    /// A further attempt on the same target.
131    Retry,
132    /// The first attempt on a declared fallback target.
133    Fallback,
134}
135
136/// The worst-case cost of one proposed attempt: the backend's disclosure and
137/// the units reserved for it before it was presented.
138#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
139#[serde(deny_unknown_fields)]
140pub struct Exposure {
141    pub bound: CostBound,
142    pub reserved_units: u64,
143}
144
145/// What the runtime presents to the host immediately before one external
146/// attempt (spec 026, 3.3): the exact decision, scope, backend, artifact,
147/// operation, canonical request identity, maximum cost exposure, remaining
148/// time and role. It carries opaque handles and identities only, never a
149/// credential.
150#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
151#[serde(deny_unknown_fields)]
152pub struct AuthorizationRequest {
153    pub decision_id: String,
154    pub subject: Handle,
155    pub scope: Scope,
156    pub step: String,
157    pub instance: Vec<String>,
158    pub operation: Operation,
159    pub target: u32,
160    pub role: AttemptRole,
161    pub backend_id: String,
162    pub artifact: ArtifactId,
163    /// The `RequestId` of this request on this target (spec 004, 3.4).
164    pub request: RequestId,
165    /// The attempt's idempotency key (spec 003, 3.9.2).
166    pub attempt_id: String,
167    pub exposure: Exposure,
168    /// Time left before the decision's deadline, by the runtime's clock.
169    pub remaining_ms: u64,
170}
171
172impl AuthorizationRequest {
173    /// `sha256(AUTHORIZATION_REQUEST_TAG || 0x00 || canonical bytes)`: what
174    /// an answer must name to answer exactly this attempt.
175    pub fn digest(&self) -> Result<ContentDigest, CanonicalError> {
176        Ok(ContentDigest::from_digest(tagged_digest(
177            AUTHORIZATION_REQUEST_TAG,
178            &canonical_bytes(self)?,
179        )))
180    }
181}
182
183/// Why an attempt was not authorized. Each prevents dispatch and is
184/// recorded as `authorization_refused` (spec 026, 3.3).
185#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
186#[serde(rename_all = "snake_case")]
187pub enum AuthorizationRefusal {
188    /// The host gave no answer.
189    Missing,
190    /// The answer's validity ended before dispatch.
191    Expired,
192    /// The answer names a different request.
193    Mismatched,
194    /// The host's authorizer failed or panicked.
195    Errored,
196    /// The host refused.
197    Refused,
198}
199
200impl AuthorizationRefusal {
201    /// The typed runtime reason every refusal is recorded under.
202    pub const CODE: &'static str = "authorization_refused";
203
204    pub fn name(self) -> &'static str {
205        match self {
206            AuthorizationRefusal::Missing => "missing",
207            AuthorizationRefusal::Expired => "expired",
208            AuthorizationRefusal::Mismatched => "mismatched",
209            AuthorizationRefusal::Errored => "errored",
210            AuthorizationRefusal::Refused => "refused",
211        }
212    }
213}
214
215impl fmt::Display for AuthorizationRefusal {
216    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
217        write!(f, "{}: {}", Self::CODE, self.name())
218    }
219}
220
221/// What the host answered for one proposed attempt, as kept for evidence.
222#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
223#[serde(rename_all = "snake_case", deny_unknown_fields)]
224pub enum AuthorizationOutcome {
225    Allowed {
226        reference: AuthorizationReference,
227    },
228    Refused {
229        refusal: AuthorizationRefusal,
230        /// The host's reference, when its answer carried one.
231        reference: Option<AuthorizationReference>,
232        /// At most 256 bytes.
233        detail: String,
234    },
235}
236
237/// Whether a proposed attempt was sent, and what it observably did.
238#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
239#[serde(rename_all = "snake_case", deny_unknown_fields)]
240pub enum Dispatch {
241    /// Nothing was sent.
242    NotDispatched,
243    Dispatched {
244        end: AttemptEnd,
245        cancellation: Cancellation,
246        remote: RemoteState,
247        charge: Charge,
248    },
249}
250
251/// One proposed external attempt of the new decision: what was presented,
252/// the host's answer, and the effect. Every attempted effect is attributable
253/// to the new decision by its attempt id (spec 026, 3.6).
254#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
255#[serde(deny_unknown_fields)]
256pub struct LiveEffect {
257    pub step: String,
258    pub instance: Vec<String>,
259    pub target: u32,
260    pub role: AttemptRole,
261    pub attempt_id: String,
262    pub backend_id: String,
263    pub artifact: ArtifactId,
264    pub request: RequestId,
265    /// The digest of the request the host was asked about.
266    pub presented: ContentDigest,
267    pub exposure: Exposure,
268    pub remaining_ms: u64,
269    pub authorization: AuthorizationOutcome,
270    pub dispatch: Dispatch,
271}
272
273// ---------------------------------------------------------------------------
274// Historical link, run link, delivery and retention (spec 026, 3.6, 3.7)
275// ---------------------------------------------------------------------------
276
277/// A historical case, linked by digest and identities only: never embedded
278/// historical content (spec 026, 3.7).
279#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
280#[serde(deny_unknown_fields)]
281pub struct HistoricalLink {
282    pub decision_id: String,
283    pub plan_id: PlanId,
284    pub snapshot_id: SnapshotId,
285    pub evaluation_time_ms: Timestamp,
286    /// The record digest of the `rustev.replay/1` bundle.
287    pub bundle_digest: ContentDigest,
288}
289
290impl HistoricalLink {
291    /// The link to a retained bundle: its identities and digest, nothing it
292    /// retains. Reading a bundle's own members resolves no item.
293    pub fn of(bundle: &ReplayBundle) -> Result<Self, CanonicalError> {
294        Ok(HistoricalLink {
295            decision_id: bundle.decision_id.clone(),
296            plan_id: bundle.plan_id.clone(),
297            snapshot_id: bundle.snapshot_id.clone(),
298            evaluation_time_ms: bundle.evaluation_time_ms,
299            bundle_digest: bundle.record_digest()?,
300        })
301    }
302}
303
304/// The historical case, or its explicit absence.
305#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
306#[serde(rename_all = "snake_case", deny_unknown_fields)]
307pub enum Historical {
308    None,
309    Linked(HistoricalLink),
310}
311
312/// The new run record, linked by digest; it stays the evidence of execution.
313#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
314#[serde(deny_unknown_fields)]
315pub struct RunLink {
316    pub termination: Termination,
317    pub cost: CostSummary,
318}
319
320/// The new decision's admission (spec 003, 3.4): a document exists only for
321/// an admitted decision.
322#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
323#[serde(deny_unknown_fields)]
324pub struct Admission {
325    pub queued_ms: u64,
326    pub deadline_ms: u64,
327}
328
329/// What became of the new run record's delivery (spec 003, 3.9.4).
330#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
331#[serde(rename_all = "snake_case", deny_unknown_fields)]
332pub enum DeliveryObservation {
333    /// The sink's receipt, at most 256 bytes; acceptance under the sink's
334    /// own terms, not verified durability.
335    Acknowledged {
336        receipt: String,
337    },
338    /// Handed to the delivery worker; the outcome arrives on the ticket.
339    Queued,
340    Dropped {
341        dropped_total: u64,
342    },
343    NotDelivered {
344        detail: String,
345        uncertain: bool,
346    },
347}
348
349/// The new run's capture, chosen independently of the historical case and
350/// off by default (spec 026, 3.7).
351#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
352#[serde(rename_all = "snake_case", deny_unknown_fields)]
353pub enum CapturePosture {
354    Off,
355    On {
356        scope: Scope,
357        max_bytes: u64,
358        status: CaptureStatus,
359    },
360}
361
362/// One historical item's availability, as observed at comparison.
363#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
364#[serde(deny_unknown_fields)]
365pub struct ItemObservation {
366    /// `bundle`, `plan`, `snapshot`, `run`, `descriptors[i]`, ...
367    pub location: String,
368    pub availability: Availability,
369}
370
371/// What is known of the historical case's retention. Live re-execution
372/// never extends it and never restores an item (spec 026, 3.7).
373#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
374#[serde(rename_all = "snake_case", deny_unknown_fields)]
375pub enum HistoricalRetention {
376    /// No historical case is linked.
377    NoHistory,
378    /// Linked, and not resolved: the runtime reads only the link.
379    NotObserved,
380    /// Observed at comparison time.
381    Observed {
382        observed_at_ms: Timestamp,
383        bundle_expires_at_ms: Timestamp,
384        /// Items that were not available, explicitly; empty when every item
385        /// that was resolved was available.
386        unavailable: Vec<ItemObservation>,
387    },
388}
389
390#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
391#[serde(deny_unknown_fields)]
392pub struct RetentionPosture {
393    pub capture: CapturePosture,
394    pub historical: HistoricalRetention,
395}
396
397// ---------------------------------------------------------------------------
398// Comparison (spec 026, 3.8)
399// ---------------------------------------------------------------------------
400
401/// Whether a compared member changed. Unknown is never unchanged.
402#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
403#[serde(rename_all = "snake_case")]
404pub enum Change {
405    Unchanged,
406    Changed,
407    Unknown,
408}
409
410impl Change {
411    pub fn of<T: PartialEq>(historical: &T, live: &T) -> Change {
412        if historical == live {
413            Change::Unchanged
414        } else {
415            Change::Changed
416        }
417    }
418}
419
420/// What is compared: an explicitly named proposal payload or outcome
421/// category, chosen by a versioned task adapter. Full judgment bytes are
422/// never the agreement metric.
423#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
424#[serde(deny_unknown_fields)]
425pub struct ComparisonDefinition {
426    pub adapter: String,
427    pub adapter_version: String,
428    /// The proposal payload fields compared beside the action; at most 64.
429    pub fields: Vec<String>,
430}
431
432/// The result of comparing the live proposal with the historical one.
433#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
434#[serde(rename_all = "snake_case", deny_unknown_fields)]
435pub enum ComparisonOutcome {
436    /// The named fields agree. Not determinism, calibration, quality,
437    /// correctness or authorization.
438    Agreement,
439    /// The named fields differ. Not a failed historical reproduction.
440    LiveDiverged,
441    /// A historical item is not available; the source is not restored.
442    Unavailable {
443        location: String,
444        availability: Availability,
445    },
446    /// The historical case or the live result cannot be compared.
447    Incomparable { reason: String },
448}
449
450impl ComparisonOutcome {
451    pub fn name(&self) -> &'static str {
452        match self {
453            ComparisonOutcome::Agreement => "agreement",
454            ComparisonOutcome::LiveDiverged => "live-diverged",
455            ComparisonOutcome::Unavailable { .. } => "unavailable",
456            ComparisonOutcome::Incomparable { .. } => "incomparable",
457        }
458    }
459}
460
461/// How the compared result changed.
462#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
463#[serde(rename_all = "snake_case")]
464pub enum ResultChange {
465    Unchanged,
466    /// A proposal on either side, and the named outcome differs.
467    ProposalChanged,
468    /// Abstentions on both sides, of different categories.
469    UnresolvedCategoryChanged,
470    /// No comparison of results was possible.
471    NotCompared,
472}
473
474/// Identity changes between the historical case and the live run.
475#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
476#[serde(deny_unknown_fields)]
477pub struct IdentityChanges {
478    pub snapshot: Change,
479    pub plan: Change,
480    pub evaluation_time: Change,
481    /// The artifacts whose outputs were supplied.
482    pub artifacts: Change,
483    /// A served identity is not recorded by `rustev.run/1`.
484    pub served_identity: Change,
485}
486
487/// Counts of runtime observations of one side.
488#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
489#[serde(deny_unknown_fields)]
490pub struct RuntimeObservations {
491    pub attempts: u64,
492    pub retries: u64,
493    pub fallbacks: u64,
494    pub cancellations_requested: u64,
495    pub possibly_continuing: u64,
496    pub deadline_expired: bool,
497}
498
499/// New backend, policy, calibration and runtime observations.
500#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
501#[serde(deny_unknown_fields)]
502pub struct ObservationChanges {
503    /// The backends attempted.
504    pub backends: Change,
505    /// The definition, whose selection policy is part of it.
506    pub policy: Change,
507    pub execution_policy: Change,
508    pub calibration: Change,
509    pub historical_runtime: Option<RuntimeObservations>,
510    pub live_runtime: RuntimeObservations,
511}
512
513/// Observed, estimated and unknown cost, kept apart, of each side.
514#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
515#[serde(deny_unknown_fields)]
516pub struct CostComparison {
517    pub historical: Option<CostSummary>,
518    pub live: CostSummary,
519}
520
521/// Cancellation, remote-state, retention and delivery differences.
522#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
523#[serde(deny_unknown_fields)]
524pub struct OtherChanges {
525    pub termination: Change,
526    pub cancellation: Change,
527    pub remote_state: Change,
528    pub retention: Change,
529    pub delivery: Change,
530}
531
532/// The comparison of a finished live run with a verified historical case.
533#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
534#[serde(deny_unknown_fields)]
535pub struct LiveComparison {
536    pub definition: ComparisonDefinition,
537    pub compared_at_ms: Timestamp,
538    pub outcome: ComparisonOutcome,
539    /// The historical and live outcome categories, by the adapter's names.
540    pub historical_result: Option<String>,
541    pub live_result: Option<String>,
542    pub result: ResultChange,
543    pub identities: IdentityChanges,
544    pub observations: ObservationChanges,
545    pub cost: CostComparison,
546    pub other: OtherChanges,
547    /// What was left out of the comparison, and why; at most 64.
548    pub exclusions: Vec<String>,
549    /// What the comparison does not establish; at most 64.
550    pub limitations: Vec<String>,
551}
552
553/// The comparison member of a live document.
554#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
555#[serde(rename_all = "snake_case", deny_unknown_fields)]
556pub enum ComparisonSection {
557    NotRequested,
558    Completed(Box<LiveComparison>),
559}
560
561// ---------------------------------------------------------------------------
562// The document
563// ---------------------------------------------------------------------------
564
565/// `rustev.live-reexecution/1`: one live re-execution, bounded by
566/// `REPLAY_V1`. Holds no credential and no secret authorization material.
567#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
568#[serde(deny_unknown_fields)]
569pub struct LiveReexecution {
570    pub schema: String,
571    /// The new decision; never the historical one.
572    pub decision_id: String,
573    pub plan_id: PlanId,
574    /// Computed from the current snapshot by the runtime.
575    pub snapshot_id: SnapshotId,
576    pub evaluation_time_ms: Timestamp,
577    pub subject: Handle,
578    pub scope: Scope,
579    pub historical: Historical,
580    pub admission: Admission,
581    /// The record digest of the new run record; absent only when the
582    /// record has no record canonical form, which nothing then matches.
583    pub run_digest: Option<ContentDigest>,
584    pub run: RunLink,
585    /// One per proposed external attempt, in proposal order.
586    pub effects: Vec<LiveEffect>,
587    /// Attempts whose charge is unknown: outstanding liability for the host
588    /// to reconcile (spec 003, 3.5.5). The record is never rewritten.
589    pub unreconciled: Vec<String>,
590    pub delivery: DeliveryObservation,
591    pub retention: RetentionPosture,
592    pub comparison: ComparisonSection,
593}
594
595crate::document::document!(LiveReexecution, schema::LIVE_REEXECUTION, REPLAY_V1);
596
597/// Why a live document is not well formed.
598#[derive(Debug, Clone, PartialEq, Eq)]
599pub enum LiveDocError {
600    Schema(String),
601    /// A decision id outside 1 to 256 bytes, or equal to the historical one.
602    DecisionId,
603    TooMany {
604        what: &'static str,
605        count: usize,
606    },
607    TextTooLong {
608        what: &'static str,
609    },
610    /// An effect or unreconciled attempt that is not the new decision's.
611    Attribution {
612        attempt_id: String,
613    },
614}
615
616impl fmt::Display for LiveDocError {
617    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
618        match self {
619            LiveDocError::Schema(s) => write!(f, "schema {s:?} is not rustev.live-reexecution/1"),
620            LiveDocError::DecisionId => {
621                f.write_str("the decision id is not a new decision id of 1 to 256 bytes")
622            }
623            LiveDocError::TooMany { what, count } => write!(f, "{count} {what} is too many"),
624            LiveDocError::TextTooLong { what } => write!(f, "{what} exceeds 256 bytes"),
625            LiveDocError::Attribution { attempt_id } => {
626                write!(f, "attempt {attempt_id:?} is not the new decision's")
627            }
628        }
629    }
630}
631
632impl std::error::Error for LiveDocError {}
633
634fn count(what: &'static str, n: usize, max: usize) -> Result<(), LiveDocError> {
635    if n > max {
636        return Err(LiveDocError::TooMany { what, count: n });
637    }
638    Ok(())
639}
640
641fn text(what: &'static str, s: &str) -> Result<(), LiveDocError> {
642    if s.len() > MAX_TEXT_BYTES {
643        return Err(LiveDocError::TextTooLong { what });
644    }
645    Ok(())
646}
647
648/// The prefix every attempt id of `decision_id` starts with (spec 003,
649/// 3.9.2: the decision id is its first, escaped, component).
650pub fn attempt_prefix(decision_id: &str) -> String {
651    let mut out = String::with_capacity(decision_id.len() + 1);
652    for c in decision_id.chars() {
653        match c {
654            '%' => out.push_str("%25"),
655            '/' => out.push_str("%2F"),
656            ',' => out.push_str("%2C"),
657            c => out.push(c),
658        }
659    }
660    out.push('/');
661    out
662}
663
664impl LiveReexecution {
665    /// Structural bounds and attribution. Every bound is also a parse bound
666    /// of `REPLAY_V1`, which applies before anything is allocated.
667    pub fn check(&self) -> Result<(), LiveDocError> {
668        if self.schema != schema::LIVE_REEXECUTION {
669            return Err(LiveDocError::Schema(self.schema.chars().take(80).collect()));
670        }
671        if self.decision_id.is_empty() || self.decision_id.len() > 256 {
672            return Err(LiveDocError::DecisionId);
673        }
674        if let Historical::Linked(h) = &self.historical
675            && (h.decision_id == self.decision_id
676                || h.decision_id.is_empty()
677                || h.decision_id.len() > 256)
678        {
679            return Err(LiveDocError::DecisionId);
680        }
681        count("effects", self.effects.len(), MAX_EFFECTS)?;
682        count(
683            "unreconciled attempts",
684            self.unreconciled.len(),
685            MAX_EFFECTS,
686        )?;
687        let prefix = attempt_prefix(&self.decision_id);
688        for e in &self.effects {
689            if !e.attempt_id.starts_with(&prefix) || e.attempt_id.len() > 4 * MAX_TEXT_BYTES {
690                return Err(LiveDocError::Attribution {
691                    attempt_id: e.attempt_id.chars().take(80).collect(),
692                });
693            }
694            if let AuthorizationOutcome::Refused { detail, .. } = &e.authorization {
695                text("an authorization detail", detail)?;
696            }
697            if let Dispatch::Dispatched {
698                end: AttemptEnd::Failed { detail, .. },
699                ..
700            } = &e.dispatch
701            {
702                text("an attempt detail", detail)?;
703            }
704        }
705        for a in &self.unreconciled {
706            if !a.starts_with(&prefix) || a.len() > 4 * MAX_TEXT_BYTES {
707                return Err(LiveDocError::Attribution {
708                    attempt_id: a.chars().take(80).collect(),
709                });
710            }
711        }
712        match &self.delivery {
713            DeliveryObservation::Acknowledged { receipt } => text("a receipt", receipt)?,
714            DeliveryObservation::NotDelivered { detail, .. } => text("a delivery detail", detail)?,
715            DeliveryObservation::Queued | DeliveryObservation::Dropped { .. } => {}
716        }
717        if let HistoricalRetention::Observed { unavailable, .. } = &self.retention.historical {
718            count(
719                "historical item observations",
720                unavailable.len(),
721                MAX_HISTORICAL_ITEMS,
722            )?;
723            for i in unavailable {
724                text("an item location", &i.location)?;
725            }
726        }
727        if let ComparisonSection::Completed(c) = &self.comparison {
728            c.check()?;
729        }
730        Ok(())
731    }
732
733    /// The new run record this document was built for: equal digest.
734    pub fn binds(&self, run: &RunRecord) -> bool {
735        run.decision_id == self.decision_id
736            && self.run_digest.is_some()
737            && run.record_digest().ok() == self.run_digest
738    }
739}
740
741impl LiveComparison {
742    pub fn check(&self) -> Result<(), LiveDocError> {
743        let d = &self.definition;
744        text("the adapter name", &d.adapter)?;
745        text("the adapter version", &d.adapter_version)?;
746        count("compared fields", d.fields.len(), MAX_NOTES)?;
747        for f in &d.fields {
748            text("a compared field", f)?;
749        }
750        count("exclusions", self.exclusions.len(), MAX_NOTES)?;
751        count("limitations", self.limitations.len(), MAX_NOTES)?;
752        for n in self.exclusions.iter().chain(&self.limitations) {
753            text("a note", n)?;
754        }
755        for r in [&self.historical_result, &self.live_result]
756            .into_iter()
757            .flatten()
758        {
759            // An outcome name carries the compared field values.
760            if r.len() > 64 * MAX_TEXT_BYTES {
761                return Err(LiveDocError::TextTooLong {
762                    what: "an outcome name",
763                });
764            }
765        }
766        match &self.outcome {
767            ComparisonOutcome::Unavailable { location, .. } => text("a location", location),
768            ComparisonOutcome::Incomparable { reason } => text("a reason", reason),
769            ComparisonOutcome::Agreement | ComparisonOutcome::LiveDiverged => Ok(()),
770        }
771    }
772}
rustev_contract/live.rs

rustev_contract/
live.rs
