@@ -283,11 +283,73 @@ struct RecoveredBoundary {
283283}
284284
285285/// Fence the previous owner through the runtime's recovery extension.
286+ /// How long a candidate keeps retrying `RecoveryHello` before giving up.
287+ /// The hello is retried BEFORE any grant is requested, so a transient
288+ /// failure (e.g. racing a relinquishing owner's connection teardown) never
289+ /// consumes the reserved epoch's single grant.
290+ const RECOVERY_HELLO_RETRY_BUDGET : Duration = Duration :: from_secs ( 4 ) ;
291+ const RECOVERY_HELLO_RETRY_INTERVAL : Duration = Duration :: from_millis ( 250 ) ;
292+ const RECOVERY_HELLO_ATTEMPT_TIMEOUT : Duration = Duration :: from_millis ( 1_500 ) ;
293+
294+ async fn recovery_hello ( client : & BoundaryClient ) -> Result < Response , BackendError > {
295+ retry_within_budget (
296+ RECOVERY_HELLO_RETRY_BUDGET ,
297+ RECOVERY_HELLO_ATTEMPT_TIMEOUT ,
298+ RECOVERY_HELLO_RETRY_INTERVAL ,
299+ || client. call_once ( Request :: RecoveryHello ) ,
300+ )
301+ . await
302+ }
303+
304+ /// Retry `attempt` on transient failures until `budget` is spent. Every
305+ /// attempt is bounded by the smaller of `attempt_timeout` and the time left,
306+ /// and no attempt or wait starts past the deadline, so the whole call never
307+ /// exceeds `budget`. A response or a definitive refusal (denied, terminated,
308+ /// unsupported) is returned immediately.
309+ async fn retry_within_budget < F , Fut > (
310+ budget : Duration ,
311+ attempt_timeout : Duration ,
312+ retry_interval : Duration ,
313+ attempt : F ,
314+ ) -> Result < Response , BackendError >
315+ where
316+ F : Fn ( ) -> Fut ,
317+ Fut : Future < Output = Result < Response , BackendError > > ,
318+ {
319+ let deadline = tokio:: time:: Instant :: now ( ) + budget;
320+ loop {
321+ let attempt_deadline = ( tokio:: time:: Instant :: now ( ) + attempt_timeout) . min ( deadline) ;
322+ let result = tokio:: time:: timeout_at ( attempt_deadline, attempt ( ) )
323+ . await
324+ . unwrap_or_else ( |_| {
325+ Err ( BackendError :: Unavailable (
326+ "recovery hello timed out" . to_string ( ) ,
327+ ) )
328+ } ) ;
329+ match result {
330+ // A response, or a definitive refusal, is final.
331+ Ok ( response) => return Ok ( response) ,
332+ Err (
333+ error @ ( BackendError :: Denied ( _)
334+ | BackendError :: Terminated ( _)
335+ | BackendError :: Unsupported ( _) ) ,
336+ ) => return Err ( error) ,
337+ Err ( error) => {
338+ if tokio:: time:: Instant :: now ( ) + retry_interval >= deadline {
339+ return Err ( error) ;
340+ }
341+ tracing:: debug!( %error, "Recovery hello failed; retrying before requesting a grant" ) ;
342+ tokio:: time:: sleep ( retry_interval) . await ;
343+ }
344+ }
345+ }
346+ }
347+
286348async fn recover_boundary (
287349 client : & BoundaryClient ,
288350 ownership : & SupervisorOwnership ,
289351) -> Result < RecoveredBoundary , BackendError > {
290- let response = client . call_once ( Request :: RecoveryHello ) . await ?;
352+ let response = recovery_hello ( client ) . await ?;
291353 let Response :: RecoveryChallenge {
292354 protocol,
293355 challenge,
@@ -4380,3 +4442,61 @@ mod tests {
43804442 let _ = server. await ;
43814443 }
43824444}
4445+
4446+ #[ cfg( test) ]
4447+ mod recovery_hello_tests {
4448+ use super :: { BackendError , Response , retry_within_budget} ;
4449+ use std:: sync:: atomic:: { AtomicUsize , Ordering } ;
4450+ use std:: time:: Duration ;
4451+
4452+ const BUDGET : Duration = Duration :: from_secs ( 4 ) ;
4453+ const ATTEMPT : Duration = Duration :: from_millis ( 1_500 ) ;
4454+ const INTERVAL : Duration = Duration :: from_millis ( 250 ) ;
4455+
4456+ #[ tokio:: test( start_paused = true ) ]
4457+ async fn a_hanging_runtime_never_exceeds_the_hello_budget ( ) {
4458+ let attempts = AtomicUsize :: new ( 0 ) ;
4459+ let started = tokio:: time:: Instant :: now ( ) ;
4460+ let result = retry_within_budget ( BUDGET , ATTEMPT , INTERVAL , || {
4461+ attempts. fetch_add ( 1 , Ordering :: SeqCst ) ;
4462+ std:: future:: pending :: < Result < Response , BackendError > > ( )
4463+ } )
4464+ . await ;
4465+ assert ! ( matches!( result, Err ( BackendError :: Unavailable ( _) ) ) ) ;
4466+ assert ! (
4467+ started. elapsed( ) <= BUDGET ,
4468+ "hello took {:?}, budget {BUDGET:?}" ,
4469+ started. elapsed( )
4470+ ) ;
4471+ // 1.5 s + 0.25 s + 1.5 s + 0.25 s + remaining 0.5 s: the last attempt
4472+ // is cut to the time left instead of a full 1.5 s.
4473+ assert_eq ! ( attempts. load( Ordering :: SeqCst ) , 3 ) ;
4474+ }
4475+
4476+ #[ tokio:: test( start_paused = true ) ]
4477+ async fn transient_failures_are_retried_and_definitive_ones_are_not ( ) {
4478+ let attempts = AtomicUsize :: new ( 0 ) ;
4479+ let result = retry_within_budget ( BUDGET , ATTEMPT , INTERVAL , || {
4480+ let n = attempts. fetch_add ( 1 , Ordering :: SeqCst ) ;
4481+ async move {
4482+ if n < 2 {
4483+ Err ( BackendError :: Unavailable ( "connection refused" . to_string ( ) ) )
4484+ } else {
4485+ Ok ( Response :: MediationReady )
4486+ }
4487+ }
4488+ } )
4489+ . await ;
4490+ assert ! ( matches!( result, Ok ( Response :: MediationReady ) ) ) ;
4491+ assert_eq ! ( attempts. load( Ordering :: SeqCst ) , 3 ) ;
4492+
4493+ let denied_attempts = AtomicUsize :: new ( 0 ) ;
4494+ let denied = retry_within_budget ( BUDGET , ATTEMPT , INTERVAL , || {
4495+ denied_attempts. fetch_add ( 1 , Ordering :: SeqCst ) ;
4496+ async { Err ( BackendError :: Terminated ( "terminating" . to_string ( ) ) ) }
4497+ } )
4498+ . await ;
4499+ assert ! ( matches!( denied, Err ( BackendError :: Terminated ( _) ) ) ) ;
4500+ assert_eq ! ( denied_attempts. load( Ordering :: SeqCst ) , 1 ) ;
4501+ }
4502+ }
0 commit comments