1use 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
37pub const MAX_REFERENCE_BYTES: usize = 256;
39pub const MAX_EFFECTS: usize = 4096;
42pub const MAX_NOTES: usize = 64;
44pub const MAX_TEXT_BYTES: usize = 256;
46pub const MAX_HISTORICAL_ITEMS: usize = 3 * crate::replay::MAX_BUNDLE_ITEMS + 4;
48
49pub const AUTHORIZATION_REQUEST_TAG: &str = "rustev.live-authorization/1";
53
54#[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#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
81pub struct AuthorizationReference(String);
82
83impl AuthorizationReference {
84 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
126#[serde(rename_all = "snake_case")]
127pub enum AttemptRole {
128 Primary,
130 Retry,
132 Fallback,
134}
135
136#[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#[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 pub request: RequestId,
165 pub attempt_id: String,
167 pub exposure: Exposure,
168 pub remaining_ms: u64,
170}
171
172impl AuthorizationRequest {
173 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
186#[serde(rename_all = "snake_case")]
187pub enum AuthorizationRefusal {
188 Missing,
190 Expired,
192 Mismatched,
194 Errored,
196 Refused,
198}
199
200impl AuthorizationRefusal {
201 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#[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 reference: Option<AuthorizationReference>,
232 detail: String,
234 },
235}
236
237#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
239#[serde(rename_all = "snake_case", deny_unknown_fields)]
240pub enum Dispatch {
241 NotDispatched,
243 Dispatched {
244 end: AttemptEnd,
245 cancellation: Cancellation,
246 remote: RemoteState,
247 charge: Charge,
248 },
249}
250
251#[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 pub presented: ContentDigest,
267 pub exposure: Exposure,
268 pub remaining_ms: u64,
269 pub authorization: AuthorizationOutcome,
270 pub dispatch: Dispatch,
271}
272
273#[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 pub bundle_digest: ContentDigest,
288}
289
290impl HistoricalLink {
291 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#[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#[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#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
331#[serde(rename_all = "snake_case", deny_unknown_fields)]
332pub enum DeliveryObservation {
333 Acknowledged {
336 receipt: String,
337 },
338 Queued,
340 Dropped {
341 dropped_total: u64,
342 },
343 NotDelivered {
344 detail: String,
345 uncertain: bool,
346 },
347}
348
349#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
364#[serde(deny_unknown_fields)]
365pub struct ItemObservation {
366 pub location: String,
368 pub availability: Availability,
369}
370
371#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
374#[serde(rename_all = "snake_case", deny_unknown_fields)]
375pub enum HistoricalRetention {
376 NoHistory,
378 NotObserved,
380 Observed {
382 observed_at_ms: Timestamp,
383 bundle_expires_at_ms: Timestamp,
384 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#[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#[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 pub fields: Vec<String>,
430}
431
432#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
434#[serde(rename_all = "snake_case", deny_unknown_fields)]
435pub enum ComparisonOutcome {
436 Agreement,
439 LiveDiverged,
441 Unavailable {
443 location: String,
444 availability: Availability,
445 },
446 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
463#[serde(rename_all = "snake_case")]
464pub enum ResultChange {
465 Unchanged,
466 ProposalChanged,
468 UnresolvedCategoryChanged,
470 NotCompared,
472}
473
474#[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 pub artifacts: Change,
483 pub served_identity: Change,
485}
486
487#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
501#[serde(deny_unknown_fields)]
502pub struct ObservationChanges {
503 pub backends: Change,
505 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#[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#[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#[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 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 pub exclusions: Vec<String>,
549 pub limitations: Vec<String>,
551}
552
553#[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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
568#[serde(deny_unknown_fields)]
569pub struct LiveReexecution {
570 pub schema: String,
571 pub decision_id: String,
573 pub plan_id: PlanId,
574 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 pub run_digest: Option<ContentDigest>,
584 pub run: RunLink,
585 pub effects: Vec<LiveEffect>,
587 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#[derive(Debug, Clone, PartialEq, Eq)]
599pub enum LiveDocError {
600 Schema(String),
601 DecisionId,
603 TooMany {
604 what: &'static str,
605 count: usize,
606 },
607 TextTooLong {
608 what: &'static str,
609 },
610 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
648pub 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 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 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 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}