use std::collections::BTreeMap;
use std::fmt;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::{Arc, Mutex};
use rustev_contract::definition::{Operation, StepBody};
use rustev_contract::ids::{ContentDigest, SnapshotId};
use rustev_contract::live::{
Admission, AttemptRole, AuthorizationOutcome, AuthorizationReference, AuthorizationRefusal,
AuthorizationRequest, CapturePosture, ComparisonSection, DeliveryObservation, Dispatch,
Exposure, Historical, HistoricalLink, HistoricalRetention, LiveEffect, LiveReexecution,
MAX_EFFECTS, RetentionPosture, RunLink,
};
use rustev_contract::plan::{PlanExecution, PlanStepDetail};
use rustev_contract::run::{Charge, CoreEvidence, CostBound, RunRecord, Termination};
use rustev_contract::scope::{Handle, Scope};
use rustev_contract::snapshot::Snapshot;
use rustev_contract::time::Timestamp;
use rustev_contract::{Document, Identified, schema};
use rustev_core::evaluate::SemanticRequest;
use rustev_core::seams::CancelSignal;
use crate::driver::{DETAIL_BYTES, DecisionCtx};
use crate::wait::cut;
use crate::{
CaptureConfig, CaptureConfigError, CaptureOutcome, Completion, DecideError, Decided,
DecisionRequest, Delivery, DeliveryFailure, PreparedPlan, Rejection, Runtime, Target, capture,
};
#[derive(Debug)]
pub enum DispatchAnswer {
Allow {
answers: ContentDigest,
reference: AuthorizationReference,
valid_for_ms: u64,
},
Refuse {
reference: Option<AuthorizationReference>,
detail: String,
},
}
pub trait LiveAuthorizer: Send + Sync {
fn authorize(&self, request: &AuthorizationRequest) -> Result<Option<DispatchAnswer>, String>;
}
#[derive(Debug, Clone)]
pub struct LiveRequest {
pub decision_id: String,
pub snapshot: Snapshot,
pub expected_snapshot_id: Option<SnapshotId>,
pub evaluation_time: Timestamp,
pub deadline_ms: Option<u64>,
pub principal_handle: Vec<u8>,
pub subject: Handle,
pub scope: Scope,
pub capture: Option<CaptureConfig>,
pub historical: Option<HistoricalLink>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LiveRefusal {
SpendAcquisitionUnsupported,
HistoricalDecisionId,
InvalidHistoricalLink,
SnapshotIdentity {
stated: SnapshotId,
computed: SnapshotId,
},
InvalidSnapshot(String),
Capture(CaptureConfigError),
Unbounded {
worst_case: u64,
max: u64,
},
Rejected(Rejection),
GenerationFailed(String),
}
impl fmt::Display for LiveRefusal {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
LiveRefusal::SpendAcquisitionUnsupported => {
f.write_str("gated live acquisition dispatch is not implemented")
}
LiveRefusal::HistoricalDecisionId => {
f.write_str("a live re-execution needs a new decision id, not the historical one")
}
LiveRefusal::InvalidHistoricalLink => {
f.write_str("the historical link's decision id is not 1 to 256 bytes")
}
LiveRefusal::SnapshotIdentity { stated, computed } => write!(
f,
"identity mismatch: the snapshot is {computed}, not the stated {stated}"
),
LiveRefusal::InvalidSnapshot(e) => write!(f, "the snapshot has no identity: {e}"),
LiveRefusal::Capture(e) => write!(f, "{e}"),
LiveRefusal::Unbounded { worst_case, max } => write!(
f,
"the plan can propose {worst_case} attempts; a live document records at most {max}"
),
LiveRefusal::Rejected(r) => write!(f, "not admitted: {r:?}"),
LiveRefusal::GenerationFailed(detail) => write!(f, "generation failed: {detail}"),
}
}
}
impl std::error::Error for LiveRefusal {}
#[derive(Debug)]
pub struct LiveResult {
pub spend_refusal: Option<crate::SpendRefusal>,
pub completion: Completion,
pub record: Arc<RunRecord>,
pub delivery: Result<Delivery, DeliveryFailure>,
pub capture: Option<CaptureOutcome>,
pub document: LiveReexecution,
}
struct Presented {
step: String,
instance: Vec<String>,
target: u32,
role: AttemptRole,
attempt_id: String,
backend_id: String,
artifact: rustev_contract::ids::ArtifactId,
request: rustev_contract::ids::RequestId,
presented: ContentDigest,
exposure: Exposure,
remaining_ms: u64,
authorization: AuthorizationOutcome,
}
pub(crate) struct LiveCtx<'l> {
authorizer: &'l dyn LiveAuthorizer,
spend: Option<crate::LiveSpend<'l>>,
spend_refusal: Mutex<Option<crate::SpendRefusal>>,
subject: &'l Handle,
scope: &'l Scope,
operations: BTreeMap<String, Operation>,
presented: Mutex<Vec<Presented>>,
}
pub(crate) struct Proposal<'p> {
pub req: &'p SemanticRequest,
pub target_index: usize,
pub target: &'p Target,
pub role: AttemptRole,
pub attempt_id: &'p str,
pub bound: CostBound,
pub reserved: u64,
}
impl LiveCtx<'_> {
pub(crate) fn spend_refusal(&self) -> Option<crate::SpendRefusal> {
self.spend_refusal
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
pub(crate) fn reserve_spend(
&self,
cx: &DecisionCtx<'_, '_>,
p: Proposal<'_>,
) -> Result<Option<crate::spend::HeldSpend<'_>>, crate::SpendRefusal> {
let Some(spend) = self.spend else {
return Ok(None);
};
let mut refusal = self.spend_refusal.lock().unwrap_or_else(|e| e.into_inner());
if let Some(r) = &*refusal {
return Err(r.clone());
}
let binding = spend
.bindings
.iter()
.find(|b| b.backend_id == p.target.backend_id && b.artifact == p.target.artifact);
let result = match binding {
None => Err(crate::SpendRefusal::new("spend-unit")),
Some(binding) => spend.gate.reserve(&crate::SpendAttempt {
decision_id: cx.decision_id.to_string(),
attempt_id: p.attempt_id.to_string(),
backend_id: p.target.backend_id.clone(),
target: p.target_index as u32,
artifact: p.target.artifact.clone(),
operation: self.operations[&p.req.step],
bound: p.bound,
cost_unit: binding.cost_unit,
}),
};
match result {
Ok(ticket) => Ok(Some(crate::spend::HeldSpend::new(spend.gate, ticket))),
Err(r) => {
*refusal = Some(r.clone());
Err(r)
}
}
}
pub(crate) fn expire_authorization(&self, attempt_id: &str) {
let mut presented = self.presented.lock().unwrap_or_else(|e| e.into_inner());
if let Some(p) = presented.iter_mut().find(|p| p.attempt_id == attempt_id)
&& let AuthorizationOutcome::Allowed { reference } = &p.authorization
{
p.authorization = AuthorizationOutcome::Refused {
refusal: AuthorizationRefusal::Expired,
reference: Some(reference.clone()),
detail: "authorization expired during spend reservation".into(),
};
}
}
pub(crate) fn authorize(
&self,
cx: &DecisionCtx<'_, '_>,
p: Proposal<'_>,
) -> Result<u64, AuthorizationRefusal> {
let now = cx.rt.clock.now();
let identity = cx
.ev()
.request_document(
&p.req.step,
&p.req.instance,
p.target_index as u32,
self.scope,
)
.map_err(|e| e.to_string())
.and_then(|d| d.id().map_err(|e| e.to_string()));
let request = match identity {
Ok(id) => id,
Err(_) => return Err(AuthorizationRefusal::Errored),
};
let asked = AuthorizationRequest {
decision_id: cx.decision_id.to_string(),
subject: self.subject.clone(),
scope: self.scope.clone(),
step: p.req.step.clone(),
instance: p.req.instance.clone(),
operation: self.operations[&p.req.step],
target: p.target_index as u32,
role: p.role,
backend_id: p.target.backend_id.clone(),
artifact: p.target.artifact.clone(),
request: request.clone(),
attempt_id: p.attempt_id.to_string(),
exposure: Exposure {
bound: p.bound,
reserved_units: p.reserved,
},
remaining_ms: cx.deadline.saturating_sub(now),
};
let Ok(digest) = asked.digest() else {
return Err(AuthorizationRefusal::Errored);
};
let refused = |refusal, reference, detail: &str| AuthorizationOutcome::Refused {
refusal,
reference,
detail: cut(detail, DETAIL_BYTES),
};
let mut valid_until = now;
let outcome = match catch_unwind(AssertUnwindSafe(|| self.authorizer.authorize(&asked))) {
Err(p) => refused(
AuthorizationRefusal::Errored,
None,
&format!(
"authorizer panicked: {}",
crate::wait::panic_detail(p.as_ref())
),
),
Ok(Err(detail)) => refused(AuthorizationRefusal::Errored, None, &detail),
Ok(Ok(None)) => refused(AuthorizationRefusal::Missing, None, "no answer"),
Ok(Ok(Some(DispatchAnswer::Refuse { reference, detail }))) => {
refused(AuthorizationRefusal::Refused, reference, &detail)
}
Ok(Ok(Some(DispatchAnswer::Allow {
answers,
reference,
valid_for_ms,
}))) => {
if answers != digest {
refused(
AuthorizationRefusal::Mismatched,
Some(reference),
"the answer names another request",
)
} else if cx.rt.clock.now() >= now.saturating_add(valid_for_ms) {
refused(
AuthorizationRefusal::Expired,
Some(reference),
&format!("the answer was valid for {valid_for_ms} ms"),
)
} else {
valid_until = now.saturating_add(valid_for_ms);
AuthorizationOutcome::Allowed { reference }
}
}
};
let result = match &outcome {
AuthorizationOutcome::Allowed { .. } => Ok(valid_until),
AuthorizationOutcome::Refused { refusal, .. } => Err(*refusal),
};
self.presented
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(Presented {
step: asked.step,
instance: asked.instance,
target: asked.target,
role: asked.role,
attempt_id: asked.attempt_id,
backend_id: asked.backend_id,
artifact: asked.artifact,
request,
presented: digest,
exposure: asked.exposure,
remaining_ms: asked.remaining_ms,
authorization: outcome,
});
result
}
}
fn worst_case_attempts(plan: &PreparedPlan) -> u64 {
match &plan.compiled().plan.execution {
PlanExecution::Declared(d) => d.policy.max_attempts_per_decision,
PlanExecution::None => plan
.compiled()
.plan
.steps
.iter()
.filter_map(|s| match &s.detail {
PlanStepDetail::Semantic(b) => Some(b.max_requests),
_ => None,
})
.fold(0, u64::saturating_add),
}
}
impl Runtime {
pub async fn reexecute_live(
&self,
plan: &PreparedPlan,
req: LiveRequest,
authorizer: &dyn LiveAuthorizer,
cancel: &CancelSignal,
) -> Result<LiveResult, LiveRefusal> {
self.reexecute_live_inner(plan, req, authorizer, cancel, None)
.await
}
pub async fn reexecute_live_with_gate(
&self,
plan: &PreparedPlan,
req: LiveRequest,
authorizer: &dyn LiveAuthorizer,
cancel: &CancelSignal,
spend: crate::LiveSpend<'_>,
) -> Result<LiveResult, LiveRefusal> {
self.reexecute_live_inner(plan, req, authorizer, cancel, Some(spend))
.await
}
async fn reexecute_live_inner(
&self,
plan: &PreparedPlan,
req: LiveRequest,
authorizer: &dyn LiveAuthorizer,
cancel: &CancelSignal,
spend: Option<crate::LiveSpend<'_>>,
) -> Result<LiveResult, LiveRefusal> {
if spend.is_some()
&& plan
.compiled()
.plan
.acquisitions
.as_ref()
.is_some_and(|a| !a.is_empty())
{
return Err(LiveRefusal::SpendAcquisitionUnsupported);
}
if req
.historical
.as_ref()
.is_some_and(|h| h.decision_id == req.decision_id)
{
return Err(LiveRefusal::HistoricalDecisionId);
}
if req
.historical
.as_ref()
.is_some_and(|h| h.decision_id.is_empty() || h.decision_id.len() > 256)
{
return Err(LiveRefusal::InvalidHistoricalLink);
}
let snapshot_id = req
.snapshot
.id()
.map_err(|e| LiveRefusal::InvalidSnapshot(e.to_string()))?;
if let Some(stated) = &req.expected_snapshot_id
&& *stated != snapshot_id
{
return Err(LiveRefusal::SnapshotIdentity {
stated: stated.clone(),
computed: snapshot_id,
});
}
if let Some(c) = &req.capture {
c.check().map_err(LiveRefusal::Capture)?;
}
let worst_case = worst_case_attempts(plan);
if worst_case > MAX_EFFECTS as u64 {
return Err(LiveRefusal::Unbounded {
worst_case,
max: MAX_EFFECTS as u64,
});
}
let operations = plan
.compiled()
.plan
.definition
.steps
.iter()
.filter_map(|s| match &s.body {
StepBody::Semantic(d) => Some((s.id.clone(), d.operation)),
_ => None,
})
.collect();
let live = LiveCtx {
authorizer,
spend,
spend_refusal: Mutex::new(None),
subject: &req.subject,
scope: &req.scope,
operations,
presented: Mutex::new(vec![]),
};
let decision = DecisionRequest {
decision_id: req.decision_id.clone(),
snapshot: req.snapshot.clone(),
evaluation_time: req.evaluation_time,
deadline_ms: req.deadline_ms,
principal_handle: req.principal_handle.clone(),
};
let mut capturing = req.capture.as_ref().map(capture::Capturing::new);
let result = self
.run(
plan,
decision,
cancel,
capturing.as_mut(),
Some(&live),
None,
None,
)
.await;
let (record, delivery) = match result {
Err(DecideError::GenerationFailed(detail)) => {
return Err(LiveRefusal::GenerationFailed(detail));
}
Err(DecideError::Rejected(r)) => return Err(LiveRefusal::Rejected(r)),
Ok(Decided {
record, delivery, ..
}) => (record, Ok(delivery)),
Err(DecideError::EvidenceNotDelivered { record, failure }) => (record, Err(failure)),
};
let capture = capturing.map(|c| {
CaptureOutcome::Captured(c.finish(
&req.decision_id,
record.termination == Termination::Cancelled,
))
});
let spend_refusal = live.spend_refusal();
let presented = live
.presented
.into_inner()
.unwrap_or_else(|e| e.into_inner());
let document = document(&req, snapshot_id, &record, &delivery, &capture, presented);
let completion = match &record.core {
CoreEvidence::Recorded(e) if record.termination == Termination::Judged => {
Completion::Judged(Box::new(e.judgment.clone()))
}
_ => Completion::Cancelled,
};
Ok(LiveResult {
spend_refusal,
completion,
record,
delivery,
capture,
document,
})
}
}
fn document(
req: &LiveRequest,
snapshot_id: SnapshotId,
record: &RunRecord,
delivery: &Result<Delivery, DeliveryFailure>,
capture: &Option<CaptureOutcome>,
presented: Vec<Presented>,
) -> LiveReexecution {
let attempts: BTreeMap<&str, &rustev_contract::run::AttemptRecord> = record
.requests
.iter()
.flat_map(|r| &r.attempts)
.map(|a| (a.attempt_id.as_str(), a))
.collect();
let effects = presented
.into_iter()
.map(|p| {
let dispatch = match attempts.get(p.attempt_id.as_str()) {
Some(a) => Dispatch::Dispatched {
end: a.end.clone(),
cancellation: a.cancellation,
remote: a.remote,
charge: a.cost.charge,
},
None => Dispatch::NotDispatched,
};
LiveEffect {
step: p.step,
instance: p.instance,
target: p.target,
role: p.role,
attempt_id: p.attempt_id,
backend_id: p.backend_id,
artifact: p.artifact,
request: p.request,
presented: p.presented,
exposure: p.exposure,
remaining_ms: p.remaining_ms,
authorization: p.authorization,
dispatch,
}
})
.collect();
let unreconciled = attempts
.values()
.filter(|a| a.cost.charge == Charge::Unknown)
.map(|a| a.attempt_id.clone())
.collect();
let delivery = match delivery {
Ok(Delivery::Acknowledged { receipt }) => DeliveryObservation::Acknowledged {
receipt: cut(receipt, DETAIL_BYTES),
},
Ok(Delivery::Queued(_)) => DeliveryObservation::Queued,
Ok(Delivery::Dropped { dropped_total }) => DeliveryObservation::Dropped {
dropped_total: *dropped_total,
},
Err(f) => DeliveryObservation::NotDelivered {
detail: cut(&f.detail, DETAIL_BYTES),
uncertain: f.uncertain,
},
};
let capture = match (&req.capture, capture) {
(Some(c), Some(CaptureOutcome::Captured(k))) => CapturePosture::On {
scope: c.scope.clone(),
max_bytes: c.max_bytes,
status: k.status,
},
_ => CapturePosture::Off,
};
LiveReexecution {
schema: schema::LIVE_REEXECUTION.into(),
decision_id: req.decision_id.clone(),
plan_id: record.plan_id.clone(),
snapshot_id,
evaluation_time_ms: req.evaluation_time,
subject: req.subject.clone(),
scope: req.scope.clone(),
historical: match &req.historical {
Some(h) => Historical::Linked(h.clone()),
None => Historical::None,
},
admission: Admission {
queued_ms: record.timing.queued_ms,
deadline_ms: record.timing.deadline_ms,
},
run_digest: record.record_digest().ok(),
run: RunLink {
termination: record.termination,
cost: record.cost,
},
effects,
unreconciled,
delivery,
retention: RetentionPosture {
capture,
historical: if req.historical.is_some() {
HistoricalRetention::NotObserved
} else {
HistoricalRetention::NoHistory
},
},
comparison: ComparisonSection::NotRequested,
}
}