1use std::collections::BTreeMap;
21use std::fmt;
22use std::panic::{AssertUnwindSafe, catch_unwind};
23use std::sync::{Arc, Mutex};
24
25use rustev_contract::definition::{Operation, StepBody};
26use rustev_contract::ids::{ContentDigest, SnapshotId};
27use rustev_contract::live::{
28 Admission, AttemptRole, AuthorizationOutcome, AuthorizationReference, AuthorizationRefusal,
29 AuthorizationRequest, CapturePosture, ComparisonSection, DeliveryObservation, Dispatch,
30 Exposure, Historical, HistoricalLink, HistoricalRetention, LiveEffect, LiveReexecution,
31 MAX_EFFECTS, RetentionPosture, RunLink,
32};
33use rustev_contract::plan::{PlanExecution, PlanStepDetail};
34use rustev_contract::run::{Charge, CoreEvidence, CostBound, RunRecord, Termination};
35use rustev_contract::scope::{Handle, Scope};
36use rustev_contract::snapshot::Snapshot;
37use rustev_contract::time::Timestamp;
38use rustev_contract::{Document, Identified, schema};
39use rustev_core::evaluate::SemanticRequest;
40use rustev_core::seams::CancelSignal;
41
42use crate::driver::{DETAIL_BYTES, DecisionCtx};
43use crate::wait::cut;
44use crate::{
45 CaptureConfig, CaptureConfigError, CaptureOutcome, Completion, DecideError, Decided,
46 DecisionRequest, Delivery, DeliveryFailure, PreparedPlan, Rejection, Runtime, Target, capture,
47};
48
49#[derive(Debug)]
53pub enum DispatchAnswer {
54 Allow {
57 answers: ContentDigest,
58 reference: AuthorizationReference,
59 valid_for_ms: u64,
60 },
61 Refuse {
62 reference: Option<AuthorizationReference>,
63 detail: String,
64 },
65}
66
67pub trait LiveAuthorizer: Send + Sync {
73 fn authorize(&self, request: &AuthorizationRequest) -> Result<Option<DispatchAnswer>, String>;
74}
75
76#[derive(Debug, Clone)]
78pub struct LiveRequest {
79 pub decision_id: String,
81 pub snapshot: Snapshot,
83 pub expected_snapshot_id: Option<SnapshotId>,
87 pub evaluation_time: Timestamp,
89 pub deadline_ms: Option<u64>,
91 pub principal_handle: Vec<u8>,
93 pub subject: Handle,
95 pub scope: Scope,
96 pub capture: Option<CaptureConfig>,
98 pub historical: Option<HistoricalLink>,
100}
101
102#[derive(Debug, Clone, PartialEq, Eq)]
106pub enum LiveRefusal {
107 SpendAcquisitionUnsupported,
110 HistoricalDecisionId,
112 InvalidHistoricalLink,
114 SnapshotIdentity {
116 stated: SnapshotId,
117 computed: SnapshotId,
118 },
119 InvalidSnapshot(String),
121 Capture(CaptureConfigError),
122 Unbounded {
124 worst_case: u64,
125 max: u64,
126 },
127 Rejected(Rejection),
129 GenerationFailed(String),
132}
133
134impl fmt::Display for LiveRefusal {
135 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136 match self {
137 LiveRefusal::SpendAcquisitionUnsupported => {
138 f.write_str("gated live acquisition dispatch is not implemented")
139 }
140 LiveRefusal::HistoricalDecisionId => {
141 f.write_str("a live re-execution needs a new decision id, not the historical one")
142 }
143 LiveRefusal::InvalidHistoricalLink => {
144 f.write_str("the historical link's decision id is not 1 to 256 bytes")
145 }
146 LiveRefusal::SnapshotIdentity { stated, computed } => write!(
147 f,
148 "identity mismatch: the snapshot is {computed}, not the stated {stated}"
149 ),
150 LiveRefusal::InvalidSnapshot(e) => write!(f, "the snapshot has no identity: {e}"),
151 LiveRefusal::Capture(e) => write!(f, "{e}"),
152 LiveRefusal::Unbounded { worst_case, max } => write!(
153 f,
154 "the plan can propose {worst_case} attempts; a live document records at most {max}"
155 ),
156 LiveRefusal::Rejected(r) => write!(f, "not admitted: {r:?}"),
157 LiveRefusal::GenerationFailed(detail) => write!(f, "generation failed: {detail}"),
158 }
159 }
160}
161
162impl std::error::Error for LiveRefusal {}
163
164#[derive(Debug)]
168pub struct LiveResult {
169 pub spend_refusal: Option<crate::SpendRefusal>,
171 pub completion: Completion,
172 pub record: Arc<RunRecord>,
174 pub delivery: Result<Delivery, DeliveryFailure>,
175 pub capture: Option<CaptureOutcome>,
177 pub document: LiveReexecution,
178}
179
180struct Presented {
182 step: String,
183 instance: Vec<String>,
184 target: u32,
185 role: AttemptRole,
186 attempt_id: String,
187 backend_id: String,
188 artifact: rustev_contract::ids::ArtifactId,
189 request: rustev_contract::ids::RequestId,
190 presented: ContentDigest,
191 exposure: Exposure,
192 remaining_ms: u64,
193 authorization: AuthorizationOutcome,
194}
195
196pub(crate) struct LiveCtx<'l> {
198 authorizer: &'l dyn LiveAuthorizer,
199 spend: Option<crate::LiveSpend<'l>>,
200 spend_refusal: Mutex<Option<crate::SpendRefusal>>,
201 subject: &'l Handle,
202 scope: &'l Scope,
203 operations: BTreeMap<String, Operation>,
204 presented: Mutex<Vec<Presented>>,
205}
206
207pub(crate) struct Proposal<'p> {
209 pub req: &'p SemanticRequest,
210 pub target_index: usize,
211 pub target: &'p Target,
212 pub role: AttemptRole,
213 pub attempt_id: &'p str,
214 pub bound: CostBound,
215 pub reserved: u64,
216}
217
218impl LiveCtx<'_> {
219 pub(crate) fn spend_refusal(&self) -> Option<crate::SpendRefusal> {
220 self.spend_refusal
221 .lock()
222 .unwrap_or_else(|e| e.into_inner())
223 .clone()
224 }
225
226 pub(crate) fn reserve_spend(
227 &self,
228 cx: &DecisionCtx<'_, '_>,
229 p: Proposal<'_>,
230 ) -> Result<Option<crate::spend::HeldSpend<'_>>, crate::SpendRefusal> {
231 let Some(spend) = self.spend else {
232 return Ok(None);
233 };
234 let mut refusal = self.spend_refusal.lock().unwrap_or_else(|e| e.into_inner());
235 if let Some(r) = &*refusal {
236 return Err(r.clone());
237 }
238 let binding = spend
239 .bindings
240 .iter()
241 .find(|b| b.backend_id == p.target.backend_id && b.artifact == p.target.artifact);
242 let result = match binding {
243 None => Err(crate::SpendRefusal::new("spend-unit")),
244 Some(binding) => spend.gate.reserve(&crate::SpendAttempt {
245 decision_id: cx.decision_id.to_string(),
246 attempt_id: p.attempt_id.to_string(),
247 backend_id: p.target.backend_id.clone(),
248 target: p.target_index as u32,
249 artifact: p.target.artifact.clone(),
250 operation: self.operations[&p.req.step],
251 bound: p.bound,
252 cost_unit: binding.cost_unit,
253 }),
254 };
255 match result {
256 Ok(ticket) => Ok(Some(crate::spend::HeldSpend::new(spend.gate, ticket))),
257 Err(r) => {
258 *refusal = Some(r.clone());
259 Err(r)
260 }
261 }
262 }
263
264 pub(crate) fn expire_authorization(&self, attempt_id: &str) {
265 let mut presented = self.presented.lock().unwrap_or_else(|e| e.into_inner());
266 if let Some(p) = presented.iter_mut().find(|p| p.attempt_id == attempt_id)
267 && let AuthorizationOutcome::Allowed { reference } = &p.authorization
268 {
269 p.authorization = AuthorizationOutcome::Refused {
270 refusal: AuthorizationRefusal::Expired,
271 reference: Some(reference.clone()),
272 detail: "authorization expired during spend reservation".into(),
273 };
274 }
275 }
276
277 pub(crate) fn authorize(
281 &self,
282 cx: &DecisionCtx<'_, '_>,
283 p: Proposal<'_>,
284 ) -> Result<u64, AuthorizationRefusal> {
285 let now = cx.rt.clock.now();
286 let identity = cx
287 .ev()
288 .request_document(
289 &p.req.step,
290 &p.req.instance,
291 p.target_index as u32,
292 self.scope,
293 )
294 .map_err(|e| e.to_string())
295 .and_then(|d| d.id().map_err(|e| e.to_string()));
296 let request = match identity {
297 Ok(id) => id,
298 Err(_) => return Err(AuthorizationRefusal::Errored),
301 };
302 let asked = AuthorizationRequest {
303 decision_id: cx.decision_id.to_string(),
304 subject: self.subject.clone(),
305 scope: self.scope.clone(),
306 step: p.req.step.clone(),
307 instance: p.req.instance.clone(),
308 operation: self.operations[&p.req.step],
309 target: p.target_index as u32,
310 role: p.role,
311 backend_id: p.target.backend_id.clone(),
312 artifact: p.target.artifact.clone(),
313 request: request.clone(),
314 attempt_id: p.attempt_id.to_string(),
315 exposure: Exposure {
316 bound: p.bound,
317 reserved_units: p.reserved,
318 },
319 remaining_ms: cx.deadline.saturating_sub(now),
320 };
321 let Ok(digest) = asked.digest() else {
322 return Err(AuthorizationRefusal::Errored);
323 };
324 let refused = |refusal, reference, detail: &str| AuthorizationOutcome::Refused {
325 refusal,
326 reference,
327 detail: cut(detail, DETAIL_BYTES),
328 };
329 let mut valid_until = now;
330 let outcome = match catch_unwind(AssertUnwindSafe(|| self.authorizer.authorize(&asked))) {
331 Err(p) => refused(
332 AuthorizationRefusal::Errored,
333 None,
334 &format!(
335 "authorizer panicked: {}",
336 crate::wait::panic_detail(p.as_ref())
337 ),
338 ),
339 Ok(Err(detail)) => refused(AuthorizationRefusal::Errored, None, &detail),
340 Ok(Ok(None)) => refused(AuthorizationRefusal::Missing, None, "no answer"),
341 Ok(Ok(Some(DispatchAnswer::Refuse { reference, detail }))) => {
342 refused(AuthorizationRefusal::Refused, reference, &detail)
343 }
344 Ok(Ok(Some(DispatchAnswer::Allow {
345 answers,
346 reference,
347 valid_for_ms,
348 }))) => {
349 if answers != digest {
350 refused(
351 AuthorizationRefusal::Mismatched,
352 Some(reference),
353 "the answer names another request",
354 )
355 } else if cx.rt.clock.now() >= now.saturating_add(valid_for_ms) {
356 refused(
357 AuthorizationRefusal::Expired,
358 Some(reference),
359 &format!("the answer was valid for {valid_for_ms} ms"),
360 )
361 } else {
362 valid_until = now.saturating_add(valid_for_ms);
363 AuthorizationOutcome::Allowed { reference }
364 }
365 }
366 };
367 let result = match &outcome {
368 AuthorizationOutcome::Allowed { .. } => Ok(valid_until),
369 AuthorizationOutcome::Refused { refusal, .. } => Err(*refusal),
370 };
371 self.presented
372 .lock()
373 .unwrap_or_else(|e| e.into_inner())
374 .push(Presented {
375 step: asked.step,
376 instance: asked.instance,
377 target: asked.target,
378 role: asked.role,
379 attempt_id: asked.attempt_id,
380 backend_id: asked.backend_id,
381 artifact: asked.artifact,
382 request,
383 presented: digest,
384 exposure: asked.exposure,
385 remaining_ms: asked.remaining_ms,
386 authorization: outcome,
387 });
388 result
389 }
390}
391
392fn worst_case_attempts(plan: &PreparedPlan) -> u64 {
396 match &plan.compiled().plan.execution {
397 PlanExecution::Declared(d) => d.policy.max_attempts_per_decision,
398 PlanExecution::None => plan
399 .compiled()
400 .plan
401 .steps
402 .iter()
403 .filter_map(|s| match &s.detail {
404 PlanStepDetail::Semantic(b) => Some(b.max_requests),
405 _ => None,
406 })
407 .fold(0, u64::saturating_add),
408 }
409}
410
411impl Runtime {
412 pub async fn reexecute_live(
450 &self,
451 plan: &PreparedPlan,
452 req: LiveRequest,
453 authorizer: &dyn LiveAuthorizer,
454 cancel: &CancelSignal,
455 ) -> Result<LiveResult, LiveRefusal> {
456 self.reexecute_live_inner(plan, req, authorizer, cancel, None)
457 .await
458 }
459
460 pub async fn reexecute_live_with_gate(
465 &self,
466 plan: &PreparedPlan,
467 req: LiveRequest,
468 authorizer: &dyn LiveAuthorizer,
469 cancel: &CancelSignal,
470 spend: crate::LiveSpend<'_>,
471 ) -> Result<LiveResult, LiveRefusal> {
472 self.reexecute_live_inner(plan, req, authorizer, cancel, Some(spend))
473 .await
474 }
475
476 async fn reexecute_live_inner(
477 &self,
478 plan: &PreparedPlan,
479 req: LiveRequest,
480 authorizer: &dyn LiveAuthorizer,
481 cancel: &CancelSignal,
482 spend: Option<crate::LiveSpend<'_>>,
483 ) -> Result<LiveResult, LiveRefusal> {
484 if spend.is_some()
485 && plan
486 .compiled()
487 .plan
488 .acquisitions
489 .as_ref()
490 .is_some_and(|a| !a.is_empty())
491 {
492 return Err(LiveRefusal::SpendAcquisitionUnsupported);
493 }
494 if req
495 .historical
496 .as_ref()
497 .is_some_and(|h| h.decision_id == req.decision_id)
498 {
499 return Err(LiveRefusal::HistoricalDecisionId);
500 }
501 if req
502 .historical
503 .as_ref()
504 .is_some_and(|h| h.decision_id.is_empty() || h.decision_id.len() > 256)
505 {
506 return Err(LiveRefusal::InvalidHistoricalLink);
507 }
508 let snapshot_id = req
509 .snapshot
510 .id()
511 .map_err(|e| LiveRefusal::InvalidSnapshot(e.to_string()))?;
512 if let Some(stated) = &req.expected_snapshot_id
513 && *stated != snapshot_id
514 {
515 return Err(LiveRefusal::SnapshotIdentity {
516 stated: stated.clone(),
517 computed: snapshot_id,
518 });
519 }
520 if let Some(c) = &req.capture {
521 c.check().map_err(LiveRefusal::Capture)?;
522 }
523 let worst_case = worst_case_attempts(plan);
524 if worst_case > MAX_EFFECTS as u64 {
525 return Err(LiveRefusal::Unbounded {
526 worst_case,
527 max: MAX_EFFECTS as u64,
528 });
529 }
530
531 let operations = plan
532 .compiled()
533 .plan
534 .definition
535 .steps
536 .iter()
537 .filter_map(|s| match &s.body {
538 StepBody::Semantic(d) => Some((s.id.clone(), d.operation)),
539 _ => None,
540 })
541 .collect();
542 let live = LiveCtx {
543 authorizer,
544 spend,
545 spend_refusal: Mutex::new(None),
546 subject: &req.subject,
547 scope: &req.scope,
548 operations,
549 presented: Mutex::new(vec![]),
550 };
551 let decision = DecisionRequest {
552 decision_id: req.decision_id.clone(),
553 snapshot: req.snapshot.clone(),
554 evaluation_time: req.evaluation_time,
555 deadline_ms: req.deadline_ms,
556 principal_handle: req.principal_handle.clone(),
557 };
558 let mut capturing = req.capture.as_ref().map(capture::Capturing::new);
559 let result = self
560 .run(
561 plan,
562 decision,
563 cancel,
564 capturing.as_mut(),
565 Some(&live),
566 None,
567 None,
568 )
569 .await;
570 let (record, delivery) = match result {
571 Err(DecideError::GenerationFailed(detail)) => {
572 return Err(LiveRefusal::GenerationFailed(detail));
573 }
574 Err(DecideError::Rejected(r)) => return Err(LiveRefusal::Rejected(r)),
575 Ok(Decided {
576 record, delivery, ..
577 }) => (record, Ok(delivery)),
578 Err(DecideError::EvidenceNotDelivered { record, failure }) => (record, Err(failure)),
579 };
580 let capture = capturing.map(|c| {
581 CaptureOutcome::Captured(c.finish(
582 &req.decision_id,
583 record.termination == Termination::Cancelled,
584 ))
585 });
586 let spend_refusal = live.spend_refusal();
587 let presented = live
588 .presented
589 .into_inner()
590 .unwrap_or_else(|e| e.into_inner());
591 let document = document(&req, snapshot_id, &record, &delivery, &capture, presented);
592 let completion = match &record.core {
593 CoreEvidence::Recorded(e) if record.termination == Termination::Judged => {
594 Completion::Judged(Box::new(e.judgment.clone()))
595 }
596 _ => Completion::Cancelled,
597 };
598 Ok(LiveResult {
599 spend_refusal,
600 completion,
601 record,
602 delivery,
603 capture,
604 document,
605 })
606 }
607}
608
609fn document(
610 req: &LiveRequest,
611 snapshot_id: SnapshotId,
612 record: &RunRecord,
613 delivery: &Result<Delivery, DeliveryFailure>,
614 capture: &Option<CaptureOutcome>,
615 presented: Vec<Presented>,
616) -> LiveReexecution {
617 let attempts: BTreeMap<&str, &rustev_contract::run::AttemptRecord> = record
618 .requests
619 .iter()
620 .flat_map(|r| &r.attempts)
621 .map(|a| (a.attempt_id.as_str(), a))
622 .collect();
623 let effects = presented
624 .into_iter()
625 .map(|p| {
626 let dispatch = match attempts.get(p.attempt_id.as_str()) {
627 Some(a) => Dispatch::Dispatched {
628 end: a.end.clone(),
629 cancellation: a.cancellation,
630 remote: a.remote,
631 charge: a.cost.charge,
632 },
633 None => Dispatch::NotDispatched,
634 };
635 LiveEffect {
636 step: p.step,
637 instance: p.instance,
638 target: p.target,
639 role: p.role,
640 attempt_id: p.attempt_id,
641 backend_id: p.backend_id,
642 artifact: p.artifact,
643 request: p.request,
644 presented: p.presented,
645 exposure: p.exposure,
646 remaining_ms: p.remaining_ms,
647 authorization: p.authorization,
648 dispatch,
649 }
650 })
651 .collect();
652 let unreconciled = attempts
653 .values()
654 .filter(|a| a.cost.charge == Charge::Unknown)
655 .map(|a| a.attempt_id.clone())
656 .collect();
657 let delivery = match delivery {
658 Ok(Delivery::Acknowledged { receipt }) => DeliveryObservation::Acknowledged {
659 receipt: cut(receipt, DETAIL_BYTES),
660 },
661 Ok(Delivery::Queued(_)) => DeliveryObservation::Queued,
662 Ok(Delivery::Dropped { dropped_total }) => DeliveryObservation::Dropped {
663 dropped_total: *dropped_total,
664 },
665 Err(f) => DeliveryObservation::NotDelivered {
666 detail: cut(&f.detail, DETAIL_BYTES),
667 uncertain: f.uncertain,
668 },
669 };
670 let capture = match (&req.capture, capture) {
671 (Some(c), Some(CaptureOutcome::Captured(k))) => CapturePosture::On {
672 scope: c.scope.clone(),
673 max_bytes: c.max_bytes,
674 status: k.status,
675 },
676 _ => CapturePosture::Off,
677 };
678 LiveReexecution {
679 schema: schema::LIVE_REEXECUTION.into(),
680 decision_id: req.decision_id.clone(),
681 plan_id: record.plan_id.clone(),
682 snapshot_id,
683 evaluation_time_ms: req.evaluation_time,
684 subject: req.subject.clone(),
685 scope: req.scope.clone(),
686 historical: match &req.historical {
687 Some(h) => Historical::Linked(h.clone()),
688 None => Historical::None,
689 },
690 admission: Admission {
691 queued_ms: record.timing.queued_ms,
692 deadline_ms: record.timing.deadline_ms,
693 },
694 run_digest: record.record_digest().ok(),
695 run: RunLink {
696 termination: record.termination,
697 cost: record.cost,
698 },
699 effects,
700 unreconciled,
701 delivery,
702 retention: RetentionPosture {
703 capture,
704 historical: if req.historical.is_some() {
705 HistoricalRetention::NotObserved
706 } else {
707 HistoricalRetention::NoHistory
708 },
709 },
710 comparison: ComparisonSection::NotRequested,
711 }
712}