hatchery_server/
mesh_runtime.rs

1//! Production wiring for [`NodeOperatorMesh`] (A2).
2//!
3//! When drain + operator control are configured, core:
4//! 1. listens for operator clients on an owner-only local endpoint;
5//! 2. opens one drain consumer per configured node id (g4a control);
6//! 3. registers each [`OperatorLink`] on the mesh so operator requests reach the node.
7//!
8//! Optional: m4a channel consumers share the same node list when a hub is present.
9//! Secrets are read from named environment variables and scrubbed; never logged.
10
11use crate::drain_consumer::{DrainConsumer, NodeStream};
12use crate::node_controller::{G4aMessage, OperatorLink};
13use crate::operator_control::{serve_local, NodeOperatorMesh};
14use hatchery_c2_client::DrainClient;
15use hatchery_core::nested::HqRecipientSecret;
16use hatchery_core::{BrowserStreamEvent, ChannelKind, NodeRequest, NodeRoute, NodeId, NodeCursor, RoutedNodeRequest};
17use std::net::SocketAddr;
18use std::path::{Path, PathBuf};
19use tokio::sync::{mpsc, watch};
20use tokio::task::JoinHandle;
21
22pub const DRAIN_ADDR_ENV: &str = "HATCHERY_CORE_DRAIN_ADDR";
23pub const DRAIN_TOKEN_VAR_ENV: &str = "HATCHERY_CORE_DRAIN_TOKEN_VAR";
24pub const OPERATOR_LISTEN_ENV: &str = "HATCHERY_CORE_OPERATOR_LISTEN";
25pub const OPERATOR_TOKEN_VAR_ENV: &str = "HATCHERY_CORE_OPERATOR_TOKEN_VAR";
26pub const NODE_IDS_ENV: &str = "HATCHERY_CORE_NODE_IDS";
27pub const RECIPIENT_SECRET_VAR_ENV: &str = "HATCHERY_CORE_RECIPIENT_SECRET_VAR";
28
29const DEFAULT_DRAIN_TOKEN_VAR: &str = "HATCHERY_CORE_DRAIN_TOKEN";
30const DEFAULT_OPERATOR_TOKEN_VAR: &str = "GATE4AGENT_CORE_OPERATOR_TOKEN";
31const G4A_CONTROL_CHANNEL: u32 = 1;
32const M4A_CHANNEL: u32 = 2;
33
34/// Optional production mesh configuration (all-or-nothing for the drain path).
35#[derive(Clone)]
36pub struct MeshConfig {
37    pub drain_addr: SocketAddr,
38    pub drain_token: String,
39    pub operator_listen: String,
40    pub operator_token: String,
41    pub node_ids: Vec<String>,
42    /// Data dir whose enroll ledger adds trusted nodes (None in tests that list nodes only).
43    pub data_dir: Option<std::path::PathBuf>,
44    pub recipient: HqRecipientSecret,
45}
46
47impl MeshConfig {
48    /// Resolve from the environment. Returns `Ok(None)` when drain is not configured.
49    /// Token values are taken from the named vars and those vars are removed from the process env.
50    pub fn from_env(data_dir: &Path) -> Result<Option<Self>, String> {
51        let Some(drain_addr) = std::env::var(DRAIN_ADDR_ENV).ok().filter(|s| !s.is_empty()) else {
52            return Ok(None);
53        };
54        let drain_addr: SocketAddr = drain_addr
55            .parse()
56            .map_err(|e| format!("{DRAIN_ADDR_ENV}: {e}"))?;
57        let operator_listen = std::env::var(OPERATOR_LISTEN_ENV)
58            .map_err(|_| format!("{OPERATOR_LISTEN_ENV} is required with {DRAIN_ADDR_ENV}"))?;
59        if operator_listen.is_empty() {
60            return Err(format!("{OPERATOR_LISTEN_ENV} is empty"));
61        }
62        let node_ids = std::env::var(NODE_IDS_ENV)
63            .unwrap_or_default()
64            .split(',')
65            .map(str::trim)
66            .filter(|s| !s.is_empty())
67            .map(str::to_owned)
68            .collect::<Vec<_>>();
69
70        let drain_token_var =
71            std::env::var(DRAIN_TOKEN_VAR_ENV).unwrap_or_else(|_| DEFAULT_DRAIN_TOKEN_VAR.into());
72        let operator_token_var = std::env::var(OPERATOR_TOKEN_VAR_ENV)
73            .unwrap_or_else(|_| DEFAULT_OPERATOR_TOKEN_VAR.into());
74        let drain_token = take_secret(&drain_token_var)?;
75        let operator_token = take_secret(&operator_token_var)?;
76        if drain_token.len() < 16 || operator_token.len() < 16 {
77            return Err("drain and operator tokens must be at least 16 bytes".into());
78        }
79        let recipient = load_or_create_recipient(data_dir)?;
80        Ok(Some(Self {
81            drain_addr,
82            drain_token,
83            operator_listen,
84            operator_token,
85            node_ids,
86            data_dir: Some(data_dir.to_path_buf()),
87            recipient,
88        }))
89    }
90}
91
92fn take_secret(var: &str) -> Result<String, String> {
93    let value = std::env::var(var).map_err(|_| format!("{var} is not set"))?;
94    std::env::remove_var(var);
95    if value.is_empty() {
96        return Err(format!("{var} is empty"));
97    }
98    Ok(value)
99}
100
101fn load_or_create_recipient(data_dir: &Path) -> Result<HqRecipientSecret, String> {
102    if let Ok(var) = std::env::var(RECIPIENT_SECRET_VAR_ENV) {
103        if !var.is_empty() {
104            let secret = take_secret(&var)?;
105            return parse_recipient_hex(&secret);
106        }
107    }
108    let path = data_dir.join("core-recipient.seed");
109    if path.exists() {
110        let bytes = std::fs::read(&path).map_err(|e| e.to_string())?;
111        if bytes.len() != 32 {
112            return Err(format!("{}: expected 32 bytes", path.display()));
113        }
114        let mut seed = [0u8; 32];
115        seed.copy_from_slice(&bytes);
116        return Ok(HqRecipientSecret::from_seed(seed));
117    }
118    let recipient = HqRecipientSecret::generate();
119    std::fs::create_dir_all(data_dir).map_err(|e| e.to_string())?;
120    // Persist only the seed bytes; never log them.
121    let seed = recipient.seed_bytes();
122    std::fs::write(&path, seed).map_err(|e| e.to_string())?;
123    #[cfg(unix)]
124    {
125        use std::os::unix::fs::PermissionsExt;
126        let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
127    }
128    Ok(recipient)
129}
130
131fn parse_recipient_hex(hex: &str) -> Result<HqRecipientSecret, String> {
132    let hex = hex.trim();
133    if hex.len() != 64 {
134        return Err("recipient secret must be 64 hex digits".into());
135    }
136    let mut seed = [0u8; 32];
137    for i in 0..32 {
138        seed[i] = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16)
139            .map_err(|_| "recipient secret is not hex".to_owned())?;
140    }
141    Ok(HqRecipientSecret::from_seed(seed))
142}
143
144/// Running mesh: operator control + per-node drain consumers.
145pub struct MeshRuntime {
146    pub mesh: NodeOperatorMesh,
147    tasks: Vec<JoinHandle<()>>,
148    stop: watch::Sender<bool>,
149}
150
151
152async fn bootstrap_cursor(
153    mesh: NodeOperatorMesh,
154    link: OperatorLink,
155    node_id: String,
156    mut shutdown: watch::Receiver<bool>,
157) {
158    use hatchery_core::{NodeCursor, NodeId, NodeIncarnationId, NodeRequest, NodeRoute, RoutedNodeRequest};
159    use crate::node_controller::G4aMessage;
160    // Retry until Snapshot lands or shutdown.
161    for attempt in 1u32..=120 {
162        if *shutdown.borrow() {
163            return;
164        }
165        let Ok(nid) = NodeId::new(node_id.clone()) else { return };
166        let route = NodeRoute {
167            node_id: nid,
168            expected_incarnation_id: NodeIncarnationId::from_bytes([0; 16]),
169        };
170        let req = RoutedNodeRequest {
171            route,
172            request: NodeRequest::Snapshot,
173        };
174        match link.request(&req, std::time::Duration::from_secs(5)).await {
175            Ok(G4aMessage::Response(routed)) => {
176                // Any routed reply carries the live incarnation (ok Snapshot or
177                // a typed failure). That is enough for HQ Spawn cursors.
178                mesh.set_cursor(
179                    &node_id,
180                    NodeCursor {
181                        incarnation_id: routed.incarnation_id,
182                        sequence: 0,
183                    },
184                );
185                if let Ok(hatchery_core::C2NodeResponse::Snapshot { snapshot, .. }) = &routed.response {
186                    if let Some(machine) = snapshot.machine.clone() {
187                        mesh.set_machine(&node_id, machine);
188                    }
189                }
190                tracing::info!(
191                    %node_id,
192                    incarnation = %routed.incarnation_id,
193                    ok = routed.response.is_ok(),
194                    "core mesh cursor from Snapshot"
195                );
196                return;
197            }
198            Ok(_) => {}
199            Err(err) => {
200                tracing::debug!(%node_id, %attempt, error = %err, "cursor bootstrap waiting for drain session");
201            }
202        }
203        tokio::select! {
204            _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => {}
205            _ = shutdown.changed() => return,
206        }
207    }
208    tracing::warn!(%node_id, "core mesh cursor bootstrap gave up");
209}
210
211
212/// What a node's drain consumers need; shared by startup and the ledger watcher.
213#[derive(Clone)]
214struct NodeCtx {
215    mesh: NodeOperatorMesh,
216    drain_addr: std::net::SocketAddr,
217    drain_token: String,
218    recipient: HqRecipientSecret,
219    m4a_handler: Option<crate::drain_consumer::CoreHandler>,
220    shutdown: watch::Receiver<bool>,
221}
222
223fn spawn_node(ctx: &NodeCtx, node_id: &String, tasks: &mut Vec<tokio::task::JoinHandle<()>>) {
224            let (events_tx, mut events_rx) = mpsc::unbounded_channel();
225    let (browser_tx, mut browser_rx) = mpsc::unbounded_channel::<BrowserStreamEvent>();
226    let (link, outbound) = OperatorLink::new_with_browser(Some(events_tx), Some(browser_tx));
227    ctx.mesh.register(node_id, link.clone());
228    // Forward node-initiated g4a events (PTY frames, etc.) onto the
229    // operator broadcast bus so every HQ client sees live bytes.
230    let mesh_ev = ctx.mesh.clone();
231    tasks.push(tokio::spawn(async move {
232        while let Some(msg) = events_rx.recv().await {
233            if let G4aMessage::Event(ev) = msg {
234                mesh_ev.publish_event(ev);
235            }
236        }
237    }));
238    let mesh_br = ctx.mesh.clone();
239    tasks.push(tokio::spawn(async move {
240        while let Some(ev) = browser_rx.recv().await {
241            mesh_br.publish_browser(ev);
242        }
243    }));
244    let consumer = DrainConsumer::new(
245        DrainClient::new(ctx.drain_addr, ctx.drain_token.clone()),
246        ctx.recipient.clone(),
247        NodeStream {
248            node_id: node_id.clone(),
249            channel_id: G4A_CONTROL_CHANNEL,
250            kind: ChannelKind::G4aControl,
251        },
252        link.clone().handler(),
253    )
254    .with_outbound(outbound)
255    .with_cursor_dir(crate::drain_consumer::CursorStore::dir_from_env());
256    let sd = ctx.shutdown.clone();
257    tasks.push(tokio::spawn(async move {
258        consumer.run(sd).await;
259    }));
260
261    // Learn incarnation once the Noise session is up so HQ Spawn has a cursor.
262    let mesh_boot = ctx.mesh.clone();
263    let link_boot = link.clone();
264    let boot_id = node_id.clone();
265    let sd = ctx.shutdown.clone();
266    tasks.push(tokio::spawn(async move {
267        bootstrap_cursor(mesh_boot, link_boot, boot_id, sd).await;
268    }));
269
270    if let Some(handler) = ctx.m4a_handler.clone() {
271        let consumer = DrainConsumer::new(
272            DrainClient::new(ctx.drain_addr, ctx.drain_token.clone()),
273            ctx.recipient.clone(),
274            NodeStream {
275                node_id: node_id.clone(),
276                channel_id: M4A_CHANNEL,
277                kind: ChannelKind::M4a,
278            },
279            handler,
280        )
281        .with_cursor_dir(crate::drain_consumer::CursorStore::dir_from_env());
282        let sd = ctx.shutdown.clone();
283        tasks.push(tokio::spawn(async move {
284            consumer.run(sd).await;
285        }));
286    }
287        }
288
289/// Nodes core drains: the env list plus every trusted or member node of the enroll ledger.
290fn wanted_nodes(env: &[String], data_dir: Option<&std::path::Path>) -> Vec<String> {
291    let mut out = env.to_vec();
292    if let Some(dir) = data_dir {
293        for e in crate::provision::read_ledger(dir).unwrap_or_default() {
294            if matches!(e.status, crate::provision::NodeStatus::Member | crate::provision::NodeStatus::Trusted) && !out.contains(&e.node_id) {
295                out.push(e.node_id);
296            }
297        }
298    }
299    out
300}
301
302impl MeshRuntime {
303    pub fn start(config: MeshConfig) -> Self {
304        Self::start_with_m4a(config, None)
305    }
306
307    pub fn start_with_m4a(
308        config: MeshConfig,
309        m4a_handler: Option<crate::drain_consumer::CoreHandler>,
310    ) -> Self {
311        let mesh = NodeOperatorMesh::new();
312        let (stop, shutdown) = watch::channel(false);
313        let mut tasks = Vec::new();
314
315        let listen = config.operator_listen.clone();
316        let token = config.operator_token.clone();
317        let mesh_serve = mesh.clone();
318        let sd = shutdown.clone();
319        tasks.push(tokio::spawn(async move {
320            if let Err(err) = serve_local(listen, token, mesh_serve, sd).await {
321                tracing::error!(%err, "core operator control stopped");
322            }
323        }));
324
325        let ctx = NodeCtx {
326            mesh: mesh.clone(),
327            drain_addr: config.drain_addr,
328            drain_token: config.drain_token.clone(),
329            recipient: config.recipient.clone(),
330            m4a_handler: m4a_handler.clone(),
331            shutdown: shutdown.clone(),
332        };
333        let mut started = wanted_nodes(&config.node_ids, config.data_dir.as_deref());
334        for node_id in &started {
335            spawn_node(&ctx, node_id, &mut tasks);
336        }
337        // A node trusted after startup gets its consumers without a core restart.
338        if let Some(dir) = config.data_dir.clone() {
339            let env = config.node_ids.clone();
340            let ctx = ctx.clone();
341            let mut sd = shutdown.clone();
342            let mut seen = started.clone();
343            tasks.push(tokio::spawn(async move {
344                let mut own = Vec::new();
345                loop {
346                    tokio::select! {
347                        _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {}
348                        _ = sd.changed() => break,
349                    }
350                    for node_id in wanted_nodes(&env, Some(&dir)) {
351                        if !seen.contains(&node_id) {
352                            tracing::info!(%node_id, "core mesh: trusted node added from the enroll ledger");
353                            spawn_node(&ctx, &node_id, &mut own);
354                            seen.push(node_id);
355                        }
356                    }
357                }
358                for t in own { t.abort(); }
359            }));
360        }
361        started.sort();
362        tracing::info!(
363            nodes = started.len(),
364            listen = %config.operator_listen,
365            "core operator mesh started (drain consumers registered)"
366        );
367        Self { mesh, tasks, stop }
368    }
369
370    pub fn shutdown(&self) {
371        let _ = self.stop.send(true);
372        for t in &self.tasks {
373            t.abort();
374        }
375    }
376}
377
378/// Public path of the persisted recipient seed (for operators to hand the pub key to C2).
379pub fn recipient_seed_path(data_dir: &Path) -> PathBuf {
380    data_dir.join("core-recipient.seed")
381}
hatchery_server/mesh_runtime.rs

hatchery_server/
mesh_runtime.rs
