fix: decouple different ipc data channels

In some situations the ipc channels from actors to the background task
can get overloaded. Currently we don't have a complete plan for how to
handle this. But the major source of messages is telemetry, which is
also one we can treat less reliably, so by dividing this from the data
messages we can deprioritize them.

Signed-off-by: Wim Looman <[email protected]>

Author: Nemo157

veecle-ipc/Cargo.toml

@@ -28,6 +28,7 @@ veecle-telemetry = { workspace = true, features = ["alloc"] }
 rand = { workspace = true, features = ["std_rng"] }
 serde = { workspace = true, features = ["derive"] }
 veecle-os = { workspace = true }
+veecle-telemetry = { workspace = true, features = ["alloc", "enable"] }
 
 [lints]
 workspace = true

veecle-ipc/src/actors/output.rs

@@ -52,13 +52,14 @@ pub async fn output<T>(
 where
     T: Storable<DataType: Serialize> + 'static,
 {
-    let output = config.connector.output();
+    let output = config.connector.storable_output();
     let send_policy = config.send_policy;
 
     loop {
-        let value = reader.wait_for_update().await.read(|value| {
-            veecle_ipc_protocol::Message::Storable(EncodedStorable::new(value).unwrap())
-        });
+        let value = reader
+            .wait_for_update()
+            .await
+            .read(|value| EncodedStorable::new(value).unwrap());
 
         match send_policy {
             SendPolicy::Drop => {

veecle-ipc/src/connector.rs

@@ -8,18 +8,77 @@ use tokio::net::UnixStream;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tokio_util::codec::Framed;
-use veecle_ipc_protocol::{Codec, ControlRequest, ControlResponse, Message};
+use veecle_ipc_protocol::{Codec, ControlRequest, ControlResponse, EncodedStorable, Message};
 
 use crate::Exporter;
 
 type Inputs = Arc<Mutex<HashMap<&'static str, mpsc::Sender<String>>>>;
 
+/// Holds various output channel senders for the [`Connector`], separated so they have decoupled
+/// buffering and prioritization.
+#[derive(Debug)]
+struct OutputTx {
+    storable: mpsc::Sender<EncodedStorable>,
+    telemetry: mpsc::Sender<veecle_telemetry::protocol::InstanceMessage<'static>>,
+    control: mpsc::Sender<ControlRequest>,
+}
+
+/// The receivers for [`OutputTx`].
+#[derive(Debug)]
+struct OutputRx {
+    storable: mpsc::Receiver<EncodedStorable>,
+    telemetry: mpsc::Receiver<veecle_telemetry::protocol::InstanceMessage<'static>>,
+    control: mpsc::Receiver<ControlRequest>,
+}
+
+impl OutputRx {
+    /// Returns the first message available on any output channel.
+    ///
+    /// Purposefully prioritizes the more important channels to drain first.
+    /// This may lead to the low priority channels never being serviced if we are not keeping up.
+    async fn recv(&mut self) -> Option<Message<'static>> {
+        Some(tokio::select! {
+            biased; // Polls all branches in order to guarantee prioritization.
+            Some(control) = self.control.recv() => Message::ControlRequest(control),
+            Some(storable) = self.storable.recv() => Message::Storable(storable),
+            Some(telemetry) = self.telemetry.recv() => Message::Telemetry(telemetry),
+            else => return None, // Only reached when all channels are closed.
+        })
+    }
+}
+
+fn outputs() -> (OutputTx, OutputRx) {
+    // Control requests are request-response so there should never be buffering as the sender will
+    // be waiting on a response.
+    let (control_tx, control_rx) = mpsc::channel(1);
+    // The output channel capacity (128) determines buffering for IPC messages.
+    // The `Output` actor uses `SendPolicy` to control behavior when this fills up:
+    // - `SendPolicy::Panic` (default): panics to make buffer exhaustion visible
+    // - `SendPolicy::Drop`: drops messages and logs a warning
+    let (storable_tx, storable_rx) = mpsc::channel(128);
+    // Telemetry can be quite chatty, so give it a large buffer, the `Exporter` will discard
+    // messages if this is filled.
+    let (telemetry_tx, telemetry_rx) = mpsc::channel(128);
+
+    (
+        OutputTx {
+            storable: storable_tx,
+            control: control_tx,
+            telemetry: telemetry_tx,
+        },
+        OutputRx {
+            storable: storable_rx,
+            control: control_rx,
+            telemetry: telemetry_rx,
+        },
+    )
+}
+
 /// Manages the connection to other runtimes via the `veecle-orchestrator`.
 #[derive(Debug)]
 pub struct Connector {
-    output: mpsc::Sender<Message<'static>>,
+    output_tx: OutputTx,
     inputs: Inputs,
-    control_requests: mpsc::Sender<ControlRequest>,
     control_responses: Mutex<Option<mpsc::Receiver<ControlResponse>>>,
     _task: JoinHandle<Result<(), Box<dyn std::error::Error + Send + Sync>>>,
 }
@@ -38,13 +97,9 @@ impl Connector {
         let stream = UnixStream::connect(&socket).await.unwrap();
         let mut stream = Framed::new(stream, Codec::new());
 
-        // The output channel capacity (128) determines buffering for IPC messages.
-        // The `Output` actor uses `SendPolicy` to control behavior when this fills up:
-        // - `SendPolicy::Panic` (default): panics to make buffer exhaustion visible
-        // - `SendPolicy::Drop`: drops messages and logs a warning
-        let (output, mut output_rx) = mpsc::channel(128);
         let inputs = Inputs::default();
-        let (control_request_tx, mut control_request_rx) = mpsc::channel(16);
+        let (output_tx, mut output_rx) = outputs();
+
         let (control_response_tx, control_response_rx) = mpsc::channel(16);
         let task = tokio::spawn({
             let inputs = inputs.clone();
@@ -55,11 +110,6 @@ impl Connector {
                             let Some(message) = message else { break };
                             stream.send(&message).await?;
                         }
-                        request = control_request_rx.recv() => {
-                            let Some(request) = request else { break };
-                            let message = Message::ControlRequest(request);
-                            stream.send(&message).await?;
-                        }
                         message = stream.next() => {
                             let Some(message) = message else { break };
                             let message = match message {
@@ -96,9 +146,8 @@ impl Connector {
         });
 
         Self {
-            output,
+            output_tx,
             inputs,
-            control_requests: control_request_tx,
             control_responses: Mutex::new(Some(control_response_rx)),
             _task: task,
         }
@@ -122,7 +171,7 @@ impl Connector {
     /// # }
     /// ```
     pub fn exporter(&self) -> Exporter {
-        Exporter::new(self.output.clone())
+        Exporter::new(self.output_tx.telemetry.clone())
     }
 
     /// Registers a new channel that will receive input from the `veecle-orchestrator` tagged with `type_name`.
@@ -138,8 +187,8 @@ impl Connector {
     }
 
     /// Gets a new sender to send values to the `veecle-orchestrator`.
-    pub(crate) fn output(&self) -> mpsc::Sender<Message<'static>> {
-        self.output.clone()
+    pub(crate) fn storable_output(&self) -> mpsc::Sender<EncodedStorable> {
+        self.output_tx.storable.clone()
     }
 
     /// Gets the sender and receiver to send control messages and receive control responses from the `veecle-orchestrator`.
@@ -152,7 +201,7 @@ impl Connector {
         mpsc::Receiver<ControlResponse>,
     ) {
         (
-            self.control_requests.clone(),
+            self.output_tx.control.clone(),
             self.control_responses
                 .lock()
                 .unwrap()

veecle-ipc/src/telemetry.rs

@@ -6,12 +6,12 @@ use veecle_telemetry::to_static::ToStatic;
 /// An [`Export`] implementer that forwards telemetry messages via IPC.
 #[derive(Debug)]
 pub struct Exporter {
-    sender: mpsc::Sender<veecle_ipc_protocol::Message<'static>>,
+    sender: mpsc::Sender<InstanceMessage<'static>>,
 }
 
 impl Exporter {
     /// Creates a new IPC telemetry exporter.
-    pub fn new(sender: mpsc::Sender<veecle_ipc_protocol::Message<'static>>) -> Self {
+    pub fn new(sender: mpsc::Sender<InstanceMessage<'static>>) -> Self {
         Self { sender }
     }
 }
@@ -23,8 +23,7 @@ impl Export for Exporter {
     /// and sends it through the IPC channel. If the channel is full or closed,
     /// the message is dropped to avoid blocking telemetry collection.
     fn export(&self, message: InstanceMessage<'_>) {
-        let message = veecle_ipc_protocol::Message::Telemetry(message.to_static());
-        let _ = self.sender.try_send(message);
+        let _ = self.sender.try_send(message.to_static());
     }
 }
 
@@ -60,20 +59,15 @@ mod tests {
 
         exporter.export(test_message);
 
-        let received = receiver.recv().await.expect("should receive message");
-        match received {
-            veecle_ipc_protocol::Message::Telemetry(message) => {
-                assert_eq!(message.thread_id, THREAD_ID);
-                match message.message {
-                    TelemetryMessage::Log(message) => {
-                        assert_eq!(message.time_unix_nano, 1000000000);
-                        assert_eq!(message.severity, Severity::Info);
-                        assert_eq!(message.body.as_ref(), "test log message");
-                    }
-                    _ => panic!("Expected Log message"),
-                }
+        let message = receiver.recv().await.expect("should receive message");
+        assert_eq!(message.thread_id, THREAD_ID);
+        match message.message {
+            TelemetryMessage::Log(message) => {
+                assert_eq!(message.time_unix_nano, 1000000000);
+                assert_eq!(message.severity, Severity::Info);
+                assert_eq!(message.body.as_ref(), "test log message");
             }
-            _ => panic!("Expected Telemetry"),
+            _ => panic!("Expected Log message"),
         }
     }