1use serde::{Deserialize, Serialize};
2use std::collections::{BTreeMap, BTreeSet};
3
4use super::{
5 canonical_fingerprint, invalid_event, validate_sha256, ExecutionEventKind, ExecutionFrameId,
6 ExecutionIdentityEnvelope, ExecutionIdentityParts, ExecutionPhase, FailureDomain,
7 FailureEnvelope, FailureEnvelopeWire, MonotonicTimestamp, NodeId, NodeInvocationId,
8 OperationParticipantCompletionReceipt, RequestIdentity, ResourceTransactionIdentity, RunId,
9 SpanId, SubmittedOperationReceipt, TrustedAbortedSequenceBinding, TrustedActiveSequenceBinding,
10 TrustedCompletedSequenceBinding, TrustedExecutionTopology, UnvalidatedExecutionIdentityParts,
11 UnvalidatedFailureEnvelope, VNextError, MAX_EXECUTION_EVENT_WIRE_BYTES,
12};
13
14#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
15pub struct IdentifiedFailure {
16 identity: ExecutionIdentityEnvelope,
17 failure: FailureEnvelope,
18}
19
20impl IdentifiedFailure {
21 pub fn new(
22 identity: ExecutionIdentityEnvelope,
23 failure: FailureEnvelope,
24 ) -> Result<Self, VNextError> {
25 failure.validate()?;
26 validate_failure_identity(failure.domain(), identity.parts())?;
27 Ok(Self { identity, failure })
28 }
29
30 pub fn identity(&self) -> &ExecutionIdentityEnvelope {
31 &self.identity
32 }
33
34 pub fn failure(&self) -> &FailureEnvelope {
35 &self.failure
36 }
37
38 pub fn fingerprint(&self) -> String {
39 canonical_fingerprint(self)
40 }
41}
42
43#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
44pub struct UnvalidatedIdentifiedFailure {
45 identity: UnvalidatedExecutionIdentityParts,
46 failure: UnvalidatedFailureEnvelope,
47}
48
49#[derive(Deserialize)]
50#[serde(deny_unknown_fields)]
51struct UnvalidatedIdentifiedFailureWire {
52 identity: UnvalidatedExecutionIdentityParts,
53 failure: FailureEnvelopeWire,
54}
55
56impl From<UnvalidatedIdentifiedFailureWire> for UnvalidatedIdentifiedFailure {
57 fn from(wire: UnvalidatedIdentifiedFailureWire) -> Self {
58 Self {
59 identity: wire.identity,
60 failure: wire.failure.into(),
61 }
62 }
63}
64
65#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
66#[serde(rename_all = "snake_case")]
67pub enum ExecutionEventDetail {
68 None,
69 MonotonicClockAnchor {
70 clock_source: String,
71 wall_anchor_unix_nanos: i64,
72 max_error_nanos: u64,
73 },
74 Counters {
75 input: u64,
76 output: u64,
77 },
78 Failure(IdentifiedFailure),
79 FailureTerminal {
80 first_failure_fingerprint: String,
81 },
82}
83
84#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
85#[serde(rename_all = "snake_case")]
86pub enum UnvalidatedExecutionEventDetail {
87 None,
88 MonotonicClockAnchor {
89 clock_source: String,
90 wall_anchor_unix_nanos: i64,
91 max_error_nanos: u64,
92 },
93 Counters {
94 input: u64,
95 output: u64,
96 },
97 Failure(UnvalidatedIdentifiedFailure),
98 FailureTerminal {
99 first_failure_fingerprint: String,
100 },
101}
102
103#[derive(Deserialize)]
104#[serde(rename_all = "snake_case")]
105enum UnvalidatedExecutionEventDetailWire {
106 None,
107 MonotonicClockAnchor {
108 clock_source: String,
109 wall_anchor_unix_nanos: i64,
110 max_error_nanos: u64,
111 },
112 Counters {
113 input: u64,
114 output: u64,
115 },
116 Failure(UnvalidatedIdentifiedFailureWire),
117 FailureTerminal {
118 first_failure_fingerprint: String,
119 },
120}
121
122impl From<UnvalidatedExecutionEventDetailWire> for UnvalidatedExecutionEventDetail {
123 fn from(wire: UnvalidatedExecutionEventDetailWire) -> Self {
124 match wire {
125 UnvalidatedExecutionEventDetailWire::None => Self::None,
126 UnvalidatedExecutionEventDetailWire::MonotonicClockAnchor {
127 clock_source,
128 wall_anchor_unix_nanos,
129 max_error_nanos,
130 } => Self::MonotonicClockAnchor {
131 clock_source,
132 wall_anchor_unix_nanos,
133 max_error_nanos,
134 },
135 UnvalidatedExecutionEventDetailWire::Counters { input, output } => {
136 Self::Counters { input, output }
137 }
138 UnvalidatedExecutionEventDetailWire::Failure(failure) => Self::Failure(failure.into()),
139 UnvalidatedExecutionEventDetailWire::FailureTerminal {
140 first_failure_fingerprint,
141 } => Self::FailureTerminal {
142 first_failure_fingerprint,
143 },
144 }
145 }
146}
147
148#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
149pub struct ExecutionEvent {
150 timestamp: MonotonicTimestamp,
151 phase: ExecutionPhase,
152 kind: ExecutionEventKind,
153 identity: ExecutionIdentityEnvelope,
154 detail: ExecutionEventDetail,
155}
156
157#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
158pub struct UnvalidatedExecutionEvent {
159 timestamp: MonotonicTimestamp,
160 phase: ExecutionPhase,
161 kind: ExecutionEventKind,
162 identity: UnvalidatedExecutionIdentityParts,
163 detail: UnvalidatedExecutionEventDetail,
164}
165
166#[derive(Deserialize)]
167#[serde(deny_unknown_fields)]
168struct ExecutionEventWire {
169 timestamp: MonotonicTimestamp,
170 phase: ExecutionPhase,
171 kind: ExecutionEventKind,
172 identity: UnvalidatedExecutionIdentityParts,
173 detail: UnvalidatedExecutionEventDetailWire,
174}
175
176impl From<ExecutionEventWire> for UnvalidatedExecutionEvent {
177 fn from(wire: ExecutionEventWire) -> Self {
178 Self {
179 timestamp: wire.timestamp,
180 phase: wire.phase,
181 kind: wire.kind,
182 identity: wire.identity,
183 detail: wire.detail.into(),
184 }
185 }
186}
187
188impl ExecutionEvent {
189 pub fn new(
190 timestamp: MonotonicTimestamp,
191 phase: ExecutionPhase,
192 kind: ExecutionEventKind,
193 identity: ExecutionIdentityEnvelope,
194 detail: ExecutionEventDetail,
195 ) -> Result<Self, VNextError> {
196 validate_event_shape(phase, kind, identity.parts(), &detail)?;
197 Ok(Self {
198 timestamp,
199 phase,
200 kind,
201 identity,
202 detail,
203 })
204 }
205
206 pub const fn timestamp(&self) -> MonotonicTimestamp {
207 self.timestamp
208 }
209
210 pub const fn phase(&self) -> ExecutionPhase {
211 self.phase
212 }
213
214 pub const fn kind(&self) -> ExecutionEventKind {
215 self.kind
216 }
217
218 pub fn identity(&self) -> &ExecutionIdentityEnvelope {
219 &self.identity
220 }
221
222 pub fn detail(&self) -> &ExecutionEventDetail {
223 &self.detail
224 }
225
226 pub fn decode_untrusted(bytes: &[u8]) -> Result<UnvalidatedExecutionEvent, VNextError> {
227 if bytes.len() > MAX_EXECUTION_EVENT_WIRE_BYTES {
228 return Err(invalid_event(
229 "untrusted execution event exceeds the wire byte limit",
230 ));
231 }
232 let raw = serde_json::from_slice::<serde_json::Value>(bytes).map_err(|error| {
233 VNextError::Serialization {
234 context: "decode untrusted execution event",
235 message: error.to_string(),
236 }
237 })?;
238 let event = serde_json::from_value::<ExecutionEventWire>(raw.clone())
239 .map(UnvalidatedExecutionEvent::from)
240 .map_err(|error| VNextError::Serialization {
241 context: "decode untrusted execution event",
242 message: error.to_string(),
243 })?;
244 let canonical =
245 serde_json::to_value(&event).map_err(|error| VNextError::Serialization {
246 context: "serialize untrusted execution event",
247 message: error.to_string(),
248 })?;
249 if canonical != raw {
250 return Err(invalid_event(
251 "execution event wire contains unknown or non-canonical nested fields",
252 ));
253 }
254 Ok(event)
255 }
256}
257
258pub struct TrustedExecutionEventContext<'a> {
259 run_id: &'a RunId,
260 request_id: &'a RequestIdentity,
261 topology: Option<&'a TrustedExecutionTopology>,
262 active: Option<&'a TrustedActiveSequenceBinding>,
263 completed: Option<&'a TrustedCompletedSequenceBinding>,
264 aborted: Option<&'a TrustedAbortedSequenceBinding>,
265 submitted_operation: Option<&'a SubmittedOperationReceipt>,
266 retired_operation: Option<&'a OperationParticipantCompletionReceipt>,
267 expected_failure: Option<&'a IdentifiedFailure>,
268 unsubmitted_recovery_identity: Option<&'a ExecutionIdentityEnvelope>,
269}
270
271impl<'a> TrustedExecutionEventContext<'a> {
272 pub fn pre_plan(run_id: &'a RunId, request_id: &'a RequestIdentity) -> Self {
273 Self {
274 run_id,
275 request_id,
276 topology: None,
277 active: None,
278 completed: None,
279 aborted: None,
280 submitted_operation: None,
281 retired_operation: None,
282 expected_failure: None,
283 unsubmitted_recovery_identity: None,
284 }
285 }
286
287 pub fn bound(
288 run_id: &'a RunId,
289 request_id: &'a RequestIdentity,
290 topology: &'a TrustedExecutionTopology,
291 ) -> Self {
292 Self {
293 run_id,
294 request_id,
295 topology: Some(topology),
296 active: None,
297 completed: None,
298 aborted: None,
299 submitted_operation: None,
300 retired_operation: None,
301 expected_failure: None,
302 unsubmitted_recovery_identity: None,
303 }
304 }
305
306 pub fn active(
307 run_id: &'a RunId,
308 request_id: &'a RequestIdentity,
309 topology: &'a TrustedExecutionTopology,
310 active: &'a TrustedActiveSequenceBinding,
311 ) -> Self {
312 Self {
313 run_id,
314 request_id,
315 topology: Some(topology),
316 active: Some(active),
317 completed: None,
318 aborted: None,
319 submitted_operation: None,
320 retired_operation: None,
321 expected_failure: None,
322 unsubmitted_recovery_identity: None,
323 }
324 }
325
326 pub fn operation_submitted(
327 run_id: &'a RunId,
328 request_id: &'a RequestIdentity,
329 topology: &'a TrustedExecutionTopology,
330 active: &'a TrustedActiveSequenceBinding,
331 submitted_operation: &'a SubmittedOperationReceipt,
332 ) -> Self {
333 Self {
334 run_id,
335 request_id,
336 topology: Some(topology),
337 active: Some(active),
338 completed: None,
339 aborted: None,
340 submitted_operation: Some(submitted_operation),
341 retired_operation: None,
342 expected_failure: None,
343 unsubmitted_recovery_identity: None,
344 }
345 }
346
347 pub(super) fn replay_operation_submitted(
348 run_id: &'a RunId,
349 request_id: &'a RequestIdentity,
350 topology: &'a TrustedExecutionTopology,
351 active: &'a TrustedActiveSequenceBinding,
352 submitted_operation: &'a SubmittedOperationReceipt,
353 ) -> Self {
354 Self {
355 run_id,
356 request_id,
357 topology: Some(topology),
358 active: Some(active),
359 completed: None,
360 aborted: None,
361 submitted_operation: Some(submitted_operation),
362 retired_operation: None,
363 expected_failure: None,
364 unsubmitted_recovery_identity: None,
365 }
366 }
367
368 pub fn node_retired(
369 run_id: &'a RunId,
370 request_id: &'a RequestIdentity,
371 topology: &'a TrustedExecutionTopology,
372 active: &'a TrustedActiveSequenceBinding,
373 retired_operation: &'a OperationParticipantCompletionReceipt,
374 ) -> Self {
375 Self {
376 run_id,
377 request_id,
378 topology: Some(topology),
379 active: Some(active),
380 completed: None,
381 aborted: None,
382 submitted_operation: None,
383 retired_operation: Some(retired_operation),
384 expected_failure: None,
385 unsubmitted_recovery_identity: None,
386 }
387 }
388
389 pub(super) fn replay_node_retired(
390 run_id: &'a RunId,
391 request_id: &'a RequestIdentity,
392 topology: &'a TrustedExecutionTopology,
393 active: &'a TrustedActiveSequenceBinding,
394 retired_operation: &'a OperationParticipantCompletionReceipt,
395 ) -> Self {
396 Self::node_retired(run_id, request_id, topology, active, retired_operation)
397 }
398
399 pub fn completed(
400 run_id: &'a RunId,
401 request_id: &'a RequestIdentity,
402 topology: &'a TrustedExecutionTopology,
403 active: &'a TrustedActiveSequenceBinding,
404 completed: &'a TrustedCompletedSequenceBinding,
405 ) -> Self {
406 Self {
407 run_id,
408 request_id,
409 topology: Some(topology),
410 active: Some(active),
411 completed: Some(completed),
412 aborted: None,
413 submitted_operation: None,
414 retired_operation: None,
415 expected_failure: None,
416 unsubmitted_recovery_identity: None,
417 }
418 }
419
420 pub fn aborted(
421 run_id: &'a RunId,
422 request_id: &'a RequestIdentity,
423 topology: &'a TrustedExecutionTopology,
424 active: &'a TrustedActiveSequenceBinding,
425 aborted: &'a TrustedAbortedSequenceBinding,
426 ) -> Self {
427 Self {
428 run_id,
429 request_id,
430 topology: Some(topology),
431 active: Some(active),
432 completed: None,
433 aborted: Some(aborted),
434 submitted_operation: None,
435 retired_operation: None,
436 expected_failure: None,
437 unsubmitted_recovery_identity: None,
438 }
439 }
440
441 pub fn failure(
442 run_id: &'a RunId,
443 request_id: &'a RequestIdentity,
444 topology: Option<&'a TrustedExecutionTopology>,
445 active: Option<&'a TrustedActiveSequenceBinding>,
446 expected_failure: &'a IdentifiedFailure,
447 ) -> Self {
448 Self {
449 run_id,
450 request_id,
451 topology,
452 active,
453 completed: None,
454 aborted: None,
455 submitted_operation: None,
456 retired_operation: None,
457 expected_failure: Some(expected_failure),
458 unsubmitted_recovery_identity: None,
459 }
460 }
461
462 pub(super) fn replay_failure(
463 run_id: &'a RunId,
464 request_id: &'a RequestIdentity,
465 topology: Option<&'a TrustedExecutionTopology>,
466 active: Option<&'a TrustedActiveSequenceBinding>,
467 expected_failure: &'a IdentifiedFailure,
468 unsubmitted_recovery_identity: Option<&'a ExecutionIdentityEnvelope>,
469 ) -> Self {
470 Self {
471 run_id,
472 request_id,
473 topology,
474 active,
475 completed: None,
476 aborted: None,
477 submitted_operation: None,
478 retired_operation: None,
479 expected_failure: Some(expected_failure),
480 unsubmitted_recovery_identity,
481 }
482 }
483
484 pub fn failure_with_disposition(
485 run_id: &'a RunId,
486 request_id: &'a RequestIdentity,
487 topology: &'a TrustedExecutionTopology,
488 active: &'a TrustedActiveSequenceBinding,
489 completed: Option<&'a TrustedCompletedSequenceBinding>,
490 aborted: Option<&'a TrustedAbortedSequenceBinding>,
491 expected_failure: &'a IdentifiedFailure,
492 ) -> Self {
493 Self {
494 run_id,
495 request_id,
496 topology: Some(topology),
497 active: Some(active),
498 completed,
499 aborted,
500 submitted_operation: None,
501 retired_operation: None,
502 expected_failure: Some(expected_failure),
503 unsubmitted_recovery_identity: None,
504 }
505 }
506
507 pub(super) const fn active_binding(&self) -> Option<&'a TrustedActiveSequenceBinding> {
508 self.active
509 }
510}
511
512impl UnvalidatedExecutionEvent {
513 pub fn revalidate(
514 self,
515 context: &TrustedExecutionEventContext<'_>,
516 ) -> Result<ExecutionEvent, VNextError> {
517 let identity = ExecutionIdentityEnvelope::new(self.identity.into())?;
518 let detail = match self.detail {
519 UnvalidatedExecutionEventDetail::None => ExecutionEventDetail::None,
520 UnvalidatedExecutionEventDetail::MonotonicClockAnchor {
521 clock_source,
522 wall_anchor_unix_nanos,
523 max_error_nanos,
524 } => ExecutionEventDetail::MonotonicClockAnchor {
525 clock_source,
526 wall_anchor_unix_nanos,
527 max_error_nanos,
528 },
529 UnvalidatedExecutionEventDetail::Counters { input, output } => {
530 ExecutionEventDetail::Counters { input, output }
531 }
532 UnvalidatedExecutionEventDetail::Failure(failure) => {
533 let expected = context.expected_failure.ok_or_else(|| {
534 invalid_event("wire failure lacks independent trusted failure evidence")
535 })?;
536 let failure_identity = ExecutionIdentityEnvelope::new(failure.identity.into())?;
537 let trusted = IdentifiedFailure::new(
538 failure_identity,
539 failure.failure.revalidate(expected.failure().domain())?,
540 )?;
541 if &trusted != expected {
542 return Err(invalid_event(
543 "wire failure differs from independent failure evidence",
544 ));
545 }
546 ExecutionEventDetail::Failure(trusted)
547 }
548 UnvalidatedExecutionEventDetail::FailureTerminal {
549 first_failure_fingerprint,
550 } => {
551 validate_sha256(&first_failure_fingerprint, "first failure fingerprint")?;
552 let expected = context.expected_failure.ok_or_else(|| {
553 invalid_event("failure terminal lacks independent first failure evidence")
554 })?;
555 if first_failure_fingerprint != expected.fingerprint() {
556 return Err(invalid_event(
557 "failure terminal differs from the first observed failure",
558 ));
559 }
560 ExecutionEventDetail::FailureTerminal {
561 first_failure_fingerprint,
562 }
563 }
564 };
565 let event = ExecutionEvent::new(self.timestamp, self.phase, self.kind, identity, detail)?;
566 validate_event_against_context(&event, context)?;
567 Ok(event)
568 }
569}
570
571fn has_pool(ids: &ExecutionIdentityParts) -> bool {
572 ids.resource_pool_id.is_some()
573}
574
575pub(super) fn has_active(ids: &ExecutionIdentityParts) -> bool {
576 ids.active_sequence_slot.is_some()
577}
578
579pub(super) fn has_completed(ids: &ExecutionIdentityParts) -> bool {
580 ids.completed_sequence_fingerprint.is_some()
581}
582
583pub(super) fn has_aborted(ids: &ExecutionIdentityParts) -> bool {
584 ids.aborted_sequence_fingerprint.is_some()
585}
586
587fn no_resource_item(ids: &ExecutionIdentityParts) -> bool {
588 ids.resource_id.is_none()
589 && ids.resource_generation.is_none()
590 && ids.resource_batch_fingerprint.is_none()
591}
592
593fn exact_plan(ids: &ExecutionIdentityParts) -> bool {
594 ids.plan_id.is_some() && ids.plan_hash.is_some() && ids.device_id.is_some()
595}
596
597pub(super) fn same_operation_authority_except_observation(
598 observation: &ExecutionIdentityParts,
599 operation: &ExecutionIdentityParts,
600) -> bool {
601 let mut normalized_observation = observation.clone();
602 normalized_observation.sequence = operation.sequence;
603 normalized_observation.span_id = operation.span_id.clone();
604 normalized_observation.parent_span_id = operation.parent_span_id.clone();
605 normalized_observation == *operation
606}
607
608fn validate_event_shape(
609 phase: ExecutionPhase,
610 kind: ExecutionEventKind,
611 ids: &ExecutionIdentityParts,
612 detail: &ExecutionEventDetail,
613) -> Result<(), VNextError> {
614 let phase_ok = match kind {
615 ExecutionEventKind::RequestAccepted => phase == ExecutionPhase::Resolution,
616 ExecutionEventKind::PlanBuilt => phase == ExecutionPhase::Planning,
617 ExecutionEventKind::FrameStarted
618 | ExecutionEventKind::NodeStarted
619 | ExecutionEventKind::OperationSubmitted
620 | ExecutionEventKind::NodeRetired
621 | ExecutionEventKind::FrameCompleted => phase == ExecutionPhase::Execution,
622 ExecutionEventKind::FailureObserved => true,
623 ExecutionEventKind::SequenceCompleted
624 | ExecutionEventKind::SequenceAborted
625 | ExecutionEventKind::RequestCompleted => phase == ExecutionPhase::Completion,
626 ExecutionEventKind::RequestFailed => true,
627 };
628 if !phase_ok {
629 return Err(invalid_event(format!(
630 "event `{kind:?}` is invalid in phase `{phase:?}`"
631 )));
632 }
633 let no_plan = ids.plan_id.is_none() && ids.plan_hash.is_none() && ids.device_id.is_none();
634 let no_frame = ids.frame_id.is_none() && ids.node_invocation_id.is_none();
635 let no_node = ids.node_id.is_none() && ids.operation_id.is_none() && ids.provider_id.is_none();
636 let no_pool = !has_pool(ids) && !has_active(ids) && no_resource_item(ids);
637 let frame_shape = exact_plan(ids)
638 && ids.frame_id.is_some()
639 && ids.node_invocation_id.is_none()
640 && no_node
641 && has_active(ids)
642 && !has_completed(ids)
643 && !has_aborted(ids)
644 && no_resource_item(ids);
645 let node_shape = exact_plan(ids)
646 && ids.frame_id.is_some()
647 && ids.node_invocation_id.is_some()
648 && ids.node_id.is_some()
649 && ids.operation_id.is_some()
650 && ids.provider_id.is_some()
651 && has_active(ids)
652 && !has_completed(ids)
653 && !has_aborted(ids)
654 && no_resource_item(ids);
655 let completed_shape = exact_plan(ids)
656 && no_frame
657 && no_node
658 && has_active(ids)
659 && has_completed(ids)
660 && !has_aborted(ids)
661 && no_resource_item(ids);
662 let aborted_shape = exact_plan(ids)
663 && no_frame
664 && no_node
665 && has_active(ids)
666 && !has_completed(ids)
667 && has_aborted(ids)
668 && no_resource_item(ids);
669 let identity_ok = match kind {
670 ExecutionEventKind::RequestAccepted => no_plan && no_frame && no_node && no_pool,
671 ExecutionEventKind::PlanBuilt => exact_plan(ids) && no_frame && no_node && no_pool,
672 ExecutionEventKind::FrameStarted | ExecutionEventKind::FrameCompleted => frame_shape,
673 ExecutionEventKind::NodeStarted
674 | ExecutionEventKind::OperationSubmitted
675 | ExecutionEventKind::NodeRetired => node_shape,
676 ExecutionEventKind::SequenceCompleted | ExecutionEventKind::RequestCompleted => {
677 completed_shape
678 }
679 ExecutionEventKind::SequenceAborted => aborted_shape,
680 ExecutionEventKind::FailureObserved => match detail {
681 ExecutionEventDetail::Failure(failure) if has_active(ids) => {
682 let failed_operation = failure.identity().parts();
683 node_shape
684 && ids.sequence > failed_operation.sequence
685 && ids.parent_span_id.as_ref() == Some(&failed_operation.span_id)
686 && same_operation_authority_except_observation(ids, failed_operation)
687 }
688 ExecutionEventDetail::Failure(failure) => failure.identity().parts() == ids,
689 _ => false,
690 },
691 ExecutionEventKind::RequestFailed => match detail {
692 ExecutionEventDetail::Failure(failure) => failure.identity().parts() == ids,
693 ExecutionEventDetail::FailureTerminal { .. } => {
694 no_frame
695 && no_node
696 && no_resource_item(ids)
697 && (no_plan || exact_plan(ids))
698 && ((!has_active(ids) && !has_completed(ids) && !has_aborted(ids))
699 || (has_active(ids) && (has_completed(ids) ^ has_aborted(ids))))
700 }
701 _ => false,
702 },
703 };
704 if !identity_ok {
705 return Err(invalid_event(format!(
706 "event `{kind:?}` has missing or extraneous identity fields"
707 )));
708 }
709 let detail_ok = match (kind, detail) {
710 (
711 ExecutionEventKind::RequestAccepted,
712 ExecutionEventDetail::MonotonicClockAnchor {
713 clock_source,
714 wall_anchor_unix_nanos,
715 ..
716 },
717 ) => !clock_source.trim().is_empty() && *wall_anchor_unix_nanos > 0,
718 (ExecutionEventKind::RequestCompleted, ExecutionEventDetail::Counters { .. }) => true,
719 (ExecutionEventKind::FailureObserved, ExecutionEventDetail::Failure(_)) => true,
720 (
721 ExecutionEventKind::RequestFailed,
722 ExecutionEventDetail::Failure(_) | ExecutionEventDetail::FailureTerminal { .. },
723 ) => true,
724 (
725 ExecutionEventKind::RequestAccepted
726 | ExecutionEventKind::PlanBuilt
727 | ExecutionEventKind::FrameStarted
728 | ExecutionEventKind::NodeStarted
729 | ExecutionEventKind::OperationSubmitted
730 | ExecutionEventKind::NodeRetired
731 | ExecutionEventKind::FrameCompleted
732 | ExecutionEventKind::SequenceCompleted
733 | ExecutionEventKind::SequenceAborted,
734 ExecutionEventDetail::None,
735 ) => true,
736 _ => false,
737 };
738 if !detail_ok {
739 return Err(invalid_event(format!(
740 "event `{kind:?}` has invalid structured detail"
741 )));
742 }
743 if let ExecutionEventDetail::FailureTerminal {
744 first_failure_fingerprint,
745 } = detail
746 {
747 validate_sha256(first_failure_fingerprint, "first failure fingerprint")?;
748 }
749 Ok(())
750}
751
752fn validate_active_identity(
753 ids: &ExecutionIdentityParts,
754 active: &TrustedActiveSequenceBinding,
755) -> Result<(), VNextError> {
756 let provisioning = active.static_provisioning_identity();
757 let pool_fingerprint = active.static_pool_identity_fingerprint_ref();
758 if &ids.run_id != active.run_id()
759 || &ids.request_id != active.request_id()
760 || ids.resource_pool_id != active.static_pool_id()
761 || ids.resource_pool_identity_fingerprint.as_deref() != pool_fingerprint
762 || ids.provisioning_run_id.as_ref() != provisioning.map(ResourceTransactionIdentity::run_id)
763 || ids.provisioning_request_id.as_ref()
764 != provisioning.map(ResourceTransactionIdentity::request_id)
765 || ids.transaction_id.as_ref()
766 != provisioning.map(ResourceTransactionIdentity::transaction_id)
767 || ids.active_sequence_slot != Some(active.sequence_authority().sparse_id())
768 || ids.admission_generation != Some(active.sequence_authority().generation())
769 || ids.activation_epoch != Some(active.activation_epoch())
770 || ids.runtime_implementation_fingerprint.as_deref()
771 != Some(active.runtime_implementation_fingerprint())
772 || ids.active_sequence_fingerprint.as_deref() != Some(active.fingerprint())
773 {
774 return Err(invalid_event(
775 "event active identity differs from pool, epoch, runtime, or provisioning evidence",
776 ));
777 }
778 Ok(())
779}
780
781fn validate_completed_identity(
782 ids: &ExecutionIdentityParts,
783 completed: &TrustedCompletedSequenceBinding,
784 active: &TrustedActiveSequenceBinding,
785) -> Result<(), VNextError> {
786 if completed.active_sequence_fingerprint() != active.fingerprint()
787 || completed.plan() != active.plan()
788 || completed.coordinator_id() != active.coordinator_id()
789 || completed.sequence_authority() != active.sequence_authority()
790 || completed.run_id() != active.run_id()
791 || completed.request_id() != active.request_id()
792 || completed.activation_epoch() != active.activation_epoch()
793 || completed.runtime_implementation_fingerprint()
794 != active.runtime_implementation_fingerprint()
795 || ids.completed_sequence_fingerprint.as_deref() != Some(completed.fingerprint())
796 {
797 return Err(invalid_event(
798 "event completion identity differs from the synchronized active sequence receipt",
799 ));
800 }
801 Ok(())
802}
803
804fn validate_aborted_identity(
805 ids: &ExecutionIdentityParts,
806 aborted: &TrustedAbortedSequenceBinding,
807 active: &TrustedActiveSequenceBinding,
808) -> Result<(), VNextError> {
809 if !active.matches_abort_disposition(aborted.disposition())
810 || aborted.active_sequence_fingerprint() != active.fingerprint()
811 || aborted.plan() != active.plan()
812 || aborted.coordinator_id() != active.coordinator_id()
813 || aborted.sequence_authority() != active.sequence_authority()
814 || aborted.run_id() != active.run_id()
815 || aborted.request_id() != active.request_id()
816 || aborted.activation_epoch() != active.activation_epoch()
817 || aborted.runtime_implementation_fingerprint()
818 != active.runtime_implementation_fingerprint()
819 || ids.aborted_sequence_fingerprint.as_deref() != Some(aborted.fingerprint())
820 {
821 return Err(invalid_event(
822 "event abort identity differs from the poisoned active sequence receipt",
823 ));
824 }
825 Ok(())
826}
827
828fn validate_event_against_context(
829 event: &ExecutionEvent,
830 context: &TrustedExecutionEventContext<'_>,
831) -> Result<(), VNextError> {
832 let ids = event.identity.parts();
833 if &ids.run_id != context.run_id || &ids.request_id != context.request_id {
834 return Err(invalid_event(
835 "event identity differs from trusted run/request context",
836 ));
837 }
838 if let Some(topology) = context.topology {
839 if ids.plan_id.as_ref() != Some(topology.plan_id())
840 || ids.plan_hash.as_ref() != Some(topology.plan_hash())
841 || ids.device_id.as_ref() != Some(topology.device_id())
842 || ids.runtime_implementation_fingerprint.as_deref()
843 != Some(topology.device_runtime_implementation_fingerprint())
844 {
845 return Err(invalid_event(
846 "event plan identity differs from trusted topology",
847 ));
848 }
849 if let Some(node_id) = &ids.node_id {
850 let node = topology
851 .nodes
852 .get(node_id)
853 .ok_or_else(|| invalid_event("event node is absent from trusted topology"))?;
854 if ids.operation_id.as_ref() != Some(&node.operation_id)
855 || ids.provider_id.as_ref() != Some(&node.provider_id)
856 {
857 return Err(invalid_event(
858 "event operation/provider differs from trusted node topology",
859 ));
860 }
861 }
862 } else if ids.plan_id.is_some() {
863 return Err(invalid_event(
864 "plan-bound event lacks trusted topology context",
865 ));
866 }
867 match (has_active(ids), context.active) {
868 (true, Some(active)) => {
869 validate_active_identity(ids, active)?;
870 let topology = context
871 .topology
872 .ok_or_else(|| invalid_event("active event lacks trusted execution topology"))?;
873 if active.runtime_implementation_fingerprint()
874 != topology.device_runtime_implementation_fingerprint()
875 || active.plan().runtime_implementation_fingerprint()
876 != topology.device_runtime_implementation_fingerprint()
877 {
878 return Err(invalid_event(
879 "plan, admission, pool, and active runtime implementations differ",
880 ));
881 }
882 }
883 (false, None) => {}
884 _ => {
885 return Err(invalid_event(
886 "event active identity presence differs from external active evidence",
887 ));
888 }
889 }
890 match (has_completed(ids), context.completed, context.active) {
891 (true, Some(completed), Some(active)) => {
892 validate_completed_identity(ids, completed, active)?;
893 }
894 (false, None, _) => {}
895 _ => {
896 return Err(invalid_event(
897 "event completion identity presence differs from external synchronized receipt",
898 ));
899 }
900 }
901 match (has_aborted(ids), context.aborted, context.active) {
902 (true, Some(aborted), Some(active)) => {
903 validate_aborted_identity(ids, aborted, active)?;
904 }
905 (false, None, _) => {}
906 _ => {
907 return Err(invalid_event(
908 "event abort identity presence differs from external poison receipt",
909 ));
910 }
911 }
912 match (event.kind, context.submitted_operation) {
913 (ExecutionEventKind::OperationSubmitted, Some(submission))
914 if submission
915 .participants()
916 .iter()
917 .any(|participant| participant.identity() == event.identity()) => {}
918 (ExecutionEventKind::OperationSubmitted, _) => {
919 return Err(invalid_event(
920 "OperationSubmitted lacks its exact external dispatch receipt",
921 ));
922 }
923 (_, None) => {}
924 (_, Some(_)) => {
925 return Err(invalid_event(
926 "operation submission receipt supplied for a different event kind",
927 ));
928 }
929 }
930 match (event.kind, context.retired_operation) {
931 (ExecutionEventKind::NodeRetired, Some(completion))
932 if same_operation_authority_except_observation(
933 event.identity().parts(),
934 completion.submission().identity().parts(),
935 ) => {}
936 (ExecutionEventKind::NodeRetired, _) => {
937 return Err(invalid_event(
938 "NodeRetired lacks its exact participant completion projection",
939 ));
940 }
941 (_, None) => {}
942 (_, Some(_)) => {
943 return Err(invalid_event(
944 "operation completion projection supplied for a different event kind",
945 ));
946 }
947 }
948 match (&event.detail, context.expected_failure) {
949 (ExecutionEventDetail::Failure(failure), Some(expected)) if failure == expected => {}
950 (
951 ExecutionEventDetail::FailureTerminal {
952 first_failure_fingerprint,
953 },
954 Some(expected),
955 ) if first_failure_fingerprint == &expected.fingerprint() => {}
956 (ExecutionEventDetail::Failure(_) | ExecutionEventDetail::FailureTerminal { .. }, _) => {
957 return Err(invalid_event(
958 "event failure differs from independent first failure evidence",
959 ));
960 }
961 (_, None) => {}
962 (_, Some(_)) => {
963 return Err(invalid_event(
964 "trusted failure evidence supplied for a non-failure event",
965 ));
966 }
967 }
968 if let Some(recovery_identity) = context.unsubmitted_recovery_identity {
969 let ExecutionEventDetail::Failure(failure) = &event.detail else {
970 return Err(invalid_event(
971 "unsubmitted recovery identity was supplied for a non-failure event",
972 ));
973 };
974 if failure.identity() != recovery_identity {
975 return Err(invalid_event(
976 "unsubmitted recovery identity differs from the exact observed failure",
977 ));
978 }
979 }
980 Ok(())
981}
982
983fn validate_failure_identity(
984 domain: FailureDomain,
985 ids: &ExecutionIdentityParts,
986) -> Result<(), VNextError> {
987 let operation_shape = exact_plan(ids)
988 && ids.frame_id.is_some()
989 && ids.node_invocation_id.is_some()
990 && ids.node_id.is_some()
991 && ids.operation_id.is_some()
992 && ids.provider_id.is_some()
993 && has_active(ids)
994 && !has_completed(ids)
995 && !has_aborted(ids)
996 && no_resource_item(ids);
997 let resource_shape = exact_plan(ids)
998 && ids.frame_id.is_none()
999 && ids.node_invocation_id.is_none()
1000 && ids.node_id.is_none()
1001 && ids.operation_id.is_none()
1002 && ids.provider_id.is_none()
1003 && has_pool(ids)
1004 && !has_active(ids)
1005 && !has_completed(ids)
1006 && !has_aborted(ids)
1007 && (ids.resource_id.is_some() ^ ids.resource_batch_fingerprint.is_some());
1008 let active_operation_resource_shape = operation_shape && has_pool(ids);
1009 let plan_shape = exact_plan(ids)
1010 && ids.frame_id.is_none()
1011 && ids.node_id.is_none()
1012 && !has_pool(ids)
1013 && !has_completed(ids)
1014 && !has_aborted(ids)
1015 && no_resource_item(ids);
1016 let device_only = ids.device_id.is_some()
1017 && ids.plan_id.is_none()
1018 && ids.frame_id.is_none()
1019 && ids.node_id.is_none()
1020 && !has_pool(ids)
1021 && !has_completed(ids)
1022 && !has_aborted(ids)
1023 && no_resource_item(ids);
1024 let pre_plan = ids.plan_id.is_none()
1025 && ids.device_id.is_none()
1026 && ids.frame_id.is_none()
1027 && ids.node_id.is_none()
1028 && !has_pool(ids)
1029 && !has_completed(ids)
1030 && !has_aborted(ids)
1031 && no_resource_item(ids);
1032 let valid = match domain {
1033 FailureDomain::Operation => operation_shape,
1034 FailureDomain::Resource => resource_shape || active_operation_resource_shape,
1035 FailureDomain::Device => device_only || plan_shape || operation_shape || resource_shape,
1036 FailureDomain::Planning => plan_shape,
1037 FailureDomain::ModelResolution | FailureDomain::Product => pre_plan || plan_shape,
1038 FailureDomain::Event => pre_plan || plan_shape || operation_shape,
1039 };
1040 if !valid {
1041 return Err(invalid_event(format!(
1042 "failure domain `{domain:?}` has missing or extraneous execution identity fields"
1043 )));
1044 }
1045 Ok(())
1046}
1047
1048#[derive(Debug, Clone)]
1049struct ActiveNodeInvocation {
1050 invocation_id: NodeInvocationId,
1051 node_span: SpanId,
1052 operation_submitted: bool,
1053}
1054
1055#[derive(Debug, Clone)]
1056struct ActiveFrame {
1057 id: ExecutionFrameId,
1058 span_id: SpanId,
1059 active_nodes: BTreeMap<NodeId, ActiveNodeInvocation>,
1060 completed_nodes: BTreeSet<NodeId>,
1061}
1062
1063#[derive(Debug, Clone)]
1064pub struct ExecutionEventCursor {
1065 run_id: RunId,
1066 request_id: RequestIdentity,
1067 last_sequence: u64,
1068 last_timestamp: Option<MonotonicTimestamp>,
1069 last_phase: Option<ExecutionPhase>,
1070 topology_fingerprint: Option<String>,
1071 active_fingerprint: Option<String>,
1072 completion_fingerprint: Option<String>,
1073 abort_fingerprint: Option<String>,
1074 observed_failure: Option<IdentifiedFailure>,
1075 accepted: bool,
1076 planned: bool,
1077 terminal: bool,
1078 root_span: Option<SpanId>,
1079 seen_spans: BTreeSet<SpanId>,
1080 next_frame: u64,
1081 next_invocation: u64,
1082 completed_frames: u64,
1083 frame: Option<ActiveFrame>,
1084}
1085
1086impl ExecutionEventCursor {
1087 pub fn new(run_id: RunId, request_id: RequestIdentity) -> Self {
1088 Self {
1089 run_id,
1090 request_id,
1091 last_sequence: 0,
1092 last_timestamp: None,
1093 last_phase: None,
1094 topology_fingerprint: None,
1095 active_fingerprint: None,
1096 completion_fingerprint: None,
1097 abort_fingerprint: None,
1098 observed_failure: None,
1099 accepted: false,
1100 planned: false,
1101 terminal: false,
1102 root_span: None,
1103 seen_spans: BTreeSet::new(),
1104 next_frame: 1,
1105 next_invocation: 1,
1106 completed_frames: 0,
1107 frame: None,
1108 }
1109 }
1110
1111 pub fn observe_against(
1112 &mut self,
1113 event: &ExecutionEvent,
1114 context: &TrustedExecutionEventContext<'_>,
1115 ) -> Result<(), VNextError> {
1116 let mut next = self.clone();
1117 next.observe_candidate_in_place(event, context)?;
1118 *self = next;
1119 Ok(())
1120 }
1121
1122 pub(super) fn observe_candidate_in_place(
1127 &mut self,
1128 event: &ExecutionEvent,
1129 context: &TrustedExecutionEventContext<'_>,
1130 ) -> Result<(), VNextError> {
1131 self.observe_inner(event, context)
1132 }
1133
1134 pub const fn last_sequence(&self) -> u64 {
1135 self.last_sequence
1136 }
1137
1138 pub const fn is_terminal(&self) -> bool {
1139 self.terminal
1140 }
1141
1142 pub const fn completed_frames(&self) -> u64 {
1143 self.completed_frames
1144 }
1145
1146 fn observe_inner(
1147 &mut self,
1148 event: &ExecutionEvent,
1149 context: &TrustedExecutionEventContext<'_>,
1150 ) -> Result<(), VNextError> {
1151 validate_event_against_context(event, context)?;
1152 let ids = event.identity.parts();
1153 if ids.run_id != self.run_id
1154 || ids.request_id != self.request_id
1155 || ids.sequence != self.last_sequence.saturating_add(1)
1156 || self
1157 .last_timestamp
1158 .is_some_and(|timestamp| event.timestamp <= timestamp)
1159 || self
1160 .last_phase
1161 .is_some_and(|phase| event.phase.rank() < phase.rank())
1162 || self.terminal
1163 {
1164 return Err(invalid_event(
1165 "request journal run, request, sequence, timestamp, phase, or terminal boundary is invalid",
1166 ));
1167 }
1168 if let Some(topology) = context.topology {
1169 if let Some(bound) = &self.topology_fingerprint {
1170 if bound != topology.fingerprint() {
1171 return Err(invalid_event("request changed trusted topology"));
1172 }
1173 }
1174 }
1175 if let Some(active) = context.active {
1176 if let Some(bound) = &self.active_fingerprint {
1177 if bound != active.fingerprint() {
1178 return Err(invalid_event(
1179 "request changed active pool/slot/epoch/runtime binding",
1180 ));
1181 }
1182 }
1183 }
1184 if self.observed_failure.is_some()
1185 && !matches!(
1186 event.kind,
1187 ExecutionEventKind::SequenceCompleted
1188 | ExecutionEventKind::SequenceAborted
1189 | ExecutionEventKind::RequestFailed
1190 )
1191 {
1192 return Err(invalid_event(
1193 "only sequence disposition and terminal failure may follow FailureObserved",
1194 ));
1195 }
1196
1197 match event.kind {
1198 ExecutionEventKind::RequestAccepted => self.accept(ids)?,
1199 ExecutionEventKind::PlanBuilt => {
1200 let topology = context
1201 .topology
1202 .ok_or_else(|| invalid_event("PlanBuilt lacks trusted topology"))?;
1203 self.plan(ids, topology)?;
1204 }
1205 ExecutionEventKind::FrameStarted => {
1206 let topology = self.require_topology(context)?;
1207 let active = self.require_active(context)?;
1208 self.start_frame(ids, topology, active)?;
1209 }
1210 ExecutionEventKind::NodeStarted => {
1211 let topology = self.require_topology(context)?;
1212 self.require_active(context)?;
1213 self.start_node(ids, topology)?;
1214 }
1215 ExecutionEventKind::OperationSubmitted => self.submit_operation(ids)?,
1216 ExecutionEventKind::NodeRetired => self.retire_node(ids)?,
1217 ExecutionEventKind::FrameCompleted => {
1218 let topology = self.require_topology(context)?;
1219 self.require_active(context)?;
1220 self.complete_frame(ids, topology)?;
1221 }
1222 ExecutionEventKind::FailureObserved => {
1223 self.observe_failure(event, context.unsubmitted_recovery_identity)?
1224 }
1225 ExecutionEventKind::SequenceCompleted => {
1226 self.require_topology(context)?;
1227 self.require_active(context)?;
1228 let completed = context.completed.ok_or_else(|| {
1229 invalid_event("SequenceCompleted lacks synchronized completion evidence")
1230 })?;
1231 self.complete_sequence(ids, completed)?;
1232 }
1233 ExecutionEventKind::SequenceAborted => {
1234 self.require_topology(context)?;
1235 self.require_active(context)?;
1236 let aborted = context.aborted.ok_or_else(|| {
1237 invalid_event("SequenceAborted lacks poisoned abort evidence")
1238 })?;
1239 self.abort_sequence(ids, aborted)?;
1240 }
1241 ExecutionEventKind::RequestCompleted => {
1242 self.require_topology(context)?;
1243 self.require_active(context)?;
1244 if context.completed.is_none() {
1245 return Err(invalid_event(
1246 "RequestCompleted lacks synchronized completion evidence",
1247 ));
1248 }
1249 self.complete_success(ids)?;
1250 }
1251 ExecutionEventKind::RequestFailed => self.fail_request(event)?,
1252 }
1253 self.last_sequence = ids.sequence;
1254 self.last_timestamp = Some(event.timestamp);
1255 self.last_phase = Some(event.phase);
1256 Ok(())
1257 }
1258
1259 fn require_topology<'a>(
1260 &self,
1261 context: &'a TrustedExecutionEventContext<'_>,
1262 ) -> Result<&'a TrustedExecutionTopology, VNextError> {
1263 if !self.planned {
1264 return Err(invalid_event("execution event precedes PlanBuilt"));
1265 }
1266 context
1267 .topology
1268 .ok_or_else(|| invalid_event("execution event lacks trusted topology"))
1269 }
1270
1271 fn require_active<'a>(
1272 &self,
1273 context: &'a TrustedExecutionEventContext<'_>,
1274 ) -> Result<&'a TrustedActiveSequenceBinding, VNextError> {
1275 context
1276 .active
1277 .ok_or_else(|| invalid_event("execution event lacks active sequence evidence"))
1278 }
1279
1280 fn accept(&mut self, ids: &ExecutionIdentityParts) -> Result<(), VNextError> {
1281 if self.accepted || self.last_sequence != 0 || ids.parent_span_id.is_some() {
1282 return Err(invalid_event(
1283 "RequestAccepted must open the first root span",
1284 ));
1285 }
1286 self.seen_spans.insert(ids.span_id.clone());
1287 self.root_span = Some(ids.span_id.clone());
1288 self.accepted = true;
1289 Ok(())
1290 }
1291
1292 fn plan(
1293 &mut self,
1294 ids: &ExecutionIdentityParts,
1295 topology: &TrustedExecutionTopology,
1296 ) -> Result<(), VNextError> {
1297 if !self.accepted
1298 || self.planned
1299 || ids.parent_span_id.as_ref() != self.root_span.as_ref()
1300 || !self.seen_spans.insert(ids.span_id.clone())
1301 {
1302 return Err(invalid_event(
1303 "PlanBuilt must uniquely bind one topology under the request root",
1304 ));
1305 }
1306 self.topology_fingerprint = Some(topology.fingerprint().to_owned());
1307 self.planned = true;
1308 Ok(())
1309 }
1310
1311 fn start_frame(
1312 &mut self,
1313 ids: &ExecutionIdentityParts,
1314 _topology: &TrustedExecutionTopology,
1315 active: &TrustedActiveSequenceBinding,
1316 ) -> Result<(), VNextError> {
1317 let frame_id = ids.frame_id.expect("frame shape validated");
1318 if self.frame.is_some()
1319 || self.completion_fingerprint.is_some()
1320 || self.abort_fingerprint.is_some()
1321 || self.observed_failure.is_some()
1322 || frame_id.get() != self.next_frame
1323 || ids.parent_span_id.as_ref() != self.root_span.as_ref()
1324 || !self.seen_spans.insert(ids.span_id.clone())
1325 {
1326 return Err(invalid_event(
1327 "frames must start once in strict contiguous order under the request root",
1328 ));
1329 }
1330 self.active_fingerprint
1331 .get_or_insert_with(|| active.fingerprint().to_owned());
1332 self.frame = Some(ActiveFrame {
1333 id: frame_id,
1334 span_id: ids.span_id.clone(),
1335 active_nodes: BTreeMap::new(),
1336 completed_nodes: BTreeSet::new(),
1337 });
1338 Ok(())
1339 }
1340
1341 fn start_node(
1342 &mut self,
1343 ids: &ExecutionIdentityParts,
1344 topology: &TrustedExecutionTopology,
1345 ) -> Result<(), VNextError> {
1346 let node_id = ids.node_id.as_ref().expect("node shape validated");
1347 let invocation_id = ids
1348 .node_invocation_id
1349 .expect("node invocation shape validated");
1350 let frame = self
1351 .frame
1352 .as_mut()
1353 .ok_or_else(|| invalid_event("node started outside an active frame"))?;
1354 let node = topology
1355 .nodes
1356 .get(node_id)
1357 .ok_or_else(|| invalid_event("node is absent from trusted topology"))?;
1358 if ids.frame_id != Some(frame.id)
1359 || invocation_id.get() != self.next_invocation
1360 || frame.active_nodes.contains_key(node_id)
1361 || frame.completed_nodes.contains(node_id)
1362 || node
1363 .dependencies
1364 .iter()
1365 .any(|dependency| !frame.completed_nodes.contains(dependency))
1366 || ids.parent_span_id.as_ref() != Some(&frame.span_id)
1367 || !self.seen_spans.insert(ids.span_id.clone())
1368 {
1369 return Err(invalid_event(
1370 "node invocation is duplicate, non-monotonic, cross-frame, or precedes same-frame dependencies",
1371 ));
1372 }
1373 self.next_invocation = self
1374 .next_invocation
1375 .checked_add(1)
1376 .ok_or_else(|| invalid_event("node invocation id overflow"))?;
1377 frame.active_nodes.insert(
1378 node_id.clone(),
1379 ActiveNodeInvocation {
1380 invocation_id,
1381 node_span: ids.span_id.clone(),
1382 operation_submitted: false,
1383 },
1384 );
1385 Ok(())
1386 }
1387
1388 fn submit_operation(&mut self, ids: &ExecutionIdentityParts) -> Result<(), VNextError> {
1389 let node_id = ids.node_id.as_ref().expect("operation shape validated");
1390 let frame = self
1391 .frame
1392 .as_mut()
1393 .ok_or_else(|| invalid_event("operation submitted outside an active frame"))?;
1394 let active = frame
1395 .active_nodes
1396 .get_mut(node_id)
1397 .ok_or_else(|| invalid_event("operation submitted without active node"))?;
1398 if ids.frame_id != Some(frame.id)
1399 || ids.node_invocation_id != Some(active.invocation_id)
1400 || active.operation_submitted
1401 || ids.parent_span_id.as_ref() != Some(&active.node_span)
1402 || !self.seen_spans.insert(ids.span_id.clone())
1403 {
1404 return Err(invalid_event(
1405 "operation submission does not match the active node invocation",
1406 ));
1407 }
1408 active.operation_submitted = true;
1409 Ok(())
1410 }
1411
1412 fn retire_node(&mut self, ids: &ExecutionIdentityParts) -> Result<(), VNextError> {
1413 let node_id = ids.node_id.as_ref().expect("node shape validated");
1414 let frame = self
1415 .frame
1416 .as_mut()
1417 .ok_or_else(|| invalid_event("node completed outside an active frame"))?;
1418 let active = frame
1419 .active_nodes
1420 .get(node_id)
1421 .ok_or_else(|| invalid_event("node completed without active invocation"))?;
1422 if ids.frame_id != Some(frame.id)
1423 || ids.node_invocation_id != Some(active.invocation_id)
1424 || ids.span_id != active.node_span
1425 || ids.parent_span_id.as_ref() != Some(&frame.span_id)
1426 || !active.operation_submitted
1427 {
1428 return Err(invalid_event(
1429 "node completion requires its exact frame, invocation, span, and operation",
1430 ));
1431 }
1432 frame.active_nodes.remove(node_id);
1433 frame.completed_nodes.insert(node_id.clone());
1434 Ok(())
1435 }
1436
1437 fn complete_sequence(
1438 &mut self,
1439 ids: &ExecutionIdentityParts,
1440 completed: &TrustedCompletedSequenceBinding,
1441 ) -> Result<(), VNextError> {
1442 let failure_cleanup = self.observed_failure.is_some();
1443 if !self.planned
1444 || !failure_cleanup && (self.completed_frames == 0 || self.frame.is_some())
1445 || failure_cleanup && self.active_fingerprint.is_none()
1446 || self.active_fingerprint.as_deref() != Some(completed.active_sequence_fingerprint())
1447 || self.completion_fingerprint.is_some()
1448 || self.abort_fingerprint.is_some()
1449 || ids.parent_span_id.as_ref() != self.root_span.as_ref()
1450 || !self.seen_spans.insert(ids.span_id.clone())
1451 {
1452 return Err(invalid_event(
1453 "SequenceCompleted requires submitted frames and one unique synchronized receipt",
1454 ));
1455 }
1456 if failure_cleanup {
1457 self.frame = None;
1458 }
1459 self.completion_fingerprint = Some(completed.fingerprint().to_owned());
1460 Ok(())
1461 }
1462
1463 fn abort_sequence(
1464 &mut self,
1465 ids: &ExecutionIdentityParts,
1466 aborted: &TrustedAbortedSequenceBinding,
1467 ) -> Result<(), VNextError> {
1468 if self.observed_failure.is_none()
1469 || self.active_fingerprint.is_none()
1470 || self.active_fingerprint.as_deref() != Some(aborted.active_sequence_fingerprint())
1471 || self.completion_fingerprint.is_some()
1472 || self.abort_fingerprint.is_some()
1473 || ids.parent_span_id.as_ref() != self.root_span.as_ref()
1474 || !self.seen_spans.insert(ids.span_id.clone())
1475 {
1476 return Err(invalid_event(
1477 "SequenceAborted requires one observed failure and one unique poison receipt",
1478 ));
1479 }
1480 self.frame = None;
1481 self.abort_fingerprint = Some(aborted.fingerprint().to_owned());
1482 Ok(())
1483 }
1484
1485 fn complete_frame(
1486 &mut self,
1487 ids: &ExecutionIdentityParts,
1488 topology: &TrustedExecutionTopology,
1489 ) -> Result<(), VNextError> {
1490 let frame = self
1491 .frame
1492 .as_ref()
1493 .ok_or_else(|| invalid_event("FrameCompleted lacks an active frame"))?;
1494 if ids.frame_id != Some(frame.id)
1495 || ids.span_id != frame.span_id
1496 || ids.parent_span_id.as_ref() != self.root_span.as_ref()
1497 || !frame.active_nodes.is_empty()
1498 || frame.completed_nodes != topology.node_ids()
1499 {
1500 return Err(invalid_event(
1501 "frame completion requires every trusted node exactly once and no active invocation",
1502 ));
1503 }
1504 self.frame = None;
1505 self.completed_frames += 1;
1506 self.next_frame = self
1507 .next_frame
1508 .checked_add(1)
1509 .ok_or_else(|| invalid_event("frame id overflow"))?;
1510 Ok(())
1511 }
1512
1513 fn observe_failure(
1514 &mut self,
1515 event: &ExecutionEvent,
1516 unsubmitted_recovery_identity: Option<&ExecutionIdentityEnvelope>,
1517 ) -> Result<(), VNextError> {
1518 if !self.accepted || self.observed_failure.is_some() {
1519 return Err(invalid_event(
1520 "FailureObserved requires one accepted non-failed request",
1521 ));
1522 }
1523 let failure = match &event.detail {
1524 ExecutionEventDetail::Failure(failure) => failure,
1525 _ => return Err(invalid_event("FailureObserved lacks identified failure")),
1526 };
1527 let ids = event.identity.parts();
1528 if has_active(ids) {
1529 let failed_operation = failure.identity().parts();
1530 self.active_fingerprint
1531 .get_or_insert_with(|| ids.active_sequence_fingerprint.clone().unwrap());
1532 let frame = self.frame.as_ref().ok_or_else(|| {
1533 invalid_event("active operation failure lacks its execution frame")
1534 })?;
1535 let node_id = ids
1536 .node_id
1537 .as_ref()
1538 .ok_or_else(|| invalid_event("active operation failure lacks its node identity"))?;
1539 let invocation = frame.active_nodes.get(node_id).ok_or_else(|| {
1540 invalid_event("active operation failure lacks its node invocation")
1541 })?;
1542 let operation_span_was_submitted = self.seen_spans.contains(&failed_operation.span_id);
1543 let is_unsubmitted_recovery = unsubmitted_recovery_identity
1544 .is_some_and(|identity| identity == failure.identity());
1545 if ids.frame_id != Some(frame.id)
1546 || ids.node_invocation_id != Some(invocation.invocation_id)
1547 || !same_operation_authority_except_observation(ids, failed_operation)
1548 || failed_operation.parent_span_id.as_ref() != Some(&invocation.node_span)
1549 || ids.parent_span_id.as_ref() != Some(&failed_operation.span_id)
1550 || operation_span_was_submitted == is_unsubmitted_recovery
1551 {
1552 return Err(invalid_event(
1553 "operation failure does not link one exact submitted operation to its observation span",
1554 ));
1555 }
1556 } else if ids.parent_span_id.as_ref() != self.root_span.as_ref()
1557 && !(ids.span_id == *self.root_span.as_ref().expect("accepted root")
1558 && ids.parent_span_id.is_none())
1559 {
1560 return Err(invalid_event(
1561 "non-active failure must be anchored under the request root",
1562 ));
1563 }
1564 if !self.seen_spans.insert(ids.span_id.clone()) {
1565 return Err(invalid_event("FailureObserved span was already used"));
1566 }
1567 self.observed_failure = Some(failure.clone());
1568 Ok(())
1569 }
1570
1571 fn complete_success(&mut self, ids: &ExecutionIdentityParts) -> Result<(), VNextError> {
1572 if self.observed_failure.is_some()
1573 || !self.planned
1574 || self.completed_frames == 0
1575 || self.frame.is_some()
1576 || self.abort_fingerprint.is_some()
1577 || self.completion_fingerprint.as_deref()
1578 != ids.completed_sequence_fingerprint.as_deref()
1579 {
1580 return Err(invalid_event(
1581 "successful request requires submitted frames and the exact synchronized sequence receipt",
1582 ));
1583 }
1584 if ids.span_id != *self.root_span.as_ref().expect("accepted root")
1585 || ids.parent_span_id.is_some()
1586 {
1587 return Err(invalid_event(
1588 "terminal request event must close the exact request root",
1589 ));
1590 }
1591 self.terminal = true;
1592 Ok(())
1593 }
1594
1595 fn fail_request(&mut self, event: &ExecutionEvent) -> Result<(), VNextError> {
1596 let ids = event.identity.parts();
1597 if !self.accepted {
1598 if self.last_sequence != 0
1599 || ids.parent_span_id.is_some()
1600 || !matches!(event.detail, ExecutionEventDetail::Failure(_))
1601 {
1602 return Err(invalid_event(
1603 "only first-event pre-plan RequestFailed may precede acceptance",
1604 ));
1605 }
1606 self.terminal = true;
1607 return Ok(());
1608 }
1609 let observed = self
1610 .observed_failure
1611 .as_ref()
1612 .ok_or_else(|| invalid_event("RequestFailed lacks FailureObserved"))?;
1613 let terminal_fingerprint = match &event.detail {
1614 ExecutionEventDetail::FailureTerminal {
1615 first_failure_fingerprint,
1616 } => first_failure_fingerprint,
1617 _ => {
1618 return Err(invalid_event(
1619 "post-acceptance RequestFailed requires FailureTerminal",
1620 ));
1621 }
1622 };
1623 if terminal_fingerprint != &observed.fingerprint() {
1624 return Err(invalid_event(
1625 "RequestFailed does not reference the first observed failure",
1626 ));
1627 }
1628 if self.active_fingerprint.is_some() {
1629 let completed_matches = self.completion_fingerprint.is_some()
1630 && self.completion_fingerprint.as_deref()
1631 == ids.completed_sequence_fingerprint.as_deref();
1632 let aborted_matches = self.abort_fingerprint.is_some()
1633 && self.abort_fingerprint.as_deref() == ids.aborted_sequence_fingerprint.as_deref();
1634 if completed_matches == aborted_matches {
1635 return Err(invalid_event(
1636 "active RequestFailed requires exactly one matching completion or abort disposition",
1637 ));
1638 }
1639 if completed_matches && ids.aborted_sequence_fingerprint.is_some()
1640 || aborted_matches && ids.completed_sequence_fingerprint.is_some()
1641 {
1642 return Err(invalid_event(
1643 "active RequestFailed carries an unexpected opposite sequence disposition",
1644 ));
1645 }
1646 } else if self.completion_fingerprint.is_some()
1647 || self.abort_fingerprint.is_some()
1648 || has_completed(ids)
1649 || has_aborted(ids)
1650 {
1651 return Err(invalid_event(
1652 "non-active RequestFailed cannot carry sequence disposition",
1653 ));
1654 }
1655 if ids.span_id != *self.root_span.as_ref().expect("accepted root")
1656 || ids.parent_span_id.is_some()
1657 {
1658 return Err(invalid_event(
1659 "terminal request event must close the exact request root",
1660 ));
1661 }
1662 self.terminal = true;
1663 Ok(())
1664 }
1665}