From b47b3bc3b630b53a457d7972a860458e6b304360 Mon Sep 17 00:00:00 2001
From: Leonardo Rodoni <rodonile@gmail.com>
Date: Mon, 7 Sep 2026 12:21:49 +0200
Subject: [PATCH] feat(flow-service,udp-notif-service): expose actor setup
 errors

Previously, if the actor task exited (e.g. due to a socket bind
error) before replying with its local address, the handle's new()
would return a generic ReceiveError, discarding the actual cause.

Race the reply channel against the actor's JoinHandle so that when
the actor dies first, its real error is propagated via the new
ActorFailed variant instead of being masked.
---
 crates/flow-service/src/flow_actor.rs | 35 ++++++++++++++++++++++-----
 crates/udp-notif-service/src/actor.rs | 29 ++++++++++++++++++++--
 2 files changed, 56 insertions(+), 8 deletions(-)

diff --git a/crates/flow-service/src/flow_actor.rs b/crates/flow-service/src/flow_actor.rs
index ebd37110..7979322f 100644
--- a/crates/flow-service/src/flow_actor.rs
+++ b/crates/flow-service/src/flow_actor.rs
@@ -892,6 +892,8 @@ pub enum FlowCollectorActorHandleError {
     SendError,
     #[strum(to_string = "Error receiving response from actor")]
     ReceiveError,
+    #[strum(to_string = "actor terminated before it could start: {0}")]
+    ActorFailed(FlowCollectorActorError),
 }
 
 impl std::error::Error for FlowCollectorActorHandleError {
@@ -899,6 +901,9 @@ impl std::error::Error for FlowCollectorActorHandleError {
         match *self {
             FlowCollectorActorHandleError::SendError => "Error sending command to actor",
             FlowCollectorActorHandleError::ReceiveError => "Error receiving response from actor",
+            FlowCollectorActorHandleError::ActorFailed(_) => {
+                "actor terminated before it could start"
+            }
         }
     }
 
@@ -954,17 +959,35 @@ impl FlowCollectorActorHandle {
             subscriber_timeout,
             stats,
         );
-        let join_handle = tokio::spawn(actor.run());
+        let mut join_handle = tokio::spawn(actor.run());
         let (tx, mut rx) = mpsc::channel(cmd_buffer_size);
         cmd_tx
             .send(FlowCollectorActorCommand::LocalAddr(tx))
             .await
             .map_err(|_| FlowCollectorActorHandleError::SendError)?;
-        let local_addr = rx
-            .recv()
-            .await
-            .ok_or(FlowCollectorActorHandleError::ReceiveError)?
-            .1;
+        let local_addr = tokio::select! {
+            biased;
+            res = &mut join_handle => {
+                return Err(match res {
+                    Ok(Err(err)) => FlowCollectorActorHandleError::ActorFailed(err),
+                    _ => FlowCollectorActorHandleError::ReceiveError,
+                });
+            }
+            local = rx.recv() => {
+                match local {
+                    Some((_, addr)) => addr,
+                    // The reply sender was dropped without a response, most likely
+                    // because the actor task exited early (e.g. socket bind
+                    // failure). Await the join handle to surface the real cause.
+                    None => {
+                        return Err(match join_handle.await {
+                            Ok(Err(err)) => FlowCollectorActorHandleError::ActorFailed(err),
+                            _ => FlowCollectorActorHandleError::ReceiveError,
+                        });
+                    }
+                }
+            }
+        };
         Ok((
             join_handle,
             Self {
diff --git a/crates/udp-notif-service/src/actor.rs b/crates/udp-notif-service/src/actor.rs
index 4afc3898..a31ac4e3 100644
--- a/crates/udp-notif-service/src/actor.rs
+++ b/crates/udp-notif-service/src/actor.rs
@@ -972,6 +972,8 @@ pub enum ActorHandleError {
     SendError,
     #[strum(to_string = "error receiving response from actor")]
     ReceiveError,
+    #[strum(to_string = "actor terminated before it could start: {0}")]
+    ActorFailed(UdpNotifActorError),
 }
 
 impl std::error::Error for ActorHandleError {
@@ -979,6 +981,7 @@ impl std::error::Error for ActorHandleError {
         match *self {
             ActorHandleError::SendError => "error sending command to actor",
             ActorHandleError::ReceiveError => "error receiving response from actor",
+            ActorHandleError::ActorFailed(_) => "actor terminated before it could start",
         }
     }
 
@@ -1034,13 +1037,35 @@ impl ActorHandle {
             reassembly_timeout,
             stats,
         );
-        let join_handle = tokio::spawn(actor.run());
+        let mut join_handle = tokio::spawn(actor.run());
         let (tx, mut rx) = mpsc::channel(cmd_buffer_size);
         cmd_tx
             .send(ActorCommand::LocalAddr(tx))
             .await
             .map_err(|_| ActorHandleError::SendError)?;
-        let local_addr = rx.recv().await.ok_or(ActorHandleError::ReceiveError)?.1;
+        let local_addr = tokio::select! {
+            biased;
+            res = &mut join_handle => {
+                return Err(match res {
+                    Ok(Err(err)) => ActorHandleError::ActorFailed(err),
+                    _ => ActorHandleError::ReceiveError,
+                });
+            }
+            local = rx.recv() => {
+                match local {
+                    Some((_, addr)) => addr,
+                    // The reply sender was dropped without a response, most likely
+                    // because the actor task exited early (e.g. socket bind
+                    // failure). Await the join handle to surface the real cause.
+                    None => {
+                        return Err(match join_handle.await {
+                            Ok(Err(err)) => ActorHandleError::ActorFailed(err),
+                            _ => ActorHandleError::ReceiveError,
+                        });
+                    }
+                }
+            }
+        };
         Ok((
             join_handle,
             Self {