@@ -157,7 +157,11 @@ async fn main() -> Result<()> {
157157 info ! ( "Shutdown complete" ) ;
158158 return Ok ( ( ) ) ;
159159 }
160- Ok ( ServeOutcome :: Disconnected ) => info ! ( "Connection closed gracefully" ) ,
160+ // The specific reason (server Close, socket error, stream end,
161+ // registration failure) is already logged at the break site inside
162+ // serve(); avoid claiming "gracefully" here since Disconnected also
163+ // covers error paths. main() only needs to know: reconnect.
164+ Ok ( ServeOutcome :: Disconnected ) => info ! ( "Connection closed, reconnecting" ) ,
161165 Err ( e) => error ! ( "Connection error: {}" , e) ,
162166 }
163167
@@ -348,10 +352,25 @@ where
348352 tokio:: pin!( shutdown) ;
349353 let outcome = loop {
350354 tokio:: select! {
351- // Prefer draining inbound messages; the signal branch still fires
352- // promptly because recv() yields between messages.
355+ // Poll shutdown FIRST. With `biased`, tokio checks branches top to
356+ // bottom and only reaches a later branch when earlier ones are
357+ // Pending; if the message branch were first and inbound messages were
358+ // continuously ready (buffered / high throughput), the shutdown branch
359+ // could starve and delay the Close frame + process exit. Shutdown is
360+ // usually Pending, so in normal operation this falls straight through
361+ // to draining messages; it can't starve them because shutdown fires
362+ // once and breaks the loop.
353363 biased;
354364
365+ _ = & mut shutdown => {
366+ // Best-effort clean close so the controller deregisters us now.
367+ let mut tx = ws_tx_clone. lock( ) . await ;
368+ if let Err ( e) = tx. send( WsMessage :: Close ( None ) ) . await {
369+ warn!( "Failed to send Close frame on shutdown: {}" , e) ;
370+ }
371+ break ServeOutcome :: Shutdown ;
372+ }
373+
355374 msg = ws_rx. next( ) => {
356375 let msg = match msg {
357376 Some ( Ok ( WsMessage :: Text ( text) ) ) => text,
@@ -391,15 +410,6 @@ where
391410 error!( "Error handling message: {}" , e) ;
392411 }
393412 }
394-
395- _ = & mut shutdown => {
396- // Best-effort clean close so the controller deregisters us now.
397- let mut tx = ws_tx_clone. lock( ) . await ;
398- if let Err ( e) = tx. send( WsMessage :: Close ( None ) ) . await {
399- warn!( "Failed to send Close frame on shutdown: {}" , e) ;
400- }
401- break ServeOutcome :: Shutdown ;
402- }
403413 }
404414 } ;
405415
@@ -892,17 +902,23 @@ mod shutdown_tests {
892902 async fn shutdown_sends_close_and_returns_shutdown ( ) {
893903 let ( client, mut server) = ws_pair ( ) . await ;
894904
895- // Fire shutdown shortly after serve starts its loop. A ready future
896- // would race registration; a short delay keeps the test deterministic
897- // without depending on wall-clock timing for correctness.
905+ // Trigger shutdown via a oneshot fired AFTER registration is acked, so
906+ // the ordering is deterministic (no wall-clock sleep racing the
907+ // handshake). `serve` awaits the receiver; mapping away the RecvError
908+ // matches its `Future<Output = ()>` bound and means a dropped sender also
909+ // resolves shutdown rather than hanging.
910+ let ( shutdown_tx, shutdown_rx) = tokio:: sync:: oneshot:: channel :: < ( ) > ( ) ;
898911 let shutdown = async {
899- tokio :: time :: sleep ( Duration :: from_millis ( 50 ) ) . await ;
912+ let _ = shutdown_rx . await ;
900913 } ;
901914
902915 let daemon = serve ( client, "test-daemon" , 3600 , HashMap :: new ( ) , shutdown) ;
903916
904917 let controller = async {
905918 ack_registration ( & mut server) . await ;
919+ // Registration is complete and serve() is now in its select! loop;
920+ // fire shutdown deterministically.
921+ shutdown_tx. send ( ( ) ) . unwrap ( ) ;
906922 // The controller side should observe a Close frame. Tolerate any
907923 // pre-close traffic (e.g. a heartbeat), though the 3600s interval
908924 // makes that unlikely in-test.
0 commit comments