fix(lifecycle): recheck host admission before queued dials

src/client/lifecycle.rs

@@ -898,7 +898,12 @@ impl Client {
                 let pause_generation = self.pause_generation.load(Ordering::SeqCst);
                 let delay = admission.delay();
                 if !self
-                    .wait_for_connect_admission(delay, pause_generation, &shutdown)
+                    .wait_for_connect_admission(
+                        admission.as_ref(),
+                        delay,
+                        pause_generation,
+                        &shutdown,
+                    )
                     .await
                 {
                     if self.is_paused()
@@ -1147,7 +1152,8 @@ impl Client {
     /// ordinary teardown wake. The generation catches a rapid pause/resume.
     async fn wait_for_connect_admission(
         &self,
-        delay: Duration,
+        admission: &dyn crate::ConnectAdmission,
+        mut delay: Duration,
         pause_generation: u64,
         shutdown: &wacore::runtime::ShutdownSignal,
     ) -> bool {
@@ -1157,23 +1163,32 @@ impl Client {
                 && !self.is_paused()
                 && self.pause_generation.load(Ordering::SeqCst) == pause_generation
         };
-        if delay.is_zero() {
-            return can_dial();
-        }
-        let sleep = self.runtime.sleep(delay).fuse();
-        let stopped = wacore::runtime::wait_for_shutdown(shutdown).fuse();
-        futures::pin_mut!(sleep, stopped);
         loop {
-            // Register before checking: pause/resume and supervision-stop
-            // notifications must not be lost between the check and select.
-            let changed = self.session_state_notifier.listen();
+            if !delay.is_zero() {
+                let sleep = self.runtime.sleep(delay).fuse();
+                let stopped = wacore::runtime::wait_for_shutdown(shutdown).fuse();
+                futures::pin_mut!(sleep, stopped);
+                loop {
+                    // Register before checking; unrelated wakes retain this timer.
+                    let changed = self.session_state_notifier.listen();
+                    if !can_dial() {
+                        return false;
+                    }
+                    futures::select! {
+                        _ = sleep => break,
+                        _ = stopped => return false,
+                        _ = changed.fuse() => {}
+                    }
+                }
+            }
             if !can_dial() {
                 return false;
             }
-            futures::select! {
-                _ = sleep => return can_dial(),
-                _ = stopped => return false,
-                _ = changed.fuse() => {}
+            // Keep the reservation and pause generation across extensions.
+            // Calling delay() again would book another slot in host budgets.
+            delay = admission.recheck();
+            if delay.is_zero() {
+                return can_dial();
             }
         }
     }

src/client/lifecycle/admission_tests.rs

@@ -10,6 +10,10 @@ impl ConnectAdmission for Wait {
 }
 
 async fn client() -> Arc<Client> {
+    client_with_policy(Wait).await
+}
+
+async fn client_with_policy(policy: impl ConnectAdmission + 'static) -> Arc<Client> {
     Client::builder()
         .with_runtime(crate::runtime_impl::TokioRuntime)
         .with_persistence_manager(Arc::new(
@@ -19,13 +23,61 @@ async fn client() -> Arc<Client> {
         ))
         .with_http_client(crate::test_utils::MockHttpClient)
         .with_transport_factory(crate::transport::mock::MockTransportFactory::new())
-        .with_connect_admission(Wait)
+        .with_connect_admission(policy)
         .build()
         .await
         .unwrap()
         .into_client()
 }
 
+struct Extend(Arc<AtomicUsize>);
+impl ConnectAdmission for Extend {
+    fn delay(&self) -> Duration {
+        Duration::ZERO
+    }
+    fn recheck(&self) -> Duration {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        Duration::from_secs(900)
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn admission_extension_retains_timer_on_unrelated_wakes_and_cancels_on_stop() {
+    let checks = Arc::new(AtomicUsize::new(0));
+    let client = client_with_policy(Extend(checks.clone())).await;
+    client.auto_reconnect_errors.store(5, Ordering::Relaxed);
+    client
+        .backoff_reset_suppressed
+        .store(true, Ordering::Relaxed);
+    let runner = client.clone();
+    let run = tokio::spawn(async move { runner.run().await });
+    crate::test_utils::wait_for_notifier_listeners(&client.session_state_notifier, 1).await;
+    for _ in 0..2 {
+        tokio::time::advance(Duration::from_secs(300)).await;
+        client.notify_connection_shutdown();
+        client.notify_session_state();
+        tokio::task::yield_now().await;
+        crate::test_utils::wait_for_notifier_listeners(&client.session_state_notifier, 1).await;
+        assert_eq!(checks.load(Ordering::SeqCst), 1);
+        assert!(!client.is_connecting.load(Ordering::Relaxed));
+        assert_eq!(client.stats().reconnects, 0);
+        assert_eq!(client.stats().reconnect_errors, 5);
+        assert!(client.backoff_reset_suppressed.load(Ordering::Relaxed));
+    }
+    tokio::time::advance(Duration::from_secs(300)).await;
+    tokio::task::yield_now().await;
+    assert_eq!(
+        checks.load(Ordering::SeqCst),
+        2,
+        "wakes must not restart the timer"
+    );
+    client.stop_supervision_loop();
+    assert!(matches!(run.await.unwrap(), RunCompletionReason::Stopped));
+    assert_eq!(client.stats().reconnects, 0);
+    assert_eq!(client.stats().reconnect_errors, 5);
+    client.shutdown().await;
+}
+
 #[tokio::test]
 async fn admission_stop_preserves_stopped_reason() {
     let client = client().await;

src/types/connect_admission.rs

@@ -2,11 +2,13 @@ use std::time::Duration;
 
 /// Optional host-owned pacing of [`Client::run`](crate::Client::run) dials.
 ///
-/// The run loop consults this policy once before starting each connect attempt,
+/// The run loop calls [`delay`](Self::delay) once before each connect attempt,
 /// including the first and forced reconnects, after its ordinary reconnect
 /// backoff and pause gate. It serves the returned delay **before** calling
 /// [`Client::connect`](crate::Client::connect), outside the transport/version
-/// connect timeouts. Zero proceeds without sleeping or consulting again.
+/// connect timeouts. It then calls [`recheck`](Self::recheck), including when
+/// the initial delay is zero, so a host can extend a wait after learning of a
+/// shared cooldown. Existing policies need only implement `delay`.
 /// Direct calls to `connect()` do not consult this policy.
 ///
 /// Shutdown or a pause (even one already resumed) abandons the wait without a
@@ -17,7 +19,7 @@ use std::time::Duration;
 /// cancellation/refund callback; hosts must account for abandoned reservations.
 /// The SDK still owns WhatsApp's backoff, stable reset, and rate-limit penalties.
 ///
-/// This synchronous callback runs inline. It must return promptly without I/O,
+/// These synchronous callbacks run inline. They must return promptly without I/O,
 /// blocking, or awaiting. Shutdown cannot interrupt a blocking callback; panics
 /// propagate to the caller driving the run loop, with no recovery guarantee.
 /// Share a policy across builders to coordinate a host-wide budget; no limiter
@@ -50,4 +52,19 @@ use std::time::Duration;
 pub trait ConnectAdmission: wacore::sync_marker::MaybeSendSync {
     /// Reserve a run-loop attempt and return how long to delay it.
     fn delay(&self) -> Duration;
+
+    /// Recheck an existing reservation before dialing, without reserving again.
+    ///
+    /// Called after the initial delay, even if zero, and after every positive
+    /// extension returned here. Return zero to proceed, or a positive duration
+    /// to wait before checking again. The default preserves one-shot policies.
+    /// Extensions remain outside connect timeouts and are cancelled by shutdown,
+    /// supervision stop, or pause just like the initial wait. Resume reserves a
+    /// new attempt through `delay`.
+    ///
+    /// This is a synchronous snapshot, not atomic exclusion with the dial:
+    /// a host update after the final zero cannot revoke that permission.
+    fn recheck(&self) -> Duration {
+        Duration::ZERO
+    }
 }

tests/api-consumer/src/lifecycle/admission.rs

@@ -17,6 +17,21 @@ impl ConnectAdmission for Policy {
     }
 }
 
+pub struct RecheckingPolicy(pub Policy);
+impl ConnectAdmission for RecheckingPolicy {
+    fn delay(&self) -> Duration {
+        self.0.delay()
+    }
+    fn recheck(&self) -> Duration {
+        // Reading the same host state remains valid with wasm's local Rc.
+        #[cfg(target_arch = "wasm32")]
+        let calls = self.0.0.get();
+        #[cfg(not(target_arch = "wasm32"))]
+        let calls = self.0.0.load(std::sync::atomic::Ordering::Relaxed);
+        Duration::from_millis(calls as u64)
+    }
+}
+
 pub fn builder(policy: Arc<dyn ConnectAdmission>) -> whatsapp_rust::ClientBuilder {
     Client::builder().with_connect_admission_arc(policy)
 }
@@ -40,5 +55,17 @@ fn shared_policy_is_object_safe() {
     let policy = Policy(std::sync::atomic::AtomicUsize::new(0));
     let boxed: Box<dyn ConnectAdmission> = Box::new(policy);
     assert_eq!(boxed.delay(), Duration::ZERO);
+    assert_eq!(boxed.recheck(), Duration::ZERO);
     let _ = builder(Arc::from(boxed));
 }
+
+#[cfg(not(target_arch = "wasm32"))]
+#[test]
+fn rechecking_policy_is_object_safe() {
+    let policy = Policy(std::sync::atomic::AtomicUsize::new(0));
+    let shared: Arc<dyn ConnectAdmission> = Arc::new(RecheckingPolicy(policy));
+    assert_eq!(shared.delay(), Duration::ZERO);
+    assert_eq!(shared.recheck(), Duration::from_millis(1));
+    assert_eq!(shared.recheck(), Duration::from_millis(1));
+    let _ = builder(shared);
+}