fix(flow-service,udp-notif-service): surface actor setup failure on early exit

crates/flow-service/src/flow_actor.rs

@@ -892,13 +892,18 @@ 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 {
     fn description(&self) -> &str {
         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 {

crates/udp-notif-service/src/actor.rs

@@ -972,13 +972,16 @@ 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 {
     fn description(&self) -> &str {
         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 {