1#[derive(Debug, thiserror::Error)]
3pub enum VMError {
4 #[error("coroutine {coro_id} faulted: {fault}")]
6 Fault {
7 coro_id: usize,
9 fault: Fault,
11 },
12 #[error("max sessions ({max}) exceeded")]
14 TooManySessions {
15 max: usize,
17 },
18 #[error("max coroutines ({max}) exceeded")]
20 TooManyCoroutines {
21 max: usize,
23 },
24 #[error("session {0} not found")]
26 SessionNotFound(SessionId),
27 #[error("effect handler error: {0}")]
29 HandlerError(EffectFailure),
30 #[error("persistence error: {0}")]
32 PersistenceError(String),
33 #[error("invalid concurrency level: {n}")]
35 InvalidConcurrency {
36 n: usize,
38 },
39 #[error("invalid VM config: {reason}")]
41 InvalidConfig {
42 reason: String,
44 },
45 #[error("thread pool build failed: {message}")]
47 ThreadPoolBuild {
48 message: String,
50 },
51 #[error("invalid code image: {reason}")]
53 InvalidCodeImage {
54 reason: String,
56 },
57 #[error("ownership contract error: {0}")]
59 OwnershipContract(String),
60}
61
62pub(crate) enum CoroUpdate {
66 AdvancePc,
68 SetPc(PC),
70 Block(BlockReason),
72 AdvancePcBlock(BlockReason),
74 Halt,
76 AdvancePcWriteReg { reg: u16, val: Value },
78}
79
80pub(crate) enum TypeUpdate {
82 Advance(LocalTypeR),
84 AdvanceWithOriginal(LocalTypeR, LocalTypeR),
86 Remove,
88}
89
90pub(crate) fn resolve_type_update(
92 cont: &LocalTypeR,
93 original: &LocalTypeR,
94 ep: &Endpoint,
95) -> (LocalTypeR, Option<(Endpoint, TypeUpdate)>) {
96 let (resolved, new_scope) = unfold_if_var_with_scope(cont, original);
97 let update = if let Some(mu) = new_scope {
98 Some((
99 ep.clone(),
100 TypeUpdate::AdvanceWithOriginal(resolved.clone(), mu),
101 ))
102 } else {
103 Some((ep.clone(), TypeUpdate::Advance(resolved.clone())))
104 };
105 (resolved, update)
106}
107
108pub(crate) struct StepPack {
113 pub(crate) coro_update: CoroUpdate,
115 pub(crate) type_update: Option<(Endpoint, TypeUpdate)>,
117 pub(crate) events: Vec<ObsEvent>,
119}
120
121#[derive(Clone, Copy)]
122pub(crate) struct GuardAcquireInput<'a> {
123 pub coro_idx: usize,
124 pub endpoint: &'a Endpoint,
125 pub role: &'a str,
126 pub sid: SessionId,
127 pub layer: &'a str,
128 pub dst: u16,
129}
130
131#[derive(Clone, Copy)]
132pub(crate) struct GuardReleaseInput<'a> {
133 pub coro_idx: usize,
134 pub endpoint: &'a Endpoint,
135 pub role: &'a str,
136 pub sid: SessionId,
137 pub layer: &'a str,
138 pub evidence: u16,
139}
140
141pub(crate) enum ExecOutcome {
143 Continue,
145 Blocked(BlockReason),
147 Halted,
149}
150
151#[derive(Debug, Clone, Serialize, Deserialize)]
155#[serde(transparent)]
156pub(crate) struct RetainedLog<T>(Vec<T>);
157
158fn default_true() -> bool {
159 true
160}
161
162impl<T> Default for RetainedLog<T> {
163 fn default() -> Self {
164 Self(Vec::new())
165 }
166}
167
168impl<T> RetainedLog<T> {
169 fn push(&mut self, item: T, config: &ObservabilityRetentionConfig) {
170 match config.mode {
171 ObservabilityRetentionMode::Disabled => {}
172 ObservabilityRetentionMode::Full => self.0.push(item),
173 ObservabilityRetentionMode::Capped => {
174 self.0.push(item);
175 self.trim_to_capacity(config.capacity);
176 }
177 }
178 }
179
180 fn extend<I>(&mut self, iter: I, config: &ObservabilityRetentionConfig)
181 where
182 I: IntoIterator<Item = T>,
183 {
184 match config.mode {
185 ObservabilityRetentionMode::Disabled => {}
186 ObservabilityRetentionMode::Full => self.0.extend(iter),
187 ObservabilityRetentionMode::Capped => {
188 self.0.extend(iter);
189 self.trim_to_capacity(config.capacity);
190 }
191 }
192 }
193
194 fn as_slice(&self) -> &[T] {
195 &self.0
196 }
197
198 fn len(&self) -> usize {
199 self.0.len()
200 }
201
202 fn drain(&mut self) -> Vec<T> {
203 self.0.drain(..).collect()
204 }
205
206 fn trim_to_capacity(&mut self, capacity: usize) {
207 if self.0.len() > capacity {
208 let overflow = self.0.len() - capacity;
209 self.0.drain(0..overflow);
210 }
211 }
212}
213
214impl<T> std::ops::Deref for RetainedLog<T> {
215 type Target = [T];
216
217 fn deref(&self) -> &Self::Target {
218 self.as_slice()
219 }
220}
221
222#[derive(Debug, Serialize, Deserialize)]
228pub struct VM<I = (), G = (), P = NoopPersistence, Nu = DefaultVerificationModel>
229where
230 P: PersistenceModel,
231{
232 config: VMConfig,
233 code: Option<Program>,
234 programs: ProgramStore,
235 identity_model: PhantomData<I>,
236 guard_model: PhantomData<G>,
237 persistence_model: PhantomData<P>,
238 persistent: P::PState,
239 verification: Nu,
240 #[serde(default)]
241 communication_consumption: DefaultCommunicationConsumption,
242 #[serde(default)]
243 communication_consumption_artifacts: RetainedLog<CommunicationConsumptionArtifact>,
244 coroutines: Vec<Coroutine>,
245 sessions: SessionStore,
246 arena: Arena,
247 resource_states: BTreeMap<ScopeId, ResourceState>,
248 sched: Scheduler,
249 monitor: SessionMonitor,
250 obs_trace: RetainedLog<ObsEvent>,
251 role_symbols: SymbolTable,
252 label_symbols: SymbolTable,
253 handler_symbols: SymbolTable,
254 edge_symbols: EdgeSymbolTable,
255 clock: SimClock,
256 next_coro_id: usize,
257 next_session_id: SessionId,
258 paused_roles: BTreeSet<String>,
259 #[serde(skip, default)]
260 coro_slots: BTreeMap<usize, usize>,
261 #[serde(skip, default)]
262 role_coroutines: BTreeMap<String, Vec<usize>>,
263 #[serde(skip, default)]
264 paused_coro_ids: BTreeSet<usize>,
265 #[serde(skip, default)]
266 timed_out_coro_ids: BTreeSet<usize>,
267 #[serde(skip, default)]
268 session_open_plans: BTreeMap<String, crate::session::SessionOpenPlan>,
269 #[serde(skip, default)]
270 eligible_ready: BTreeSet<usize>,
271 #[serde(skip, default = "default_true")]
272 eligibility_dirty: bool,
273 guard_layer: InMemoryGuardLayer,
274 effect_trace: RetainedLog<EffectTraceEntry>,
275 next_effect_id: u64,
276 output_condition_checks: RetainedLog<OutputConditionCheck>,
277 delegation_audit_log: RetainedLog<DelegationAuditRecord>,
278 next_delegation_receipt_id: u64,
279 authority_audit_log: RetainedLog<AuthorityAuditRecord>,
280 next_authority_witness_id: u64,
281 crashed_sites: BTreeSet<SiteId>,
282 partitioned_edges: BTreeSet<(SiteId, SiteId)>,
283 corrupted_edges: BTreeMap<(SiteId, SiteId), CorruptionType>,
284 timed_out_sites: BTreeMap<SiteId, u64>,
285 last_sched_step: Option<SchedStepDebug>,
286 handler_identity_anchor: Option<String>,
287}
288
289pub type VMState<I = (), G = (), P = NoopPersistence, Nu = DefaultVerificationModel> =
291 VM<I, G, P, Nu>;
292
293impl<I, G, P, Nu> VM<I, G, P, Nu>
294where
295 P: PersistenceModel,
296{
297 #[must_use]
299 pub fn new_with_models(config: VMConfig) -> Self
300 where
301 P::PState: Default,
302 Nu: VerificationModel + Default,
303 {
304 config.assert_invariants();
305 let tick_duration = config.tick_duration;
306 let communication_replay_mode = config.communication_replay_mode;
307 let sched = Scheduler::new(config.sched_policy.clone());
308 let mut guard_resources = BTreeMap::new();
309 for layer in &config.guard_layers {
310 guard_resources.insert(layer.id.clone(), Value::Unit);
311 }
312 Self {
313 config,
314 code: None,
315 programs: ProgramStore::new(),
316 identity_model: PhantomData,
317 guard_model: PhantomData,
318 persistence_model: PhantomData,
319 persistent: P::PState::default(),
320 verification: Nu::default(),
321 communication_consumption: DefaultCommunicationConsumption::new(
322 communication_replay_mode,
323 ),
324 communication_consumption_artifacts: RetainedLog::default(),
325 coroutines: Vec::new(),
326 sessions: SessionStore::new(),
327 arena: Arena::default(),
328 resource_states: BTreeMap::new(),
329 sched,
330 monitor: SessionMonitor::default(),
331 obs_trace: RetainedLog::default(),
332 role_symbols: SymbolTable::new(),
333 label_symbols: SymbolTable::new(),
334 handler_symbols: SymbolTable::new(),
335 edge_symbols: EdgeSymbolTable::new(),
336 clock: SimClock::new(tick_duration),
337 next_coro_id: 0,
338 next_session_id: 0,
339 paused_roles: BTreeSet::new(),
340 coro_slots: BTreeMap::new(),
341 role_coroutines: BTreeMap::new(),
342 paused_coro_ids: BTreeSet::new(),
343 timed_out_coro_ids: BTreeSet::new(),
344 session_open_plans: BTreeMap::new(),
345 eligible_ready: BTreeSet::new(),
346 eligibility_dirty: true,
347 guard_layer: InMemoryGuardLayer {
348 resources: guard_resources
349 .into_iter()
350 .map(|(k, v)| (LayerId(k), v))
351 .collect(),
352 },
353 effect_trace: RetainedLog::default(),
354 next_effect_id: 0,
355 output_condition_checks: RetainedLog::default(),
356 delegation_audit_log: RetainedLog::default(),
357 next_delegation_receipt_id: 0,
358 authority_audit_log: RetainedLog::default(),
359 next_authority_witness_id: 0,
360 crashed_sites: BTreeSet::new(),
361 partitioned_edges: BTreeSet::new(),
362 corrupted_edges: BTreeMap::new(),
363 timed_out_sites: BTreeMap::new(),
364 last_sched_step: None,
365 handler_identity_anchor: None,
366 }
367 }
368
369 #[must_use]
371 pub fn persistent_state(&self) -> &P::PState {
372 &self.persistent
373 }
374
375 pub fn persistent_state_mut(&mut self) -> &mut P::PState {
377 &mut self.persistent
378 }
379
380 fn apply_open_delta(&mut self, sid: SessionId) -> Result<(), String> {
381 let delta = P::open_delta(sid);
382 P::apply(&mut self.persistent, &delta)
383 }
384
385 fn apply_close_delta(&mut self, sid: SessionId) -> Result<(), String> {
386 let delta = P::close_delta(sid);
387 P::apply(&mut self.persistent, &delta)
388 }
389
390 fn apply_invoke_delta(&mut self, sid: SessionId, action: &str) -> Result<(), String> {
391 if let Some(delta) = P::invoke_delta(sid, action) {
392 P::apply(&mut self.persistent, &delta)?;
393 }
394 Ok(())
395 }
396
397 #[must_use]
399 pub fn bridge_guard_layer_for_participant<B>(
400 &self,
401 bridge: &B,
402 participant: &I::ParticipantId,
403 ) -> LayerId
404 where
405 I: IdentityModel,
406 G: GuardLayer,
407 B: IdentityGuardBridge<I, G>,
408 {
409 bridge.guard_layer_for_participant(participant)
410 }
411
412 #[must_use]
414 pub fn bridge_verifying_key_for_participant<B>(
415 &self,
416 bridge: &B,
417 participant: &I::ParticipantId,
418 ) -> Nu::VerifyingKey
419 where
420 I: IdentityModel,
421 Nu: VerificationModel,
422 B: IdentityVerificationBridge<I, Nu>,
423 {
424 bridge.verification_key_for_participant(participant)
425 }
426}