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
//! Which single writer a timer event belongs to (aion#213 R3).
//!
//! Split out of `nif_timer_bridge` for the same two reasons `nif_timer_fire`
//! was: that file is at the 500-line production cap this codebase holds itself
//! to, and this is a real seam rather than a slice taken to hit a number.
//! Everything here answers ONE question — given a timer event, whose recorder
//! may append it — and the answer is what aion#213 got wrong: the bridge used
//! to take the first handle a registry scan yielded for the workflow id, which
//! is correct only while invariant 3 holds and silent exactly when it does not.
use aion_core::{RunId, WorkflowId};
use crate::EngineError;
use crate::registry::WorkflowHandle;
use crate::engine_seam::{EngineSeamError, WorkflowProcessHandle};
use crate::runtime::nif_timer_bridge::{TimerNifBridge, TimerOutcome};
impl TimerNifBridge {
/// The workflow a wheel entry's process belongs to, resolved through the
/// process's OWN RUN and refused when that run is no longer the workflow's
/// live generation (aion#213 R3).
///
/// # 🔴 A PID IS RUN-SCOPED; A WORKFLOW ID IS NOT
///
/// This used to answer with the workflow id alone, and the fire task it
/// feeds then resolved the recorder by that id — so across a
/// continue-as-new window, when one workflow id had two registered
/// handles, the fire could be routed through the OTHER generation's
/// recorder. Answering with the run as well lets the arm refuse a process
/// that is no longer the workflow's live generation instead of arming a
/// durable writer for a run that is over.
///
/// The sole-handle cross-check is what makes that refusal real rather than
/// nominal: it is [`Registry::sole_handle`] that reports two live handles
/// for one workflow as the invariant breach it is, instead of letting a
/// map iteration pick one.
///
/// # Errors
///
/// Returns [`EngineSeamError::TimerWheel`] when no handle matches the pid,
/// when the registry cannot be read, and when the pid's run is not the
/// workflow's live generation.
pub(super) fn live_generation_workflow_for_process(
&self,
process: WorkflowProcessHandle,
) -> Result<WorkflowId, EngineSeamError> {
let handle = self
.registry
.list()
.map_err(|error| EngineSeamError::TimerWheel {
reason: error.to_string(),
})?
.into_iter()
.find(|handle| handle.pid() == process.pid())
.ok_or_else(|| EngineSeamError::TimerWheel {
reason: format!("unknown workflow process {}", process.pid()),
})?;
let live = self
.sole_registered_handle(handle.workflow_id())
.map_err(|error| EngineSeamError::TimerWheel {
reason: error.to_string(),
})?;
match live {
Some(live) if live.run_id() == handle.run_id() => Ok(handle.workflow_id().clone()),
Some(live) => Err(EngineSeamError::TimerWheel {
reason: format!(
"timer for process {} belongs to run `{}` of workflow `{}`, whose live \
generation is now run `{}`: a timer armed for a superseded generation would \
record into the live one",
process.pid(),
handle.run_id(),
handle.workflow_id(),
live.run_id()
),
}),
// Unreachable while the handle above is registered — it was read
// from the same map — but reported rather than assumed: a
// deregistration racing this read means the run is going away, and
// arming a durable writer for it is exactly what must not happen.
None => Err(EngineSeamError::TimerWheel {
reason: format!(
"workflow `{}` has no live handle; its timer is not armed here",
handle.workflow_id()
),
}),
}
}
/// The workflow's ONE registered handle, refusing to guess when there are
/// two (aion#213 R3).
///
/// Every place this bridge needs "the workflow's writer" goes through
/// here. The previous shape — `registry.list().find(|h| h.workflow_id() ==
/// id)` — took whichever handle the map yielded first, which is correct
/// only while invariant 3 holds and silent exactly when it does not: with a
/// predecessor and a successor both registered, a durable timer append went
/// through an arbitrary one of them.
///
/// # Errors
///
/// Propagates [`EngineError::WorkflowWritersAmbiguous`] and registry
/// poisoning from [`Registry::sole_handle`].
pub(super) fn sole_registered_handle(
&self,
workflow_id: &WorkflowId,
) -> Result<Option<WorkflowHandle>, EngineError> {
self.registry.sole_handle(workflow_id)
}
}
/// The run a reserved `deadline:{run}` timer names, when that run is NOT the
/// one `handle` is the live generation of (aion#213 R3).
///
/// The deadline family is the one timer family whose id carries its run, so it
/// is the one place a workflow-id lookup can be checked against the run the
/// caller actually meant without threading a run through the whole engine
/// seam. `None` for an ordinary timer (whose id names no run) and for a
/// deadline belonging to the live generation.
pub(super) fn deadline_for_another_run(
outcome: &TimerOutcome,
handle: &WorkflowHandle,
) -> Option<RunId> {
let timer_id = match outcome {
TimerOutcome::Fired(timer_id) | TimerOutcome::Cancelled(timer_id, _) => timer_id,
};
let named_run = crate::time::deadline_run_id(timer_id)?;
(&named_run != handle.run_id()).then_some(named_run)
}