rust_supervisor/dashboard/
runtime.rs

1//! Dashboard IPC runtime lifecycle.
2//!
3//! The runtime owns the target-side Unix socket accept loop and the dynamic
4//! registration heartbeat used by the relay integration.
5
6use crate::config::audit::AuditConfig;
7use crate::control::handle::SupervisorHandle;
8use crate::dashboard::config::ValidatedDashboardIpcConfig;
9use crate::dashboard::error::DashboardError;
10use crate::dashboard::ipc_server::{DashboardIpcService, bind_dashboard_listener};
11use crate::dashboard::protocol::{IpcResponse, parse_request_line, response_to_line};
12use crate::dashboard::registration::run_registration_heartbeat;
13use crate::dashboard::state::declared_state_from_spec;
14use crate::ipc::security::IpcSecurityPipeline;
15use crate::ipc::security::peer_identity::{PeerIdentity, extract_peer_identity};
16use crate::journal::ring::EventJournal;
17use crate::spec::supervisor::SupervisorSpec;
18use std::fmt;
19use std::os::unix::io::AsRawFd;
20use std::path::PathBuf;
21use std::sync::Arc;
22use std::sync::atomic::{AtomicU64, Ordering};
23use tokio::io::{AsyncReadExt, AsyncWriteExt};
24use tokio::net::{UnixListener, UnixStream};
25use tokio::task::{JoinHandle, JoinSet};
26
27/// Default maximum frame size for bounded frame reader: 1 MiB.
28pub(crate) const DEFAULT_MAX_FRAME_BYTES: usize = 1_048_576;
29
30/// Maximum concurrent IPC connections (C10 resource boundary).
31const MAX_CONCURRENT_CONNECTIONS: usize = 64;
32
33/// Per-connection idle timeout: drop connections that send no complete
34/// frame within this duration (C10 resource boundary).
35const CONNECTION_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);
36
37/// Per-process connection counter for unique connection_id generation.
38static CONNECTION_COUNTER: AtomicU64 = AtomicU64::new(0);
39
40/// Guard that owns dashboard IPC background tasks and socket cleanup.
41pub struct DashboardIpcRuntimeGuard {
42    /// Socket path created by this runtime.
43    ipc_path: PathBuf,
44    /// Target-side IPC accept task.
45    ipc_task: JoinHandle<()>,
46    /// Optional registration heartbeat task.
47    heartbeat_task: Option<JoinHandle<()>>,
48}
49
50impl fmt::Debug for DashboardIpcRuntimeGuard {
51    /// Formats guard diagnostics without exposing task internals.
52    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
53        formatter
54            .debug_struct("DashboardIpcRuntimeGuard")
55            .field("ipc_path", &self.ipc_path)
56            .field("has_heartbeat_task", &self.heartbeat_task.is_some())
57            .finish_non_exhaustive()
58    }
59}
60
61impl Drop for DashboardIpcRuntimeGuard {
62    /// Stops background tasks and removes the socket created by this runtime.
63    fn drop(&mut self) {
64        self.ipc_task.abort();
65        if let Some(task) = self.heartbeat_task.as_ref() {
66            task.abort();
67        }
68        if let Err(error) = std::fs::remove_file(&self.ipc_path)
69            && error.kind() != std::io::ErrorKind::NotFound
70        {
71            tracing::warn!(
72                ipc_path = %self.ipc_path.display(),
73                ?error,
74                "failed to remove dashboard IPC socket"
75            );
76        }
77    }
78}
79
80/// Starts the dashboard IPC runtime for an enabled target configuration.
81///
82/// # Arguments
83///
84/// - `config`: Validated dashboard IPC configuration.
85/// - `audit_config`: Root audit persistence configuration.
86/// - `spec`: Supervisor declaration used to build dashboard state.
87/// - `handle`: Runtime control handle used by command requests.
88///
89/// # Returns
90///
91/// Returns a guard that stops runtime tasks and removes the socket on drop.
92pub fn start_dashboard_ipc_runtime(
93    config: ValidatedDashboardIpcConfig,
94    audit_config: AuditConfig,
95    spec: SupervisorSpec,
96    handle: SupervisorHandle,
97) -> Result<Arc<DashboardIpcRuntimeGuard>, DashboardError> {
98    let listener = bind_dashboard_listener(&config)?;
99    let ipc_path = config.path.clone();
100    let target_id = config.target_id.clone();
101    let service = dashboard_service(config.clone(), audit_config, spec, handle);
102    let ipc_task = tokio::spawn(run_accept_loop(listener, service, target_id));
103    let heartbeat_task = start_heartbeat_task(config);
104
105    Ok(Arc::new(DashboardIpcRuntimeGuard {
106        ipc_path,
107        ipc_task,
108        heartbeat_task,
109    }))
110}
111
112/// Builds the service used by all socket connections.
113///
114/// When `config.security_config` is present, an IPC security pipeline is
115/// constructed with the root audit config and wired into the service via
116/// `with_security_pipeline`.
117fn dashboard_service(
118    config: ValidatedDashboardIpcConfig,
119    audit_config: AuditConfig,
120    spec: SupervisorSpec,
121    handle: SupervisorHandle,
122) -> Arc<DashboardIpcService> {
123    let state = declared_state_from_spec(&spec);
124    let journal = EventJournal::new(spec.event_channel_capacity);
125    let mut service =
126        DashboardIpcService::new(config.clone(), spec, state, journal).with_handle(handle);
127    if let Some(security_config) = config.security_config {
128        let pipeline = IpcSecurityPipeline::new(security_config, audit_config);
129        service = service.with_security_pipeline(pipeline);
130    }
131    Arc::new(service)
132}
133
134/// Starts the dynamic registration heartbeat when registration is enabled.
135fn start_heartbeat_task(config: ValidatedDashboardIpcConfig) -> Option<JoinHandle<()>> {
136    config.registration.as_ref()?;
137    Some(tokio::spawn(async move {
138        if let Err(error) = run_registration_heartbeat(config).await {
139            tracing::warn!(?error, "dashboard registration heartbeat stopped");
140        }
141    }))
142}
143
144/// Accepts target-side IPC connections until the listener fails or is aborted.
145///
146/// Enforces a maximum concurrent connection limit (`MAX_CONCURRENT_CONNECTIONS`)
147/// to prevent resource exhaustion. When the limit is reached, new connections
148/// are accepted but immediately dropped with a log warning.
149async fn run_accept_loop(
150    listener: UnixListener,
151    service: Arc<DashboardIpcService>,
152    target_id: String,
153) {
154    let mut connections = JoinSet::new();
155    loop {
156        tokio::select! {
157            accepted = listener.accept() => {
158                match accepted {
159                    Ok((stream, _)) => {
160                        if connections.len() >= MAX_CONCURRENT_CONNECTIONS {
161                            tracing::warn!(
162                                target: "rust_supervisor::dashboard::ipc",
163                                "dashboard IPC connection limit reached ({MAX_CONCURRENT_CONNECTIONS}), dropping new connection"
164                            );
165                            drop(stream);
166                            continue;
167                        }
168                        let service = Arc::clone(&service);
169                        let target_id = target_id.clone();
170                        connections.spawn(async move {
171                            handle_connection(stream, service, target_id).await
172                        });
173                    }
174                    Err(error) => {
175                        tracing::warn!(?error, "dashboard IPC accept loop stopped");
176                        break;
177                    }
178                }
179            }
180            Some(joined) = connections.join_next() => {
181                match joined {
182                    Ok(Ok(())) => {}
183                    Ok(Err(error)) => {
184                        tracing::warn!(?error, "dashboard IPC connection ended with error");
185                    }
186                    Err(error) => {
187                        tracing::warn!(?error, "dashboard IPC connection task failed");
188                    }
189                }
190            }
191        }
192    }
193}
194
195/// Handles one IPC connection with bounded frame reading, real peer
196/// credential extraction, per-connection unique identifier,
197/// and idle timeout (C10 resource boundary).
198async fn handle_connection(
199    stream: UnixStream,
200    service: Arc<DashboardIpcService>,
201    target_id: String,
202) -> Result<(), DashboardError> {
203    // ---- extract real peer credential before wrapping into tokio ----
204    let std_stream = stream.into_std().map_err(|error| {
205        io_error(
206            "ipc_into_std_failed",
207            "ipc_connect",
208            Some(target_id.clone()),
209            error,
210        )
211    })?;
212    let peer = extract_peer_identity(&std_stream)?;
213    let raw_fd = std_stream.as_raw_fd();
214    let connection_id = format!(
215        "conn-{raw_fd}-{}",
216        CONNECTION_COUNTER.fetch_add(1, Ordering::Relaxed)
217    );
218    let stream = UnixStream::from_std(std_stream).map_err(|error| {
219        io_error(
220            "ipc_from_std_failed",
221            "ipc_connect",
222            Some(target_id.clone()),
223            error,
224        )
225    })?;
226
227    // Use C5 request size limit when security pipeline is configured,
228    // otherwise fall back to 1 MiB default.
229    let max_frame_bytes = service.max_frame_bytes();
230    let mut reader = BoundedFrameReader::new(stream, max_frame_bytes);
231    loop {
232        // Enforce per-connection idle timeout: if no complete frame
233        // arrives within CONNECTION_IDLE_TIMEOUT, drop the connection.
234        let frame = tokio::time::timeout(CONNECTION_IDLE_TIMEOUT, reader.read_frame()).await;
235        match frame {
236            Ok(Ok(Some(raw_frame))) => {
237                let raw_body_len = raw_frame.len();
238                let response =
239                    response_for_line(&service, &raw_frame, &peer, &connection_id, raw_body_len)
240                        .await;
241                write_response(&mut reader, &response, &target_id).await?;
242            }
243            Ok(Ok(None)) => {
244                // EOF — peer closed connection gracefully
245                return Ok(());
246            }
247            Ok(Err(error)) => {
248                return Err(error);
249            }
250            Err(_elapsed) => {
251                tracing::warn!(
252                    target: "rust_supervisor::dashboard::ipc",
253                    %connection_id,
254                    "dashboard IPC connection idle timeout after {}s",
255                    CONNECTION_IDLE_TIMEOUT.as_secs(),
256                );
257                return Err(DashboardError::new(
258                    "ipc_idle_timeout",
259                    "ipc_read",
260                    Some(target_id),
261                    "connection idle timeout".to_owned(),
262                    false,
263                ));
264            }
265        }
266    }
267}
268
269/// Bounded frame reader that limits each frame to `max_bytes` before
270/// allocating the target buffer.
271struct BoundedFrameReader {
272    /// Inner tokio stream.
273    stream: UnixStream,
274    /// Maximum frame size in bytes.
275    max_bytes: usize,
276    /// Read buffer reused across frames.
277    buf: Vec<u8>,
278}
279
280impl BoundedFrameReader {
281    /// Creates a new bounded frame reader.
282    fn new(stream: UnixStream, max_bytes: usize) -> Self {
283        Self {
284            stream,
285            max_bytes,
286            buf: Vec::with_capacity(max_bytes.min(4096)),
287        }
288    }
289
290    /// Reads one newline-delimited frame.
291    ///
292    /// Returns `Ok(Some(frame))` for a complete frame, `Ok(None)` for EOF
293    /// before any data, or `Err` when the frame exceeds `max_bytes` or a
294    /// read error occurs.
295    async fn read_frame(&mut self) -> Result<Option<String>, DashboardError> {
296        self.buf.clear();
297        loop {
298            let mut byte = [0u8; 1];
299            match self.stream.read_exact(&mut byte).await {
300                Ok(_bytes_read) => {
301                    if byte[0] == b'\n' {
302                        let frame = String::from_utf8(self.buf.clone()).map_err(|_| {
303                            DashboardError::new(
304                                "invalid_utf8",
305                                "ipc_read",
306                                None,
307                                "frame is not valid UTF-8".to_owned(),
308                                false,
309                            )
310                        })?;
311                        return Ok(Some(frame));
312                    }
313                    self.buf.push(byte[0]);
314                    if self.buf.len() > self.max_bytes {
315                        return Err(DashboardError::new(
316                            "frame_too_large",
317                            "ipc_read",
318                            None,
319                            format!(
320                                "frame exceeded maximum size of {max} bytes",
321                                max = self.max_bytes
322                            ),
323                            false,
324                        ));
325                    }
326                }
327                Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => {
328                    if self.buf.is_empty() {
329                        return Ok(None);
330                    }
331                    return Err(DashboardError::new(
332                        "incomplete_frame",
333                        "ipc_read",
334                        None,
335                        "connection closed before newline delimiter".to_owned(),
336                        false,
337                    ));
338                }
339                Err(err) => {
340                    return Err(io_error("ipc_read_failed", "ipc_read", None, err));
341                }
342            }
343        }
344    }
345
346    /// Returns a mutable reference to the inner stream for writing.
347    fn stream_mut(&mut self) -> &mut UnixStream {
348        &mut self.stream
349    }
350}
351
352impl std::os::unix::io::AsRawFd for BoundedFrameReader {
353    /// Returns the raw file descriptor for readiness polling.
354    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
355        self.stream.as_raw_fd()
356    }
357}
358
359/// Converts one request line into a response, passing connection context.
360async fn response_for_line(
361    service: &DashboardIpcService,
362    line: &str,
363    peer: &PeerIdentity,
364    connection_id: &str,
365    raw_body_len: usize,
366) -> IpcResponse {
367    match parse_request_line(line) {
368        Ok(request) => {
369            service
370                .handle_request(request, peer, connection_id, raw_body_len)
371                .await
372        }
373        Err(error) => IpcResponse::error("invalid-request", error),
374    }
375}
376
377/// Writes one response line to the socket.
378async fn write_response(
379    reader: &mut BoundedFrameReader,
380    response: &IpcResponse,
381    target_id: &str,
382) -> Result<(), DashboardError> {
383    let line = response_to_line(response)?;
384    reader
385        .stream_mut()
386        .write_all(line.as_bytes())
387        .await
388        .map_err(|error| {
389            io_error(
390                "ipc_write_failed",
391                "ipc_write",
392                Some(target_id.to_owned()),
393                error,
394            )
395        })
396}
397
398/// Creates a structured IPC runtime I/O error.
399fn io_error(
400    code: &str,
401    stage: &str,
402    target_id: Option<String>,
403    error: std::io::Error,
404) -> DashboardError {
405    DashboardError::new(code, stage, target_id, error.to_string(), true)
406}
rust_supervisor/dashboard/runtime.rs

rust_supervisor/dashboard/
runtime.rs
