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
236
237
238
use std::{
collections::{BTreeMap, BTreeSet},
path::PathBuf,
time::Duration,
};
use crate::{
BuildId, DataflowId, SessionId,
common::{DaemonId, GitSource},
descriptor::{Descriptor, ResolvedNode},
id::{DataId, NodeId, OperatorId},
};
// ---------------------------------------------------------------------------
// State catch-up types (incremental replay for reconnecting daemons)
// ---------------------------------------------------------------------------
/// A single state mutation that a reconnecting daemon may have missed.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct StateCatchUpEntry {
pub sequence: u64,
pub operation: StateCatchUpOperation,
}
/// The kind of state mutation recorded in the replication log.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum StateCatchUpOperation {
SetParam {
node_id: NodeId,
key: String,
value: serde_json::Value,
},
DeleteParam {
node_id: NodeId,
key: String,
},
}
pub use crate::common::Timestamped;
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub enum RegisterResult {
Ok {
/// unique ID assigned by the coordinator
daemon_id: DaemonId,
},
Err(String),
}
impl RegisterResult {
pub fn to_result(self) -> eyre::Result<DaemonId> {
match self {
RegisterResult::Ok { daemon_id } => Ok(daemon_id),
RegisterResult::Err(err) => Err(eyre::eyre!(err)),
}
}
}
/// Reply to `CoordinatorRequest::ResolveMachine` — sent by the coordinator
/// to the requesting daemon over the same request/response channel used for
/// `RegisterResult`.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum ResolveMachineReply {
/// Reply to `CoordinatorRequest::ResolveMachine`.
ResolveMachineResult {
found: bool,
/// The target daemon's WS peer address as seen by the coordinator
/// (set at registration). Used by the memory-pool direct-TCP data
/// plane to reach the mirror daemon's data listener.
address: Option<std::net::SocketAddr>,
},
}
#[allow(clippy::large_enum_variant)]
#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub enum DaemonCoordinatorEvent {
Build(BuildDataflowNodes),
Spawn(SpawnDataflowNodes),
AllNodesReady {
dataflow_id: DataflowId,
exited_before_subscribe: Vec<NodeId>,
},
StopDataflow {
dataflow_id: DataflowId,
grace_duration: Option<Duration>,
#[serde(default)]
force: bool,
},
ReloadDataflow {
dataflow_id: DataflowId,
node_id: NodeId,
operator_id: Option<OperatorId>,
},
Logs {
dataflow_id: DataflowId,
node_id: NodeId,
tail: Option<usize>,
},
RestartNode {
dataflow_id: DataflowId,
node_id: NodeId,
grace_duration: Option<Duration>,
},
StopNode {
dataflow_id: DataflowId,
node_id: NodeId,
grace_duration: Option<Duration>,
},
SetParam {
dataflow_id: DataflowId,
node_id: NodeId,
key: String,
value: serde_json::Value,
},
DeleteParam {
dataflow_id: DataflowId,
node_id: NodeId,
key: String,
},
Destroy,
Heartbeat,
PeerDaemonDisconnected {
daemon_id: DaemonId,
},
// --- Dynamic Topology ---
/// Add a node to a running dataflow on this daemon.
AddNode {
dataflow_id: DataflowId,
node: crate::descriptor::ResolvedNode,
uv: bool,
},
/// Remove a node from a running dataflow on this daemon.
RemoveNode {
dataflow_id: DataflowId,
node_id: NodeId,
grace_duration: Option<Duration>,
},
/// Atomically replace a running node on this daemon with a new
/// definition under the same id (dora-rs/dora#2927): spawn the
/// replacement first (a failure leaves the current incarnation
/// untouched), then swap the entry and stop the outgoing incarnation.
///
/// `unresolved_node` is the original YAML-shape [`crate::descriptor::Node`]
/// (as `dora node replace` received it); the daemon assigns it wholesale
/// onto the stored descriptor entry so the child's `DORA_NODE_CONFIG` /
/// `DoraNode::dataflow_descriptor()` reflect the replacement's definition
/// end-to-end. The field is separate from `node` because the two shapes
/// differ (`Node` is the flat YAML surface; `ResolvedNode` is the nested
/// resolved form used for spawning), and a hand-written back-conversion
/// would silently drift as fields are added to either type
/// (dora-rs/dora#2988 review, finding 2).
ReplaceNode {
dataflow_id: DataflowId,
node: crate::descriptor::ResolvedNode,
unresolved_node: crate::descriptor::Node,
uv: bool,
grace_duration: Option<Duration>,
},
/// Add a mapping (connection) in a running dataflow.
AddMapping {
dataflow_id: DataflowId,
source_node: NodeId,
source_output: DataId,
target_node: NodeId,
target_input: DataId,
},
/// Remove a mapping (connection) in a running dataflow.
RemoveMapping {
dataflow_id: DataflowId,
source_node: NodeId,
source_output: DataId,
target_node: NodeId,
target_input: DataId,
},
/// Start forwarding matching output frames back to the coordinator over
/// the daemon control channel for CLI topic inspection.
StartTopicDebugStream {
dataflow_id: DataflowId,
outputs: Vec<(NodeId, DataId)>,
subscription_id: uuid::Uuid,
},
/// Stop forwarding output frames for a previously registered CLI topic
/// inspection subscription.
StopTopicDebugStream {
dataflow_id: DataflowId,
subscription_id: uuid::Uuid,
},
/// Incremental state catch-up: replays missed state mutations to a
/// reconnecting daemon.
StateCatchUp {
dataflow_id: DataflowId,
/// The entries the daemon missed, ordered by sequence number.
entries: Vec<StateCatchUpEntry>,
},
}
#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub struct BuildDataflowNodes {
pub build_id: BuildId,
pub session_id: SessionId,
/// Allows overwriting the base working dir when CLI and daemon are
/// running on the same machine.
///
/// Must not be used for multi-machine dataflows.
///
/// Note that nodes with git sources still use a subdirectory of
/// the base working dir.
pub local_working_dir: Option<PathBuf>,
pub git_sources: BTreeMap<NodeId, GitSource>,
pub prev_git_sources: BTreeMap<NodeId, GitSource>,
pub dataflow_descriptor: Descriptor,
pub nodes_on_machine: BTreeSet<NodeId>,
pub uv: bool,
}
#[derive(Debug, serde::Deserialize, serde::Serialize)]
pub struct SpawnDataflowNodes {
pub build_id: Option<BuildId>,
pub session_id: SessionId,
pub dataflow_id: DataflowId,
/// Allows overwriting the base working dir when CLI and daemon are
/// running on the same machine.
///
/// Must not be used for multi-machine dataflows.
///
/// Note that nodes with git sources still use a subdirectory of
/// the base working dir.
pub local_working_dir: Option<PathBuf>,
pub nodes: BTreeMap<NodeId, ResolvedNode>,
pub dataflow_descriptor: Descriptor,
pub spawn_nodes: BTreeSet<NodeId>,
pub uv: bool,
pub write_events_to: Option<PathBuf>,
/// Base URL for downloading artifacts from the coordinator (HTTP distribution mode).
/// When set, daemons can pull binaries from `{artifact_base_url}/{build_id}/{node_id}`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub artifact_base_url: Option<String>,
}