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
use crate::error::{CommandMetadata, NotificationMetadata};
use crate::service_protocol::messages::AwaitingOnMessage;
use crate::service_protocol::{MessageType, NotificationId, CANCEL_SIGNAL_ID};
use crate::vm::async_results_state::ResolveFutureResult;
use crate::vm::context::Context;
use crate::vm::errors::UncompletedDoProgressDuringReplay;
use crate::vm::transitions::{HitSuspensionPoint, Transition, TransitionAndReturn};
use crate::vm::{awakeable_id_str, State};
use crate::{
AwaitResponse, AwaitingOnPolicy, Error, NotificationHandle, UnresolvedFuture, Value, Version,
};
use std::collections::HashMap;
pub(crate) struct Suspended;
pub(crate) struct DoProgress(pub(crate) UnresolvedFuture);
impl TransitionAndReturn<Context, DoProgress> for State {
type Output = Result<AwaitResponse, Suspended>;
fn transition_and_return(
mut self,
context: &mut Context,
DoProgress(unresolved_future): DoProgress,
) -> Result<(Self, Self::Output), Error> {
match self {
State::Replaying {
ref mut async_results,
ref run_state,
..
} => {
let ResolveFutureResult::WaitExternalInput(unresolved_future) =
async_results.try_resolve_future(unresolved_future)
else {
// We're good, let's give back control to user code
return Ok((self, Ok(AwaitResponse::AnyCompleted)));
};
// This assertion proves the user mutated the code, adding an await point.
//
// Proof by contradiction:
//
// 1. During replay, we transition to processing AFTER replaying all COMMANDS
// (not after replaying the entire journal, which includes both commands and notifications)
// 2. This code path handles awaits ONLY during replay
// 3. If we reach this point, none of the previous checks succeeded, meaning we don't have enough notifications to complete this await point
// 4. But if this await cannot be completed during replay, then in previous replays/execution attempts, no progress should have been made afterward,
// meaning there should be no more commands to replay in the state machine
// 5. However, we ARE still replaying (as evidenced by being in this code path), which means
// there ARE commands to replay after this await point
//
// This contradiction proves the code was mutated: an await must have been added after
// the journal was originally created.
// Prepare error metadata here, we gotta be nice to make sure users can debug this
let awaiting_on_handles = unresolved_future.handles();
let notification_ids =
async_results.resolve_notification_handles(&awaiting_on_handles);
let mut known_notification_metadata = HashMap::with_capacity(2);
let mut known_command_metadata: Option<CommandMetadata> = None;
// Collect run info
for handle in awaiting_on_handles {
if let Some((command_index, name)) = run_state.get_run_info(&handle) {
let notification_id =
async_results.must_resolve_notification_handle(&handle);
let command_metadata = CommandMetadata::new_named(
name.to_owned(),
command_index,
MessageType::RunCommand,
);
known_command_metadata = Some(command_metadata.clone());
known_notification_metadata.insert(
notification_id,
NotificationMetadata::RelatedToCommand(command_metadata),
);
}
}
// For awakeables, prep ids
for notification_id in ¬ification_ids {
if let NotificationId::SignalId(id) = notification_id {
if *id == CANCEL_SIGNAL_ID {
known_notification_metadata.insert(
notification_id.clone(),
NotificationMetadata::Cancellation,
);
} else if *id > 16 {
known_notification_metadata.insert(
notification_id.clone(),
NotificationMetadata::Awakeable(awakeable_id_str(
&context.expect_start_info().id,
*id,
)),
);
}
}
}
let mut error = Error::from(UncompletedDoProgressDuringReplay::new(
notification_ids,
known_notification_metadata,
));
if let Some(command_metadata) = known_command_metadata {
error = error.with_related_command_metadata(command_metadata)
}
Err(error)
}
State::Processing {
ref mut async_results,
ref mut run_state,
..
} => {
let ResolveFutureResult::WaitExternalInput(unresolved_future) =
async_results.try_resolve_future(unresolved_future)
else {
// We're good, let's give back control to user code
return Ok((self, Ok(AwaitResponse::AnyCompleted)));
};
let awaiting_on_handles = unresolved_future.handles();
// We couldn't find any notification for the given ids, let's check if there's some run to execute
if let Some(run_to_execute) = run_state.try_execute_run(&awaiting_on_handles) {
return Ok((self, Ok(AwaitResponse::ExecuteRun(run_to_execute))));
}
// Maybe any of the handles in the awaiting on set is executing?
let waiting_run_proposal =
run_state.any_executing_in_this_set(&awaiting_on_handles);
// Check suspension condition
if context.input_is_closed {
// Some run still executing; it's not time to suspend yet!
if waiting_run_proposal {
return Ok((
self,
Ok(AwaitResponse::WaitingExternalProgress {
waiting_input: false,
waiting_run_proposal: true,
}),
));
}
let state = self.transition(context, HitSuspensionPoint(unresolved_future))?;
return Ok((state, Err(Suspended)));
};
// The only thing we can do at this point is to wait for some notification from the runtime,
// which will be received reading from input.
// Before returning, let's see if we need to send AwaitingOnMessage
if context.negotiated_protocol_version >= Version::V7
&& match context.awaiting_on_policy {
AwaitingOnPolicy::SendAlways => true,
AwaitingOnPolicy::DontSendWhenExecutingRun => !run_state.any_executing(),
AwaitingOnPolicy::DontSend => false,
}
{
context.output.send(&AwaitingOnMessage {
awaiting_on: Some(
async_results.resolve_unresolved_future(unresolved_future),
),
executing_side_effects: false,
})
}
// Nothing else can be done, we need more input
Ok((
self,
Ok(AwaitResponse::WaitingExternalProgress {
waiting_input: true,
waiting_run_proposal,
}),
))
}
s => Err(s.as_unexpected_state("await")),
}
}
}
pub(crate) struct TakeNotification(pub(crate) NotificationHandle);
impl TransitionAndReturn<Context, TakeNotification> for State {
type Output = Result<Option<Value>, Suspended>;
fn transition_and_return(
mut self,
_: &mut Context,
TakeNotification(handle): TakeNotification,
) -> Result<(Self, Self::Output), Error> {
match self {
State::Processing {
ref mut async_results,
..
}
| State::Replaying {
ref mut async_results,
..
} => {
let opt = async_results.take_handle(handle);
Ok((self, Ok(opt.map(Into::into))))
}
s => Err(s.as_unexpected_state("TakeNotification")),
}
}
}
pub(crate) struct CopyNotification(pub(crate) NotificationHandle);
impl TransitionAndReturn<Context, CopyNotification> for State {
type Output = Result<Option<Value>, Suspended>;
fn transition_and_return(
mut self,
_: &mut Context,
CopyNotification(handle): CopyNotification,
) -> Result<(Self, Self::Output), Error> {
match self {
State::Processing {
ref mut async_results,
..
}
| State::Replaying {
ref mut async_results,
..
} => {
let opt = async_results.copy_handle(handle);
Ok((self, Ok(opt.map(Into::into))))
}
s => Err(s.as_unexpected_state("CopyNotification")),
}
}
}