1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
//! The options of one run.
use super::message::AuditMode;
use crate::base::{CancelToken, Instant};
use crate::function::CallLayer;
use crate::observe::{Observer, TraceOptions};
use std::fmt;
use std::sync::Arc;
/// What a caller chooses about one run.
#[derive(Clone)]
#[non_exhaustive]
pub struct RunOptions<'a> {
/// Run the Active workflows of this channel; `None` runs every Active one.
pub channel: Option<&'a str>,
/// The bucket a rollout admits, `0..=99`; a larger one is served as 99.
/// `None` is admitted by every rollout.
pub routing_bucket: Option<u8>,
/// Record each write's `Change` in its task's audit entry. `true` by
/// default (#63). It governs only the audit trail: a trace records its
/// own changes whatever it says.
pub capture_changes: bool,
/// Which audit entries the message keeps. `Full` by default.
pub audit: AuditMode,
/// When the run must end; `None` by default. It is checked at every
/// host call, every fan-out element's start, every sweep and every
/// [`CHECK_EVERY`](crate::run::CHECK_EVERY) ops. Natively each host call also races it, on a Tokio timer, so a
/// run with a deadline is polled inside a Tokio runtime with its time
/// driver enabled. A call it cuts short, or that returns `Err` once it
/// has passed, is recorded as `TIMEOUT_ERROR` on its task; either way
/// the run ends [`TimedOut`].
///
/// On wasm no timer exists, so nothing races a host call and it is
/// never cut short mid-flight: a deadline that passes while a call runs
/// is seen at the next check, after the call returns.
///
/// [`TimedOut`]: crate::run::RunStatus::TimedOut
pub deadline: Option<Instant>,
/// A token that cancels the run when it, or an ancestor, is cancelled;
/// `None` by default. It is checked and raced where the deadline is,
/// and first. A call it cuts short, or that returns `Err` once it is
/// cancelled, is recorded as `CANCELLED` on its task; either way the
/// run ends [`Cancelled`].
///
/// [`Cancelled`]: crate::run::RunStatus::Cancelled
pub cancel: Option<&'a CancelToken>,
/// Layers around every host call of this run, outside the registry's;
/// the first is the outermost. None by default.
pub layers: &'a [Arc<dyn CallLayer>],
/// An observer of this run, beside the engine's
/// ([`EngineConfig::observer`]); `None` by default.
///
/// [`EngineConfig::observer`]: crate::run::EngineConfig::observer()
pub observer: Option<&'a dyn Observer>,
/// Trace the run, as these options say: its [`Run::trace`] is then the
/// initial context plus each step's changes. `None` by default: no
/// trace, and no copy of the context.
///
/// [`Run::trace`]: crate::Run::trace()
pub trace: Option<TraceOptions>,
}
impl RunOptions<'static> {
/// The defaults, as a constant: `Engine::run` borrows a static copy.
pub(crate) const DEFAULT: Self = Self {
channel: None,
routing_bucket: None,
capture_changes: true,
audit: AuditMode::Full,
deadline: None,
cancel: None,
layers: &[],
observer: None,
trace: None,
};
}
impl Default for RunOptions<'_> {
fn default() -> Self {
RunOptions::DEFAULT
}
}
impl fmt::Debug for RunOptions<'_> {
/// Every option, the layers as a count and the observer as a presence.
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let observer = if self.observer.is_some() {
"present"
} else {
"absent"
};
f.debug_struct("RunOptions")
.field("channel", &self.channel)
.field("routing_bucket", &self.routing_bucket)
.field("capture_changes", &self.capture_changes)
.field("audit", &self.audit)
.field("deadline", &self.deadline)
.field("cancel", &self.cancel)
.field("layers", &self.layers.len())
.field("observer", &format_args!("{observer}"))
.field("trace", &self.trace)
.finish()
}
}
impl<'a> RunOptions<'a> {
/// Runs the Active workflows of channel `c`.
#[must_use]
pub fn channel(mut self, c: &'a str) -> Self {
self.channel = Some(c);
self
}
/// Sets the routing bucket rollouts are tested against: `0..=99`. A
/// larger one is served as bucket 99, as in 3.x, so derive it as a
/// hash `% 100`, never `% 256`.
#[must_use]
pub fn routing_bucket(mut self, b: u8) -> Self {
self.routing_bucket = Some(b);
self
}
/// Turns the audit trail's change capture on or off.
#[must_use]
pub fn capture_changes(mut self, on: bool) -> Self {
self.capture_changes = on;
self
}
/// Sets which audit entries the message keeps.
#[must_use]
pub fn audit(mut self, mode: AuditMode) -> Self {
self.audit = mode;
self
}
/// Ends the run at `deadline`, as the `deadline` field says.
#[must_use]
pub fn deadline(mut self, deadline: Instant) -> Self {
self.deadline = Some(deadline);
self
}
/// Cancels the run when `token` is cancelled, as the `cancel` field
/// says.
#[must_use]
pub fn cancel(mut self, token: &'a CancelToken) -> Self {
self.cancel = Some(token);
self
}
/// Wraps every host call of the run in `layers`, outside the registry's
/// layers; the first is the outermost.
#[must_use]
pub fn layers(mut self, layers: &'a [Arc<dyn CallLayer>]) -> Self {
self.layers = layers;
self
}
/// Reports this run to `o` too, beside the engine's observer.
#[must_use]
pub fn observer(mut self, o: &'a dyn Observer) -> Self {
self.observer = Some(o);
self
}
/// Traces the run under `opts`, as the `trace` field says.
#[must_use]
pub fn trace(mut self, opts: TraceOptions) -> Self {
self.trace = Some(opts);
self
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn run_options_default_captures_changes_and_keeps_the_full_audit() {
let o = RunOptions::default();
assert!(o.capture_changes);
assert_eq!(o.audit, AuditMode::Full);
assert_eq!(o.channel, None);
assert_eq!(o.routing_bucket, None);
assert!(o.deadline.is_none() && o.cancel.is_none());
assert!(o.layers.is_empty());
assert!(o.observer.is_none());
assert!(o.trace.is_none());
let layers: Vec<Arc<dyn CallLayer>> = vec![Arc::new(Pass)];
let token = CancelToken::new();
let o = o
.channel("c")
.routing_bucket(7)
.capture_changes(false)
.audit(AuditMode::Off)
.cancel(&token)
.layers(&layers);
assert_eq!((o.channel, o.routing_bucket), (Some("c"), Some(7)));
assert!(!o.capture_changes);
assert_eq!(o.audit, AuditMode::Off);
assert_eq!(o.layers.len(), 1);
assert_eq!(
format!("{o:?}"),
r#"RunOptions { channel: Some("c"), routing_bucket: Some(7), capture_changes: false, audit: Off, deadline: None, cancel: Some(CancelToken { cancelled: false, .. }), layers: 1, observer: absent, trace: None }"#
);
let watcher = Quiet;
let o = o.observer(&watcher);
assert!(format!("{o:?}").ends_with("layers: 1, observer: present, trace: None }"));
let o = o.trace(TraceOptions::default().timings(true));
assert!(o.trace.as_ref().is_some_and(|t| t.timings && t.changes));
let at = Instant::now();
assert_eq!(o.deadline(at).deadline, Some(at));
}
/// An observer that keeps nothing.
struct Quiet;
impl Observer for Quiet {
fn on(&self, _event: &crate::observe::Event<'_>) {}
}
/// A layer that goes straight on.
struct Pass;
impl CallLayer for Pass {
fn call<'a>(
&'a self,
ctx: &'a mut crate::function::Ctx<'_>,
next: crate::function::Next<'a>,
) -> crate::function::BoxFuture<'a, crate::base::Result<crate::function::Outcome>> {
next.run(ctx)
}
}
}