pub struct WorkflowRun<CS: StateStoreBackend, A: AgentBackend, X: ActionBackend = NoActions, C: ConditionBackend = NoConditions, T: StateTyping = JsonState, H: LifecycleHooks = NoHooks, E: ChildExecutor = InlineExecutor> { /* private fields */ }Expand description
A durable workflow-run handle.
This handle observes durable state and requests explicit controls. Dropping it
or a wait/snapshot future does not cancel accepted work. Self::execution
borrows it exclusively and returns an owner whose drop does revoke uncommitted
advancement. Use Self::cancel for durable, monotonic cancellation.
§Example
let terminal = run.execution(ExecutionOptions::default()).await?.finish().await?;
assert!(matches!(terminal.status, RunStatus::Succeeded));
assert_eq!(run.snapshot().await?.state.values()["count"], 1);Implementations§
Source§impl<CS: StateStoreBackend, A: AgentBackend, X: ActionBackend, C: ConditionBackend, T: StateTyping, H: LifecycleHooks, E: ChildExecutor> WorkflowRun<CS, A, X, C, T, H, E>
impl<CS: StateStoreBackend, A: AgentBackend, X: ActionBackend, C: ConditionBackend, T: StateTyping, H: LifecycleHooks, E: ChildExecutor> WorkflowRun<CS, A, X, C, T, H, E>
Sourcepub fn id(&self) -> &RunId
pub fn id(&self) -> &RunId
Examples found in repository?
examples/history_and_recovery.rs (line 100)
63 async fn create_fork(
64 &self,
65 app: &App<StateStore, Agent>,
66 ) -> Result<(WorkspaceId, RunSelector, RunSelector), Box<dyn Error + Send + Sync>> {
67 let workspace = app
68 .create_workspace(WorkspaceConfig {
69 name: Some("history example".into()),
70 root: std::env::current_dir()?,
71 })
72 .await?;
73 let workflow = workspace.workflow("history-counter".parse()?).await?;
74 let mut source = workflow.prepare(WorkflowState::default()).await?;
75 let completed = source
76 .execution(ExecutionOptions::default())
77 .await?
78 .finish()
79 .await?;
80 if !matches!(completed.status, RunStatus::Succeeded)
81 || completed.state.values()["count"] != 2
82 {
83 return Err(
84 AppError::Integration("source did not complete both increments".into()).into(),
85 );
86 }
87
88 // An action ordinal selects the whole committed boundary containing it;
89 // it is independent of the checkpoint sequence and storage revision.
90 let first = source.checkpoint(CheckpointSelector::Action(0)).await?;
91 if !matches!(first.kind, CheckpointKind::Action)
92 || first.position.as_deref() != Some("first")
93 || first.state.values()["count"] != 1
94 {
95 return Err(
96 AppError::Integration("first action history is inconsistent".into()).into(),
97 );
98 }
99 let source_selector = RunSelector::Exact {
100 id: source.id().clone(),
101 occurrence: first.occurrence.clone(),
102 };
103 let mut fork = workflow
104 .restore(RestoreOptions {
105 source: source_selector.clone(),
106 checkpoint: CheckpointSelector::Action(0),
107 overrides: WorkflowState::new(Map::from_iter([("count".into(), json!(40))]))?,
108 position: ResumePosition::AfterCompleted,
109 fallback: None,
110 })
111 .await?;
112 let parked = fork.snapshot().await?;
113 if !matches!(parked.status, RunStatus::Paused)
114 || parked.checkpoint != 0
115 || parked.state.values()["count"] != 40
116 || !parked.outcomes.is_empty()
117 {
118 return Err(
119 AppError::Integration("restore did not create a parked clean copy".into()).into(),
120 );
121 }
122
123 source.delete(completed.checkpoint).await?;
124 if !matches!(source.snapshot().await, Err(AppError::NotFound(_))) {
125 return Err(
126 AppError::Integration("explicit source deletion did not finish".into()).into(),
127 );
128 }
129 // Deletion does not invalidate the copied state. AfterCompleted skips
130 // the first increment, so only the second changes the override: 40 -> 41.
131 let completed = fork
132 .execution(ExecutionOptions::default())
133 .await?
134 .finish()
135 .await?;
136 if !matches!(completed.status, RunStatus::Succeeded)
137 || completed.state.values()["count"] != 41
138 || completed.last_action_ordinal != Some(1)
139 {
140 return Err(
141 AppError::Integration("fork resumed from the wrong boundary".into()).into(),
142 );
143 }
144 println!("Source count: 2; copied count: 40; completed fork count: 41");
145 Ok((
146 workspace.id().clone(),
147 source_selector,
148 RunSelector::Exact {
149 id: fork.id().clone(),
150 occurrence: completed.occurrence,
151 },
152 ))
153 }
154
155 async fn read_reopened(
156 &self,
157 app: &App<StateStore, Agent>,
158 workspace: WorkspaceId,
159 source: RunSelector,
160 fork: RunSelector,
161 ) -> Result<(), AppError> {
162 let workspace = app.workspace(workspace).await?;
163 let workflow = workspace.workflow("history-counter".parse()?).await?;
164 let fork = workflow.open_run(fork).await?;
165 let committed = fork.checkpoint(CheckpointSelector::Action(1)).await?;
166 let lineage = committed
167 .lineage
168 .as_ref()
169 .ok_or_else(|| AppError::Integration("reopened fork lost its source lineage".into()))?;
170 if committed.state.values()["count"] != 41
171 || !matches!(&source, RunSelector::Exact { id, occurrence }
172 if &lineage.source == id && &lineage.source_occurrence == occurrence)
173 || lineage.action_ordinal != Some(0)
174 {
175 return Err(AppError::Integration(
176 "reopened fork history changed".into(),
177 ));
178 }
179 println!(
180 "Reopened fork {} at count 41; source lineage remains readable",
181 fork.id()
182 );
183 Ok(())
184 }More examples
examples/performance.rs (line 299)
265 async fn history(
266 &self,
267 run: &WorkflowRun<StateStore, Agent, WorkloadAction>,
268 actions: u64,
269 linear: bool,
270 terminal_sequence: u64,
271 ) -> Result<Value, AppError> {
272 let ceiling = actions * 16 + 16;
273 let mut cursor = None;
274 let mut prior_sequence = None;
275 let mut rows = 0_u64;
276 let mut observed_actions = 0_u64;
277 let mut initial = 0;
278 let mut terminal = 0;
279 let mut controls = 0;
280 let mut halts = 0;
281 let mut samples = Vec::new();
282 let started = Instant::now();
283 loop {
284 assert!(
285 samples.len() < ceiling.div_ceil(PAGE_SIZE as u64) as usize + 1,
286 "history page ceiling exhausted without cursor completion"
287 );
288 let page_started = Instant::now();
289 let page = run
290 .history(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
291 .await?;
292 samples.push(page_started.elapsed());
293 assert!(page.checkpoints.len() <= PAGE_SIZE);
294 assert!(
295 !page.checkpoints.is_empty() || page.next.is_none(),
296 "history cursor made no progress"
297 );
298 for row in &page.checkpoints {
299 assert_eq!(&row.run, run.id());
300 assert!(prior_sequence.is_none_or(|prior| row.sequence > prior));
301 prior_sequence = Some(row.sequence);
302 rows += 1;
303 assert!(
304 rows <= ceiling,
305 "declared retained-row workload ceiling exceeded"
306 );
307 match row.kind {
308 CheckpointKind::Initial => {
309 initial += 1;
310 assert_eq!(observed_actions, 0);
311 assert_eq!(row.sequence, 0);
312 }
313 CheckpointKind::Action => {
314 assert_eq!(row.outcomes.len(), 1);
315 let outcome = &row.outcomes[0];
316 assert_eq!(outcome.ordinal, observed_actions);
317 assert_eq!(
318 outcome.node,
319 if linear {
320 format!("n{observed_actions:04}")
321 } else {
322 "count".into()
323 }
324 );
325 assert!(outcome.result.is_empty());
326 observed_actions += 1;
327 assert_eq!(row.action_ordinal, Some(observed_actions - 1));
328 }
329 CheckpointKind::Terminal => {
330 terminal += 1;
331 assert!(matches!(row.status, RunStatus::Succeeded));
332 }
333 CheckpointKind::Control => {
334 assert!(row.outcomes.is_empty());
335 controls += 1;
336 if let RunStatus::Halted { node, result } = &row.status {
337 assert!(!linear);
338 assert_eq!(row.sequence, 4);
339 assert_eq!(node, "count");
340 assert_eq!(result, &Some(Map::new()));
341 halts += 1;
342 }
343 }
344 CheckpointKind::Failed { .. } => {
345 panic!("workload history contains a failed action")
346 }
347 }
348 assert_eq!(
349 row.state.values(),
350 &Map::from_iter([("count".into(), json!(observed_actions))])
351 );
352 if !linear && matches!(row.sequence, 4 | 5) {
353 assert!(matches!(row.kind, CheckpointKind::Control));
354 assert_eq!(observed_actions, 1);
355 assert_eq!(row.action_ordinal, Some(0));
356 assert_eq!(row.position.as_deref(), Some("count"));
357 if row.sequence == 5 {
358 assert!(matches!(row.status, RunStatus::Paused));
359 }
360 }
361 }
362 cursor = page.next;
363 // Each complete Checkpoint/WorkflowState is released with this page.
364 if cursor.is_none() {
365 break;
366 }
367 }
368 assert_eq!(initial, 1);
369 assert_eq!(terminal, 1);
370 assert_eq!(observed_actions, actions);
371 assert_eq!(prior_sequence, Some(terminal_sequence));
372 if !linear {
373 assert_eq!(rows, actions * 3 + 4);
374 assert_eq!(terminal_sequence, rows - 1);
375 assert_eq!(controls, actions * 2 + 2);
376 assert_eq!(halts, 1);
377 assert_eq!(samples.len(), (rows as usize).div_ceil(PAGE_SIZE));
378 }
379 let elapsed = started.elapsed();
380 let first = samples.first().copied().unwrap_or_default();
381 let subsequent: Duration = samples.iter().skip(1).copied().sum();
382 Ok(
383 json!({"rows":rows,"action_rows":observed_actions,"pages":samples.len(),
384 "elapsed_ns":elapsed.as_nanos(),"first_page_ns":first.as_nanos(),
385 "subsequent_pages_total_ns":subsequent.as_nanos(),
386 "rows_per_second":rows as f64 / elapsed.as_secs_f64(),"page_latency":Self::samples(samples)}),
387 )
388 }
389
390 async fn seed(
391 &self,
392 app: &App<StateStore, Agent, WorkloadAction>,
393 actions: u64,
394 linear: bool,
395 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
396 let workspace = app
397 .create_workspace(WorkspaceConfig {
398 name: Some("performance consumer".into()),
399 root: self.root.join("workspace"),
400 })
401 .await?;
402 let workflow = workspace.workflow("performance".parse()?).await?;
403 let mut run = workflow.prepare(WorkflowState::default()).await?;
404 let mut execution = run
405 .execution(ExecutionOptions {
406 halt_after: if linear {
407 Vec::new()
408 } else {
409 vec!["count".into()]
410 },
411 ..ExecutionOptions::default()
412 })
413 .await?;
414 let mut samples = Vec::with_capacity(actions as usize);
415 let started = Instant::now();
416 for ordinal in 0..actions {
417 let step_started = Instant::now();
418 let event = execution.step().await?;
419 samples.push(step_started.elapsed());
420 let Some(ExecutionEvent::Committed { outcomes, .. }) = event else {
421 panic!("each serial workload step must commit one action");
422 };
423 assert_eq!(outcomes.len(), 1);
424 assert_eq!(outcomes[0].ordinal, ordinal);
425 assert_eq!(
426 outcomes[0].node,
427 if linear {
428 format!("n{ordinal:04}")
429 } else {
430 "count".into()
431 }
432 );
433 assert!(outcomes[0].result.is_empty());
434 if !linear && ordinal == 0 {
435 // One public halt and explicit resume add two control rows to
436 // Initial + three checkpoints per action + Terminal.
437 let halted = execution.finish().await?;
438 assert!(matches!(&halted.status, RunStatus::Halted { node, result }
439 if node == "count" && result == &Some(Map::new())));
440 assert_eq!(halted.checkpoint, 4);
441 assert_eq!(
442 halted.state.values(),
443 &Map::from_iter([("count".into(), json!(1))])
444 );
445 assert_eq!(halted.last_action_ordinal, Some(0));
446 assert!(halted.outcomes == outcomes);
447 assert!(halted.interactions.is_empty());
448 execution = run.execution(ExecutionOptions::default()).await?;
449 }
450 }
451 let snapshot = execution.finish().await?;
452 Self::verify_snapshot(&snapshot, actions);
453 let execution_elapsed = started.elapsed();
454 let history = self
455 .history(&run, actions, linear, snapshot.checkpoint)
456 .await?;
457 if let Command::SeedHistory { rows } = self.command {
458 assert_eq!(history["rows"], rows);
459 }
460 Ok(
461 json!({"dataset":if linear {"linear"} else {"history"},"actions":actions,
462 "workspace":workspace.id().to_string(),"run":run.id().to_string(),
463 "occurrence":snapshot.occurrence,
464 "expected_state":snapshot.state.values(),"last_action_ordinal":snapshot.last_action_ordinal,
465 "terminal_checkpoint":snapshot.checkpoint,"retained_rows":history["rows"],
466 "execution_ns":execution_elapsed.as_nanos(),
467 "actions_per_second":actions as f64 / execution_elapsed.as_secs_f64(),
468 "step_latency":Self::samples(samples),"history":history}),
469 )
470 }
471
472 async fn reopen(
473 &self,
474 app: &App<StateStore, Agent, WorkloadAction>,
475 manifest: &Value,
476 actions: u64,
477 open_only: bool,
478 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
479 let started = Instant::now();
480 let workspace: WorkspaceId = manifest["workspace"]
481 .as_str()
482 .ok_or("missing workspace ID")?
483 .parse()?;
484 let run_id: RunId = manifest["run"].as_str().ok_or("missing run ID")?.parse()?;
485 let occurrence = manifest["occurrence"]
486 .as_str()
487 .ok_or("missing run occurrence")?
488 .to_owned();
489 let workspace = app.workspace(workspace).await?;
490 let workflow = workspace.workflow("performance".parse()?).await?;
491 let run = workflow
492 .open_run(RunSelector::Exact {
493 id: run_id,
494 occurrence: occurrence.clone(),
495 })
496 .await?;
497 let snapshot = run.snapshot().await?;
498 Self::verify_snapshot(&snapshot, actions);
499 assert_eq!(snapshot.occurrence, occurrence);
500 assert_eq!(json!(snapshot.checkpoint), manifest["terminal_checkpoint"]);
501 assert_eq!(json!(snapshot.state.values()), manifest["expected_state"]);
502 assert_eq!(
503 json!(snapshot.last_action_ordinal),
504 manifest["last_action_ordinal"]
505 );
506 let lookup = started.elapsed();
507 if open_only {
508 return Ok(
509 json!({"dataset":"history","actions":actions,"mode":"open-only",
510 "lookup_and_snapshot_ns":lookup.as_nanos(),
511 "seed_retained_rows":manifest["retained_rows"],"observed_history_rows":null}),
512 );
513 }
514 let history = self
515 .history(&run, actions, false, snapshot.checkpoint)
516 .await?;
517 assert_eq!(history["rows"], manifest["retained_rows"]);
518 Ok(
519 json!({"dataset":"history","actions":actions,"mode":"full-history",
520 "lookup_and_snapshot_ns":lookup.as_nanos(),"history":history}),
521 )
522 }
523
524 fn definitions(&self, actions: u64) -> Result<(Vec<WorkflowDefinition>, usize), AppError> {
525 if matches!(
526 self.command,
527 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. }
528 ) {
529 let (definition, bytes) =
530 Self::definition(matches!(self.command, Command::Linear), actions)?;
531 return Ok((vec![definition], bytes));
532 }
533 let leaf = json!({
534 "schema":"app-kit.workflow/v2","id":"performance-leaf","revision":"r1",
535 "state":{"value":0,"result":0},"required":["value","result"],
536 "reducers":{"value":"reject_concurrent","result":"reject_concurrent"},
537 "entry":"work","nodes":[{"id":"work","deadline_ms":NODE_DEADLINE_MS,
538 "step":{"kind":"action","action":"performance-action","revision":"r1"}}],"routes":{}
539 });
540 let documents = match self.command {
541 Command::Ready => {
542 let mut nodes = vec![json!({"id":"fork","step":{"kind":"state","mutations":[]}})];
543 let mut branches = Vec::new();
544 for index in 0..32 {
545 let node = format!("n{index:02}");
546 nodes.push(json!({"id":node,"deadline_ms":NODE_DEADLINE_MS,
547 "step":{"kind":"action","action":"performance-action","revision":"r1"}}));
548 branches.push(json!({"target":{"kind":"node","id":node}}));
549 }
550 vec![json!({
551 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
552 "state":{"value":0,"result":0},"required":["value","result"],
553 "reducers":{"value":"reject_concurrent","result":"collect"},
554 "entry":"fork","nodes":nodes,
555 "routes":{"fork":{"mode":"broadcast","branches":branches}}
556 })]
557 }
558 Command::Map | Command::NestedCancel => {
559 let nested = matches!(self.command, Command::NestedCancel);
560 let items = if nested { 2 } else { 4_096 };
561 let mut documents = vec![
562 json!({
563 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
564 "state":{"value":0,"result":0},"required":["value","result"],
565 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
566 "entry":"map","nodes":[{"id":"map","deadline_ms":if nested {NODE_DEADLINE_MS} else {WORKLOAD_BUDGET.as_millis() as u64},
567 "step":{"kind":"map","workflow":if nested {"performance-mid"} else {"performance-leaf"},
568 "items":{"kind":"literal","value":(0..items).collect::<Vec<_>>()},
569 "input_key":"value","max_parallel":if nested {2} else {8},
570 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
571 }),
572 leaf,
573 ];
574 if nested {
575 documents.push(json!({
576 "schema":"app-kit.workflow/v2","id":"performance-mid","revision":"r1",
577 "state":{"value":0,"result":0},"required":["value","result"],
578 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
579 "entry":"map","nodes":[{"id":"map","deadline_ms":NODE_DEADLINE_MS,
580 "step":{"kind":"map","workflow":"performance-leaf",
581 "items":{"kind":"literal","value":(0..8).collect::<Vec<_>>()},
582 "input_key":"value","max_parallel":8,
583 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
584 }));
585 }
586 documents
587 }
588 Command::Repeated | Command::DelayedObserver => vec![leaf],
589 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. } => {
590 unreachable!()
591 }
592 };
593 let mut bytes = 0;
594 let mut definitions = Vec::with_capacity(documents.len());
595 for document in documents {
596 let source = document.to_string();
597 assert!(source.len() <= 1_048_576);
598 bytes += source.len();
599 definitions.push(WorkflowDefinition::parse(&source)?);
600 }
601 Ok((definitions, bytes))
602 }
603
604 async fn ready(
605 &self,
606 app: &App<StateStore, Agent, WorkloadAction>,
607 probe: &WorkloadProbe,
608 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
609 let workspace = app
610 .create_workspace(WorkspaceConfig {
611 name: None,
612 root: self.root.join("workspace"),
613 })
614 .await?;
615 let workflow = workspace.workflow("performance".parse()?).await?;
616 let mut waves = Vec::with_capacity(32);
617 let mut releases = Vec::with_capacity(32 * 31);
618 let mut barriers = Vec::with_capacity(32);
619 let started = Instant::now();
620 for permutation in 0..32 {
621 probe.assert_idle();
622 let mut run = workflow.prepare(WorkflowState::default()).await?;
623 let observer = workflow
624 .open_run(RunSelector::Latest(run.id().clone()))
625 .await?;
626 let mut execution = run.execution(ExecutionOptions::default()).await?;
627 let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
628 panic!("the native fork must commit before the ready wave");
629 };
630 assert_eq!(outcomes.len(), 1);
631 assert_eq!(
632 (outcomes[0].node.as_str(), outcomes[0].ordinal),
633 ("fork", 0)
634 );
635 let wave = Instant::now();
636 let mut finish = Box::pin(execution.finish());
637 tokio::select! {
638 _ = probe.wait_for(&probe.constructed, (permutation + 1) * 32) => {}
639 result = &mut finish => panic!("gated ready wave finished before admission: {}", result.is_ok()),
640 }
641 assert_eq!(probe.active.load(Ordering::SeqCst), 32);
642 let order = (0..32)
643 .map(|offset| (permutation + 31 - offset) % 32)
644 .collect::<Vec<_>>();
645 for (released, index) in order.iter().take(31).enumerate() {
646 let releasing = Instant::now();
647 probe.release(*index);
648 tokio::select! {
649 _ = probe.wait_for(&probe.completed, permutation * 32 + released + 1) => {}
650 result = &mut finish => panic!("a partial ready wave committed: {}", result.is_ok()),
651 }
652 releases.push(releasing.elapsed());
653 let visible = observer.snapshot().await?;
654 assert_eq!(
655 visible.state.values()["result"],
656 0,
657 "staged siblings cannot expose a partial barrier"
658 );
659 assert_eq!(visible.last_action_ordinal, Some(0));
660 }
661 let barrier = Instant::now();
662 probe.release(order[31]);
663 let snapshot = finish.await?;
664 barriers.push(barrier.elapsed());
665 waves.push(wave.elapsed());
666 assert!(matches!(snapshot.status, RunStatus::Succeeded));
667 assert!(snapshot.interactions.is_empty());
668 assert_eq!(
669 snapshot.state.values()["result"],
670 json!((1..=32).collect::<Vec<_>>())
671 );
672 assert_eq!(snapshot.last_action_ordinal, Some(32));
673 assert_eq!(snapshot.outcomes.len(), 32);
674 for (index, outcome) in snapshot.outcomes.iter().enumerate() {
675 assert_eq!(outcome.node, format!("n{index:02}"));
676 assert_eq!(outcome.ordinal, index as u64 + 1);
677 assert_eq!(
678 outcome.result,
679 Map::from_iter([("value".into(), json!(index + 1))])
680 );
681 }
682 assert_eq!(std::mem::take(&mut *probe.order.lock().unwrap()), order);
683 assert_eq!(
684 probe.completed.load(Ordering::SeqCst),
685 (permutation + 1) * 32
686 );
687 probe.assert_idle();
688 }
689 assert_eq!(probe.peak.load(Ordering::SeqCst), 32);
690 let elapsed = started.elapsed();
691 Ok(
692 json!({"dataset":"ready32","permutations":32,"ready_actions_per_wave":32,
693 "caller_action_commits":1024,"native_fork_commits":32,
694 "execution_ns":elapsed.as_nanos(),"caller_actions_per_second":1024.0 / elapsed.as_secs_f64(),
695 "wave_latency_including_snapshot_probes":Self::samples(waves),
696 "release_to_source_completion_latency":Self::samples(releases),
697 "last_release_to_committed_barrier_latency":Self::samples(barriers)}),
698 )
699 }
700
701 async fn mapped(
702 &self,
703 app: &App<StateStore, Agent, WorkloadAction>,
704 probe: &WorkloadProbe,
705 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
706 let workspace = app
707 .create_workspace(WorkspaceConfig {
708 name: None,
709 root: self.root.join("workspace"),
710 })
711 .await?;
712 let workflow = workspace.workflow("performance".parse()?).await?;
713 let mut run = workflow.prepare(WorkflowState::default()).await?;
714 let observer = workflow
715 .open_run(RunSelector::Latest(run.id().clone()))
716 .await?;
717 let started = Instant::now();
718 let mut finish = Box::pin(run.execution(ExecutionOptions::default()).await?.finish());
719 tokio::select! {
720 _ = probe.wait_for(&probe.constructed, 8) => {}
721 result = &mut finish => panic!("map finished before its first eight blocked sources: {}", result.is_ok()),
722 }
723 assert_eq!(probe.active.load(Ordering::SeqCst), 8);
724 assert_eq!(probe.constructed.load(Ordering::SeqCst), 8);
725 let unchanged = observer.snapshot().await?;
726 assert_eq!(unchanged.state.values()["result"], 0);
727 assert!(unchanged.last_action_ordinal.is_none());
728 probe.release_all();
729 let snapshot = finish.await?;
730 let elapsed = started.elapsed();
731 assert!(matches!(snapshot.status, RunStatus::Succeeded));
732 assert!(snapshot.interactions.is_empty());
733 assert_eq!(snapshot.state.values()["result"], 4_096_u64 * 4_097 / 2);
734 assert_eq!(snapshot.last_action_ordinal, Some(0));
735 assert_eq!(snapshot.outcomes.len(), 1);
736 assert_eq!(snapshot.outcomes[0].node, "map");
737 assert_eq!(probe.constructed.load(Ordering::SeqCst), 4_096);
738 assert_eq!(probe.completed.load(Ordering::SeqCst), 4_096);
739 assert_eq!(probe.peak.load(Ordering::SeqCst), 8);
740 probe.assert_idle();
741 let child = workspace.workflow("performance-leaf".parse()?).await?;
742 let mut cursor = None;
743 let mut values = BTreeSet::new();
744 let verifying = Instant::now();
745 loop {
746 let page = child
747 .list_runs(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
748 .await?;
749 assert!(page.len() <= PAGE_SIZE);
750 if page.is_empty() {
751 break;
752 }
753 for row in &page {
754 assert!(matches!(row.status, RunStatus::Succeeded));
755 let value = row.state.values()["value"].as_u64().unwrap();
756 assert!(
757 value < 4_096 && values.insert(value),
758 "every mapped input has exactly one retained child"
759 );
760 assert_eq!(row.state.values()["result"], value + 1);
761 assert_eq!(row.last_action_ordinal, Some(0));
762 assert_eq!(row.outcomes.len(), 1);
763 assert_eq!(row.outcomes[0].node, "work");
764 }
765 let last = page.last().unwrap();
766 cursor = Some(RunSelector::Exact {
767 id: last.id.clone(),
768 occurrence: last.occurrence.clone(),
769 });
770 assert!(values.len() <= 4_096);
771 }
772 assert_eq!(values.len(), 4_096);
773 Ok(json!({"dataset":"map4096","items":4096,"max_parallel":8,
774 "parent_node_deadline_ms":WORKLOAD_BUDGET.as_millis(),
775 "execution_ns":elapsed.as_nanos(),"child_actions_per_second":4096.0 / elapsed.as_secs_f64(),
776 "map_completion_latency":Self::samples(vec![elapsed]),
777 "child_inventory_verification_ns":verifying.elapsed().as_nanos(),"verified_children":values.len()}))
778 }
779
780 async fn repeated(
781 &self,
782 app: &App<StateStore, Agent, WorkloadAction>,
783 probe: &WorkloadProbe,
784 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
785 let workspace = app
786 .create_workspace(WorkspaceConfig {
787 name: None,
788 root: self.root.join("workspace"),
789 })
790 .await?;
791 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
792 let started = Instant::now();
793 // Admit one run at a time, then directly poll at most 16 original
794 // executions together. No consumer task is spawned or detached.
795 let mut runs = std::pin::pin!(
796 stream::iter(0..100_u64)
797 .then(|value| {
798 let workflow = &workflow;
799 async move {
800 let started = Instant::now();
801 let run = workflow
802 .prepare(WorkflowState::new(Map::from_iter([(
803 "value".into(),
804 json!(value),
805 )]))?)
806 .await?;
807 Ok::<_, AppError>((value, run, started))
808 }
809 })
810 .map(|prepared| async move {
811 let (value, mut run, started) = prepared?;
812 let snapshot = run
813 .execution(ExecutionOptions::default())
814 .await?
815 .finish()
816 .await?;
817 assert!(matches!(snapshot.status, RunStatus::Succeeded));
818 assert!(snapshot.interactions.is_empty());
819 assert_eq!(snapshot.state.values()["value"], value);
820 assert_eq!(snapshot.state.values()["result"], value + 1);
821 assert_eq!(snapshot.last_action_ordinal, Some(0));
822 assert_eq!(snapshot.outcomes.len(), 1);
823 Ok::<_, AppError>((run.id().clone(), started.elapsed()))
824 })
825 .buffer_unordered(16)
826 );
827 tokio::select! {
828 _ = probe.wait_for(&probe.constructed, 16) => {}
829 result = runs.next() => panic!(
830 "run finished before the first 16 blocked sources: {:?}",
831 result.map(|result| result.map(|_| ())),
832 ),
833 }
834 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
835 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
836 probe.release_all();
837 let mut ids = BTreeSet::new();
838 let mut samples = Vec::with_capacity(100);
839 while let Some(result) = runs.next().await {
840 let (id, elapsed) = result?;
841 assert!(ids.insert(id));
842 samples.push(elapsed);
843 }
844 assert_eq!(ids.len(), 100);
845 assert_eq!(probe.constructed.load(Ordering::SeqCst), 100);
846 assert_eq!(probe.completed.load(Ordering::SeqCst), 100);
847 assert_eq!(probe.peak.load(Ordering::SeqCst), 16);
848 probe.assert_idle();
849 let rows = workflow
850 .list_runs(None, NonZeroUsize::new(128).unwrap())
851 .await?;
852 assert_eq!(rows.len(), 100);
853 assert_eq!(
854 rows.iter()
855 .map(|row| row.id.clone())
856 .collect::<BTreeSet<_>>(),
857 ids
858 );
859 assert!(
860 rows.iter()
861 .all(|row| matches!(row.status, RunStatus::Succeeded))
862 );
863 let elapsed = started.elapsed();
864 Ok(
865 json!({"dataset":"repeated100","runs":100,"caller_in_flight_limit":16,"admission":"sequential",
866 "production_active_limit":"unchanged; this measures only the caller's 16-journey envelope",
867 "execution_ns":elapsed.as_nanos(),"runs_per_second":100.0 / elapsed.as_secs_f64(),
868 "run_latency_including_admission":Self::samples(samples),"indefinite_stability_proof":false}),
869 )
870 }
871
872 async fn nested_cancel(
873 &self,
874 app: &App<StateStore, Agent, WorkloadAction>,
875 probe: &WorkloadProbe,
876 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
877 let workspace = app
878 .create_workspace(WorkspaceConfig {
879 name: None,
880 root: self.root.join("workspace"),
881 })
882 .await?;
883 let workflow = workspace.workflow("performance".parse()?).await?;
884 let run = workflow.start(ExecutionOptions::default()).await?;
885 tokio::select! {
886 _ = probe.wait_for(&probe.constructed, 16) => {}
887 result = run.wait() => panic!("nested run ended before its held leaf sources: {}", result.is_ok()),
888 }
889 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
890 let cancelling = Instant::now();
891 let snapshot = tokio::time::timeout(Duration::from_secs(30), async {
892 loop {
893 let before = run.snapshot().await?;
894 assert_eq!(before.state.values()["result"], 0);
895 assert!(before.outcomes.is_empty() && before.last_action_ordinal.is_none());
896 match run.cancel(before.checkpoint).await {
897 Ok(_) => break,
898 Err(AppError::Conflict(_)) => tokio::task::yield_now().await,
899 Err(error) => return Err(error),
900 }
901 }
902 run.wait().await
903 })
904 .await
905 .map_err(|_| {
906 AppError::Timeout("nested cancellation exceeded its declared 30-second grace".into())
907 })??;
908 let elapsed = cancelling.elapsed();
909 assert!(matches!(snapshot.status, RunStatus::Cancelled));
910 assert!(snapshot.interactions.is_empty() && snapshot.outcomes.is_empty());
911 assert!(snapshot.last_action_ordinal.is_none());
912 assert_eq!(snapshot.state.values()["result"], 0);
913 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
914 assert_eq!(probe.completed.load(Ordering::SeqCst), 0);
915 probe.assert_idle();
916 for (id, count) in [
917 ("performance", 1),
918 ("performance-mid", 2),
919 ("performance-leaf", 16),
920 ] {
921 let child = workspace.workflow(id.parse()?).await?;
922 let rows = child
923 .list_runs(None, NonZeroUsize::new(128).unwrap())
924 .await?;
925 assert_eq!(rows.len(), count);
926 for row in rows {
927 assert!(matches!(row.status, RunStatus::Cancelled));
928 assert!(row.outcomes.is_empty() && row.last_action_ordinal.is_none());
929 assert_eq!(row.state.values()["result"], 0);
930 }
931 }
932 Ok(
933 json!({"dataset":"nested-cancel","depth":2,"intermediate_children":2,"leaf_children":16,
934 "cooperative_cancel_grace_ms":30_000,"cancellation_latency":Self::samples(vec![elapsed]),
935 "source_replays":0,"caller_sources_dropped":16,"agent_unknown_replay":"not applicable: no Agent attempt"}),
936 )
937 }
938
939 async fn delayed_observer(
940 &self,
941 app: &App<StateStore, Agent, WorkloadAction>,
942 probe: &WorkloadProbe,
943 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
944 let workspace = app
945 .create_workspace(WorkspaceConfig {
946 name: None,
947 root: self.root.join("workspace"),
948 })
949 .await?;
950 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
951 let run = workflow.start(ExecutionOptions::default()).await?;
952 let observer = workflow
953 .open_run(RunSelector::Latest(run.id().clone()))
954 .await?;
955 // This public observation is deliberately never polled while the source
956 // produces its 256 partials and commits. It owns no execution work.
957 let delayed = observer.wait();
958 let produced = run.wait().await?;
959 assert!(matches!(produced.status, RunStatus::Succeeded));
960 assert_eq!(probe.completed.load(Ordering::SeqCst), 1);
961 probe.assert_idle();
962 tokio::time::sleep(Duration::from_secs(2)).await;
963 let observing = Instant::now();
964 let snapshot = delayed.await?;
965 let elapsed = observing.elapsed();
966 assert!(matches!(snapshot.status, RunStatus::Succeeded));
967 assert_eq!(snapshot.checkpoint, produced.checkpoint);
968 assert_eq!(snapshot.state.values()["result"], 1);
969 assert_eq!(snapshot.last_action_ordinal, Some(0));
970 assert_eq!(snapshot.outcomes.len(), 1);
971 assert_eq!(
972 snapshot.outcomes[0].result,
973 Map::from_iter([("value".into(), json!(1))])
974 );
975 assert_eq!(probe.constructed.load(Ordering::SeqCst), 1);
976 assert_eq!(probe.partials.load(Ordering::SeqCst), 256);
977 probe.assert_idle();
978 let terminal = run.snapshot().await?;
979 assert_eq!(terminal.checkpoint, snapshot.checkpoint);
980 assert_eq!(terminal.state, snapshot.state);
981 let mut cursor = None;
982 let mut actions = 0;
983 let mut rows = 0;
984 loop {
985 let page = run
986 .history(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
987 .await?;
988 for row in &page.checkpoints {
989 rows += 1;
990 assert!(rows <= 32);
991 if matches!(row.kind, CheckpointKind::Action) {
992 actions += 1;
993 assert_eq!(row.action_ordinal, Some(0));
994 assert_eq!(row.outcomes.len(), 1);
995 }
996 }
997 cursor = page.next;
998 if cursor.is_none() {
999 break;
1000 }
1001 }
1002 assert_eq!(
1003 actions, 1,
1004 "partials cannot become extra durable action commits"
1005 );
1006 Ok(
1007 json!({"dataset":"delayed-observer","unobserved_partial_events":256,"delay_after_terminal_confirmation_ms":2_000,
1008 "observation_latency":Self::samples(vec![elapsed]),"retained_rows":rows,
1009 "memory_scope":"external process RSS; this consumer retains no partial payloads"}),
1010 )
1011 }Sourcepub async fn execution(
&mut self,
options: ExecutionOptions,
) -> Result<Execution<'_, CS, A, X, C, T, H, E>, AppError>
pub async fn execution( &mut self, options: ExecutionOptions, ) -> Result<Execution<'_, CS, A, X, C, T, H, E>, AppError>
Acquires exclusive caller-local advancement without constructing a source.
A live execution keeps that borrow until its last use:
ⓘ
use agents_kit::AgentBackend;
use app_kit::{AppError, workflow::{ExecutionOptions, RunSnapshot, WorkflowRun}};
use storage_kit::StateStoreBackend;
async fn overlapping<CS: StateStoreBackend, A: AgentBackend>(
run: &mut WorkflowRun<CS, A>,
) -> Result<RunSnapshot, AppError> {
let execution = run.execution(ExecutionOptions::default()).await?;
run.snapshot().await?;
execution.finish().await
}§Errors
Returns InvalidInput for a nonempty options state or invalid inputs/selectors,
Busy if advancement or admission is already owned, Conflict for stale
authority, and Closed after revocation. Storage failures use Integration.
§Example
let terminal = run.execution(ExecutionOptions::default()).await?.finish().await?;
assert_eq!(run.snapshot().await?.checkpoint, terminal.checkpoint);Examples found in repository?
examples/performance.rs (lines 405-412)
390 async fn seed(
391 &self,
392 app: &App<StateStore, Agent, WorkloadAction>,
393 actions: u64,
394 linear: bool,
395 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
396 let workspace = app
397 .create_workspace(WorkspaceConfig {
398 name: Some("performance consumer".into()),
399 root: self.root.join("workspace"),
400 })
401 .await?;
402 let workflow = workspace.workflow("performance".parse()?).await?;
403 let mut run = workflow.prepare(WorkflowState::default()).await?;
404 let mut execution = run
405 .execution(ExecutionOptions {
406 halt_after: if linear {
407 Vec::new()
408 } else {
409 vec!["count".into()]
410 },
411 ..ExecutionOptions::default()
412 })
413 .await?;
414 let mut samples = Vec::with_capacity(actions as usize);
415 let started = Instant::now();
416 for ordinal in 0..actions {
417 let step_started = Instant::now();
418 let event = execution.step().await?;
419 samples.push(step_started.elapsed());
420 let Some(ExecutionEvent::Committed { outcomes, .. }) = event else {
421 panic!("each serial workload step must commit one action");
422 };
423 assert_eq!(outcomes.len(), 1);
424 assert_eq!(outcomes[0].ordinal, ordinal);
425 assert_eq!(
426 outcomes[0].node,
427 if linear {
428 format!("n{ordinal:04}")
429 } else {
430 "count".into()
431 }
432 );
433 assert!(outcomes[0].result.is_empty());
434 if !linear && ordinal == 0 {
435 // One public halt and explicit resume add two control rows to
436 // Initial + three checkpoints per action + Terminal.
437 let halted = execution.finish().await?;
438 assert!(matches!(&halted.status, RunStatus::Halted { node, result }
439 if node == "count" && result == &Some(Map::new())));
440 assert_eq!(halted.checkpoint, 4);
441 assert_eq!(
442 halted.state.values(),
443 &Map::from_iter([("count".into(), json!(1))])
444 );
445 assert_eq!(halted.last_action_ordinal, Some(0));
446 assert!(halted.outcomes == outcomes);
447 assert!(halted.interactions.is_empty());
448 execution = run.execution(ExecutionOptions::default()).await?;
449 }
450 }
451 let snapshot = execution.finish().await?;
452 Self::verify_snapshot(&snapshot, actions);
453 let execution_elapsed = started.elapsed();
454 let history = self
455 .history(&run, actions, linear, snapshot.checkpoint)
456 .await?;
457 if let Command::SeedHistory { rows } = self.command {
458 assert_eq!(history["rows"], rows);
459 }
460 Ok(
461 json!({"dataset":if linear {"linear"} else {"history"},"actions":actions,
462 "workspace":workspace.id().to_string(),"run":run.id().to_string(),
463 "occurrence":snapshot.occurrence,
464 "expected_state":snapshot.state.values(),"last_action_ordinal":snapshot.last_action_ordinal,
465 "terminal_checkpoint":snapshot.checkpoint,"retained_rows":history["rows"],
466 "execution_ns":execution_elapsed.as_nanos(),
467 "actions_per_second":actions as f64 / execution_elapsed.as_secs_f64(),
468 "step_latency":Self::samples(samples),"history":history}),
469 )
470 }
471
472 async fn reopen(
473 &self,
474 app: &App<StateStore, Agent, WorkloadAction>,
475 manifest: &Value,
476 actions: u64,
477 open_only: bool,
478 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
479 let started = Instant::now();
480 let workspace: WorkspaceId = manifest["workspace"]
481 .as_str()
482 .ok_or("missing workspace ID")?
483 .parse()?;
484 let run_id: RunId = manifest["run"].as_str().ok_or("missing run ID")?.parse()?;
485 let occurrence = manifest["occurrence"]
486 .as_str()
487 .ok_or("missing run occurrence")?
488 .to_owned();
489 let workspace = app.workspace(workspace).await?;
490 let workflow = workspace.workflow("performance".parse()?).await?;
491 let run = workflow
492 .open_run(RunSelector::Exact {
493 id: run_id,
494 occurrence: occurrence.clone(),
495 })
496 .await?;
497 let snapshot = run.snapshot().await?;
498 Self::verify_snapshot(&snapshot, actions);
499 assert_eq!(snapshot.occurrence, occurrence);
500 assert_eq!(json!(snapshot.checkpoint), manifest["terminal_checkpoint"]);
501 assert_eq!(json!(snapshot.state.values()), manifest["expected_state"]);
502 assert_eq!(
503 json!(snapshot.last_action_ordinal),
504 manifest["last_action_ordinal"]
505 );
506 let lookup = started.elapsed();
507 if open_only {
508 return Ok(
509 json!({"dataset":"history","actions":actions,"mode":"open-only",
510 "lookup_and_snapshot_ns":lookup.as_nanos(),
511 "seed_retained_rows":manifest["retained_rows"],"observed_history_rows":null}),
512 );
513 }
514 let history = self
515 .history(&run, actions, false, snapshot.checkpoint)
516 .await?;
517 assert_eq!(history["rows"], manifest["retained_rows"]);
518 Ok(
519 json!({"dataset":"history","actions":actions,"mode":"full-history",
520 "lookup_and_snapshot_ns":lookup.as_nanos(),"history":history}),
521 )
522 }
523
524 fn definitions(&self, actions: u64) -> Result<(Vec<WorkflowDefinition>, usize), AppError> {
525 if matches!(
526 self.command,
527 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. }
528 ) {
529 let (definition, bytes) =
530 Self::definition(matches!(self.command, Command::Linear), actions)?;
531 return Ok((vec![definition], bytes));
532 }
533 let leaf = json!({
534 "schema":"app-kit.workflow/v2","id":"performance-leaf","revision":"r1",
535 "state":{"value":0,"result":0},"required":["value","result"],
536 "reducers":{"value":"reject_concurrent","result":"reject_concurrent"},
537 "entry":"work","nodes":[{"id":"work","deadline_ms":NODE_DEADLINE_MS,
538 "step":{"kind":"action","action":"performance-action","revision":"r1"}}],"routes":{}
539 });
540 let documents = match self.command {
541 Command::Ready => {
542 let mut nodes = vec![json!({"id":"fork","step":{"kind":"state","mutations":[]}})];
543 let mut branches = Vec::new();
544 for index in 0..32 {
545 let node = format!("n{index:02}");
546 nodes.push(json!({"id":node,"deadline_ms":NODE_DEADLINE_MS,
547 "step":{"kind":"action","action":"performance-action","revision":"r1"}}));
548 branches.push(json!({"target":{"kind":"node","id":node}}));
549 }
550 vec![json!({
551 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
552 "state":{"value":0,"result":0},"required":["value","result"],
553 "reducers":{"value":"reject_concurrent","result":"collect"},
554 "entry":"fork","nodes":nodes,
555 "routes":{"fork":{"mode":"broadcast","branches":branches}}
556 })]
557 }
558 Command::Map | Command::NestedCancel => {
559 let nested = matches!(self.command, Command::NestedCancel);
560 let items = if nested { 2 } else { 4_096 };
561 let mut documents = vec![
562 json!({
563 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
564 "state":{"value":0,"result":0},"required":["value","result"],
565 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
566 "entry":"map","nodes":[{"id":"map","deadline_ms":if nested {NODE_DEADLINE_MS} else {WORKLOAD_BUDGET.as_millis() as u64},
567 "step":{"kind":"map","workflow":if nested {"performance-mid"} else {"performance-leaf"},
568 "items":{"kind":"literal","value":(0..items).collect::<Vec<_>>()},
569 "input_key":"value","max_parallel":if nested {2} else {8},
570 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
571 }),
572 leaf,
573 ];
574 if nested {
575 documents.push(json!({
576 "schema":"app-kit.workflow/v2","id":"performance-mid","revision":"r1",
577 "state":{"value":0,"result":0},"required":["value","result"],
578 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
579 "entry":"map","nodes":[{"id":"map","deadline_ms":NODE_DEADLINE_MS,
580 "step":{"kind":"map","workflow":"performance-leaf",
581 "items":{"kind":"literal","value":(0..8).collect::<Vec<_>>()},
582 "input_key":"value","max_parallel":8,
583 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
584 }));
585 }
586 documents
587 }
588 Command::Repeated | Command::DelayedObserver => vec![leaf],
589 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. } => {
590 unreachable!()
591 }
592 };
593 let mut bytes = 0;
594 let mut definitions = Vec::with_capacity(documents.len());
595 for document in documents {
596 let source = document.to_string();
597 assert!(source.len() <= 1_048_576);
598 bytes += source.len();
599 definitions.push(WorkflowDefinition::parse(&source)?);
600 }
601 Ok((definitions, bytes))
602 }
603
604 async fn ready(
605 &self,
606 app: &App<StateStore, Agent, WorkloadAction>,
607 probe: &WorkloadProbe,
608 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
609 let workspace = app
610 .create_workspace(WorkspaceConfig {
611 name: None,
612 root: self.root.join("workspace"),
613 })
614 .await?;
615 let workflow = workspace.workflow("performance".parse()?).await?;
616 let mut waves = Vec::with_capacity(32);
617 let mut releases = Vec::with_capacity(32 * 31);
618 let mut barriers = Vec::with_capacity(32);
619 let started = Instant::now();
620 for permutation in 0..32 {
621 probe.assert_idle();
622 let mut run = workflow.prepare(WorkflowState::default()).await?;
623 let observer = workflow
624 .open_run(RunSelector::Latest(run.id().clone()))
625 .await?;
626 let mut execution = run.execution(ExecutionOptions::default()).await?;
627 let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
628 panic!("the native fork must commit before the ready wave");
629 };
630 assert_eq!(outcomes.len(), 1);
631 assert_eq!(
632 (outcomes[0].node.as_str(), outcomes[0].ordinal),
633 ("fork", 0)
634 );
635 let wave = Instant::now();
636 let mut finish = Box::pin(execution.finish());
637 tokio::select! {
638 _ = probe.wait_for(&probe.constructed, (permutation + 1) * 32) => {}
639 result = &mut finish => panic!("gated ready wave finished before admission: {}", result.is_ok()),
640 }
641 assert_eq!(probe.active.load(Ordering::SeqCst), 32);
642 let order = (0..32)
643 .map(|offset| (permutation + 31 - offset) % 32)
644 .collect::<Vec<_>>();
645 for (released, index) in order.iter().take(31).enumerate() {
646 let releasing = Instant::now();
647 probe.release(*index);
648 tokio::select! {
649 _ = probe.wait_for(&probe.completed, permutation * 32 + released + 1) => {}
650 result = &mut finish => panic!("a partial ready wave committed: {}", result.is_ok()),
651 }
652 releases.push(releasing.elapsed());
653 let visible = observer.snapshot().await?;
654 assert_eq!(
655 visible.state.values()["result"],
656 0,
657 "staged siblings cannot expose a partial barrier"
658 );
659 assert_eq!(visible.last_action_ordinal, Some(0));
660 }
661 let barrier = Instant::now();
662 probe.release(order[31]);
663 let snapshot = finish.await?;
664 barriers.push(barrier.elapsed());
665 waves.push(wave.elapsed());
666 assert!(matches!(snapshot.status, RunStatus::Succeeded));
667 assert!(snapshot.interactions.is_empty());
668 assert_eq!(
669 snapshot.state.values()["result"],
670 json!((1..=32).collect::<Vec<_>>())
671 );
672 assert_eq!(snapshot.last_action_ordinal, Some(32));
673 assert_eq!(snapshot.outcomes.len(), 32);
674 for (index, outcome) in snapshot.outcomes.iter().enumerate() {
675 assert_eq!(outcome.node, format!("n{index:02}"));
676 assert_eq!(outcome.ordinal, index as u64 + 1);
677 assert_eq!(
678 outcome.result,
679 Map::from_iter([("value".into(), json!(index + 1))])
680 );
681 }
682 assert_eq!(std::mem::take(&mut *probe.order.lock().unwrap()), order);
683 assert_eq!(
684 probe.completed.load(Ordering::SeqCst),
685 (permutation + 1) * 32
686 );
687 probe.assert_idle();
688 }
689 assert_eq!(probe.peak.load(Ordering::SeqCst), 32);
690 let elapsed = started.elapsed();
691 Ok(
692 json!({"dataset":"ready32","permutations":32,"ready_actions_per_wave":32,
693 "caller_action_commits":1024,"native_fork_commits":32,
694 "execution_ns":elapsed.as_nanos(),"caller_actions_per_second":1024.0 / elapsed.as_secs_f64(),
695 "wave_latency_including_snapshot_probes":Self::samples(waves),
696 "release_to_source_completion_latency":Self::samples(releases),
697 "last_release_to_committed_barrier_latency":Self::samples(barriers)}),
698 )
699 }
700
701 async fn mapped(
702 &self,
703 app: &App<StateStore, Agent, WorkloadAction>,
704 probe: &WorkloadProbe,
705 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
706 let workspace = app
707 .create_workspace(WorkspaceConfig {
708 name: None,
709 root: self.root.join("workspace"),
710 })
711 .await?;
712 let workflow = workspace.workflow("performance".parse()?).await?;
713 let mut run = workflow.prepare(WorkflowState::default()).await?;
714 let observer = workflow
715 .open_run(RunSelector::Latest(run.id().clone()))
716 .await?;
717 let started = Instant::now();
718 let mut finish = Box::pin(run.execution(ExecutionOptions::default()).await?.finish());
719 tokio::select! {
720 _ = probe.wait_for(&probe.constructed, 8) => {}
721 result = &mut finish => panic!("map finished before its first eight blocked sources: {}", result.is_ok()),
722 }
723 assert_eq!(probe.active.load(Ordering::SeqCst), 8);
724 assert_eq!(probe.constructed.load(Ordering::SeqCst), 8);
725 let unchanged = observer.snapshot().await?;
726 assert_eq!(unchanged.state.values()["result"], 0);
727 assert!(unchanged.last_action_ordinal.is_none());
728 probe.release_all();
729 let snapshot = finish.await?;
730 let elapsed = started.elapsed();
731 assert!(matches!(snapshot.status, RunStatus::Succeeded));
732 assert!(snapshot.interactions.is_empty());
733 assert_eq!(snapshot.state.values()["result"], 4_096_u64 * 4_097 / 2);
734 assert_eq!(snapshot.last_action_ordinal, Some(0));
735 assert_eq!(snapshot.outcomes.len(), 1);
736 assert_eq!(snapshot.outcomes[0].node, "map");
737 assert_eq!(probe.constructed.load(Ordering::SeqCst), 4_096);
738 assert_eq!(probe.completed.load(Ordering::SeqCst), 4_096);
739 assert_eq!(probe.peak.load(Ordering::SeqCst), 8);
740 probe.assert_idle();
741 let child = workspace.workflow("performance-leaf".parse()?).await?;
742 let mut cursor = None;
743 let mut values = BTreeSet::new();
744 let verifying = Instant::now();
745 loop {
746 let page = child
747 .list_runs(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
748 .await?;
749 assert!(page.len() <= PAGE_SIZE);
750 if page.is_empty() {
751 break;
752 }
753 for row in &page {
754 assert!(matches!(row.status, RunStatus::Succeeded));
755 let value = row.state.values()["value"].as_u64().unwrap();
756 assert!(
757 value < 4_096 && values.insert(value),
758 "every mapped input has exactly one retained child"
759 );
760 assert_eq!(row.state.values()["result"], value + 1);
761 assert_eq!(row.last_action_ordinal, Some(0));
762 assert_eq!(row.outcomes.len(), 1);
763 assert_eq!(row.outcomes[0].node, "work");
764 }
765 let last = page.last().unwrap();
766 cursor = Some(RunSelector::Exact {
767 id: last.id.clone(),
768 occurrence: last.occurrence.clone(),
769 });
770 assert!(values.len() <= 4_096);
771 }
772 assert_eq!(values.len(), 4_096);
773 Ok(json!({"dataset":"map4096","items":4096,"max_parallel":8,
774 "parent_node_deadline_ms":WORKLOAD_BUDGET.as_millis(),
775 "execution_ns":elapsed.as_nanos(),"child_actions_per_second":4096.0 / elapsed.as_secs_f64(),
776 "map_completion_latency":Self::samples(vec![elapsed]),
777 "child_inventory_verification_ns":verifying.elapsed().as_nanos(),"verified_children":values.len()}))
778 }
779
780 async fn repeated(
781 &self,
782 app: &App<StateStore, Agent, WorkloadAction>,
783 probe: &WorkloadProbe,
784 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
785 let workspace = app
786 .create_workspace(WorkspaceConfig {
787 name: None,
788 root: self.root.join("workspace"),
789 })
790 .await?;
791 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
792 let started = Instant::now();
793 // Admit one run at a time, then directly poll at most 16 original
794 // executions together. No consumer task is spawned or detached.
795 let mut runs = std::pin::pin!(
796 stream::iter(0..100_u64)
797 .then(|value| {
798 let workflow = &workflow;
799 async move {
800 let started = Instant::now();
801 let run = workflow
802 .prepare(WorkflowState::new(Map::from_iter([(
803 "value".into(),
804 json!(value),
805 )]))?)
806 .await?;
807 Ok::<_, AppError>((value, run, started))
808 }
809 })
810 .map(|prepared| async move {
811 let (value, mut run, started) = prepared?;
812 let snapshot = run
813 .execution(ExecutionOptions::default())
814 .await?
815 .finish()
816 .await?;
817 assert!(matches!(snapshot.status, RunStatus::Succeeded));
818 assert!(snapshot.interactions.is_empty());
819 assert_eq!(snapshot.state.values()["value"], value);
820 assert_eq!(snapshot.state.values()["result"], value + 1);
821 assert_eq!(snapshot.last_action_ordinal, Some(0));
822 assert_eq!(snapshot.outcomes.len(), 1);
823 Ok::<_, AppError>((run.id().clone(), started.elapsed()))
824 })
825 .buffer_unordered(16)
826 );
827 tokio::select! {
828 _ = probe.wait_for(&probe.constructed, 16) => {}
829 result = runs.next() => panic!(
830 "run finished before the first 16 blocked sources: {:?}",
831 result.map(|result| result.map(|_| ())),
832 ),
833 }
834 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
835 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
836 probe.release_all();
837 let mut ids = BTreeSet::new();
838 let mut samples = Vec::with_capacity(100);
839 while let Some(result) = runs.next().await {
840 let (id, elapsed) = result?;
841 assert!(ids.insert(id));
842 samples.push(elapsed);
843 }
844 assert_eq!(ids.len(), 100);
845 assert_eq!(probe.constructed.load(Ordering::SeqCst), 100);
846 assert_eq!(probe.completed.load(Ordering::SeqCst), 100);
847 assert_eq!(probe.peak.load(Ordering::SeqCst), 16);
848 probe.assert_idle();
849 let rows = workflow
850 .list_runs(None, NonZeroUsize::new(128).unwrap())
851 .await?;
852 assert_eq!(rows.len(), 100);
853 assert_eq!(
854 rows.iter()
855 .map(|row| row.id.clone())
856 .collect::<BTreeSet<_>>(),
857 ids
858 );
859 assert!(
860 rows.iter()
861 .all(|row| matches!(row.status, RunStatus::Succeeded))
862 );
863 let elapsed = started.elapsed();
864 Ok(
865 json!({"dataset":"repeated100","runs":100,"caller_in_flight_limit":16,"admission":"sequential",
866 "production_active_limit":"unchanged; this measures only the caller's 16-journey envelope",
867 "execution_ns":elapsed.as_nanos(),"runs_per_second":100.0 / elapsed.as_secs_f64(),
868 "run_latency_including_admission":Self::samples(samples),"indefinite_stability_proof":false}),
869 )
870 }More examples
examples/history_and_recovery.rs (line 76)
63 async fn create_fork(
64 &self,
65 app: &App<StateStore, Agent>,
66 ) -> Result<(WorkspaceId, RunSelector, RunSelector), Box<dyn Error + Send + Sync>> {
67 let workspace = app
68 .create_workspace(WorkspaceConfig {
69 name: Some("history example".into()),
70 root: std::env::current_dir()?,
71 })
72 .await?;
73 let workflow = workspace.workflow("history-counter".parse()?).await?;
74 let mut source = workflow.prepare(WorkflowState::default()).await?;
75 let completed = source
76 .execution(ExecutionOptions::default())
77 .await?
78 .finish()
79 .await?;
80 if !matches!(completed.status, RunStatus::Succeeded)
81 || completed.state.values()["count"] != 2
82 {
83 return Err(
84 AppError::Integration("source did not complete both increments".into()).into(),
85 );
86 }
87
88 // An action ordinal selects the whole committed boundary containing it;
89 // it is independent of the checkpoint sequence and storage revision.
90 let first = source.checkpoint(CheckpointSelector::Action(0)).await?;
91 if !matches!(first.kind, CheckpointKind::Action)
92 || first.position.as_deref() != Some("first")
93 || first.state.values()["count"] != 1
94 {
95 return Err(
96 AppError::Integration("first action history is inconsistent".into()).into(),
97 );
98 }
99 let source_selector = RunSelector::Exact {
100 id: source.id().clone(),
101 occurrence: first.occurrence.clone(),
102 };
103 let mut fork = workflow
104 .restore(RestoreOptions {
105 source: source_selector.clone(),
106 checkpoint: CheckpointSelector::Action(0),
107 overrides: WorkflowState::new(Map::from_iter([("count".into(), json!(40))]))?,
108 position: ResumePosition::AfterCompleted,
109 fallback: None,
110 })
111 .await?;
112 let parked = fork.snapshot().await?;
113 if !matches!(parked.status, RunStatus::Paused)
114 || parked.checkpoint != 0
115 || parked.state.values()["count"] != 40
116 || !parked.outcomes.is_empty()
117 {
118 return Err(
119 AppError::Integration("restore did not create a parked clean copy".into()).into(),
120 );
121 }
122
123 source.delete(completed.checkpoint).await?;
124 if !matches!(source.snapshot().await, Err(AppError::NotFound(_))) {
125 return Err(
126 AppError::Integration("explicit source deletion did not finish".into()).into(),
127 );
128 }
129 // Deletion does not invalidate the copied state. AfterCompleted skips
130 // the first increment, so only the second changes the override: 40 -> 41.
131 let completed = fork
132 .execution(ExecutionOptions::default())
133 .await?
134 .finish()
135 .await?;
136 if !matches!(completed.status, RunStatus::Succeeded)
137 || completed.state.values()["count"] != 41
138 || completed.last_action_ordinal != Some(1)
139 {
140 return Err(
141 AppError::Integration("fork resumed from the wrong boundary".into()).into(),
142 );
143 }
144 println!("Source count: 2; copied count: 40; completed fork count: 41");
145 Ok((
146 workspace.id().clone(),
147 source_selector,
148 RunSelector::Exact {
149 id: fork.id().clone(),
150 occurrence: completed.occurrence,
151 },
152 ))
153 }examples/actions_and_inputs.rs (lines 54-60)
23 async fn run() -> Result<(), Box<dyn Error + Send + Sync>> {
24 let definition = WorkflowDefinition::parse(
25 &json!({
26 "schema":"app-kit.workflow/v2", "id":"invoice", "revision":"r1",
27 "state":{"unit_price":7,"total":0}, "required":["unit_price","total"],
28 "reducers":{"unit_price":"reject_concurrent","total":"reject_concurrent"},
29 "entry":"calculate",
30 "nodes":[{"id":"calculate","step":{
31 "kind":"action","action":"invoice","revision":"r1","project_result":false
32 }}], "routes":{}
33 })
34 .to_string(),
35 )?;
36 let store =
37 StateStore::connect(StateStoreConfig::Lance(LanceStateStoreConfig::memory())).await?;
38 // Native caller actions need no Agent value or provider service.
39 let mut app = App::open(
40 AppConfig::<_, Agent>::new(store, Vec::new(), vec![definition])
41 .with_actions(vec![InvoiceAction::new()]),
42 )
43 .await?;
44 let observed = tokio::time::timeout(Duration::from_secs(15), async {
45 let workspace = app
46 .create_workspace(WorkspaceConfig {
47 name: Some("invoice example".into()),
48 root: PathBuf::from(env!("CARGO_MANIFEST_DIR")),
49 })
50 .await?;
51 let workflow = workspace.workflow("invoice".parse()?).await?;
52 let mut run = workflow.prepare(WorkflowState::default()).await?;
53 let mut execution = run
54 .execution(ExecutionOptions {
55 inputs: Map::from_iter([
56 ("quantity".into(), json!(6)),
57 ("trace".into(), json!("sample")),
58 ]),
59 ..ExecutionOptions::default()
60 })
61 .await?;
62 let Some(ExecutionEvent::Partial { result, .. }) = execution.next().await? else {
63 return Err(AppError::Integration(
64 "expected the real action partial".into(),
65 ));
66 };
67 if result.get("progress") != Some(&json!("calculated")) {
68 return Err(AppError::Integration("partial changed".into()));
69 }
70 let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
71 return Err(AppError::Integration(
72 "expected one committed boundary".into(),
73 ));
74 };
75 let expected =
76 json!({"quantity":6,"label":"order","currency":"USD","trace":"sample","total":42});
77 if outcomes.len() != 1 || Value::Object(outcomes[0].result.clone()) != expected {
78 return Err(AppError::Integration(
79 "resolved inputs or action result changed".into(),
80 ));
81 }
82 let terminal = execution.finish().await?;
83 if !matches!(terminal.status, RunStatus::Succeeded)
84 || terminal.state.values()
85 != json!({"unit_price":7,"total":42}).as_object().unwrap()
86 {
87 return Err(AppError::Integration(
88 "only the explicit total write should persist".into(),
89 ));
90 }
91 println!(
92 "Partial: calculated; committed total: 42; transient inputs stayed out of state"
93 );
94 Ok::<(), AppError>(())
95 })
96 .await;
97 let closed = tokio::time::timeout(
98 Duration::from_secs(15),
99 app.shutdown(Shutdown::CancelAndDrain {
100 timeout: Duration::from_secs(10),
101 }),
102 )
103 .await;
104 drop(app);
105 observed.map_err(|_| AppError::Timeout("invoice execution timed out".into()))??;
106 closed.map_err(|_| AppError::Timeout("invoice cleanup timed out".into()))??;
107 Ok(())
108 }Sourcepub async fn snapshot(&self) -> Result<RunSnapshot, AppError>
pub async fn snapshot(&self) -> Result<RunSnapshot, AppError>
Loads the current run snapshot.
§Errors
Returns Closed for revoked App ownership or sealed/deleted scope, NotFound
for absent retained evidence, or Integration for inconsistent or unreadable
records. A newer App generation can invalidate this handle.
§Example
let snapshot = run.snapshot().await?;
assert_eq!(snapshot.checkpoint, 0);
assert!(matches!(snapshot.status, RunStatus::Paused));Examples found in repository?
examples/performance.rs (line 497)
472 async fn reopen(
473 &self,
474 app: &App<StateStore, Agent, WorkloadAction>,
475 manifest: &Value,
476 actions: u64,
477 open_only: bool,
478 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
479 let started = Instant::now();
480 let workspace: WorkspaceId = manifest["workspace"]
481 .as_str()
482 .ok_or("missing workspace ID")?
483 .parse()?;
484 let run_id: RunId = manifest["run"].as_str().ok_or("missing run ID")?.parse()?;
485 let occurrence = manifest["occurrence"]
486 .as_str()
487 .ok_or("missing run occurrence")?
488 .to_owned();
489 let workspace = app.workspace(workspace).await?;
490 let workflow = workspace.workflow("performance".parse()?).await?;
491 let run = workflow
492 .open_run(RunSelector::Exact {
493 id: run_id,
494 occurrence: occurrence.clone(),
495 })
496 .await?;
497 let snapshot = run.snapshot().await?;
498 Self::verify_snapshot(&snapshot, actions);
499 assert_eq!(snapshot.occurrence, occurrence);
500 assert_eq!(json!(snapshot.checkpoint), manifest["terminal_checkpoint"]);
501 assert_eq!(json!(snapshot.state.values()), manifest["expected_state"]);
502 assert_eq!(
503 json!(snapshot.last_action_ordinal),
504 manifest["last_action_ordinal"]
505 );
506 let lookup = started.elapsed();
507 if open_only {
508 return Ok(
509 json!({"dataset":"history","actions":actions,"mode":"open-only",
510 "lookup_and_snapshot_ns":lookup.as_nanos(),
511 "seed_retained_rows":manifest["retained_rows"],"observed_history_rows":null}),
512 );
513 }
514 let history = self
515 .history(&run, actions, false, snapshot.checkpoint)
516 .await?;
517 assert_eq!(history["rows"], manifest["retained_rows"]);
518 Ok(
519 json!({"dataset":"history","actions":actions,"mode":"full-history",
520 "lookup_and_snapshot_ns":lookup.as_nanos(),"history":history}),
521 )
522 }
523
524 fn definitions(&self, actions: u64) -> Result<(Vec<WorkflowDefinition>, usize), AppError> {
525 if matches!(
526 self.command,
527 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. }
528 ) {
529 let (definition, bytes) =
530 Self::definition(matches!(self.command, Command::Linear), actions)?;
531 return Ok((vec![definition], bytes));
532 }
533 let leaf = json!({
534 "schema":"app-kit.workflow/v2","id":"performance-leaf","revision":"r1",
535 "state":{"value":0,"result":0},"required":["value","result"],
536 "reducers":{"value":"reject_concurrent","result":"reject_concurrent"},
537 "entry":"work","nodes":[{"id":"work","deadline_ms":NODE_DEADLINE_MS,
538 "step":{"kind":"action","action":"performance-action","revision":"r1"}}],"routes":{}
539 });
540 let documents = match self.command {
541 Command::Ready => {
542 let mut nodes = vec![json!({"id":"fork","step":{"kind":"state","mutations":[]}})];
543 let mut branches = Vec::new();
544 for index in 0..32 {
545 let node = format!("n{index:02}");
546 nodes.push(json!({"id":node,"deadline_ms":NODE_DEADLINE_MS,
547 "step":{"kind":"action","action":"performance-action","revision":"r1"}}));
548 branches.push(json!({"target":{"kind":"node","id":node}}));
549 }
550 vec![json!({
551 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
552 "state":{"value":0,"result":0},"required":["value","result"],
553 "reducers":{"value":"reject_concurrent","result":"collect"},
554 "entry":"fork","nodes":nodes,
555 "routes":{"fork":{"mode":"broadcast","branches":branches}}
556 })]
557 }
558 Command::Map | Command::NestedCancel => {
559 let nested = matches!(self.command, Command::NestedCancel);
560 let items = if nested { 2 } else { 4_096 };
561 let mut documents = vec![
562 json!({
563 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
564 "state":{"value":0,"result":0},"required":["value","result"],
565 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
566 "entry":"map","nodes":[{"id":"map","deadline_ms":if nested {NODE_DEADLINE_MS} else {WORKLOAD_BUDGET.as_millis() as u64},
567 "step":{"kind":"map","workflow":if nested {"performance-mid"} else {"performance-leaf"},
568 "items":{"kind":"literal","value":(0..items).collect::<Vec<_>>()},
569 "input_key":"value","max_parallel":if nested {2} else {8},
570 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
571 }),
572 leaf,
573 ];
574 if nested {
575 documents.push(json!({
576 "schema":"app-kit.workflow/v2","id":"performance-mid","revision":"r1",
577 "state":{"value":0,"result":0},"required":["value","result"],
578 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
579 "entry":"map","nodes":[{"id":"map","deadline_ms":NODE_DEADLINE_MS,
580 "step":{"kind":"map","workflow":"performance-leaf",
581 "items":{"kind":"literal","value":(0..8).collect::<Vec<_>>()},
582 "input_key":"value","max_parallel":8,
583 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
584 }));
585 }
586 documents
587 }
588 Command::Repeated | Command::DelayedObserver => vec![leaf],
589 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. } => {
590 unreachable!()
591 }
592 };
593 let mut bytes = 0;
594 let mut definitions = Vec::with_capacity(documents.len());
595 for document in documents {
596 let source = document.to_string();
597 assert!(source.len() <= 1_048_576);
598 bytes += source.len();
599 definitions.push(WorkflowDefinition::parse(&source)?);
600 }
601 Ok((definitions, bytes))
602 }
603
604 async fn ready(
605 &self,
606 app: &App<StateStore, Agent, WorkloadAction>,
607 probe: &WorkloadProbe,
608 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
609 let workspace = app
610 .create_workspace(WorkspaceConfig {
611 name: None,
612 root: self.root.join("workspace"),
613 })
614 .await?;
615 let workflow = workspace.workflow("performance".parse()?).await?;
616 let mut waves = Vec::with_capacity(32);
617 let mut releases = Vec::with_capacity(32 * 31);
618 let mut barriers = Vec::with_capacity(32);
619 let started = Instant::now();
620 for permutation in 0..32 {
621 probe.assert_idle();
622 let mut run = workflow.prepare(WorkflowState::default()).await?;
623 let observer = workflow
624 .open_run(RunSelector::Latest(run.id().clone()))
625 .await?;
626 let mut execution = run.execution(ExecutionOptions::default()).await?;
627 let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
628 panic!("the native fork must commit before the ready wave");
629 };
630 assert_eq!(outcomes.len(), 1);
631 assert_eq!(
632 (outcomes[0].node.as_str(), outcomes[0].ordinal),
633 ("fork", 0)
634 );
635 let wave = Instant::now();
636 let mut finish = Box::pin(execution.finish());
637 tokio::select! {
638 _ = probe.wait_for(&probe.constructed, (permutation + 1) * 32) => {}
639 result = &mut finish => panic!("gated ready wave finished before admission: {}", result.is_ok()),
640 }
641 assert_eq!(probe.active.load(Ordering::SeqCst), 32);
642 let order = (0..32)
643 .map(|offset| (permutation + 31 - offset) % 32)
644 .collect::<Vec<_>>();
645 for (released, index) in order.iter().take(31).enumerate() {
646 let releasing = Instant::now();
647 probe.release(*index);
648 tokio::select! {
649 _ = probe.wait_for(&probe.completed, permutation * 32 + released + 1) => {}
650 result = &mut finish => panic!("a partial ready wave committed: {}", result.is_ok()),
651 }
652 releases.push(releasing.elapsed());
653 let visible = observer.snapshot().await?;
654 assert_eq!(
655 visible.state.values()["result"],
656 0,
657 "staged siblings cannot expose a partial barrier"
658 );
659 assert_eq!(visible.last_action_ordinal, Some(0));
660 }
661 let barrier = Instant::now();
662 probe.release(order[31]);
663 let snapshot = finish.await?;
664 barriers.push(barrier.elapsed());
665 waves.push(wave.elapsed());
666 assert!(matches!(snapshot.status, RunStatus::Succeeded));
667 assert!(snapshot.interactions.is_empty());
668 assert_eq!(
669 snapshot.state.values()["result"],
670 json!((1..=32).collect::<Vec<_>>())
671 );
672 assert_eq!(snapshot.last_action_ordinal, Some(32));
673 assert_eq!(snapshot.outcomes.len(), 32);
674 for (index, outcome) in snapshot.outcomes.iter().enumerate() {
675 assert_eq!(outcome.node, format!("n{index:02}"));
676 assert_eq!(outcome.ordinal, index as u64 + 1);
677 assert_eq!(
678 outcome.result,
679 Map::from_iter([("value".into(), json!(index + 1))])
680 );
681 }
682 assert_eq!(std::mem::take(&mut *probe.order.lock().unwrap()), order);
683 assert_eq!(
684 probe.completed.load(Ordering::SeqCst),
685 (permutation + 1) * 32
686 );
687 probe.assert_idle();
688 }
689 assert_eq!(probe.peak.load(Ordering::SeqCst), 32);
690 let elapsed = started.elapsed();
691 Ok(
692 json!({"dataset":"ready32","permutations":32,"ready_actions_per_wave":32,
693 "caller_action_commits":1024,"native_fork_commits":32,
694 "execution_ns":elapsed.as_nanos(),"caller_actions_per_second":1024.0 / elapsed.as_secs_f64(),
695 "wave_latency_including_snapshot_probes":Self::samples(waves),
696 "release_to_source_completion_latency":Self::samples(releases),
697 "last_release_to_committed_barrier_latency":Self::samples(barriers)}),
698 )
699 }
700
701 async fn mapped(
702 &self,
703 app: &App<StateStore, Agent, WorkloadAction>,
704 probe: &WorkloadProbe,
705 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
706 let workspace = app
707 .create_workspace(WorkspaceConfig {
708 name: None,
709 root: self.root.join("workspace"),
710 })
711 .await?;
712 let workflow = workspace.workflow("performance".parse()?).await?;
713 let mut run = workflow.prepare(WorkflowState::default()).await?;
714 let observer = workflow
715 .open_run(RunSelector::Latest(run.id().clone()))
716 .await?;
717 let started = Instant::now();
718 let mut finish = Box::pin(run.execution(ExecutionOptions::default()).await?.finish());
719 tokio::select! {
720 _ = probe.wait_for(&probe.constructed, 8) => {}
721 result = &mut finish => panic!("map finished before its first eight blocked sources: {}", result.is_ok()),
722 }
723 assert_eq!(probe.active.load(Ordering::SeqCst), 8);
724 assert_eq!(probe.constructed.load(Ordering::SeqCst), 8);
725 let unchanged = observer.snapshot().await?;
726 assert_eq!(unchanged.state.values()["result"], 0);
727 assert!(unchanged.last_action_ordinal.is_none());
728 probe.release_all();
729 let snapshot = finish.await?;
730 let elapsed = started.elapsed();
731 assert!(matches!(snapshot.status, RunStatus::Succeeded));
732 assert!(snapshot.interactions.is_empty());
733 assert_eq!(snapshot.state.values()["result"], 4_096_u64 * 4_097 / 2);
734 assert_eq!(snapshot.last_action_ordinal, Some(0));
735 assert_eq!(snapshot.outcomes.len(), 1);
736 assert_eq!(snapshot.outcomes[0].node, "map");
737 assert_eq!(probe.constructed.load(Ordering::SeqCst), 4_096);
738 assert_eq!(probe.completed.load(Ordering::SeqCst), 4_096);
739 assert_eq!(probe.peak.load(Ordering::SeqCst), 8);
740 probe.assert_idle();
741 let child = workspace.workflow("performance-leaf".parse()?).await?;
742 let mut cursor = None;
743 let mut values = BTreeSet::new();
744 let verifying = Instant::now();
745 loop {
746 let page = child
747 .list_runs(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
748 .await?;
749 assert!(page.len() <= PAGE_SIZE);
750 if page.is_empty() {
751 break;
752 }
753 for row in &page {
754 assert!(matches!(row.status, RunStatus::Succeeded));
755 let value = row.state.values()["value"].as_u64().unwrap();
756 assert!(
757 value < 4_096 && values.insert(value),
758 "every mapped input has exactly one retained child"
759 );
760 assert_eq!(row.state.values()["result"], value + 1);
761 assert_eq!(row.last_action_ordinal, Some(0));
762 assert_eq!(row.outcomes.len(), 1);
763 assert_eq!(row.outcomes[0].node, "work");
764 }
765 let last = page.last().unwrap();
766 cursor = Some(RunSelector::Exact {
767 id: last.id.clone(),
768 occurrence: last.occurrence.clone(),
769 });
770 assert!(values.len() <= 4_096);
771 }
772 assert_eq!(values.len(), 4_096);
773 Ok(json!({"dataset":"map4096","items":4096,"max_parallel":8,
774 "parent_node_deadline_ms":WORKLOAD_BUDGET.as_millis(),
775 "execution_ns":elapsed.as_nanos(),"child_actions_per_second":4096.0 / elapsed.as_secs_f64(),
776 "map_completion_latency":Self::samples(vec![elapsed]),
777 "child_inventory_verification_ns":verifying.elapsed().as_nanos(),"verified_children":values.len()}))
778 }
779
780 async fn repeated(
781 &self,
782 app: &App<StateStore, Agent, WorkloadAction>,
783 probe: &WorkloadProbe,
784 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
785 let workspace = app
786 .create_workspace(WorkspaceConfig {
787 name: None,
788 root: self.root.join("workspace"),
789 })
790 .await?;
791 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
792 let started = Instant::now();
793 // Admit one run at a time, then directly poll at most 16 original
794 // executions together. No consumer task is spawned or detached.
795 let mut runs = std::pin::pin!(
796 stream::iter(0..100_u64)
797 .then(|value| {
798 let workflow = &workflow;
799 async move {
800 let started = Instant::now();
801 let run = workflow
802 .prepare(WorkflowState::new(Map::from_iter([(
803 "value".into(),
804 json!(value),
805 )]))?)
806 .await?;
807 Ok::<_, AppError>((value, run, started))
808 }
809 })
810 .map(|prepared| async move {
811 let (value, mut run, started) = prepared?;
812 let snapshot = run
813 .execution(ExecutionOptions::default())
814 .await?
815 .finish()
816 .await?;
817 assert!(matches!(snapshot.status, RunStatus::Succeeded));
818 assert!(snapshot.interactions.is_empty());
819 assert_eq!(snapshot.state.values()["value"], value);
820 assert_eq!(snapshot.state.values()["result"], value + 1);
821 assert_eq!(snapshot.last_action_ordinal, Some(0));
822 assert_eq!(snapshot.outcomes.len(), 1);
823 Ok::<_, AppError>((run.id().clone(), started.elapsed()))
824 })
825 .buffer_unordered(16)
826 );
827 tokio::select! {
828 _ = probe.wait_for(&probe.constructed, 16) => {}
829 result = runs.next() => panic!(
830 "run finished before the first 16 blocked sources: {:?}",
831 result.map(|result| result.map(|_| ())),
832 ),
833 }
834 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
835 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
836 probe.release_all();
837 let mut ids = BTreeSet::new();
838 let mut samples = Vec::with_capacity(100);
839 while let Some(result) = runs.next().await {
840 let (id, elapsed) = result?;
841 assert!(ids.insert(id));
842 samples.push(elapsed);
843 }
844 assert_eq!(ids.len(), 100);
845 assert_eq!(probe.constructed.load(Ordering::SeqCst), 100);
846 assert_eq!(probe.completed.load(Ordering::SeqCst), 100);
847 assert_eq!(probe.peak.load(Ordering::SeqCst), 16);
848 probe.assert_idle();
849 let rows = workflow
850 .list_runs(None, NonZeroUsize::new(128).unwrap())
851 .await?;
852 assert_eq!(rows.len(), 100);
853 assert_eq!(
854 rows.iter()
855 .map(|row| row.id.clone())
856 .collect::<BTreeSet<_>>(),
857 ids
858 );
859 assert!(
860 rows.iter()
861 .all(|row| matches!(row.status, RunStatus::Succeeded))
862 );
863 let elapsed = started.elapsed();
864 Ok(
865 json!({"dataset":"repeated100","runs":100,"caller_in_flight_limit":16,"admission":"sequential",
866 "production_active_limit":"unchanged; this measures only the caller's 16-journey envelope",
867 "execution_ns":elapsed.as_nanos(),"runs_per_second":100.0 / elapsed.as_secs_f64(),
868 "run_latency_including_admission":Self::samples(samples),"indefinite_stability_proof":false}),
869 )
870 }
871
872 async fn nested_cancel(
873 &self,
874 app: &App<StateStore, Agent, WorkloadAction>,
875 probe: &WorkloadProbe,
876 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
877 let workspace = app
878 .create_workspace(WorkspaceConfig {
879 name: None,
880 root: self.root.join("workspace"),
881 })
882 .await?;
883 let workflow = workspace.workflow("performance".parse()?).await?;
884 let run = workflow.start(ExecutionOptions::default()).await?;
885 tokio::select! {
886 _ = probe.wait_for(&probe.constructed, 16) => {}
887 result = run.wait() => panic!("nested run ended before its held leaf sources: {}", result.is_ok()),
888 }
889 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
890 let cancelling = Instant::now();
891 let snapshot = tokio::time::timeout(Duration::from_secs(30), async {
892 loop {
893 let before = run.snapshot().await?;
894 assert_eq!(before.state.values()["result"], 0);
895 assert!(before.outcomes.is_empty() && before.last_action_ordinal.is_none());
896 match run.cancel(before.checkpoint).await {
897 Ok(_) => break,
898 Err(AppError::Conflict(_)) => tokio::task::yield_now().await,
899 Err(error) => return Err(error),
900 }
901 }
902 run.wait().await
903 })
904 .await
905 .map_err(|_| {
906 AppError::Timeout("nested cancellation exceeded its declared 30-second grace".into())
907 })??;
908 let elapsed = cancelling.elapsed();
909 assert!(matches!(snapshot.status, RunStatus::Cancelled));
910 assert!(snapshot.interactions.is_empty() && snapshot.outcomes.is_empty());
911 assert!(snapshot.last_action_ordinal.is_none());
912 assert_eq!(snapshot.state.values()["result"], 0);
913 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
914 assert_eq!(probe.completed.load(Ordering::SeqCst), 0);
915 probe.assert_idle();
916 for (id, count) in [
917 ("performance", 1),
918 ("performance-mid", 2),
919 ("performance-leaf", 16),
920 ] {
921 let child = workspace.workflow(id.parse()?).await?;
922 let rows = child
923 .list_runs(None, NonZeroUsize::new(128).unwrap())
924 .await?;
925 assert_eq!(rows.len(), count);
926 for row in rows {
927 assert!(matches!(row.status, RunStatus::Cancelled));
928 assert!(row.outcomes.is_empty() && row.last_action_ordinal.is_none());
929 assert_eq!(row.state.values()["result"], 0);
930 }
931 }
932 Ok(
933 json!({"dataset":"nested-cancel","depth":2,"intermediate_children":2,"leaf_children":16,
934 "cooperative_cancel_grace_ms":30_000,"cancellation_latency":Self::samples(vec![elapsed]),
935 "source_replays":0,"caller_sources_dropped":16,"agent_unknown_replay":"not applicable: no Agent attempt"}),
936 )
937 }
938
939 async fn delayed_observer(
940 &self,
941 app: &App<StateStore, Agent, WorkloadAction>,
942 probe: &WorkloadProbe,
943 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
944 let workspace = app
945 .create_workspace(WorkspaceConfig {
946 name: None,
947 root: self.root.join("workspace"),
948 })
949 .await?;
950 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
951 let run = workflow.start(ExecutionOptions::default()).await?;
952 let observer = workflow
953 .open_run(RunSelector::Latest(run.id().clone()))
954 .await?;
955 // This public observation is deliberately never polled while the source
956 // produces its 256 partials and commits. It owns no execution work.
957 let delayed = observer.wait();
958 let produced = run.wait().await?;
959 assert!(matches!(produced.status, RunStatus::Succeeded));
960 assert_eq!(probe.completed.load(Ordering::SeqCst), 1);
961 probe.assert_idle();
962 tokio::time::sleep(Duration::from_secs(2)).await;
963 let observing = Instant::now();
964 let snapshot = delayed.await?;
965 let elapsed = observing.elapsed();
966 assert!(matches!(snapshot.status, RunStatus::Succeeded));
967 assert_eq!(snapshot.checkpoint, produced.checkpoint);
968 assert_eq!(snapshot.state.values()["result"], 1);
969 assert_eq!(snapshot.last_action_ordinal, Some(0));
970 assert_eq!(snapshot.outcomes.len(), 1);
971 assert_eq!(
972 snapshot.outcomes[0].result,
973 Map::from_iter([("value".into(), json!(1))])
974 );
975 assert_eq!(probe.constructed.load(Ordering::SeqCst), 1);
976 assert_eq!(probe.partials.load(Ordering::SeqCst), 256);
977 probe.assert_idle();
978 let terminal = run.snapshot().await?;
979 assert_eq!(terminal.checkpoint, snapshot.checkpoint);
980 assert_eq!(terminal.state, snapshot.state);
981 let mut cursor = None;
982 let mut actions = 0;
983 let mut rows = 0;
984 loop {
985 let page = run
986 .history(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
987 .await?;
988 for row in &page.checkpoints {
989 rows += 1;
990 assert!(rows <= 32);
991 if matches!(row.kind, CheckpointKind::Action) {
992 actions += 1;
993 assert_eq!(row.action_ordinal, Some(0));
994 assert_eq!(row.outcomes.len(), 1);
995 }
996 }
997 cursor = page.next;
998 if cursor.is_none() {
999 break;
1000 }
1001 }
1002 assert_eq!(
1003 actions, 1,
1004 "partials cannot become extra durable action commits"
1005 );
1006 Ok(
1007 json!({"dataset":"delayed-observer","unobserved_partial_events":256,"delay_after_terminal_confirmation_ms":2_000,
1008 "observation_latency":Self::samples(vec![elapsed]),"retained_rows":rows,
1009 "memory_scope":"external process RSS; this consumer retains no partial payloads"}),
1010 )
1011 }More examples
examples/history_and_recovery.rs (line 112)
63 async fn create_fork(
64 &self,
65 app: &App<StateStore, Agent>,
66 ) -> Result<(WorkspaceId, RunSelector, RunSelector), Box<dyn Error + Send + Sync>> {
67 let workspace = app
68 .create_workspace(WorkspaceConfig {
69 name: Some("history example".into()),
70 root: std::env::current_dir()?,
71 })
72 .await?;
73 let workflow = workspace.workflow("history-counter".parse()?).await?;
74 let mut source = workflow.prepare(WorkflowState::default()).await?;
75 let completed = source
76 .execution(ExecutionOptions::default())
77 .await?
78 .finish()
79 .await?;
80 if !matches!(completed.status, RunStatus::Succeeded)
81 || completed.state.values()["count"] != 2
82 {
83 return Err(
84 AppError::Integration("source did not complete both increments".into()).into(),
85 );
86 }
87
88 // An action ordinal selects the whole committed boundary containing it;
89 // it is independent of the checkpoint sequence and storage revision.
90 let first = source.checkpoint(CheckpointSelector::Action(0)).await?;
91 if !matches!(first.kind, CheckpointKind::Action)
92 || first.position.as_deref() != Some("first")
93 || first.state.values()["count"] != 1
94 {
95 return Err(
96 AppError::Integration("first action history is inconsistent".into()).into(),
97 );
98 }
99 let source_selector = RunSelector::Exact {
100 id: source.id().clone(),
101 occurrence: first.occurrence.clone(),
102 };
103 let mut fork = workflow
104 .restore(RestoreOptions {
105 source: source_selector.clone(),
106 checkpoint: CheckpointSelector::Action(0),
107 overrides: WorkflowState::new(Map::from_iter([("count".into(), json!(40))]))?,
108 position: ResumePosition::AfterCompleted,
109 fallback: None,
110 })
111 .await?;
112 let parked = fork.snapshot().await?;
113 if !matches!(parked.status, RunStatus::Paused)
114 || parked.checkpoint != 0
115 || parked.state.values()["count"] != 40
116 || !parked.outcomes.is_empty()
117 {
118 return Err(
119 AppError::Integration("restore did not create a parked clean copy".into()).into(),
120 );
121 }
122
123 source.delete(completed.checkpoint).await?;
124 if !matches!(source.snapshot().await, Err(AppError::NotFound(_))) {
125 return Err(
126 AppError::Integration("explicit source deletion did not finish".into()).into(),
127 );
128 }
129 // Deletion does not invalidate the copied state. AfterCompleted skips
130 // the first increment, so only the second changes the override: 40 -> 41.
131 let completed = fork
132 .execution(ExecutionOptions::default())
133 .await?
134 .finish()
135 .await?;
136 if !matches!(completed.status, RunStatus::Succeeded)
137 || completed.state.values()["count"] != 41
138 || completed.last_action_ordinal != Some(1)
139 {
140 return Err(
141 AppError::Integration("fork resumed from the wrong boundary".into()).into(),
142 );
143 }
144 println!("Source count: 2; copied count: 40; completed fork count: 41");
145 Ok((
146 workspace.id().clone(),
147 source_selector,
148 RunSelector::Exact {
149 id: fork.id().clone(),
150 occurrence: completed.occurrence,
151 },
152 ))
153 }Sourcepub async fn checkpoint(
&self,
selector: CheckpointSelector,
) -> Result<Checkpoint, AppError>
pub async fn checkpoint( &self, selector: CheckpointSelector, ) -> Result<Checkpoint, AppError>
Reads one immutable checkpoint without advancing the workflow.
An action selector returns the latest successful or failed observation of that ordinal. Missing or explicitly deleted history returns NotFound.
§Errors
Returns Closed for revoked App ownership or sealed/deleted scope, NotFound
for absent retained evidence, or Integration for inconsistent or unreadable
records. A newer App generation can invalidate this handle.
An explicit missing sequence/action is NotFound; action ordinals are distinct
from checkpoint sequences and select the latest matching saved observation.
§Example
let terminal = run.execution(ExecutionOptions::default()).await?.finish().await?;
let initial = run.checkpoint(CheckpointSelector::Sequence(0)).await?;
let action = run.checkpoint(CheckpointSelector::Action(0)).await?;
assert_eq!(initial.state.values()["count"], 0);
assert_eq!(action.state.values()["count"], 1);Examples found in repository?
examples/history_and_recovery.rs (line 90)
63 async fn create_fork(
64 &self,
65 app: &App<StateStore, Agent>,
66 ) -> Result<(WorkspaceId, RunSelector, RunSelector), Box<dyn Error + Send + Sync>> {
67 let workspace = app
68 .create_workspace(WorkspaceConfig {
69 name: Some("history example".into()),
70 root: std::env::current_dir()?,
71 })
72 .await?;
73 let workflow = workspace.workflow("history-counter".parse()?).await?;
74 let mut source = workflow.prepare(WorkflowState::default()).await?;
75 let completed = source
76 .execution(ExecutionOptions::default())
77 .await?
78 .finish()
79 .await?;
80 if !matches!(completed.status, RunStatus::Succeeded)
81 || completed.state.values()["count"] != 2
82 {
83 return Err(
84 AppError::Integration("source did not complete both increments".into()).into(),
85 );
86 }
87
88 // An action ordinal selects the whole committed boundary containing it;
89 // it is independent of the checkpoint sequence and storage revision.
90 let first = source.checkpoint(CheckpointSelector::Action(0)).await?;
91 if !matches!(first.kind, CheckpointKind::Action)
92 || first.position.as_deref() != Some("first")
93 || first.state.values()["count"] != 1
94 {
95 return Err(
96 AppError::Integration("first action history is inconsistent".into()).into(),
97 );
98 }
99 let source_selector = RunSelector::Exact {
100 id: source.id().clone(),
101 occurrence: first.occurrence.clone(),
102 };
103 let mut fork = workflow
104 .restore(RestoreOptions {
105 source: source_selector.clone(),
106 checkpoint: CheckpointSelector::Action(0),
107 overrides: WorkflowState::new(Map::from_iter([("count".into(), json!(40))]))?,
108 position: ResumePosition::AfterCompleted,
109 fallback: None,
110 })
111 .await?;
112 let parked = fork.snapshot().await?;
113 if !matches!(parked.status, RunStatus::Paused)
114 || parked.checkpoint != 0
115 || parked.state.values()["count"] != 40
116 || !parked.outcomes.is_empty()
117 {
118 return Err(
119 AppError::Integration("restore did not create a parked clean copy".into()).into(),
120 );
121 }
122
123 source.delete(completed.checkpoint).await?;
124 if !matches!(source.snapshot().await, Err(AppError::NotFound(_))) {
125 return Err(
126 AppError::Integration("explicit source deletion did not finish".into()).into(),
127 );
128 }
129 // Deletion does not invalidate the copied state. AfterCompleted skips
130 // the first increment, so only the second changes the override: 40 -> 41.
131 let completed = fork
132 .execution(ExecutionOptions::default())
133 .await?
134 .finish()
135 .await?;
136 if !matches!(completed.status, RunStatus::Succeeded)
137 || completed.state.values()["count"] != 41
138 || completed.last_action_ordinal != Some(1)
139 {
140 return Err(
141 AppError::Integration("fork resumed from the wrong boundary".into()).into(),
142 );
143 }
144 println!("Source count: 2; copied count: 40; completed fork count: 41");
145 Ok((
146 workspace.id().clone(),
147 source_selector,
148 RunSelector::Exact {
149 id: fork.id().clone(),
150 occurrence: completed.occurrence,
151 },
152 ))
153 }
154
155 async fn read_reopened(
156 &self,
157 app: &App<StateStore, Agent>,
158 workspace: WorkspaceId,
159 source: RunSelector,
160 fork: RunSelector,
161 ) -> Result<(), AppError> {
162 let workspace = app.workspace(workspace).await?;
163 let workflow = workspace.workflow("history-counter".parse()?).await?;
164 let fork = workflow.open_run(fork).await?;
165 let committed = fork.checkpoint(CheckpointSelector::Action(1)).await?;
166 let lineage = committed
167 .lineage
168 .as_ref()
169 .ok_or_else(|| AppError::Integration("reopened fork lost its source lineage".into()))?;
170 if committed.state.values()["count"] != 41
171 || !matches!(&source, RunSelector::Exact { id, occurrence }
172 if &lineage.source == id && &lineage.source_occurrence == occurrence)
173 || lineage.action_ordinal != Some(0)
174 {
175 return Err(AppError::Integration(
176 "reopened fork history changed".into(),
177 ));
178 }
179 println!(
180 "Reopened fork {} at count 41; source lineage remains readable",
181 fork.id()
182 );
183 Ok(())
184 }Sourcepub async fn history(
&self,
after: Option<HistoryCursor>,
limit: NonZeroUsize,
) -> Result<HistoryPage, AppError>
pub async fn history( &self, after: Option<HistoryCursor>, limit: NonZeroUsize, ) -> Result<HistoryPage, AppError>
Reads a bounded page from a fixed retained-history cutoff.
Limits are 1 through 1,000. A returned cursor preserves its cutoff and is valid only for this run creation and App generation. Reading a page never invokes actions, state typing, codecs or lower-agent recovery.
§Errors
Returns Closed for revoked App ownership or sealed/deleted scope, NotFound
for absent retained evidence, or Integration for inconsistent or unreadable
records. A newer App generation can invalidate this handle.
An excessive page limit or a cursor from another run/occurrence/generation is
InvalidInput. A captured cutoff never includes later appended checkpoints.
§Example
let terminal = run.execution(ExecutionOptions::default()).await?.finish().await?;
let page = run.history(None, std::num::NonZeroUsize::new(1).unwrap()).await?;
assert_eq!(page.checkpoints[0].sequence, 0);
let next = run.history(page.next, std::num::NonZeroUsize::new(100).unwrap()).await?;
assert!(next.checkpoints.iter().all(|checkpoint| checkpoint.sequence > 0));Examples found in repository?
examples/performance.rs (line 290)
265 async fn history(
266 &self,
267 run: &WorkflowRun<StateStore, Agent, WorkloadAction>,
268 actions: u64,
269 linear: bool,
270 terminal_sequence: u64,
271 ) -> Result<Value, AppError> {
272 let ceiling = actions * 16 + 16;
273 let mut cursor = None;
274 let mut prior_sequence = None;
275 let mut rows = 0_u64;
276 let mut observed_actions = 0_u64;
277 let mut initial = 0;
278 let mut terminal = 0;
279 let mut controls = 0;
280 let mut halts = 0;
281 let mut samples = Vec::new();
282 let started = Instant::now();
283 loop {
284 assert!(
285 samples.len() < ceiling.div_ceil(PAGE_SIZE as u64) as usize + 1,
286 "history page ceiling exhausted without cursor completion"
287 );
288 let page_started = Instant::now();
289 let page = run
290 .history(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
291 .await?;
292 samples.push(page_started.elapsed());
293 assert!(page.checkpoints.len() <= PAGE_SIZE);
294 assert!(
295 !page.checkpoints.is_empty() || page.next.is_none(),
296 "history cursor made no progress"
297 );
298 for row in &page.checkpoints {
299 assert_eq!(&row.run, run.id());
300 assert!(prior_sequence.is_none_or(|prior| row.sequence > prior));
301 prior_sequence = Some(row.sequence);
302 rows += 1;
303 assert!(
304 rows <= ceiling,
305 "declared retained-row workload ceiling exceeded"
306 );
307 match row.kind {
308 CheckpointKind::Initial => {
309 initial += 1;
310 assert_eq!(observed_actions, 0);
311 assert_eq!(row.sequence, 0);
312 }
313 CheckpointKind::Action => {
314 assert_eq!(row.outcomes.len(), 1);
315 let outcome = &row.outcomes[0];
316 assert_eq!(outcome.ordinal, observed_actions);
317 assert_eq!(
318 outcome.node,
319 if linear {
320 format!("n{observed_actions:04}")
321 } else {
322 "count".into()
323 }
324 );
325 assert!(outcome.result.is_empty());
326 observed_actions += 1;
327 assert_eq!(row.action_ordinal, Some(observed_actions - 1));
328 }
329 CheckpointKind::Terminal => {
330 terminal += 1;
331 assert!(matches!(row.status, RunStatus::Succeeded));
332 }
333 CheckpointKind::Control => {
334 assert!(row.outcomes.is_empty());
335 controls += 1;
336 if let RunStatus::Halted { node, result } = &row.status {
337 assert!(!linear);
338 assert_eq!(row.sequence, 4);
339 assert_eq!(node, "count");
340 assert_eq!(result, &Some(Map::new()));
341 halts += 1;
342 }
343 }
344 CheckpointKind::Failed { .. } => {
345 panic!("workload history contains a failed action")
346 }
347 }
348 assert_eq!(
349 row.state.values(),
350 &Map::from_iter([("count".into(), json!(observed_actions))])
351 );
352 if !linear && matches!(row.sequence, 4 | 5) {
353 assert!(matches!(row.kind, CheckpointKind::Control));
354 assert_eq!(observed_actions, 1);
355 assert_eq!(row.action_ordinal, Some(0));
356 assert_eq!(row.position.as_deref(), Some("count"));
357 if row.sequence == 5 {
358 assert!(matches!(row.status, RunStatus::Paused));
359 }
360 }
361 }
362 cursor = page.next;
363 // Each complete Checkpoint/WorkflowState is released with this page.
364 if cursor.is_none() {
365 break;
366 }
367 }
368 assert_eq!(initial, 1);
369 assert_eq!(terminal, 1);
370 assert_eq!(observed_actions, actions);
371 assert_eq!(prior_sequence, Some(terminal_sequence));
372 if !linear {
373 assert_eq!(rows, actions * 3 + 4);
374 assert_eq!(terminal_sequence, rows - 1);
375 assert_eq!(controls, actions * 2 + 2);
376 assert_eq!(halts, 1);
377 assert_eq!(samples.len(), (rows as usize).div_ceil(PAGE_SIZE));
378 }
379 let elapsed = started.elapsed();
380 let first = samples.first().copied().unwrap_or_default();
381 let subsequent: Duration = samples.iter().skip(1).copied().sum();
382 Ok(
383 json!({"rows":rows,"action_rows":observed_actions,"pages":samples.len(),
384 "elapsed_ns":elapsed.as_nanos(),"first_page_ns":first.as_nanos(),
385 "subsequent_pages_total_ns":subsequent.as_nanos(),
386 "rows_per_second":rows as f64 / elapsed.as_secs_f64(),"page_latency":Self::samples(samples)}),
387 )
388 }
389
390 async fn seed(
391 &self,
392 app: &App<StateStore, Agent, WorkloadAction>,
393 actions: u64,
394 linear: bool,
395 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
396 let workspace = app
397 .create_workspace(WorkspaceConfig {
398 name: Some("performance consumer".into()),
399 root: self.root.join("workspace"),
400 })
401 .await?;
402 let workflow = workspace.workflow("performance".parse()?).await?;
403 let mut run = workflow.prepare(WorkflowState::default()).await?;
404 let mut execution = run
405 .execution(ExecutionOptions {
406 halt_after: if linear {
407 Vec::new()
408 } else {
409 vec!["count".into()]
410 },
411 ..ExecutionOptions::default()
412 })
413 .await?;
414 let mut samples = Vec::with_capacity(actions as usize);
415 let started = Instant::now();
416 for ordinal in 0..actions {
417 let step_started = Instant::now();
418 let event = execution.step().await?;
419 samples.push(step_started.elapsed());
420 let Some(ExecutionEvent::Committed { outcomes, .. }) = event else {
421 panic!("each serial workload step must commit one action");
422 };
423 assert_eq!(outcomes.len(), 1);
424 assert_eq!(outcomes[0].ordinal, ordinal);
425 assert_eq!(
426 outcomes[0].node,
427 if linear {
428 format!("n{ordinal:04}")
429 } else {
430 "count".into()
431 }
432 );
433 assert!(outcomes[0].result.is_empty());
434 if !linear && ordinal == 0 {
435 // One public halt and explicit resume add two control rows to
436 // Initial + three checkpoints per action + Terminal.
437 let halted = execution.finish().await?;
438 assert!(matches!(&halted.status, RunStatus::Halted { node, result }
439 if node == "count" && result == &Some(Map::new())));
440 assert_eq!(halted.checkpoint, 4);
441 assert_eq!(
442 halted.state.values(),
443 &Map::from_iter([("count".into(), json!(1))])
444 );
445 assert_eq!(halted.last_action_ordinal, Some(0));
446 assert!(halted.outcomes == outcomes);
447 assert!(halted.interactions.is_empty());
448 execution = run.execution(ExecutionOptions::default()).await?;
449 }
450 }
451 let snapshot = execution.finish().await?;
452 Self::verify_snapshot(&snapshot, actions);
453 let execution_elapsed = started.elapsed();
454 let history = self
455 .history(&run, actions, linear, snapshot.checkpoint)
456 .await?;
457 if let Command::SeedHistory { rows } = self.command {
458 assert_eq!(history["rows"], rows);
459 }
460 Ok(
461 json!({"dataset":if linear {"linear"} else {"history"},"actions":actions,
462 "workspace":workspace.id().to_string(),"run":run.id().to_string(),
463 "occurrence":snapshot.occurrence,
464 "expected_state":snapshot.state.values(),"last_action_ordinal":snapshot.last_action_ordinal,
465 "terminal_checkpoint":snapshot.checkpoint,"retained_rows":history["rows"],
466 "execution_ns":execution_elapsed.as_nanos(),
467 "actions_per_second":actions as f64 / execution_elapsed.as_secs_f64(),
468 "step_latency":Self::samples(samples),"history":history}),
469 )
470 }
471
472 async fn reopen(
473 &self,
474 app: &App<StateStore, Agent, WorkloadAction>,
475 manifest: &Value,
476 actions: u64,
477 open_only: bool,
478 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
479 let started = Instant::now();
480 let workspace: WorkspaceId = manifest["workspace"]
481 .as_str()
482 .ok_or("missing workspace ID")?
483 .parse()?;
484 let run_id: RunId = manifest["run"].as_str().ok_or("missing run ID")?.parse()?;
485 let occurrence = manifest["occurrence"]
486 .as_str()
487 .ok_or("missing run occurrence")?
488 .to_owned();
489 let workspace = app.workspace(workspace).await?;
490 let workflow = workspace.workflow("performance".parse()?).await?;
491 let run = workflow
492 .open_run(RunSelector::Exact {
493 id: run_id,
494 occurrence: occurrence.clone(),
495 })
496 .await?;
497 let snapshot = run.snapshot().await?;
498 Self::verify_snapshot(&snapshot, actions);
499 assert_eq!(snapshot.occurrence, occurrence);
500 assert_eq!(json!(snapshot.checkpoint), manifest["terminal_checkpoint"]);
501 assert_eq!(json!(snapshot.state.values()), manifest["expected_state"]);
502 assert_eq!(
503 json!(snapshot.last_action_ordinal),
504 manifest["last_action_ordinal"]
505 );
506 let lookup = started.elapsed();
507 if open_only {
508 return Ok(
509 json!({"dataset":"history","actions":actions,"mode":"open-only",
510 "lookup_and_snapshot_ns":lookup.as_nanos(),
511 "seed_retained_rows":manifest["retained_rows"],"observed_history_rows":null}),
512 );
513 }
514 let history = self
515 .history(&run, actions, false, snapshot.checkpoint)
516 .await?;
517 assert_eq!(history["rows"], manifest["retained_rows"]);
518 Ok(
519 json!({"dataset":"history","actions":actions,"mode":"full-history",
520 "lookup_and_snapshot_ns":lookup.as_nanos(),"history":history}),
521 )
522 }
523
524 fn definitions(&self, actions: u64) -> Result<(Vec<WorkflowDefinition>, usize), AppError> {
525 if matches!(
526 self.command,
527 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. }
528 ) {
529 let (definition, bytes) =
530 Self::definition(matches!(self.command, Command::Linear), actions)?;
531 return Ok((vec![definition], bytes));
532 }
533 let leaf = json!({
534 "schema":"app-kit.workflow/v2","id":"performance-leaf","revision":"r1",
535 "state":{"value":0,"result":0},"required":["value","result"],
536 "reducers":{"value":"reject_concurrent","result":"reject_concurrent"},
537 "entry":"work","nodes":[{"id":"work","deadline_ms":NODE_DEADLINE_MS,
538 "step":{"kind":"action","action":"performance-action","revision":"r1"}}],"routes":{}
539 });
540 let documents = match self.command {
541 Command::Ready => {
542 let mut nodes = vec![json!({"id":"fork","step":{"kind":"state","mutations":[]}})];
543 let mut branches = Vec::new();
544 for index in 0..32 {
545 let node = format!("n{index:02}");
546 nodes.push(json!({"id":node,"deadline_ms":NODE_DEADLINE_MS,
547 "step":{"kind":"action","action":"performance-action","revision":"r1"}}));
548 branches.push(json!({"target":{"kind":"node","id":node}}));
549 }
550 vec![json!({
551 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
552 "state":{"value":0,"result":0},"required":["value","result"],
553 "reducers":{"value":"reject_concurrent","result":"collect"},
554 "entry":"fork","nodes":nodes,
555 "routes":{"fork":{"mode":"broadcast","branches":branches}}
556 })]
557 }
558 Command::Map | Command::NestedCancel => {
559 let nested = matches!(self.command, Command::NestedCancel);
560 let items = if nested { 2 } else { 4_096 };
561 let mut documents = vec![
562 json!({
563 "schema":"app-kit.workflow/v2","id":"performance","revision":"r1",
564 "state":{"value":0,"result":0},"required":["value","result"],
565 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
566 "entry":"map","nodes":[{"id":"map","deadline_ms":if nested {NODE_DEADLINE_MS} else {WORKLOAD_BUDGET.as_millis() as u64},
567 "step":{"kind":"map","workflow":if nested {"performance-mid"} else {"performance-leaf"},
568 "items":{"kind":"literal","value":(0..items).collect::<Vec<_>>()},
569 "input_key":"value","max_parallel":if nested {2} else {8},
570 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
571 }),
572 leaf,
573 ];
574 if nested {
575 documents.push(json!({
576 "schema":"app-kit.workflow/v2","id":"performance-mid","revision":"r1",
577 "state":{"value":0,"result":0},"required":["value","result"],
578 "reducers":{"value":"reject_concurrent","result":"sum_integer"},
579 "entry":"map","nodes":[{"id":"map","deadline_ms":NODE_DEADLINE_MS,
580 "step":{"kind":"map","workflow":"performance-leaf",
581 "items":{"kind":"literal","value":(0..8).collect::<Vec<_>>()},
582 "input_key":"value","max_parallel":8,
583 "reductions":[{"source":"result","target":"result","reducer":"sum_integer"}]}}],"routes":{}
584 }));
585 }
586 documents
587 }
588 Command::Repeated | Command::DelayedObserver => vec![leaf],
589 Command::Linear | Command::SeedHistory { .. } | Command::ReopenHistory { .. } => {
590 unreachable!()
591 }
592 };
593 let mut bytes = 0;
594 let mut definitions = Vec::with_capacity(documents.len());
595 for document in documents {
596 let source = document.to_string();
597 assert!(source.len() <= 1_048_576);
598 bytes += source.len();
599 definitions.push(WorkflowDefinition::parse(&source)?);
600 }
601 Ok((definitions, bytes))
602 }
603
604 async fn ready(
605 &self,
606 app: &App<StateStore, Agent, WorkloadAction>,
607 probe: &WorkloadProbe,
608 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
609 let workspace = app
610 .create_workspace(WorkspaceConfig {
611 name: None,
612 root: self.root.join("workspace"),
613 })
614 .await?;
615 let workflow = workspace.workflow("performance".parse()?).await?;
616 let mut waves = Vec::with_capacity(32);
617 let mut releases = Vec::with_capacity(32 * 31);
618 let mut barriers = Vec::with_capacity(32);
619 let started = Instant::now();
620 for permutation in 0..32 {
621 probe.assert_idle();
622 let mut run = workflow.prepare(WorkflowState::default()).await?;
623 let observer = workflow
624 .open_run(RunSelector::Latest(run.id().clone()))
625 .await?;
626 let mut execution = run.execution(ExecutionOptions::default()).await?;
627 let Some(ExecutionEvent::Committed { outcomes, .. }) = execution.step().await? else {
628 panic!("the native fork must commit before the ready wave");
629 };
630 assert_eq!(outcomes.len(), 1);
631 assert_eq!(
632 (outcomes[0].node.as_str(), outcomes[0].ordinal),
633 ("fork", 0)
634 );
635 let wave = Instant::now();
636 let mut finish = Box::pin(execution.finish());
637 tokio::select! {
638 _ = probe.wait_for(&probe.constructed, (permutation + 1) * 32) => {}
639 result = &mut finish => panic!("gated ready wave finished before admission: {}", result.is_ok()),
640 }
641 assert_eq!(probe.active.load(Ordering::SeqCst), 32);
642 let order = (0..32)
643 .map(|offset| (permutation + 31 - offset) % 32)
644 .collect::<Vec<_>>();
645 for (released, index) in order.iter().take(31).enumerate() {
646 let releasing = Instant::now();
647 probe.release(*index);
648 tokio::select! {
649 _ = probe.wait_for(&probe.completed, permutation * 32 + released + 1) => {}
650 result = &mut finish => panic!("a partial ready wave committed: {}", result.is_ok()),
651 }
652 releases.push(releasing.elapsed());
653 let visible = observer.snapshot().await?;
654 assert_eq!(
655 visible.state.values()["result"],
656 0,
657 "staged siblings cannot expose a partial barrier"
658 );
659 assert_eq!(visible.last_action_ordinal, Some(0));
660 }
661 let barrier = Instant::now();
662 probe.release(order[31]);
663 let snapshot = finish.await?;
664 barriers.push(barrier.elapsed());
665 waves.push(wave.elapsed());
666 assert!(matches!(snapshot.status, RunStatus::Succeeded));
667 assert!(snapshot.interactions.is_empty());
668 assert_eq!(
669 snapshot.state.values()["result"],
670 json!((1..=32).collect::<Vec<_>>())
671 );
672 assert_eq!(snapshot.last_action_ordinal, Some(32));
673 assert_eq!(snapshot.outcomes.len(), 32);
674 for (index, outcome) in snapshot.outcomes.iter().enumerate() {
675 assert_eq!(outcome.node, format!("n{index:02}"));
676 assert_eq!(outcome.ordinal, index as u64 + 1);
677 assert_eq!(
678 outcome.result,
679 Map::from_iter([("value".into(), json!(index + 1))])
680 );
681 }
682 assert_eq!(std::mem::take(&mut *probe.order.lock().unwrap()), order);
683 assert_eq!(
684 probe.completed.load(Ordering::SeqCst),
685 (permutation + 1) * 32
686 );
687 probe.assert_idle();
688 }
689 assert_eq!(probe.peak.load(Ordering::SeqCst), 32);
690 let elapsed = started.elapsed();
691 Ok(
692 json!({"dataset":"ready32","permutations":32,"ready_actions_per_wave":32,
693 "caller_action_commits":1024,"native_fork_commits":32,
694 "execution_ns":elapsed.as_nanos(),"caller_actions_per_second":1024.0 / elapsed.as_secs_f64(),
695 "wave_latency_including_snapshot_probes":Self::samples(waves),
696 "release_to_source_completion_latency":Self::samples(releases),
697 "last_release_to_committed_barrier_latency":Self::samples(barriers)}),
698 )
699 }
700
701 async fn mapped(
702 &self,
703 app: &App<StateStore, Agent, WorkloadAction>,
704 probe: &WorkloadProbe,
705 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
706 let workspace = app
707 .create_workspace(WorkspaceConfig {
708 name: None,
709 root: self.root.join("workspace"),
710 })
711 .await?;
712 let workflow = workspace.workflow("performance".parse()?).await?;
713 let mut run = workflow.prepare(WorkflowState::default()).await?;
714 let observer = workflow
715 .open_run(RunSelector::Latest(run.id().clone()))
716 .await?;
717 let started = Instant::now();
718 let mut finish = Box::pin(run.execution(ExecutionOptions::default()).await?.finish());
719 tokio::select! {
720 _ = probe.wait_for(&probe.constructed, 8) => {}
721 result = &mut finish => panic!("map finished before its first eight blocked sources: {}", result.is_ok()),
722 }
723 assert_eq!(probe.active.load(Ordering::SeqCst), 8);
724 assert_eq!(probe.constructed.load(Ordering::SeqCst), 8);
725 let unchanged = observer.snapshot().await?;
726 assert_eq!(unchanged.state.values()["result"], 0);
727 assert!(unchanged.last_action_ordinal.is_none());
728 probe.release_all();
729 let snapshot = finish.await?;
730 let elapsed = started.elapsed();
731 assert!(matches!(snapshot.status, RunStatus::Succeeded));
732 assert!(snapshot.interactions.is_empty());
733 assert_eq!(snapshot.state.values()["result"], 4_096_u64 * 4_097 / 2);
734 assert_eq!(snapshot.last_action_ordinal, Some(0));
735 assert_eq!(snapshot.outcomes.len(), 1);
736 assert_eq!(snapshot.outcomes[0].node, "map");
737 assert_eq!(probe.constructed.load(Ordering::SeqCst), 4_096);
738 assert_eq!(probe.completed.load(Ordering::SeqCst), 4_096);
739 assert_eq!(probe.peak.load(Ordering::SeqCst), 8);
740 probe.assert_idle();
741 let child = workspace.workflow("performance-leaf".parse()?).await?;
742 let mut cursor = None;
743 let mut values = BTreeSet::new();
744 let verifying = Instant::now();
745 loop {
746 let page = child
747 .list_runs(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
748 .await?;
749 assert!(page.len() <= PAGE_SIZE);
750 if page.is_empty() {
751 break;
752 }
753 for row in &page {
754 assert!(matches!(row.status, RunStatus::Succeeded));
755 let value = row.state.values()["value"].as_u64().unwrap();
756 assert!(
757 value < 4_096 && values.insert(value),
758 "every mapped input has exactly one retained child"
759 );
760 assert_eq!(row.state.values()["result"], value + 1);
761 assert_eq!(row.last_action_ordinal, Some(0));
762 assert_eq!(row.outcomes.len(), 1);
763 assert_eq!(row.outcomes[0].node, "work");
764 }
765 let last = page.last().unwrap();
766 cursor = Some(RunSelector::Exact {
767 id: last.id.clone(),
768 occurrence: last.occurrence.clone(),
769 });
770 assert!(values.len() <= 4_096);
771 }
772 assert_eq!(values.len(), 4_096);
773 Ok(json!({"dataset":"map4096","items":4096,"max_parallel":8,
774 "parent_node_deadline_ms":WORKLOAD_BUDGET.as_millis(),
775 "execution_ns":elapsed.as_nanos(),"child_actions_per_second":4096.0 / elapsed.as_secs_f64(),
776 "map_completion_latency":Self::samples(vec![elapsed]),
777 "child_inventory_verification_ns":verifying.elapsed().as_nanos(),"verified_children":values.len()}))
778 }
779
780 async fn repeated(
781 &self,
782 app: &App<StateStore, Agent, WorkloadAction>,
783 probe: &WorkloadProbe,
784 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
785 let workspace = app
786 .create_workspace(WorkspaceConfig {
787 name: None,
788 root: self.root.join("workspace"),
789 })
790 .await?;
791 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
792 let started = Instant::now();
793 // Admit one run at a time, then directly poll at most 16 original
794 // executions together. No consumer task is spawned or detached.
795 let mut runs = std::pin::pin!(
796 stream::iter(0..100_u64)
797 .then(|value| {
798 let workflow = &workflow;
799 async move {
800 let started = Instant::now();
801 let run = workflow
802 .prepare(WorkflowState::new(Map::from_iter([(
803 "value".into(),
804 json!(value),
805 )]))?)
806 .await?;
807 Ok::<_, AppError>((value, run, started))
808 }
809 })
810 .map(|prepared| async move {
811 let (value, mut run, started) = prepared?;
812 let snapshot = run
813 .execution(ExecutionOptions::default())
814 .await?
815 .finish()
816 .await?;
817 assert!(matches!(snapshot.status, RunStatus::Succeeded));
818 assert!(snapshot.interactions.is_empty());
819 assert_eq!(snapshot.state.values()["value"], value);
820 assert_eq!(snapshot.state.values()["result"], value + 1);
821 assert_eq!(snapshot.last_action_ordinal, Some(0));
822 assert_eq!(snapshot.outcomes.len(), 1);
823 Ok::<_, AppError>((run.id().clone(), started.elapsed()))
824 })
825 .buffer_unordered(16)
826 );
827 tokio::select! {
828 _ = probe.wait_for(&probe.constructed, 16) => {}
829 result = runs.next() => panic!(
830 "run finished before the first 16 blocked sources: {:?}",
831 result.map(|result| result.map(|_| ())),
832 ),
833 }
834 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
835 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
836 probe.release_all();
837 let mut ids = BTreeSet::new();
838 let mut samples = Vec::with_capacity(100);
839 while let Some(result) = runs.next().await {
840 let (id, elapsed) = result?;
841 assert!(ids.insert(id));
842 samples.push(elapsed);
843 }
844 assert_eq!(ids.len(), 100);
845 assert_eq!(probe.constructed.load(Ordering::SeqCst), 100);
846 assert_eq!(probe.completed.load(Ordering::SeqCst), 100);
847 assert_eq!(probe.peak.load(Ordering::SeqCst), 16);
848 probe.assert_idle();
849 let rows = workflow
850 .list_runs(None, NonZeroUsize::new(128).unwrap())
851 .await?;
852 assert_eq!(rows.len(), 100);
853 assert_eq!(
854 rows.iter()
855 .map(|row| row.id.clone())
856 .collect::<BTreeSet<_>>(),
857 ids
858 );
859 assert!(
860 rows.iter()
861 .all(|row| matches!(row.status, RunStatus::Succeeded))
862 );
863 let elapsed = started.elapsed();
864 Ok(
865 json!({"dataset":"repeated100","runs":100,"caller_in_flight_limit":16,"admission":"sequential",
866 "production_active_limit":"unchanged; this measures only the caller's 16-journey envelope",
867 "execution_ns":elapsed.as_nanos(),"runs_per_second":100.0 / elapsed.as_secs_f64(),
868 "run_latency_including_admission":Self::samples(samples),"indefinite_stability_proof":false}),
869 )
870 }
871
872 async fn nested_cancel(
873 &self,
874 app: &App<StateStore, Agent, WorkloadAction>,
875 probe: &WorkloadProbe,
876 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
877 let workspace = app
878 .create_workspace(WorkspaceConfig {
879 name: None,
880 root: self.root.join("workspace"),
881 })
882 .await?;
883 let workflow = workspace.workflow("performance".parse()?).await?;
884 let run = workflow.start(ExecutionOptions::default()).await?;
885 tokio::select! {
886 _ = probe.wait_for(&probe.constructed, 16) => {}
887 result = run.wait() => panic!("nested run ended before its held leaf sources: {}", result.is_ok()),
888 }
889 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
890 let cancelling = Instant::now();
891 let snapshot = tokio::time::timeout(Duration::from_secs(30), async {
892 loop {
893 let before = run.snapshot().await?;
894 assert_eq!(before.state.values()["result"], 0);
895 assert!(before.outcomes.is_empty() && before.last_action_ordinal.is_none());
896 match run.cancel(before.checkpoint).await {
897 Ok(_) => break,
898 Err(AppError::Conflict(_)) => tokio::task::yield_now().await,
899 Err(error) => return Err(error),
900 }
901 }
902 run.wait().await
903 })
904 .await
905 .map_err(|_| {
906 AppError::Timeout("nested cancellation exceeded its declared 30-second grace".into())
907 })??;
908 let elapsed = cancelling.elapsed();
909 assert!(matches!(snapshot.status, RunStatus::Cancelled));
910 assert!(snapshot.interactions.is_empty() && snapshot.outcomes.is_empty());
911 assert!(snapshot.last_action_ordinal.is_none());
912 assert_eq!(snapshot.state.values()["result"], 0);
913 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
914 assert_eq!(probe.completed.load(Ordering::SeqCst), 0);
915 probe.assert_idle();
916 for (id, count) in [
917 ("performance", 1),
918 ("performance-mid", 2),
919 ("performance-leaf", 16),
920 ] {
921 let child = workspace.workflow(id.parse()?).await?;
922 let rows = child
923 .list_runs(None, NonZeroUsize::new(128).unwrap())
924 .await?;
925 assert_eq!(rows.len(), count);
926 for row in rows {
927 assert!(matches!(row.status, RunStatus::Cancelled));
928 assert!(row.outcomes.is_empty() && row.last_action_ordinal.is_none());
929 assert_eq!(row.state.values()["result"], 0);
930 }
931 }
932 Ok(
933 json!({"dataset":"nested-cancel","depth":2,"intermediate_children":2,"leaf_children":16,
934 "cooperative_cancel_grace_ms":30_000,"cancellation_latency":Self::samples(vec![elapsed]),
935 "source_replays":0,"caller_sources_dropped":16,"agent_unknown_replay":"not applicable: no Agent attempt"}),
936 )
937 }
938
939 async fn delayed_observer(
940 &self,
941 app: &App<StateStore, Agent, WorkloadAction>,
942 probe: &WorkloadProbe,
943 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
944 let workspace = app
945 .create_workspace(WorkspaceConfig {
946 name: None,
947 root: self.root.join("workspace"),
948 })
949 .await?;
950 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
951 let run = workflow.start(ExecutionOptions::default()).await?;
952 let observer = workflow
953 .open_run(RunSelector::Latest(run.id().clone()))
954 .await?;
955 // This public observation is deliberately never polled while the source
956 // produces its 256 partials and commits. It owns no execution work.
957 let delayed = observer.wait();
958 let produced = run.wait().await?;
959 assert!(matches!(produced.status, RunStatus::Succeeded));
960 assert_eq!(probe.completed.load(Ordering::SeqCst), 1);
961 probe.assert_idle();
962 tokio::time::sleep(Duration::from_secs(2)).await;
963 let observing = Instant::now();
964 let snapshot = delayed.await?;
965 let elapsed = observing.elapsed();
966 assert!(matches!(snapshot.status, RunStatus::Succeeded));
967 assert_eq!(snapshot.checkpoint, produced.checkpoint);
968 assert_eq!(snapshot.state.values()["result"], 1);
969 assert_eq!(snapshot.last_action_ordinal, Some(0));
970 assert_eq!(snapshot.outcomes.len(), 1);
971 assert_eq!(
972 snapshot.outcomes[0].result,
973 Map::from_iter([("value".into(), json!(1))])
974 );
975 assert_eq!(probe.constructed.load(Ordering::SeqCst), 1);
976 assert_eq!(probe.partials.load(Ordering::SeqCst), 256);
977 probe.assert_idle();
978 let terminal = run.snapshot().await?;
979 assert_eq!(terminal.checkpoint, snapshot.checkpoint);
980 assert_eq!(terminal.state, snapshot.state);
981 let mut cursor = None;
982 let mut actions = 0;
983 let mut rows = 0;
984 loop {
985 let page = run
986 .history(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
987 .await?;
988 for row in &page.checkpoints {
989 rows += 1;
990 assert!(rows <= 32);
991 if matches!(row.kind, CheckpointKind::Action) {
992 actions += 1;
993 assert_eq!(row.action_ordinal, Some(0));
994 assert_eq!(row.outcomes.len(), 1);
995 }
996 }
997 cursor = page.next;
998 if cursor.is_none() {
999 break;
1000 }
1001 }
1002 assert_eq!(
1003 actions, 1,
1004 "partials cannot become extra durable action commits"
1005 );
1006 Ok(
1007 json!({"dataset":"delayed-observer","unobserved_partial_events":256,"delay_after_terminal_confirmation_ms":2_000,
1008 "observation_latency":Self::samples(vec![elapsed]),"retained_rows":rows,
1009 "memory_scope":"external process RSS; this consumer retains no partial payloads"}),
1010 )
1011 }Sourcepub async fn delete(&mut self, expected_checkpoint: u64) -> Result<(), AppError>
pub async fn delete(&mut self, expected_checkpoint: u64) -> Result<(), AppError>
Deletes terminal history through bounded, resumable pages.
The expected checkpoint fences the terminal boundary. Dropping this operation’s observer preserves admitted deletion and native commits.
§Errors
Returns Conflict for a stale checkpoint, Busy for nonterminal/unknown work
or a child still referenced by an active parent, and Closed after sealing or
revocation. Storage failures use Integration. A successful seal survives
observer loss and deletion resumes in bounded pages after reopen.
§Example
let terminal = run.execution(ExecutionOptions::default()).await?.finish().await?;
let id = run.id().clone();
run.delete(terminal.checkpoint).await?;
assert!(matches!(workflow.open_run(RunSelector::Latest(id)).await, Err(app_kit::AppError::NotFound(_))));Examples found in repository?
examples/history_and_recovery.rs (line 123)
63 async fn create_fork(
64 &self,
65 app: &App<StateStore, Agent>,
66 ) -> Result<(WorkspaceId, RunSelector, RunSelector), Box<dyn Error + Send + Sync>> {
67 let workspace = app
68 .create_workspace(WorkspaceConfig {
69 name: Some("history example".into()),
70 root: std::env::current_dir()?,
71 })
72 .await?;
73 let workflow = workspace.workflow("history-counter".parse()?).await?;
74 let mut source = workflow.prepare(WorkflowState::default()).await?;
75 let completed = source
76 .execution(ExecutionOptions::default())
77 .await?
78 .finish()
79 .await?;
80 if !matches!(completed.status, RunStatus::Succeeded)
81 || completed.state.values()["count"] != 2
82 {
83 return Err(
84 AppError::Integration("source did not complete both increments".into()).into(),
85 );
86 }
87
88 // An action ordinal selects the whole committed boundary containing it;
89 // it is independent of the checkpoint sequence and storage revision.
90 let first = source.checkpoint(CheckpointSelector::Action(0)).await?;
91 if !matches!(first.kind, CheckpointKind::Action)
92 || first.position.as_deref() != Some("first")
93 || first.state.values()["count"] != 1
94 {
95 return Err(
96 AppError::Integration("first action history is inconsistent".into()).into(),
97 );
98 }
99 let source_selector = RunSelector::Exact {
100 id: source.id().clone(),
101 occurrence: first.occurrence.clone(),
102 };
103 let mut fork = workflow
104 .restore(RestoreOptions {
105 source: source_selector.clone(),
106 checkpoint: CheckpointSelector::Action(0),
107 overrides: WorkflowState::new(Map::from_iter([("count".into(), json!(40))]))?,
108 position: ResumePosition::AfterCompleted,
109 fallback: None,
110 })
111 .await?;
112 let parked = fork.snapshot().await?;
113 if !matches!(parked.status, RunStatus::Paused)
114 || parked.checkpoint != 0
115 || parked.state.values()["count"] != 40
116 || !parked.outcomes.is_empty()
117 {
118 return Err(
119 AppError::Integration("restore did not create a parked clean copy".into()).into(),
120 );
121 }
122
123 source.delete(completed.checkpoint).await?;
124 if !matches!(source.snapshot().await, Err(AppError::NotFound(_))) {
125 return Err(
126 AppError::Integration("explicit source deletion did not finish".into()).into(),
127 );
128 }
129 // Deletion does not invalidate the copied state. AfterCompleted skips
130 // the first increment, so only the second changes the override: 40 -> 41.
131 let completed = fork
132 .execution(ExecutionOptions::default())
133 .await?
134 .finish()
135 .await?;
136 if !matches!(completed.status, RunStatus::Succeeded)
137 || completed.state.values()["count"] != 41
138 || completed.last_action_ordinal != Some(1)
139 {
140 return Err(
141 AppError::Integration("fork resumed from the wrong boundary".into()).into(),
142 );
143 }
144 println!("Source count: 2; copied count: 40; completed fork count: 41");
145 Ok((
146 workspace.id().clone(),
147 source_selector,
148 RunSelector::Exact {
149 id: fork.id().clone(),
150 occurrence: completed.occurrence,
151 },
152 ))
153 }Sourcepub async fn wait(&self) -> Result<RunSnapshot, AppError>
pub async fn wait(&self) -> Result<RunSnapshot, AppError>
Waits for the next actionable or terminal snapshot.
This observes accepted work; it does not acquire execution ownership or resume a prepared run. A caller interaction or unknown effect can return before a terminal result. Dropping the observation future never requests cancellation.
§Errors
Returns Closed for revoked App ownership or sealed/deleted scope, NotFound
for absent retained evidence, or Integration for inconsistent or unreadable
records. A newer App generation can invalidate this handle.
§Example
let background = workflow.start(ExecutionOptions::default()).await?;
let observed = background.wait().await?;
assert!(matches!(observed.status, RunStatus::Succeeded));Examples found in repository?
examples/performance.rs (line 887)
872 async fn nested_cancel(
873 &self,
874 app: &App<StateStore, Agent, WorkloadAction>,
875 probe: &WorkloadProbe,
876 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
877 let workspace = app
878 .create_workspace(WorkspaceConfig {
879 name: None,
880 root: self.root.join("workspace"),
881 })
882 .await?;
883 let workflow = workspace.workflow("performance".parse()?).await?;
884 let run = workflow.start(ExecutionOptions::default()).await?;
885 tokio::select! {
886 _ = probe.wait_for(&probe.constructed, 16) => {}
887 result = run.wait() => panic!("nested run ended before its held leaf sources: {}", result.is_ok()),
888 }
889 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
890 let cancelling = Instant::now();
891 let snapshot = tokio::time::timeout(Duration::from_secs(30), async {
892 loop {
893 let before = run.snapshot().await?;
894 assert_eq!(before.state.values()["result"], 0);
895 assert!(before.outcomes.is_empty() && before.last_action_ordinal.is_none());
896 match run.cancel(before.checkpoint).await {
897 Ok(_) => break,
898 Err(AppError::Conflict(_)) => tokio::task::yield_now().await,
899 Err(error) => return Err(error),
900 }
901 }
902 run.wait().await
903 })
904 .await
905 .map_err(|_| {
906 AppError::Timeout("nested cancellation exceeded its declared 30-second grace".into())
907 })??;
908 let elapsed = cancelling.elapsed();
909 assert!(matches!(snapshot.status, RunStatus::Cancelled));
910 assert!(snapshot.interactions.is_empty() && snapshot.outcomes.is_empty());
911 assert!(snapshot.last_action_ordinal.is_none());
912 assert_eq!(snapshot.state.values()["result"], 0);
913 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
914 assert_eq!(probe.completed.load(Ordering::SeqCst), 0);
915 probe.assert_idle();
916 for (id, count) in [
917 ("performance", 1),
918 ("performance-mid", 2),
919 ("performance-leaf", 16),
920 ] {
921 let child = workspace.workflow(id.parse()?).await?;
922 let rows = child
923 .list_runs(None, NonZeroUsize::new(128).unwrap())
924 .await?;
925 assert_eq!(rows.len(), count);
926 for row in rows {
927 assert!(matches!(row.status, RunStatus::Cancelled));
928 assert!(row.outcomes.is_empty() && row.last_action_ordinal.is_none());
929 assert_eq!(row.state.values()["result"], 0);
930 }
931 }
932 Ok(
933 json!({"dataset":"nested-cancel","depth":2,"intermediate_children":2,"leaf_children":16,
934 "cooperative_cancel_grace_ms":30_000,"cancellation_latency":Self::samples(vec![elapsed]),
935 "source_replays":0,"caller_sources_dropped":16,"agent_unknown_replay":"not applicable: no Agent attempt"}),
936 )
937 }
938
939 async fn delayed_observer(
940 &self,
941 app: &App<StateStore, Agent, WorkloadAction>,
942 probe: &WorkloadProbe,
943 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
944 let workspace = app
945 .create_workspace(WorkspaceConfig {
946 name: None,
947 root: self.root.join("workspace"),
948 })
949 .await?;
950 let workflow = workspace.workflow("performance-leaf".parse()?).await?;
951 let run = workflow.start(ExecutionOptions::default()).await?;
952 let observer = workflow
953 .open_run(RunSelector::Latest(run.id().clone()))
954 .await?;
955 // This public observation is deliberately never polled while the source
956 // produces its 256 partials and commits. It owns no execution work.
957 let delayed = observer.wait();
958 let produced = run.wait().await?;
959 assert!(matches!(produced.status, RunStatus::Succeeded));
960 assert_eq!(probe.completed.load(Ordering::SeqCst), 1);
961 probe.assert_idle();
962 tokio::time::sleep(Duration::from_secs(2)).await;
963 let observing = Instant::now();
964 let snapshot = delayed.await?;
965 let elapsed = observing.elapsed();
966 assert!(matches!(snapshot.status, RunStatus::Succeeded));
967 assert_eq!(snapshot.checkpoint, produced.checkpoint);
968 assert_eq!(snapshot.state.values()["result"], 1);
969 assert_eq!(snapshot.last_action_ordinal, Some(0));
970 assert_eq!(snapshot.outcomes.len(), 1);
971 assert_eq!(
972 snapshot.outcomes[0].result,
973 Map::from_iter([("value".into(), json!(1))])
974 );
975 assert_eq!(probe.constructed.load(Ordering::SeqCst), 1);
976 assert_eq!(probe.partials.load(Ordering::SeqCst), 256);
977 probe.assert_idle();
978 let terminal = run.snapshot().await?;
979 assert_eq!(terminal.checkpoint, snapshot.checkpoint);
980 assert_eq!(terminal.state, snapshot.state);
981 let mut cursor = None;
982 let mut actions = 0;
983 let mut rows = 0;
984 loop {
985 let page = run
986 .history(cursor, NonZeroUsize::new(PAGE_SIZE).unwrap())
987 .await?;
988 for row in &page.checkpoints {
989 rows += 1;
990 assert!(rows <= 32);
991 if matches!(row.kind, CheckpointKind::Action) {
992 actions += 1;
993 assert_eq!(row.action_ordinal, Some(0));
994 assert_eq!(row.outcomes.len(), 1);
995 }
996 }
997 cursor = page.next;
998 if cursor.is_none() {
999 break;
1000 }
1001 }
1002 assert_eq!(
1003 actions, 1,
1004 "partials cannot become extra durable action commits"
1005 );
1006 Ok(
1007 json!({"dataset":"delayed-observer","unobserved_partial_events":256,"delay_after_terminal_confirmation_ms":2_000,
1008 "observation_latency":Self::samples(vec![elapsed]),"retained_rows":rows,
1009 "memory_scope":"external process RSS; this consumer retains no partial payloads"}),
1010 )
1011 }Sourcepub async fn cancel(
&self,
expected_checkpoint: u64,
) -> Result<RunSnapshot, AppError>
pub async fn cancel( &self, expected_checkpoint: u64, ) -> Result<RunSnapshot, AppError>
Commits explicit checkpoint-fenced cancellation.
§Errors
Returns Conflict for a stale checkpoint, Busy for unavailable admission,
Closed after revocation, or Integration for unreadable/failed storage.
Cancellation admission is durable; observer loss does not reverse it.
§Example
let before = run.snapshot().await?;
run.cancel(before.checkpoint).await?;
let cancelled = run.wait().await?;
assert!(matches!(cancelled.status, RunStatus::Cancelled));Examples found in repository?
examples/performance.rs (line 896)
872 async fn nested_cancel(
873 &self,
874 app: &App<StateStore, Agent, WorkloadAction>,
875 probe: &WorkloadProbe,
876 ) -> Result<Value, Box<dyn Error + Send + Sync>> {
877 let workspace = app
878 .create_workspace(WorkspaceConfig {
879 name: None,
880 root: self.root.join("workspace"),
881 })
882 .await?;
883 let workflow = workspace.workflow("performance".parse()?).await?;
884 let run = workflow.start(ExecutionOptions::default()).await?;
885 tokio::select! {
886 _ = probe.wait_for(&probe.constructed, 16) => {}
887 result = run.wait() => panic!("nested run ended before its held leaf sources: {}", result.is_ok()),
888 }
889 assert_eq!(probe.active.load(Ordering::SeqCst), 16);
890 let cancelling = Instant::now();
891 let snapshot = tokio::time::timeout(Duration::from_secs(30), async {
892 loop {
893 let before = run.snapshot().await?;
894 assert_eq!(before.state.values()["result"], 0);
895 assert!(before.outcomes.is_empty() && before.last_action_ordinal.is_none());
896 match run.cancel(before.checkpoint).await {
897 Ok(_) => break,
898 Err(AppError::Conflict(_)) => tokio::task::yield_now().await,
899 Err(error) => return Err(error),
900 }
901 }
902 run.wait().await
903 })
904 .await
905 .map_err(|_| {
906 AppError::Timeout("nested cancellation exceeded its declared 30-second grace".into())
907 })??;
908 let elapsed = cancelling.elapsed();
909 assert!(matches!(snapshot.status, RunStatus::Cancelled));
910 assert!(snapshot.interactions.is_empty() && snapshot.outcomes.is_empty());
911 assert!(snapshot.last_action_ordinal.is_none());
912 assert_eq!(snapshot.state.values()["result"], 0);
913 assert_eq!(probe.constructed.load(Ordering::SeqCst), 16);
914 assert_eq!(probe.completed.load(Ordering::SeqCst), 0);
915 probe.assert_idle();
916 for (id, count) in [
917 ("performance", 1),
918 ("performance-mid", 2),
919 ("performance-leaf", 16),
920 ] {
921 let child = workspace.workflow(id.parse()?).await?;
922 let rows = child
923 .list_runs(None, NonZeroUsize::new(128).unwrap())
924 .await?;
925 assert_eq!(rows.len(), count);
926 for row in rows {
927 assert!(matches!(row.status, RunStatus::Cancelled));
928 assert!(row.outcomes.is_empty() && row.last_action_ordinal.is_none());
929 assert_eq!(row.state.values()["result"], 0);
930 }
931 }
932 Ok(
933 json!({"dataset":"nested-cancel","depth":2,"intermediate_children":2,"leaf_children":16,
934 "cooperative_cancel_grace_ms":30_000,"cancellation_latency":Self::samples(vec![elapsed]),
935 "source_replays":0,"caller_sources_dropped":16,"agent_unknown_replay":"not applicable: no Agent attempt"}),
936 )
937 }Sourcepub async fn resolve(
&self,
sequence: u64,
generation: u64,
resolution: Resolution,
) -> Result<RunSnapshot, AppError>
pub async fn resolve( &self, sequence: u64, generation: u64, resolution: Resolution, ) -> Result<RunSnapshot, AppError>
Settles one exact interaction using its original target generation.
Sibling checkpoints do not invalidate that generation. An identical decision replays its retained receipt; a different decision conflicts. Transient inputs replay only while their execution retains the values.
§Errors
Returns InvalidInput for a mismatched resolution or invalid schema/input,
Conflict for a stale target generation, a different repeated decision, or an
unsafe retry. Capacity, revocation, lower/storage failure and observation
deadlines report Busy, Closed, Integration, or Timeout.
§Example
use app_kit::workflow::{Interaction, Resolution};
let run = workflow.start(ExecutionOptions::default()).await?;
let waiting = run.wait().await?;
let Interaction::Input { sequence, generation, .. } = waiting.interactions[0] else {
panic!("the input workflow must request a value");
};
run.resolve(sequence, generation, Resolution::Input(serde_json::json!("ownership"))).await?;
assert_eq!(run.wait().await?.state.values()["answer"], "ownership");Auto Trait Implementations§
impl<CS, A, X = NoActions, C = NoConditions, T = JsonState, H = NoHooks, E = InlineExecutor> !RefUnwindSafe for WorkflowRun<CS, A, X, C, T, H, E>
impl<CS, A, X = NoActions, C = NoConditions, T = JsonState, H = NoHooks, E = InlineExecutor> !UnwindSafe for WorkflowRun<CS, A, X, C, T, H, E>
impl<CS, A, X, C, T, H, E> Freeze for WorkflowRun<CS, A, X, C, T, H, E>where
WorkflowRunRuntime<CS, A, X, C, T, H, E>: Freeze,
impl<CS, A, X, C, T, H, E> Send for WorkflowRun<CS, A, X, C, T, H, E>where
WorkflowRunRuntime<CS, A, X, C, T, H, E>: Send,
impl<CS, A, X, C, T, H, E> Sync for WorkflowRun<CS, A, X, C, T, H, E>where
WorkflowRunRuntime<CS, A, X, C, T, H, E>: Sync,
impl<CS, A, X, C, T, H, E> Unpin for WorkflowRun<CS, A, X, C, T, H, E>where
WorkflowRunRuntime<CS, A, X, C, T, H, E>: Unpin,
impl<CS, A, X, C, T, H, E> UnsafeUnpin for WorkflowRun<CS, A, X, C, T, H, E>where
WorkflowRunRuntime<CS, A, X, C, T, H, E>: UnsafeUnpin,
Blanket Implementations§
Source§impl<T> ArchivePointee for T
impl<T> ArchivePointee for T
Source§type ArchivedMetadata = ()
type ArchivedMetadata = ()
The archived version of the pointer metadata for this type.
Source§fn pointer_metadata(
_: &<T as ArchivePointee>::ArchivedMetadata,
) -> <T as Pointee>::Metadata
fn pointer_metadata( _: &<T as ArchivePointee>::ArchivedMetadata, ) -> <T as Pointee>::Metadata
Converts some archived metadata to the pointer metadata for itself.
Source§impl<T> BorrowMut<T> for Twhere
T: ?Sized,
impl<T> BorrowMut<T> for Twhere
T: ?Sized,
Source§fn borrow_mut(&mut self) -> &mut T
fn borrow_mut(&mut self) -> &mut T
Mutably borrows from an owned value. Read more
impl<ST, DT> CastableFrom<ST, Initialized, Initialized> for DT
impl<ST, DT> CastableFrom<ST, Uninit, Uninit> for DT
impl<T> ErasedDestructor for Twhere
T: 'static,
Source§impl<T> FmtForward for T
impl<T> FmtForward for T
Source§fn fmt_binary(self) -> FmtBinary<Self>where
Self: Binary,
fn fmt_binary(self) -> FmtBinary<Self>where
Self: Binary,
Causes
self to use its Binary implementation when Debug-formatted.Source§fn fmt_display(self) -> FmtDisplay<Self>where
Self: Display,
fn fmt_display(self) -> FmtDisplay<Self>where
Self: Display,
Causes
self to use its Display implementation when
Debug-formatted.Source§fn fmt_lower_exp(self) -> FmtLowerExp<Self>where
Self: LowerExp,
fn fmt_lower_exp(self) -> FmtLowerExp<Self>where
Self: LowerExp,
Causes
self to use its LowerExp implementation when
Debug-formatted.Source§fn fmt_lower_hex(self) -> FmtLowerHex<Self>where
Self: LowerHex,
fn fmt_lower_hex(self) -> FmtLowerHex<Self>where
Self: LowerHex,
Causes
self to use its LowerHex implementation when
Debug-formatted.Source§fn fmt_octal(self) -> FmtOctal<Self>where
Self: Octal,
fn fmt_octal(self) -> FmtOctal<Self>where
Self: Octal,
Causes
self to use its Octal implementation when Debug-formatted.Source§fn fmt_pointer(self) -> FmtPointer<Self>where
Self: Pointer,
fn fmt_pointer(self) -> FmtPointer<Self>where
Self: Pointer,
Causes
self to use its Pointer implementation when
Debug-formatted.Source§fn fmt_upper_exp(self) -> FmtUpperExp<Self>where
Self: UpperExp,
fn fmt_upper_exp(self) -> FmtUpperExp<Self>where
Self: UpperExp,
Causes
self to use its UpperExp implementation when
Debug-formatted.Source§fn fmt_upper_hex(self) -> FmtUpperHex<Self>where
Self: UpperHex,
fn fmt_upper_hex(self) -> FmtUpperHex<Self>where
Self: UpperHex,
Causes
self to use its UpperHex implementation when
Debug-formatted.Source§impl<T> Instrument for T
impl<T> Instrument for T
Source§fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
fn instrument(self, span: Span) -> Instrumented<Self> ⓘ
Source§fn in_current_span(self) -> Instrumented<Self> ⓘ
fn in_current_span(self) -> Instrumented<Self> ⓘ
Source§impl<T> IntoEither for T
impl<T> IntoEither for T
Source§fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ
Converts
self into a Left variant of Either<Self, Self>
if into_left is true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read moreSource§fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
Converts
self into a Left variant of Either<Self, Self>
if into_left(&self) returns true.
Converts self into a Right variant of Either<Self, Self>
otherwise. Read moreSource§impl<T> IntoRequest<T> for T
impl<T> IntoRequest<T> for T
Source§fn into_request(self) -> Request<T>
fn into_request(self) -> Request<T>
Wrap the input message
T in a tonic::RequestSource§impl<T> LayoutRaw for T
impl<T> LayoutRaw for T
Source§fn layout_raw(_: <T as Pointee>::Metadata) -> Result<Layout, LayoutError>
fn layout_raw(_: <T as Pointee>::Metadata) -> Result<Layout, LayoutError>
Returns the layout of the type.
Source§impl<T, N1, N2> Niching<NichedOption<T, N1>> for N2
impl<T, N1, N2> Niching<NichedOption<T, N1>> for N2
Source§unsafe fn is_niched(niched: *const NichedOption<T, N1>) -> bool
unsafe fn is_niched(niched: *const NichedOption<T, N1>) -> bool
Returns whether the given value has been niched. Read more
Source§fn resolve_niched(out: Place<NichedOption<T, N1>>)
fn resolve_niched(out: Place<NichedOption<T, N1>>)
Writes data to
out indicating that a T is niched.Source§impl<T> Pipe for Twhere
T: ?Sized,
impl<T> Pipe for Twhere
T: ?Sized,
Source§fn pipe<R>(self, func: impl FnOnce(Self) -> R) -> Rwhere
Self: Sized,
fn pipe<R>(self, func: impl FnOnce(Self) -> R) -> Rwhere
Self: Sized,
Pipes by value. This is generally the method you want to use. Read more
Source§fn pipe_ref<'a, R>(&'a self, func: impl FnOnce(&'a Self) -> R) -> Rwhere
R: 'a,
fn pipe_ref<'a, R>(&'a self, func: impl FnOnce(&'a Self) -> R) -> Rwhere
R: 'a,
Borrows
self and passes that borrow into the pipe function. Read moreSource§fn pipe_ref_mut<'a, R>(&'a mut self, func: impl FnOnce(&'a mut Self) -> R) -> Rwhere
R: 'a,
fn pipe_ref_mut<'a, R>(&'a mut self, func: impl FnOnce(&'a mut Self) -> R) -> Rwhere
R: 'a,
Mutably borrows
self and passes that borrow into the pipe function. Read moreSource§fn pipe_borrow<'a, B, R>(&'a self, func: impl FnOnce(&'a B) -> R) -> R
fn pipe_borrow<'a, B, R>(&'a self, func: impl FnOnce(&'a B) -> R) -> R
Source§fn pipe_borrow_mut<'a, B, R>(
&'a mut self,
func: impl FnOnce(&'a mut B) -> R,
) -> R
fn pipe_borrow_mut<'a, B, R>( &'a mut self, func: impl FnOnce(&'a mut B) -> R, ) -> R
Source§fn pipe_as_ref<'a, U, R>(&'a self, func: impl FnOnce(&'a U) -> R) -> R
fn pipe_as_ref<'a, U, R>(&'a self, func: impl FnOnce(&'a U) -> R) -> R
Borrows
self, then passes self.as_ref() into the pipe function.Source§fn pipe_as_mut<'a, U, R>(&'a mut self, func: impl FnOnce(&'a mut U) -> R) -> R
fn pipe_as_mut<'a, U, R>(&'a mut self, func: impl FnOnce(&'a mut U) -> R) -> R
Mutably borrows
self, then passes self.as_mut() into the pipe
function.Source§fn pipe_deref<'a, T, R>(&'a self, func: impl FnOnce(&'a T) -> R) -> R
fn pipe_deref<'a, T, R>(&'a self, func: impl FnOnce(&'a T) -> R) -> R
Borrows
self, then passes self.deref() into the pipe function.Source§impl<T> Pointable for T
impl<T> Pointable for T
Source§impl<T> PolicyExt for Twhere
T: ?Sized,
impl<T> PolicyExt for Twhere
T: ?Sized,
impl<T> Read<Exclusive, BecauseExclusive> for Twhere
T: ?Sized,
Source§impl<T> Tap for T
impl<T> Tap for T
Source§fn tap_borrow<B>(self, func: impl FnOnce(&B)) -> Self
fn tap_borrow<B>(self, func: impl FnOnce(&B)) -> Self
Immutable access to the
Borrow<B> of a value. Read moreSource§fn tap_borrow_mut<B>(self, func: impl FnOnce(&mut B)) -> Self
fn tap_borrow_mut<B>(self, func: impl FnOnce(&mut B)) -> Self
Mutable access to the
BorrowMut<B> of a value. Read moreSource§fn tap_ref<R>(self, func: impl FnOnce(&R)) -> Self
fn tap_ref<R>(self, func: impl FnOnce(&R)) -> Self
Immutable access to the
AsRef<R> view of a value. Read moreSource§fn tap_ref_mut<R>(self, func: impl FnOnce(&mut R)) -> Self
fn tap_ref_mut<R>(self, func: impl FnOnce(&mut R)) -> Self
Mutable access to the
AsMut<R> view of a value. Read moreSource§fn tap_deref<T>(self, func: impl FnOnce(&T)) -> Self
fn tap_deref<T>(self, func: impl FnOnce(&T)) -> Self
Immutable access to the
Deref::Target of a value. Read moreSource§fn tap_deref_mut<T>(self, func: impl FnOnce(&mut T)) -> Self
fn tap_deref_mut<T>(self, func: impl FnOnce(&mut T)) -> Self
Mutable access to the
Deref::Target of a value. Read moreSource§fn tap_dbg(self, func: impl FnOnce(&Self)) -> Self
fn tap_dbg(self, func: impl FnOnce(&Self)) -> Self
Calls
.tap() only in debug builds, and is erased in release builds.Source§fn tap_mut_dbg(self, func: impl FnOnce(&mut Self)) -> Self
fn tap_mut_dbg(self, func: impl FnOnce(&mut Self)) -> Self
Calls
.tap_mut() only in debug builds, and is erased in release
builds.Source§fn tap_borrow_dbg<B>(self, func: impl FnOnce(&B)) -> Self
fn tap_borrow_dbg<B>(self, func: impl FnOnce(&B)) -> Self
Calls
.tap_borrow() only in debug builds, and is erased in release
builds.Source§fn tap_borrow_mut_dbg<B>(self, func: impl FnOnce(&mut B)) -> Self
fn tap_borrow_mut_dbg<B>(self, func: impl FnOnce(&mut B)) -> Self
Calls
.tap_borrow_mut() only in debug builds, and is erased in release
builds.Source§fn tap_ref_dbg<R>(self, func: impl FnOnce(&R)) -> Self
fn tap_ref_dbg<R>(self, func: impl FnOnce(&R)) -> Self
Calls
.tap_ref() only in debug builds, and is erased in release
builds.Source§fn tap_ref_mut_dbg<R>(self, func: impl FnOnce(&mut R)) -> Self
fn tap_ref_mut_dbg<R>(self, func: impl FnOnce(&mut R)) -> Self
Calls
.tap_ref_mut() only in debug builds, and is erased in release
builds.Source§fn tap_deref_dbg<T>(self, func: impl FnOnce(&T)) -> Self
fn tap_deref_dbg<T>(self, func: impl FnOnce(&T)) -> Self
Calls
.tap_deref() only in debug builds, and is erased in release
builds.