@@ -130,11 +130,73 @@ struct RecoveredBoundary {
130130}
131131
132132/// Fence the previous owner through the runtime's recovery extension.
133+ /// How long a candidate keeps retrying `RecoveryHello` before giving up.
134+ /// The hello is retried BEFORE any grant is requested, so a transient
135+ /// failure (e.g. racing a relinquishing owner's connection teardown) never
136+ /// consumes the reserved epoch's single grant.
137+ const RECOVERY_HELLO_RETRY_BUDGET : Duration = Duration :: from_secs ( 4 ) ;
138+ const RECOVERY_HELLO_RETRY_INTERVAL : Duration = Duration :: from_millis ( 250 ) ;
139+ const RECOVERY_HELLO_ATTEMPT_TIMEOUT : Duration = Duration :: from_millis ( 1_500 ) ;
140+
141+ async fn recovery_hello ( client : & BoundaryClient ) -> Result < Response , BackendError > {
142+ retry_within_budget (
143+ RECOVERY_HELLO_RETRY_BUDGET ,
144+ RECOVERY_HELLO_ATTEMPT_TIMEOUT ,
145+ RECOVERY_HELLO_RETRY_INTERVAL ,
146+ || client. call_once ( Request :: RecoveryHello ) ,
147+ )
148+ . await
149+ }
150+
151+ /// Retry `attempt` on transient failures until `budget` is spent. Every
152+ /// attempt is bounded by the smaller of `attempt_timeout` and the time left,
153+ /// and no attempt or wait starts past the deadline, so the whole call never
154+ /// exceeds `budget`. A response or a definitive refusal (denied, terminated,
155+ /// unsupported) is returned immediately.
156+ async fn retry_within_budget < F , Fut > (
157+ budget : Duration ,
158+ attempt_timeout : Duration ,
159+ retry_interval : Duration ,
160+ attempt : F ,
161+ ) -> Result < Response , BackendError >
162+ where
163+ F : Fn ( ) -> Fut ,
164+ Fut : Future < Output = Result < Response , BackendError > > ,
165+ {
166+ let deadline = tokio:: time:: Instant :: now ( ) + budget;
167+ loop {
168+ let attempt_deadline = ( tokio:: time:: Instant :: now ( ) + attempt_timeout) . min ( deadline) ;
169+ let result = tokio:: time:: timeout_at ( attempt_deadline, attempt ( ) )
170+ . await
171+ . unwrap_or_else ( |_| {
172+ Err ( BackendError :: Unavailable (
173+ "recovery hello timed out" . to_string ( ) ,
174+ ) )
175+ } ) ;
176+ match result {
177+ // A response, or a definitive refusal, is final.
178+ Ok ( response) => return Ok ( response) ,
179+ Err (
180+ error @ ( BackendError :: Denied ( _)
181+ | BackendError :: Terminated ( _)
182+ | BackendError :: Unsupported ( _) ) ,
183+ ) => return Err ( error) ,
184+ Err ( error) => {
185+ if tokio:: time:: Instant :: now ( ) + retry_interval >= deadline {
186+ return Err ( error) ;
187+ }
188+ tracing:: debug!( %error, "Recovery hello failed; retrying before requesting a grant" ) ;
189+ tokio:: time:: sleep ( retry_interval) . await ;
190+ }
191+ }
192+ }
193+ }
194+
133195async fn recover_boundary (
134196 client : & BoundaryClient ,
135197 ownership : & SupervisorOwnership ,
136198) -> Result < RecoveredBoundary , BackendError > {
137- let response = client . call_once ( Request :: RecoveryHello ) . await ?;
199+ let response = recovery_hello ( client ) . await ?;
138200 let Response :: RecoveryChallenge {
139201 protocol,
140202 challenge,
@@ -3920,3 +3982,61 @@ mod tests {
39203982 let _ = server. await ;
39213983 }
39223984}
3985+
3986+ #[ cfg( test) ]
3987+ mod recovery_hello_tests {
3988+ use super :: { BackendError , Response , retry_within_budget} ;
3989+ use std:: sync:: atomic:: { AtomicUsize , Ordering } ;
3990+ use std:: time:: Duration ;
3991+
3992+ const BUDGET : Duration = Duration :: from_secs ( 4 ) ;
3993+ const ATTEMPT : Duration = Duration :: from_millis ( 1_500 ) ;
3994+ const INTERVAL : Duration = Duration :: from_millis ( 250 ) ;
3995+
3996+ #[ tokio:: test( start_paused = true ) ]
3997+ async fn a_hanging_runtime_never_exceeds_the_hello_budget ( ) {
3998+ let attempts = AtomicUsize :: new ( 0 ) ;
3999+ let started = tokio:: time:: Instant :: now ( ) ;
4000+ let result = retry_within_budget ( BUDGET , ATTEMPT , INTERVAL , || {
4001+ attempts. fetch_add ( 1 , Ordering :: SeqCst ) ;
4002+ std:: future:: pending :: < Result < Response , BackendError > > ( )
4003+ } )
4004+ . await ;
4005+ assert ! ( matches!( result, Err ( BackendError :: Unavailable ( _) ) ) ) ;
4006+ assert ! (
4007+ started. elapsed( ) <= BUDGET ,
4008+ "hello took {:?}, budget {BUDGET:?}" ,
4009+ started. elapsed( )
4010+ ) ;
4011+ // 1.5 s + 0.25 s + 1.5 s + 0.25 s + remaining 0.5 s: the last attempt
4012+ // is cut to the time left instead of a full 1.5 s.
4013+ assert_eq ! ( attempts. load( Ordering :: SeqCst ) , 3 ) ;
4014+ }
4015+
4016+ #[ tokio:: test( start_paused = true ) ]
4017+ async fn transient_failures_are_retried_and_definitive_ones_are_not ( ) {
4018+ let attempts = AtomicUsize :: new ( 0 ) ;
4019+ let result = retry_within_budget ( BUDGET , ATTEMPT , INTERVAL , || {
4020+ let n = attempts. fetch_add ( 1 , Ordering :: SeqCst ) ;
4021+ async move {
4022+ if n < 2 {
4023+ Err ( BackendError :: Unavailable ( "connection refused" . to_string ( ) ) )
4024+ } else {
4025+ Ok ( Response :: MediationReady )
4026+ }
4027+ }
4028+ } )
4029+ . await ;
4030+ assert ! ( matches!( result, Ok ( Response :: MediationReady ) ) ) ;
4031+ assert_eq ! ( attempts. load( Ordering :: SeqCst ) , 3 ) ;
4032+
4033+ let denied_attempts = AtomicUsize :: new ( 0 ) ;
4034+ let denied = retry_within_budget ( BUDGET , ATTEMPT , INTERVAL , || {
4035+ denied_attempts. fetch_add ( 1 , Ordering :: SeqCst ) ;
4036+ async { Err ( BackendError :: Terminated ( "terminating" . to_string ( ) ) ) }
4037+ } )
4038+ . await ;
4039+ assert ! ( matches!( denied, Err ( BackendError :: Terminated ( _) ) ) ) ;
4040+ assert_eq ! ( denied_attempts. load( Ordering :: SeqCst ) , 1 ) ;
4041+ }
4042+ }
0 commit comments