From cc7b298cf303fb0815bd26d02eee77727c4282b6 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 15 Nov 2026 15:36:33 +0100 Subject: [PATCH] swarm-queue-client: never rotate the queue token more often than a floor, and not at all for a token too short to rotate --- swarm-queue-client/src/lib.rs | 79 ++++++++++++++++++++++++++--------- 1 file changed, 59 insertions(+), 20 deletions(-) diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 79af64be4..e7173a1fe 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -620,10 +620,16 @@ impl TokenCache { /// How long until the cached token has only [`TOKEN_REFRESH_SKEW`] of its /// life left: zero once it has, or when the clock reads earlier than the - /// mint (see [`MintedToken::usable_at`]). `None` while nothing is cached. + /// mint (see [`MintedToken::usable_at`]). `None` while nothing is cached, + /// and for a token that lives less than twice the skew: it is never + /// reusable (`usable_at`) and rotating it would reconnect at least every + /// life minus skew, so the reconnect path mints for it as it always did. async fn rotate_in(&self) -> Option { let slot = self.slot.lock().await; let minted = slot.as_ref()?; + if minted.token.lifetime < TOKEN_REFRESH_SKEW * 2 { + return None; + } let Ok(age) = (self.now)().duration_since(minted.minted_at) else { return Some(std::time::Duration::ZERO); }; @@ -1198,9 +1204,20 @@ pub async fn connect(cfg: QueueConfig) -> Result { Ok(client) } -/// How long [`rotate_before_expiry`] waits to look again when no token is -/// cached yet (the first connect is still pending) or minting one failed. -const ROTATION_RETRY: std::time::Duration = std::time::Duration::from_secs(30); +/// The least [`rotate_before_expiry`] waits between looks, and how long it +/// waits when no token is cached yet (the first connect is still pending). +/// +/// A floor, so that no state of the cache (a clock that reads earlier than the +/// mint, a failed mint, a token that is due again at once) can make the task a +/// tight loop against the identity provider and the queue. +const ROTATION_FLOOR: std::time::Duration = std::time::Duration::from_secs(30); + +/// How long [`rotate_before_expiry`] sleeps before looking again, given +/// [`TokenCache::rotate_in`]: until the rotation is due, but never less than +/// [`ROTATION_FLOOR`]. +fn rotation_wait(rotate_in: Option) -> std::time::Duration { + rotate_in.map_or(ROTATION_FLOOR, |wait| wait.max(ROTATION_FLOOR)) +} /// Replace the connection's token before the queue expires it. /// @@ -1214,8 +1231,8 @@ const ROTATION_RETRY: std::time::Duration = std::time::Duration::from_secs(30); /// identity provider. /// /// A failed mint costs nothing yet, the current token has [`TOKEN_REFRESH_SKEW`] -/// left, so it is retried at a fixed [`ROTATION_RETRY`] interval, not hammered. -/// If every try fails the connection still ends at the expiry and the +/// left, so it is retried no sooner than [`ROTATION_FLOOR`] later, not +/// hammered; the same floor holds after a rotation that worked. If every try fails the connection still ends at the expiry and the /// reconnect path takes over, as it did before this task existed. /// /// Runs for the life of the process, like the client it keeps. @@ -1226,20 +1243,19 @@ async fn rotate_before_expiry( cfg: QueueConfig, ) { loop { - match cache.rotate_in().await { - None => tokio::time::sleep(ROTATION_RETRY).await, - Some(wait) if !wait.is_zero() => tokio::time::sleep(wait).await, - Some(_) => { - if let Err(e) = cache.rotate(|| mint_token(&http, &cfg, None, None)).await { - tracing::warn!( - error = %chain(&e), - "minting the next queue token ahead of expiry failed; retrying" - ); - tokio::time::sleep(ROTATION_RETRY).await; - } else if let Err(e) = client.force_reconnect().await { - tracing::warn!(error = %e, "reconnecting to the queue with the next token failed"); - } - } + tokio::time::sleep(rotation_wait(cache.rotate_in().await)).await; + // Look again after the sleep: the token may have been replaced by a + // reconnect meanwhile, and the clock may have moved. + if cache.rotate_in().await != Some(std::time::Duration::ZERO) { + continue; + } + if let Err(e) = cache.rotate(|| mint_token(&http, &cfg, None, None)).await { + tracing::warn!( + error = %chain(&e), + "minting the next queue token ahead of expiry failed; retrying" + ); + } else if let Err(e) = client.force_reconnect().await { + tracing::warn!(error = %e, "reconnecting to the queue with the next token failed"); } } } @@ -1776,6 +1792,29 @@ mod tests { assert_eq!(cache.rotate_in().await, Some(std::time::Duration::ZERO)); } + /// A token too short-lived to rotate is left to the reconnect path, and + /// the task never waits less than the floor, however due the rotation is. + #[tokio::test] + async fn a_short_lived_token_is_not_rotated_and_the_wait_has_a_floor() { + let clock = FakeClock::starting_at(START); + let cache = clock.cache(); + let secs = std::time::Duration::from_secs; + cache + .token(|| async { + Ok(CachedToken { + token: "short".to_owned(), + lifetime: secs(239), + }) + }) + .await + .expect("the stub mint succeeds"); + assert_eq!(cache.rotate_in().await, None, "under twice the skew"); + + assert_eq!(rotation_wait(None), ROTATION_FLOOR); + assert_eq!(rotation_wait(Some(secs(0))), ROTATION_FLOOR); + assert_eq!(rotation_wait(Some(secs(3000))), secs(3000)); + } + /// A rotation replaces the token even though the cached one is usable, so /// the next connect presents it without asking the identity provider. #[tokio::test]