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    pub recipient: HqRecipientSecret,
43}
44
45impl MeshConfig {
46    /// Resolve from the environment. Returns `Ok(None)` when drain is not configured.
47    /// Token values are taken from the named vars and those vars are removed from the process env.
48    pub fn from_env(data_dir: &Path) -> Result<Option<Self>, String> {
49        let Some(drain_addr) = std::env::var(DRAIN_ADDR_ENV).ok().filter(|s| !s.is_empty()) else {
50            return Ok(None);
51        };
52        let drain_addr: SocketAddr = drain_addr
53            .parse()
54            .map_err(|e| format!("{DRAIN_ADDR_ENV}: {e}"))?;
55        let operator_listen = std::env::var(OPERATOR_LISTEN_ENV)
56            .map_err(|_| format!("{OPERATOR_LISTEN_ENV} is required with {DRAIN_ADDR_ENV}"))?;
57        if operator_listen.is_empty() {
58            return Err(format!("{OPERATOR_LISTEN_ENV} is empty"));
59        }
60        let node_ids = std::env::var(NODE_IDS_ENV)
61            .map_err(|_| format!("{NODE_IDS_ENV} is required with {DRAIN_ADDR_ENV}"))?
62            .split(',')
63            .map(str::trim)
64            .filter(|s| !s.is_empty())
65            .map(str::to_owned)
66            .collect::<Vec<_>>();
67        if node_ids.is_empty() {
68            return Err(format!("{NODE_IDS_ENV} must list at least one node id"));
69        }
70
71        let drain_token_var =
72            std::env::var(DRAIN_TOKEN_VAR_ENV).unwrap_or_else(|_| DEFAULT_DRAIN_TOKEN_VAR.into());
73        let operator_token_var = std::env::var(OPERATOR_TOKEN_VAR_ENV)
74            .unwrap_or_else(|_| DEFAULT_OPERATOR_TOKEN_VAR.into());
75        let drain_token = take_secret(&drain_token_var)?;
76        let operator_token = take_secret(&operator_token_var)?;
77        if drain_token.len() < 16 || operator_token.len() < 16 {
78            return Err("drain and operator tokens must be at least 16 bytes".into());
79        }
80        let recipient = load_or_create_recipient(data_dir)?;
81        Ok(Some(Self {
82            drain_addr,
83            drain_token,
84            operator_listen,
85            operator_token,
86            node_ids,
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
211impl MeshRuntime {
212    pub fn start(config: MeshConfig) -> Self {
213        Self::start_with_m4a(config, None)
214    }
215
216    pub fn start_with_m4a(
217        config: MeshConfig,
218        m4a_handler: Option<crate::drain_consumer::CoreHandler>,
219    ) -> Self {
220        let mesh = NodeOperatorMesh::new();
221        let (stop, shutdown) = watch::channel(false);
222        let mut tasks = Vec::new();
223
224        let listen = config.operator_listen.clone();
225        let token = config.operator_token.clone();
226        let mesh_serve = mesh.clone();
227        let sd = shutdown.clone();
228        tasks.push(tokio::spawn(async move {
229            if let Err(err) = serve_local(listen, token, mesh_serve, sd).await {
230                tracing::error!(%err, "core operator control stopped");
231            }
232        }));
233
234        for node_id in &config.node_ids {
235            let (events_tx, mut events_rx) = mpsc::unbounded_channel();
236            let (browser_tx, mut browser_rx) = mpsc::unbounded_channel::<BrowserStreamEvent>();
237            let (link, outbound) = OperatorLink::new_with_browser(Some(events_tx), Some(browser_tx));
238            mesh.register(node_id, link.clone());
239            // Forward node-initiated g4a events (PTY frames, etc.) onto the
240            // operator broadcast bus so every HQ client sees live bytes.
241            let mesh_ev = mesh.clone();
242            tasks.push(tokio::spawn(async move {
243                while let Some(msg) = events_rx.recv().await {
244                    if let G4aMessage::Event(ev) = msg {
245                        mesh_ev.publish_event(ev);
246                    }
247                }
248            }));
249            let mesh_br = mesh.clone();
250            tasks.push(tokio::spawn(async move {
251                while let Some(ev) = browser_rx.recv().await {
252                    mesh_br.publish_browser(ev);
253                }
254            }));
255            let consumer = DrainConsumer::new(
256                DrainClient::new(config.drain_addr, config.drain_token.clone()),
257                config.recipient.clone(),
258                NodeStream {
259                    node_id: node_id.clone(),
260                    channel_id: G4A_CONTROL_CHANNEL,
261                    kind: ChannelKind::G4aControl,
262                },
263                link.clone().handler(),
264            )
265            .with_outbound(outbound)
266            .with_cursor_dir(crate::drain_consumer::CursorStore::dir_from_env());
267            let sd = shutdown.clone();
268            tasks.push(tokio::spawn(async move {
269                consumer.run(sd).await;
270            }));
271
272            // Learn incarnation once the Noise session is up so HQ Spawn has a cursor.
273            let mesh_boot = mesh.clone();
274            let link_boot = link.clone();
275            let boot_id = node_id.clone();
276            let sd = shutdown.clone();
277            tasks.push(tokio::spawn(async move {
278                bootstrap_cursor(mesh_boot, link_boot, boot_id, sd).await;
279            }));
280
281            if let Some(handler) = m4a_handler.clone() {
282                let consumer = DrainConsumer::new(
283                    DrainClient::new(config.drain_addr, config.drain_token.clone()),
284                    config.recipient.clone(),
285                    NodeStream {
286                        node_id: node_id.clone(),
287                        channel_id: M4A_CHANNEL,
288                        kind: ChannelKind::M4a,
289                    },
290                    handler,
291                )
292                .with_cursor_dir(crate::drain_consumer::CursorStore::dir_from_env());
293                let sd = shutdown.clone();
294                tasks.push(tokio::spawn(async move {
295                    consumer.run(sd).await;
296                }));
297            }
298        }
299
300        tracing::info!(
301            nodes = config.node_ids.len(),
302            listen = %config.operator_listen,
303            "core operator mesh started (drain consumers registered)"
304        );
305        Self { mesh, tasks, stop }
306    }
307
308    pub fn shutdown(&self) {
309        let _ = self.stop.send(true);
310        for t in &self.tasks {
311            t.abort();
312        }
313    }
314}
315
316/// Public path of the persisted recipient seed (for operators to hand the pub key to C2).
317pub fn recipient_seed_path(data_dir: &Path) -> PathBuf {
318    data_dir.join("core-recipient.seed")
319}
hatchery_server/mesh_runtime.rs

hatchery_server/
mesh_runtime.rs
