From aa885795c8f5250f7f516c1cf4ef6effd012a914 Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Tue, 13 Oct 2026 00:21:03 +1100 Subject: [PATCH 1/7] fix(stack-auth)!: finish token refreshes after caller cancellation A token refresh ran inside the caller's get_token future, so cancelling that future (a request timeout, a dropped connection, a select! that lost) dropped the refresh with it. With a rotating device-session refresh token, a refresh the server had already accepted could be lost before it was saved: the old credential was spent, the new one was gone, and the client could not authenticate again until the process restarted. On native targets the refresh now runs in a spawned Tokio task that owns the exchange, persistence, token installation and waiter notification. Cancelling get_token abandons only that caller's wait. wasm32 refreshes still run inline, since browser futures are not Send and WASI hosts drive the export to completion. This also closes a missed-wakeup race: a caller waiting on an in-flight refresh now joins the notify list while it still holds the state lock and rechecks the in-progress flag before parking. Cancellation and seeded soak tests cover both, and CI runs the soak on PRs and before a crates release. BREAKING CHANGE: On native targets a token refresh now needs a current Tokio runtime, and `TokenStore` (with the closures given to `TokenStoreFn`) must be `Send + Sync + 'static`, because the spawned refresh task holds the store until the new token is saved. A store that borrows request state must own it instead (for example an `Arc` captured by `move` closures), and save callbacks run in the refresh task, not the caller's. Callers that drive `get_token` with another executor (`futures::executor::block_on`, smol) must run it inside a Tokio runtime. wasm32 is unchanged. Refs #1170. --- .cargo/mutants.toml | 2 +- .changeset/cancelled-auth-refresh.md | 5 + .github/workflows/_test-auth-refresh-soak.yml | 37 ++ .github/workflows/release-plz.yml | 24 +- .github/workflows/tests-auth-refresh.yml | 50 ++ AGENTS.md | 5 + packages/stack-auth/src/auto_refresh.rs | 568 +++++------------- .../src/auto_refresh/tests/cancellation.rs | 341 +++++++++++ .../stack-auth/src/auto_refresh/tests/soak.rs | 268 +++++++++ packages/stack-auth/src/lib.rs | 7 + .../src/oidc_federation_strategy.rs | 10 +- packages/stack-auth/src/refresher.rs | 4 +- packages/stack-auth/src/token_store.rs | 15 +- packages/stack-auth/tasks.toml | 4 + .../__tests__/auth-refresh-soak-ci.test.mjs | 66 ++ 15 files changed, 958 insertions(+), 448 deletions(-) create mode 100644 .changeset/cancelled-auth-refresh.md create mode 100644 .github/workflows/_test-auth-refresh-soak.yml create mode 100644 .github/workflows/tests-auth-refresh.yml create mode 100644 packages/stack-auth/src/auto_refresh/tests/cancellation.rs create mode 100644 packages/stack-auth/src/auto_refresh/tests/soak.rs create mode 100644 scripts/__tests__/auth-refresh-soak-ci.test.mjs diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index 1fae9ba39..6cbd58933 100644 --- a/.cargo/mutants.toml +++ b/.cargo/mutants.toml @@ -63,7 +63,7 @@ exclude_re = [ # `a_builder_without_a_transport_is_refused_when_there_is_no_bundled_one`. 'stack-auth/src/error\.rs:\d+:\d+: replace RequestError::is_no_transport -> bool with false$', 'stack-auth/src/auto_strategy\.rs:150:9: replace AutoStrategy::detect_inner -> Result with Ok\(Default::default\(\)\)$', - 'stack-auth/src/token_store\.rs:(258|266):9: replace >::(load|save)', + 'stack-auth/src/token_store\.rs:(261|269):9: replace >::(load|save)', # stack-auth — equivalent. 'stack-auth/src/(access_key|oidc)_refresher\.rs:\d+:\d+: replace )?>::try_credential -> Option with Some\(Default::default\(\)\)$', 'stack-auth/src/error\.rs:\d+:\d+: replace AuthErrorKind::payload -> serde_json::Map with Default::default\(\)$', diff --git a/.changeset/cancelled-auth-refresh.md b/.changeset/cancelled-auth-refresh.md new file mode 100644 index 000000000..bc25e17e6 --- /dev/null +++ b/.changeset/cancelled-auth-refresh.md @@ -0,0 +1,5 @@ +--- +'@cipherstash/auth': patch +--- + +Keep native token refreshes running when their caller is cancelled, so rotated device-session credentials are saved and subsequent requests can still authenticate. Close a refresh-waiter notification race and avoid returning a token that expired during refresh. diff --git a/.github/workflows/_test-auth-refresh-soak.yml b/.github/workflows/_test-auth-refresh-soak.yml new file mode 100644 index 000000000..b129d8d78 --- /dev/null +++ b/.github/workflows/_test-auth-refresh-soak.yml @@ -0,0 +1,37 @@ +name: Auth refresh cancellation soak + +on: + workflow_call: + inputs: + long: + type: boolean + default: false + +permissions: + contents: read + +jobs: + soak: + runs-on: blacksmith-16vcpu-ubuntu-2204 + timeout-minutes: 30 + env: + MISE_ENV: test + NEXTEST_PROFILE: ci + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 + with: + persist-credentials: false + - uses: jdx/mise-action@1648a7812b9aeae629881980618f079932869151 # v4 + with: + version: 2026.4.0 + install: true + working_directory: . + cache: true + cache_key_prefix: mise-v1-ubuntu-2204 + - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + - name: Short seeded soak + if: ${{ !inputs.long }} + run: cargo nextest run --locked -p stack-auth --all-features -E 'test(cancellation_soak_short)' --success-output final + - name: Long seeded soak + if: inputs.long + run: mise run test:soak:stack-auth diff --git a/.github/workflows/release-plz.yml b/.github/workflows/release-plz.yml index 1885d2b13..50265415a 100644 --- a/.github/workflows/release-plz.yml +++ b/.github/workflows/release-plz.yml @@ -225,9 +225,31 @@ jobs: id: armed run: node scripts/eql-pipeline-armed.mjs crates + auth-refresh-soak: + name: Auth refresh cancellation soak before publishing + runs-on: ubuntu-latest + timeout-minutes: 60 + env: + MISE_ENV: test + NEXTEST_PROFILE: ci + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 + with: + persist-credentials: false + # Release workflows never restore caches. The PR workflow runs the + # same mise task with caches; only toolchain setup differs here. + - uses: jdx/mise-action@1648a7812b9aeae629881980618f079932869151 # v4 + with: + version: 2026.4.0 + install: true + working_directory: . + cache: false + - name: Long seeded soak + run: mise run test:soak:stack-auth + release-crates: name: "Release the stack-* crates" - needs: [crates-armed] + needs: [crates-armed, auth-refresh-soak] if: needs.crates-armed.outputs.armed == 'true' runs-on: ubuntu-latest timeout-minutes: 30 diff --git a/.github/workflows/tests-auth-refresh.yml b/.github/workflows/tests-auth-refresh.yml new file mode 100644 index 000000000..505979d30 --- /dev/null +++ b/.github/workflows/tests-auth-refresh.yml @@ -0,0 +1,50 @@ +name: Tests (auth refresh cancellation) + +# The short run covers every PR, including Version Packages. The longer run +# also covers release PRs and changes to the engine, refreshers or toolchain. +on: + pull_request: {} + workflow_dispatch: {} + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + short: + uses: ./.github/workflows/_test-auth-refresh-soak.yml + + changes: + runs-on: ubuntu-latest + timeout-minutes: 5 + permissions: + contents: read + pull-requests: read + outputs: + refresh: ${{ steps.paths.outputs.refresh }} + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 + with: + persist-credentials: false + - uses: dorny/paths-filter@0e4a8c6effa4802afeda77dc8d303f8176d7dfad # v3 + id: paths + with: + filters: | + refresh: + - 'packages/stack-auth/**' + - 'Cargo.toml' + - 'Cargo.lock' + - 'mise*.toml' + - '.config/nextest.toml' + - '.github/workflows/*auth-refresh*.yml' + - '.github/workflows/release-plz.yml' + + long: + needs: changes + if: github.event_name == 'workflow_dispatch' || github.head_ref == 'changeset-release/main' || needs.changes.outputs.refresh == 'true' + uses: ./.github/workflows/_test-auth-refresh-soak.yml + with: + long: true diff --git a/AGENTS.md b/AGENTS.md index 6e9020f11..f38e905d9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -706,6 +706,11 @@ exact pin would stop the suite sharing one `cts-common` with it. `mutants.yml`, `fuzz.yml`, `miri.yml` and `udeps.yml`. `scripts/__tests__/crates-ci.test.mjs` fails when a mise task in the root `mise.toml` or the five crate `tasks.toml` files loses its last CI caller. +- **Auth refresh cancellation soak:** `tests-auth-refresh.yml` runs a short, + seeded, virtual-time soak on every PR and the long run on auth changes and + Version Packages PRs. `release-plz.yml` also requires the long run before + `release-crates` can publish. Run it locally with `mise run + test:soak:stack-auth`; it needs no credentials or live services. ### Fuzzing diff --git a/packages/stack-auth/src/auto_refresh.rs b/packages/stack-auth/src/auto_refresh.rs index cf9ea042e..a264dd38b 100644 --- a/packages/stack-auth/src/auto_refresh.rs +++ b/packages/stack-auth/src/auto_refresh.rs @@ -1,6 +1,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; -use tokio::sync::{Mutex, MutexGuard, Notify}; +use tokio::sync::{Mutex, Notify, OwnedMutexGuard}; use crate::clock::{system_clock, SharedClock}; use crate::refresher::Refresher; @@ -43,8 +44,13 @@ impl From for crate::AuthError { /// See the [crate-level documentation](crate#token-refresh) for a full /// description of the concurrency model and flow diagram. pub(crate) struct AutoRefresh { + shared: Arc>, +} + +/// Owned by both the strategy and any refresh task that outlives its caller. +struct RefreshState { refresher: R, - state: Mutex, + state: Arc>, store: S, /// Set to `true` while a refresh HTTP call is in-flight. /// @@ -82,7 +88,7 @@ struct State { /// Unlike `denial`, this is not TTL'd, is not restricted to account-level /// refusals, and is never consulted by `get_token`'s own retry path — it /// exists solely so a caller parked in - /// [`wait_for_in_flight_refresh`](AutoRefresh::wait_for_in_flight_refresh) + /// [`wait_for_in_flight_refresh`](RefreshState::wait_for_in_flight_refresh) /// sees the *same* outcome as whoever actually performed the refresh it /// was waiting on, instead of a generic `Expired` that discards why the /// wait ended in failure. @@ -139,22 +145,16 @@ fn credential_consumed(err: &crate::AuthError) -> bool { matches!(err, crate::AuthError::Store(_)) } -/// Ensures [`AutoRefresh::refresh_in_progress`] is cleared and waiters are +/// Ensures [`RefreshState::refresh_in_progress`] is cleared and waiters are /// notified if the refresh future is cancelled (dropped) before completing. /// /// On the normal path (success or handled error), the guard is defused before /// drop so that the regular cleanup code runs instead. /// -/// Unlike the normal paths, `Drop` is synchronous and so notifies without -/// taking the state lock. A caller that has read `refresh_in_progress` as -/// `true` and is on its way into -/// [`wait_for_in_flight_refresh`](AutoRefresh::wait_for_in_flight_refresh) -/// holds that lock, which does not block this notification — so if the refresh -/// future is cancelled in that window, the wake still lands on an empty list -/// and that caller parks with nothing left to wake it. The `enable()` call in -/// `wait_for_in_flight_refresh` does not close this narrower variant; doing so -/// needs the notify moved under the state lock or a bounded wait, and is -/// tracked separately. +/// Native caller cancellation never drops this guard: the refresh task owns it. +/// It still releases waiters on a task panic, runtime shutdown, or an inline +/// wasm future being dropped. Waiters register *before* rechecking the flag, +/// so this synchronous notification cannot be lost even without the state lock. struct CancelGuard<'a> { in_progress: &'a AtomicBool, notify: &'a Notify, @@ -244,68 +244,53 @@ impl State { } impl AutoRefresh { - /// Create a new `AutoRefresh` with a pre-loaded token and no external store. - /// - /// Use this for refreshers that cannot self-authenticate (e.g. OAuth, - /// which needs a refresh token from a prior device code flow). + /// Create an engine with a pre-loaded token and no external store. pub(crate) fn with_token(refresher: R, token: Token) -> Self { - Self { - refresher, - state: Mutex::new(State::new(Some(token))), - store: NoStore, - refresh_in_progress: AtomicBool::new(false), - refresh_notify: Notify::new(), - clock: system_clock(), - } + Self::new(refresher, NoStore, Some(token), system_clock()) } - /// Like [`with_token`](Self::with_token) but with an injected clock, so tests - /// can drive token expiry deterministically. #[cfg(test)] #[cfg(feature = "http")] pub(crate) fn with_token_and_clock(refresher: R, token: Token, clock: SharedClock) -> Self { - Self { - refresher, - state: Mutex::new(State::new(Some(token))), - store: NoStore, - refresh_in_progress: AtomicBool::new(false), - refresh_notify: Notify::new(), - clock, - } + Self::new(refresher, NoStore, Some(token), clock) } } -impl AutoRefresh { - /// Create a new `AutoRefresh` backed by `store` and no in-memory token. - /// - /// On the first `get_token` call the store is consulted before falling - /// through to initial authentication via `try_credential(None)`; every - /// successful refresh writes the new token back via `store.save()`. Pass - /// [`NoStore`] for the no-external-cache case — it's the default and - /// elides to a zero-cost no-op. - pub(crate) fn with_store(refresher: R, store: S) -> Self { +impl AutoRefresh { + fn new(refresher: R, store: S, token: Option, clock: SharedClock) -> Self { Self { - refresher, - state: Mutex::new(State::new(None)), - store, - refresh_in_progress: AtomicBool::new(false), - refresh_notify: Notify::new(), - clock: system_clock(), + shared: Arc::new(RefreshState { + refresher, + state: Arc::new(Mutex::new(State::new(token))), + store, + refresh_in_progress: AtomicBool::new(false), + refresh_notify: Notify::new(), + clock, + }), } } -} -impl AutoRefresh { - /// The refresher this engine renews through. + /// Create an engine that consults `store` on cold start and saves every + /// successful refresh. `NoStore` disables external caching. + pub(crate) fn with_store(refresher: R, store: S) -> Self { + Self::new(refresher, store, None, system_clock()) + } + pub(crate) fn refresher(&self) -> &R { - &self.refresher + &self.shared.refresher } } impl AutoRefresh { - /// Retrieve a valid access token, refreshing or re-authenticating as needed. pub(crate) async fn get_token(&self) -> Result { - let mut state = self.state.lock().await; + self.shared.get_token().await + } +} + +impl RefreshState { + /// Retrieve a valid access token, refreshing or re-authenticating as needed. + async fn get_token(self: &Arc) -> Result { + let mut state = self.state.clone().lock_owned().await; if state.token.is_none() { // A settled account-level refusal is checked before the store @@ -321,7 +306,7 @@ impl AutoRefresh { // case another caller populated it while we awaited. drop(state); let loaded = self.store.load().await; - state = self.state.lock().await; + state = self.state.clone().lock_owned().await; if state.token.is_none() { state.token = loaded; } @@ -336,7 +321,10 @@ impl AutoRefresh { if let Some(err) = state.fresh_denial(now) { return Err(AutoRefreshError::Auth(err)); } - return self.initial_auth(&mut state).await; + let Some(credential) = self.refresher.try_credential(None) else { + return Err(AutoRefreshError::NotFound); + }; + return self.start_refresh(state, credential).await; } if !state.token.as_ref().is_some_and(|t| t.is_expired_at(now)) { @@ -362,44 +350,51 @@ impl AutoRefresh { return state.require_usable_token(now); }; + self.start_refresh(state, credential).await + } + + /// There is no suspension between extracting the credential and spawning + /// its owner. Dropping the JoinHandle cancels only the caller's wait, not + /// the exchange, persistence, state installation, or waiter notification. + async fn start_refresh( + self: &Arc, + state: OwnedMutexGuard, + credential: R::Credential, + ) -> Result { self.refresh_in_progress.store(true, Ordering::Release); + #[cfg(not(target_arch = "wasm32"))] + { + let shared = Arc::clone(self); + tokio::spawn(async move { shared.run_refresh(state, credential).await }) + .await + .map_err(|_| { + crate::AuthError::Internal(crate::error::InternalError( + "token refresh task terminated before completion".into(), + )) + })? + } + // WASI hosts drive the export to completion with block_on; browser + // wasm futures are not Send and have no Tokio task executor. + #[cfg(target_arch = "wasm32")] + self.run_refresh(state, credential).await + } - if state.token.as_ref().is_some_and(|t| t.is_usable_at(now)) { + async fn run_refresh( + &self, + mut state: OwnedMutexGuard, + credential: R::Credential, + ) -> Result { + if state + .token + .as_ref() + .is_some_and(|t| t.is_usable_at(self.clock.now_unix_secs())) + { self.refresh_non_blocking(state, credential).await } else { self.refresh_blocking(&mut state, credential).await } } - /// No cached token — authenticate via `try_credential(None)`. - /// - /// The lock is held throughout to prevent concurrent initial-auth attempts. - async fn initial_auth(&self, state: &mut State) -> Result { - let Some(credential) = self.refresher.try_credential(None) else { - return Err(AutoRefreshError::NotFound); - }; - self.refresh_in_progress.store(true, Ordering::Release); - let mut guard = CancelGuard { - in_progress: &self.refresh_in_progress, - notify: &self.refresh_notify, - defused: false, - }; - match self.refresher.refresh(&credential).await { - Ok(new_token) => { - self.save_refreshed_token(&new_token).await; - let token = self.install_refreshed_token(state, new_token); - guard.defuse(); - Ok(token) - } - Err(err) => { - guard.defuse(); - self.refresh_in_progress.store(false, Ordering::Release); - state.record_refusal(&err, self.clock.now_unix_secs()); - Err(AutoRefreshError::Auth(err)) - } - } - } - /// Persist a freshly refreshed token to the per-refresher sink and the /// user-supplied `TokenStore`. Awaits the store write, so callers should /// drop the state lock before invoking this where possible (the @@ -431,7 +426,7 @@ impl AutoRefresh { /// the notification. async fn wait_for_in_flight_refresh( &self, - state: MutexGuard<'_, State>, + state: OwnedMutexGuard, now: u64, ) -> Result { if let Ok(token) = state.service_token() { @@ -451,15 +446,20 @@ impl AutoRefresh { // `enable()` joins the list while the state lock is still held, and // `refresh_non_blocking` takes that same lock to record its outcome // before it notifies, so the notification cannot land before we are - // listed. This does not cover `CancelGuard::drop`, which notifies - // without the lock — see the note on that impl. + // listed. The flag is rechecked below for `CancelGuard::drop`, which + // does not acquire the lock. let mut notified = std::pin::pin!(self.refresh_notify.notified()); // The `bool` reports whether a stored permit was consumed, which only // `notify_one` produces; this `Notify` is only ever driven by // `notify_waiters`, so there is nothing to act on. let _ = notified.as_mut().enable(); drop(state); - notified.await; + // CancelGuard can notify without this lock. Registering first and + // then rechecking closes the cancellation/panic race too: either its + // notification reaches us, or its cleared flag makes waiting needless. + if self.refresh_in_progress.load(Ordering::Acquire) { + notified.await; + } // Re-check after wake — refresh may have failed. Re-read the clock: an // arbitrary amount of time may have passed while awaiting the refresh. let now = self.clock.now_unix_secs(); @@ -499,26 +499,15 @@ impl AutoRefresh { } } - /// Token is expiring but still usable — drop the lock, refresh in the - /// background of this call, and return the old (still-valid) token. - /// - /// Takes `MutexGuard` by value because the lock is dropped before the HTTP - /// request. Notifies waiters after the refresh completes (success or error). - /// - /// A [`CancelGuard`] ensures that if this future is cancelled at any point - /// before the new token is installed — including the post-HTTP save + - /// install window — `refresh_in_progress` is cleared and waiters are - /// notified, so subsequent callers don't hang in - /// [`wait_for_in_flight_refresh`](Self::wait_for_in_flight_refresh). - /// The credential is not restored on cancellation (it's already gone from - /// `state.token`), so the next caller will get whatever the cached token - /// offers — usable, expired, or absent. + /// Release the lock while refreshing so concurrent callers can use the + /// cached token. The initiating caller waits for this attempt's outcome. async fn refresh_non_blocking( &self, - state: MutexGuard<'_, State>, + state: OwnedMutexGuard, credential: R::Credential, ) -> Result { let current_service_token = state.service_token()?; + let current_expires_at = state.token.as_ref().map_or(0, Token::expires_at); drop(state); let mut guard = CancelGuard { @@ -533,7 +522,12 @@ impl AutoRefresh { let mut state = self.state.lock().await; let _ = self.install_refreshed_token(&mut state, new_token); guard.defuse(); - Ok(current_service_token) + let now = self.clock.now_unix_secs(); + if now < current_expires_at { + Ok(current_service_token) + } else { + state.require_usable_token(now) + } } Err(err) => { let consumed = credential_consumed(&err); @@ -565,7 +559,7 @@ impl AutoRefresh { // call still succeeds. A consumed credential does not: the // rotation happened upstream but was lost, and succeeding // would hide that until the cached token expires. - if consumed { + if consumed || self.clock.now_unix_secs() >= current_expires_at { Err(AutoRefreshError::Auth(err)) } else { Ok(current_service_token) @@ -577,15 +571,8 @@ impl AutoRefresh { result } - /// Token is fully expired — refresh while holding the lock so concurrent - /// callers block on `lock().await` until the new token is available. - /// - /// A [`CancelGuard`] ensures that if this future is cancelled at any point - /// before the new token is installed — including the post-HTTP save - /// window — `refresh_in_progress` is cleared and waiters are notified so - /// they don't hang indefinitely. (The credential is lost on cancel — - /// see [`CancelGuard`] docs — but subsequent callers will get `Expired` - /// rather than blocking forever.) + /// Initial auth or a fully expired token: keep the owned lock until the + /// task finishes, so concurrent callers all observe the installed result. async fn refresh_blocking( &self, state: &mut State, @@ -1061,7 +1048,7 @@ mod tests { ); // Verify the refresh token was restored so a retry is possible. - let state = strategy.state.lock().await; + let state = strategy.shared.state.lock().await; assert!( state.token.is_some(), "token should still be cached after failed refresh" @@ -1195,7 +1182,7 @@ mod tests { ); // Verify the access token and refresh token are still present. - let state = strategy.state.lock().await; + let state = strategy.shared.state.lock().await; assert!(state.token.is_some(), "token should still be cached"); assert_eq!( state.token.as_ref().unwrap().access_token().as_str(), @@ -1245,7 +1232,7 @@ mod tests { ); // Verify the cache now holds the refreshed token. - let state = strategy.state.lock().await; + let state = strategy.shared.state.lock().await; assert_eq!( state.token.as_ref().unwrap().access_token().as_str(), "refreshed-token", @@ -1299,7 +1286,7 @@ mod tests { "a lost rotation must fail the call, got: {result:?}" ); - let state = strategy.state.lock().await; + let state = strategy.shared.state.lock().await; assert!( state.token.as_ref().unwrap().refresh_token().is_none(), "the spent refresh token must not be restored for replay" @@ -1321,7 +1308,7 @@ mod tests { "a lost rotation must fail the call, got: {result:?}" ); - let state = strategy.state.lock().await; + let state = strategy.shared.state.lock().await; assert!( state.token.as_ref().unwrap().refresh_token().is_none(), "the spent refresh token must not be restored for replay" @@ -1774,7 +1761,7 @@ mod stress_tests { ); } - let state = strategy.state.lock().await; + let state = strategy.shared.state.lock().await; assert!( state.token.as_ref().unwrap().refresh_token().is_some(), "refresh token should be restored after failed refresh" @@ -1865,173 +1852,6 @@ mod stress_tests { assert_eq!(stats2.total(), 1, "only one retry refresh should be made"); } } - - mod given_cancelled_refresh { - use super::*; - - /// If a blocking refresh (fully expired token) is cancelled mid-flight, - /// the `CancelGuard` must reset `refresh_in_progress` and notify waiters - /// so the next caller doesn't hang in `wait_for_in_flight_refresh`. - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn blocked_callers_recover_after_cancellation() { - let counting = CountingState::new(); - let state = DelayedRefreshState { - counting: counting.clone(), - delay: Duration::from_secs(10), // Very slow — will be cancelled - }; - let (base_url, _) = start_axum_server(delayed_refresh_handler, state).await; - let dir = tempfile::tempdir().unwrap(); - let strategy = Arc::new(auto_refresh_with_token( - &dir, - &base_url, - make_token("expired-token", 0, true), - )); - - // Spawn get_token and let the blocking refresh start. - let s = Arc::clone(&strategy); - let handle = tokio::spawn(async move { s.get_token().await }); - tokio::time::sleep(Duration::from_millis(100)).await; - - // Cancel the refresh mid-flight. - handle.abort(); - let _ = handle.await; - - // The next caller must not hang. The credential is lost (refresh - // token was taken before the HTTP call), so the result is Expired, - // but the important thing is that it completes promptly. - let s = Arc::clone(&strategy); - let result = tokio::time::timeout(Duration::from_secs(2), s.get_token()).await; - - assert!( - result.is_ok(), - "get_token() should not hang after cancelled blocking refresh" - ); - } - - /// Regression test: cancellation in the window *after* the upstream - /// HTTP refresh succeeds but *before* the new token is installed must - /// still clear `refresh_in_progress` and notify waiters. The previous - /// implementation defused the [`CancelGuard`] before - /// `save_refreshed_token`, so a drop during the (async) store-save or - /// the subsequent state-lock acquire would strand the flag — wedging - /// any caller that later hit `wait_for_in_flight_refresh`. - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn save_phase_cancellation_does_not_strand_in_progress_flag() { - use crate::token_store::TokenStore; - - /// Store that returns a single pre-loaded token from `load()` and - /// delays inside `save()` long enough for a test to cancel. - struct SlowSaveStore { - initial: tokio::sync::Mutex>, - delay: Duration, - } - - impl TokenStore for SlowSaveStore { - async fn load(&self) -> Option { - self.initial.lock().await.take() - } - - async fn save(&self, _token: &Token) { - tokio::time::sleep(self.delay).await; - } - } - - // Fast upstream HTTP — refresh succeeds in <50ms. - let counting = CountingState::new(); - let state = DelayedRefreshState { - counting: counting.clone(), - delay: Duration::from_millis(10), - }; - let (base_url, _) = start_axum_server(delayed_refresh_handler, state).await; - let dir = tempfile::tempdir().unwrap(); - let store = ProfileStore::new(dir.path()); - store.init_workspace("ZVATKW3VHMFG27DY").unwrap(); - let ws_store = store.current_workspace_store().unwrap(); - let refresher = DeviceSessionRefresher::new( - Some(ws_store), - base_url, - "cli", - "ap-southeast-2.aws", - None, - crate::transport::default_transport(), - ); - // Slow async save — cancellation reliably lands here, in the - // post-HTTP / pre-install window. - let slow_store = SlowSaveStore { - initial: tokio::sync::Mutex::new(Some(make_token("expired-token", 0, true))), - delay: Duration::from_secs(10), - }; - let strategy = Arc::new(AutoRefresh::with_store(refresher, slow_store)); - - // Trigger refresh; the task will complete the HTTP exchange and - // then block inside store.save (the slow async path). - let s = Arc::clone(&strategy); - let handle = tokio::spawn(async move { s.get_token().await }); - // 200ms is comfortably past the 10ms HTTP delay but well inside - // the 10s save delay — so abort() lands during save_refreshed_token. - tokio::time::sleep(Duration::from_millis(200)).await; - handle.abort(); - let _ = handle.await; - - // The CancelGuard must have cleared refresh_in_progress on drop. - // If the old (pre-fix) code regresses, the flag stays true and a - // subsequent caller wedges on wait_for_in_flight_refresh waiting - // for a notify that will never come — the timeout below catches it. - let s = Arc::clone(&strategy); - let result = tokio::time::timeout(Duration::from_secs(2), s.get_token()).await; - - assert!( - result.is_ok(), - "get_token() should not hang after cancellation in the save/install window" - ); - } - - /// If a non-blocking refresh (expiring-but-usable token) is cancelled - /// mid-flight, the `CancelGuard` must reset `refresh_in_progress` and - /// notify waiters so they don't hang once the token crosses real expiry. - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn non_blocking_callers_recover_after_cancellation() { - let counting = CountingState::new(); - let state = DelayedRefreshState { - counting: counting.clone(), - delay: Duration::from_secs(10), // Very slow — will be cancelled - }; - let (base_url, _) = start_axum_server(delayed_refresh_handler, state).await; - let dir = tempfile::tempdir().unwrap(); - // Token expires in 30s — is_expired() = true, is_usable() = true. - let strategy = Arc::new(auto_refresh_with_token( - &dir, - &base_url, - make_token("still-usable", 30, true), - )); - - // Spawn get_token — triggers non-blocking refresh, drops lock, then - // blocks on the slow HTTP call. - let s = Arc::clone(&strategy); - let handle = tokio::spawn(async move { s.get_token().await }); - tokio::time::sleep(Duration::from_millis(100)).await; - - // Cancel the refresh mid-flight. - handle.abort(); - let _ = handle.await; - - // The next caller must not hang. The token is still usable so it - // should be returned even though the refresh was cancelled. - let s = Arc::clone(&strategy); - let result = tokio::time::timeout(Duration::from_secs(2), s.get_token()).await; - - assert!( - result.is_ok(), - "get_token() should not hang after cancelled non-blocking refresh" - ); - let result = result.unwrap(); - assert!( - result.is_ok(), - "expected Ok with still-usable token, got: {:?}", - result.unwrap_err() - ); - } - } } /// Deterministic regression test for the "token crosses real expiry while a @@ -2175,8 +1995,8 @@ mod expiry_crossing_regression { let first = first.await.unwrap().unwrap(); assert_eq!( first.as_str(), - "expiring-soon", - "first caller receives the old token (still usable when it was called)" + "refreshed-token", + "the old token expired while the first caller waited" ); for (i, waiter) in waiters.into_iter().enumerate() { @@ -2241,7 +2061,7 @@ mod expiry_crossing_regression { /// The failure counterpart to /// [`waiters_wait_for_refresh_when_token_crosses_expiry`]: when the in-flight /// refresh *fails* and the clock has crossed real expiry, waiters waking in - /// [`AutoRefresh::wait_for_in_flight_refresh`] must re-read the clock, find + /// [`RefreshState::wait_for_in_flight_refresh`] must re-read the clock, find /// the cached token unusable via `require_usable_token(now)`, and return /// `Expired` — they must not hang, and must not hand back a stale token. This /// is exactly the branch the post-wake clock re-read (the `now` re-read after @@ -2292,17 +2112,15 @@ mod expiry_crossing_regression { tokio::task::yield_now().await; } - // 4. Release the refresh → it returns Err. The first caller still returns - // the old token it captured while it was usable; the waiters re-read - // the now-advanced clock, find the token unusable, and get Expired. + // 4. Release the failed refresh. No caller may receive the old token: + // it expired while the exchange was in flight. gate.notify_one(); let first = first.await.unwrap(); - assert_eq!( - first.unwrap().as_str(), - "expiring-soon", - "first caller keeps the old token it captured before the refresh failed" - ); + assert!(matches!( + first, + Err(AutoRefreshError::Auth(crate::AuthError::TokenExpired(_))) + )); for (i, waiter) in waiters.into_iter().enumerate() { let result = waiter.await.unwrap(); @@ -2384,11 +2202,10 @@ mod expiry_crossing_regression { gate.notify_one(); let first = first.await.unwrap(); - assert_eq!( - first.unwrap().as_str(), - "expiring-soon", - "first caller keeps the old token it captured before the refusal" - ); + assert!(matches!( + first, + Err(AutoRefreshError::Auth(AuthError::UsageLimitExceeded(_))) + )); for (i, waiter) in waiters.into_iter().enumerate() { let result = waiter.await.unwrap(); @@ -2491,11 +2308,10 @@ mod expiry_crossing_regression { gate.notify_one(); let first = first.await.unwrap(); - assert_eq!( - first.unwrap().as_str(), - "expiring-soon", - "first caller keeps the old token it captured before the refusal" - ); + assert!(matches!( + first, + Err(AutoRefreshError::Auth(AuthError::InvalidGrant(_))) + )); for (i, waiter) in waiters.into_iter().enumerate() { let result = waiter.await.unwrap(); @@ -2541,124 +2357,10 @@ mod expiry_crossing_regression { } } -#[cfg(test)] -#[cfg(feature = "http")] -#[allow(clippy::unwrap_used)] -mod regression_cip_3159 { - use super::*; - use crate::access_key_refresher::AccessKeyRefresher; - use crate::SecretToken; - use std::sync::atomic::Ordering; - use std::sync::Arc; - use std::time::{Duration, SystemTime, UNIX_EPOCH}; - - /// `/api/authorise` handler that sleeps `delay` before returning a valid - /// access-key token response (with an ABSOLUTE-epoch `expiry`, as CTS - /// returns), giving the test a window to cancel in. - async fn delayed_authorise_handler( - axum::extract::State(delay): axum::extract::State, - ) -> axum::Json { - tokio::time::sleep(delay).await; - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs(); - axum::Json(serde_json::json!({ - "accessToken": "refreshed-token", - "expiry": now + 3600 - })) - } +#[cfg(all(test, feature = "http", not(target_arch = "wasm32")))] +#[path = "auto_refresh/tests/cancellation.rs"] +mod cancellation_tests; - async fn start_authorise_server(delay: Duration) -> url::Url { - let app = axum::Router::new() - .route( - "/api/authorise", - axum::routing::post(delayed_authorise_handler), - ) - .with_state(delay); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { - axum::serve(listener, app).await.unwrap(); - }); - url::Url::parse(&format!("http://{addr}")).unwrap() - } - - /// is_expired() == true (within the 90s leeway, so `get_token` refreshes), - /// but is_usable() == true for `secs_until_expiry` (so it takes the - /// non-blocking path). - fn expiring_but_usable_token(access: &str, secs_until_expiry: u64) -> Token { - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs(); - Token { - access_token: SecretToken::new(access), - token_type: "Bearer".to_string(), - expires_at: now + secs_until_expiry, - refresh_token: None, - region: None, - client_id: None, - device_instance_id: None, - federated_from: None, - } - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn cancellation_in_relock_window_does_not_strand_refresh() { - let http_delay = Duration::from_millis(400); - let base_url = start_authorise_server(http_delay).await; - - let strategy = Arc::new(AutoRefresh::with_token( - AccessKeyRefresher::new( - SecretToken::new("CSAKtestKeyId.testKeySecret"), - base_url, - None, - crate::transport::default_transport(), - ), - expiring_but_usable_token("old-usable", 2), - )); - - // Caller A drives the refresh: it locks state, sets the in-progress - // flag, drops the lock, then awaits the (slow) HTTP authorise call. - let a = Arc::clone(&strategy); - let handle = tokio::spawn(async move { a.get_token().await }); - - // Let A reach the HTTP await, then take the state lock so that when A's - // request completes it parks on its post-HTTP `state.lock().await` - // instead of installing the new token. - tokio::time::sleep(Duration::from_millis(100)).await; - let held = strategy.state.lock().await; - - // A's HTTP completes (~400ms) and blocks on the lock we hold. - tokio::time::sleep(http_delay + Duration::from_millis(200)).await; - assert!( - strategy.refresh_in_progress.load(Ordering::Acquire), - "precondition: a refresh should be in flight while caller A is parked", - ); - - // Cancel A precisely in the post-HTTP, pre-install window. - handle.abort(); - let _ = handle.await; - drop(held); - - // The CancelGuard's Drop must have cleared the flag on cancellation. - // Pre-fix, defuse() ran before the re-lock, so this stays `true`. - assert!( - !strategy.refresh_in_progress.load(Ordering::Acquire), - "refresh_in_progress stranded `true` after cancellation in the re-lock window (CIP-3159)", - ); - - // End-to-end: once the cached token crosses real expiry, a stranded flag - // would route the next caller into wait_for_in_flight_refresh and hang on - // a notify that never comes. With the fix, the caller re-authenticates. - tokio::time::sleep(Duration::from_millis(2100)).await; - let b = Arc::clone(&strategy); - let result = - tokio::time::timeout(Duration::from_secs(3), async move { b.get_token().await }).await; - assert!( - matches!(result, Ok(Ok(_))), - "get_token() hung or failed after cancellation — refresh wedged (CIP-3159): {result:?}", - ); - } -} +#[cfg(all(test, feature = "http", not(target_arch = "wasm32")))] +#[path = "auto_refresh/tests/soak.rs"] +mod cancellation_soak; diff --git a/packages/stack-auth/src/auto_refresh/tests/cancellation.rs b/packages/stack-auth/src/auto_refresh/tests/cancellation.rs new file mode 100644 index 000000000..821a034d1 --- /dev/null +++ b/packages/stack-auth/src/auto_refresh/tests/cancellation.rs @@ -0,0 +1,341 @@ +//! Cancellation must leave the rotating session usable, including after the +//! issuer has spent the old credential. Gates pin each await without sleeps. +#![allow(clippy::unwrap_used)] + +use super::*; +use crate::clock::TestClock; +use crate::{AuthError, SecretToken}; +use std::sync::atomic::AtomicUsize; +use std::time::Duration; +use tokio::sync::Semaphore; + +fn token(generation: usize, expires_at: u64) -> Token { + Token { + access_token: SecretToken::new(format!("access-{generation}")), + refresh_token: Some(SecretToken::new(generation.to_string())), + token_type: "Bearer".into(), + expires_at, + region: None, + client_id: None, + device_instance_id: None, + federated_from: None, + } +} + +struct Issuer { + started: Notify, + exchange: Semaphore, + answered: Notify, + persist: Semaphore, + generation: AtomicUsize, + saved: AtomicUsize, + calls: AtomicUsize, + fail_once: AtomicBool, + initial_auth: bool, +} + +impl Issuer { + fn new(initial_auth: bool) -> Self { + Self { + started: Notify::new(), + exchange: Semaphore::new(0), + answered: Notify::new(), + persist: Semaphore::new(0), + generation: AtomicUsize::new(0), + saved: AtomicUsize::new(0), + calls: AtomicUsize::new(0), + fail_once: AtomicBool::new(false), + initial_auth, + } + } +} + +struct RotatingRefresher(Arc); + +impl Refresher for RotatingRefresher { + type Credential = SecretToken; + + fn save(&self, _: &Token) {} + + fn try_credential(&self, token: Option<&mut Token>) -> Option { + match token { + Some(token) => token.take_refresh_token(), + None if self.0.initial_auth => Some(SecretToken::new("0")), + None => None, + } + } + + fn restore(&self, token: &mut Token, credential: SecretToken) { + token.refresh_token = Some(credential); + } + + async fn refresh(&self, credential: &SecretToken) -> Result { + let issuer = &self.0; + issuer.calls.fetch_add(1, Ordering::SeqCst); + issuer.started.notify_one(); + issuer.exchange.acquire().await.unwrap().forget(); + if issuer.fail_once.swap(false, Ordering::SeqCst) { + return Err(crate::error::ServerError("injected transient failure".into()).into()); + } + let previous = issuer.generation.fetch_add(1, Ordering::SeqCst); + assert_eq!( + credential.as_str(), + previous.to_string(), + "credential replay" + ); + // The issuer has already rotated, but the response is not persisted. + issuer.answered.notify_one(); + issuer.persist.acquire().await.unwrap().forget(); + issuer.saved.store(previous + 1, Ordering::SeqCst); + Ok(token(previous + 1, (previous as u64 + 1) * 10_000)) + } +} + +struct GatedStore { + entered: Notify, + save: Semaphore, + saved: AtomicUsize, +} + +impl TokenStore for GatedStore { + async fn load(&self) -> Option { + None + } + + async fn save(&self, token: &Token) { + self.entered.notify_one(); + self.save.acquire().await.unwrap().forget(); + self.saved.store( + token.refresh_token().unwrap().as_str().parse().unwrap(), + Ordering::SeqCst, + ); + } +} + +#[derive(Clone, Copy)] +enum Phase { + Exchange, + PersistRotation, + SaveStore, + Install, +} + +async fn cancelled_at(phase: Phase, usable: bool, initial: bool) { + let clock = TestClock::new(1_000); + let issuer = Arc::new(Issuer::new(initial)); + let store = Arc::new(GatedStore { + entered: Notify::new(), + save: Semaphore::new(0), + saved: AtomicUsize::new(0), + }); + let strategy = Arc::new(AutoRefresh::new( + RotatingRefresher(issuer.clone()), + store.clone(), + (!initial).then(|| token(0, if usable { 1_030 } else { 999 })), + clock.shared(), + )); + let caller = tokio::spawn({ + let strategy = strategy.clone(); + async move { strategy.get_token().await } + }); + issuer.started.notified().await; + if !matches!(phase, Phase::Exchange) { + issuer.exchange.add_permits(1); + issuer.answered.notified().await; + } + if matches!(phase, Phase::SaveStore | Phase::Install) { + issuer.persist.add_permits(1); + store.entered.notified().await; + } + // Only the non-blocking path releases the state lock during the exchange. + let held = if matches!(phase, Phase::Install) { + let held = strategy.shared.state.lock().await; + store.save.add_permits(1); + while store.saved.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + Some(held) + } else { + None + }; + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + // A later caller must wait for the rotated token, even on the formerly + // non-blocking path. Cancelling that waiter must not affect other waiters. + clock.advance(31); + assert!( + tokio::time::timeout(Duration::from_secs(1), strategy.get_token()) + .await + .is_err() + ); + let mut waiters = Vec::new(); + for _ in 0..32 { + let strategy = strategy.clone(); + waiters.push(tokio::spawn(async move { strategy.get_token().await })); + } + issuer.exchange.add_permits(1); + issuer.persist.add_permits(1); + store.save.add_permits(1); + drop(held); + for waiter in waiters { + let result = tokio::time::timeout(Duration::from_secs(2), waiter) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(result.as_str(), "access-1"); + } + assert_eq!(issuer.calls.load(Ordering::SeqCst), 1); + assert_eq!(issuer.saved.load(Ordering::SeqCst), 1); + assert_eq!(store.saved.load(Ordering::SeqCst), 1); + // A second rotation proves the new refresh credential was installed too. + clock.set(10_001); + issuer.exchange.add_permits(1); + issuer.persist.add_permits(1); + store.save.add_permits(1); + let result = strategy.get_token().await.unwrap(); + assert_eq!(result.as_str(), "access-2"); + assert_eq!(issuer.saved.load(Ordering::SeqCst), 2); + assert_eq!(store.saved.load(Ordering::SeqCst), 2); +} + +#[tokio::test(start_paused = true)] +async fn cancelled_blocking_refresh_finishes_every_phase() { + for phase in [Phase::Exchange, Phase::PersistRotation, Phase::SaveStore] { + cancelled_at(phase, false, false).await; + } +} + +#[tokio::test(start_paused = true)] +async fn cancelled_non_blocking_refresh_finishes_every_phase() { + for phase in [ + Phase::Exchange, + Phase::PersistRotation, + Phase::SaveStore, + Phase::Install, + ] { + cancelled_at(phase, true, false).await; + } +} + +#[tokio::test(start_paused = true)] +async fn cancelled_initial_auth_finishes_and_is_shared() { + for phase in [Phase::Exchange, Phase::SaveStore] { + cancelled_at(phase, false, true).await; + } +} + +#[tokio::test(start_paused = true)] +async fn cancelled_failed_refresh_restores_the_credential_for_retry() { + for usable in [false, true] { + let clock = TestClock::new(1_000); + let issuer = Arc::new(Issuer::new(false)); + issuer.fail_once.store(true, Ordering::SeqCst); + let strategy = Arc::new(AutoRefresh::with_token_and_clock( + RotatingRefresher(issuer.clone()), + token(0, if usable { 1_030 } else { 999 }), + clock.shared(), + )); + let caller = tokio::spawn({ + let strategy = strategy.clone(); + async move { strategy.get_token().await } + }); + issuer.started.notified().await; + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + issuer.exchange.add_permits(1); + // Let the failing task settle before retrying, on either lock path. + while strategy.shared.refresh_in_progress.load(Ordering::Acquire) { + tokio::task::yield_now().await; + } + clock.advance(31); + issuer.exchange.add_permits(1); + issuer.persist.add_permits(1); + assert_eq!(strategy.get_token().await.unwrap().as_str(), "access-1"); + assert_eq!(issuer.calls.load(Ordering::SeqCst), 2); + } +} + +#[tokio::test(start_paused = true)] +async fn rotation_is_saved_even_after_all_callers_drop_the_strategy() { + let issuer = Arc::new(Issuer::new(false)); + let strategy = Arc::new(AutoRefresh::with_token_and_clock( + RotatingRefresher(issuer.clone()), + token(0, 999), + TestClock::new(1_000).shared(), + )); + let caller = tokio::spawn(async move { strategy.get_token().await }); + issuer.started.notified().await; + issuer.exchange.add_permits(1); + issuer.answered.notified().await; + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + issuer.persist.add_permits(1); + tokio::time::timeout(Duration::from_secs(1), async { + while issuer.saved.load(Ordering::SeqCst) != 1 { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .unwrap(); + assert_eq!(issuer.calls.load(Ordering::SeqCst), 1); +} + +#[tokio::test(start_paused = true)] +async fn a_panicking_refresh_reports_an_error_and_releases_waiters() { + let issuer = Arc::new(Issuer::new(false)); + let clock = TestClock::new(1_000); + let strategy = Arc::new(AutoRefresh::with_token_and_clock( + RotatingRefresher(issuer.clone()), + token(0, 1_030), + clock.shared(), + )); + let caller = tokio::spawn({ + let strategy = strategy.clone(); + async move { strategy.get_token().await } + }); + issuer.started.notified().await; + clock.advance(31); + let waiter = tokio::spawn({ + let strategy = strategy.clone(); + async move { strategy.get_token().await } + }); + tokio::task::yield_now().await; + // Closing the gate makes the refresher panic at acquire().unwrap(). + issuer.exchange.close(); + assert!(matches!( + caller.await.unwrap(), + Err(AutoRefreshError::Auth(AuthError::Internal(_))) + )); + let result = tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .unwrap() + .unwrap(); + assert!(matches!(result, Err(AutoRefreshError::Expired))); +} + +#[tokio::test(start_paused = true)] +async fn guard_notification_before_waiter_registration_is_not_lost() { + let strategy = AutoRefresh::with_token_and_clock( + RotatingRefresher(Arc::new(Issuer::new(false))), + token(0, 999), + TestClock::new(1_000).shared(), + ); + let shared = &strategy.shared; + let state = shared.state.clone().lock_owned().await; + shared.refresh_in_progress.store(true, Ordering::Release); + // The waiter observed true while holding the lock, then the guard dropped + // before the waiter could register its Notified future. + drop(CancelGuard { + in_progress: &shared.refresh_in_progress, + notify: &shared.refresh_notify, + defused: false, + }); + let result = tokio::time::timeout( + Duration::from_secs(1), + shared.wait_for_in_flight_refresh(state, 1_000), + ) + .await + .unwrap(); + assert!(matches!(result, Err(AutoRefreshError::Expired))); +} diff --git a/packages/stack-auth/src/auto_refresh/tests/soak.rs b/packages/stack-auth/src/auto_refresh/tests/soak.rs new file mode 100644 index 000000000..3721d2be7 --- /dev/null +++ b/packages/stack-auth/src/auto_refresh/tests/soak.rs @@ -0,0 +1,268 @@ +//! Seeded, virtual-time cancellation soak. No credentials or network needed. +#![allow(clippy::unwrap_used)] + +use super::*; +use crate::clock::Clock; +use crate::{AuthError, SecretToken}; +use std::collections::HashSet; +use std::sync::Mutex as StdMutex; +use std::time::Duration; +use tokio::task::JoinSet; +use tokio::time::{sleep, timeout, Instant}; + +// One deterministic stream per caller and one for the issuer. There is no +// dependency on thread scheduling or the process-global RNG. +struct Random(u64); + +impl Random { + fn below(&mut self, limit: u64) -> u64 { + self.0 ^= self.0 << 13; + self.0 ^= self.0 >> 7; + self.0 ^= self.0 << 17; + self.0 % limit + } +} + +struct PausedClock(Instant); + +impl Clock for PausedClock { + fn now_unix_secs(&self) -> u64 { + 1_000_000 + self.0.elapsed().as_secs() + } +} + +struct IdentityProvider { + random: Random, + generation: u64, + used: HashSet, + revoked: bool, + saved: u64, + errors: u64, +} + +struct RotatingRefresher { + idp: Arc>, + clock: SharedClock, +} + +fn token(generation: u64, expires_at: u64) -> Token { + Token { + access_token: SecretToken::new(format!("{generation}:{expires_at}")), + refresh_token: Some(SecretToken::new(generation.to_string())), + token_type: "Bearer".into(), + expires_at, + region: None, + client_id: None, + device_instance_id: None, + federated_from: None, + } +} + +impl Refresher for RotatingRefresher { + type Credential = SecretToken; + + fn save(&self, _: &Token) {} + + fn try_credential(&self, token: Option<&mut Token>) -> Option { + token.and_then(Token::take_refresh_token) + } + + fn restore(&self, token: &mut Token, credential: SecretToken) { + token.refresh_token = Some(credential); + } + + async fn refresh(&self, credential: &SecretToken) -> Result { + let (delay, fail, lifetime) = { + let mut idp = self.idp.lock().unwrap(); + ( + idp.random.below(3_000), + idp.random.below(8) == 0, + 95 + idp.random.below(30), + ) + }; + sleep(Duration::from_millis(delay)).await; + let generation = { + let mut idp = self.idp.lock().unwrap(); + if fail { + // A definite rejection before spending the credential. An + // ambiguous network failure after rotation needs IdP support + // (idempotency/grace), and is not caused by caller cancellation. + idp.errors += 1; + return Err(crate::error::ServerError("injected transient failure".into()).into()); + } + let presented: u64 = credential.as_str().parse().unwrap(); + if presented != idp.generation || !idp.used.insert(presented) { + idp.revoked = true; + } + if idp.revoked { + return Err(crate::error::InvalidGrant.into()); + } + idp.generation += 1; + idp.generation + }; + Ok(token(generation, self.clock.now_unix_secs() + lifetime)) + } +} + +struct RotationStore(Arc>); + +impl TokenStore for RotationStore { + async fn load(&self) -> Option { + None + } + + async fn save(&self, token: &Token) { + let delay = self.0.lock().unwrap().random.below(2_000); + // Callers also cancel after the IdP has rotated, during persistence. + sleep(Duration::from_millis(delay)).await; + let mut idp = self.0.lock().unwrap(); + let generation: u64 = token.refresh_token().unwrap().as_str().parse().unwrap(); + assert_eq!(generation, idp.generation, "saved a stale rotation"); + assert_eq!(generation, idp.saved + 1, "lost or reordered a rotation"); + idp.saved = generation; + } +} + +async fn soak(seed: u64, rounds: usize) { + let clock: SharedClock = Arc::new(PausedClock(Instant::now())); + let idp = Arc::new(StdMutex::new(IdentityProvider { + random: Random(seed), + generation: 0, + used: HashSet::new(), + revoked: false, + saved: 0, + errors: 0, + })); + let strategy = Arc::new(AutoRefresh::new( + RotatingRefresher { + idp: idp.clone(), + clock: clock.clone(), + }, + RotationStore(idp.clone()), + Some(token(0, clock.now_unix_secs())), + clock.clone(), + )); + let mut callers = JoinSet::new(); + for caller in 1..=256 { + let strategy = strategy.clone(); + let clock = clock.clone(); + callers.spawn(async move { + let mut random = Random(seed.wrapping_add(caller)); + let (mut completed, mut timed_out, mut selected_out) = (0, 0, 0); + for _ in 0..rounds { + sleep(Duration::from_millis(random.below(60_000))).await; + let acquisition = async { + loop { + // Read before the call: the engine promises a token + // usable at the `now` it reads inside, which is no + // earlier than this. Read after, a busy runner's + // preemption in the paced run could pass expiry. + let before = clock.now_unix_secs(); + match strategy.get_token().await { + Ok(token) => { + let (_, expiry) = token.as_str().split_once(':').unwrap(); + assert!( + before < expiry.parse::().unwrap(), + "returned an expired token (seed {seed})" + ); + break; + } + // Injected transient errors may reach a caller + // with no usable token. Retry within the bound. + Err(AutoRefreshError::Auth(AuthError::Server(_))) => { + sleep(Duration::from_millis(10)).await; + } + Err(err) => panic!("session lost (seed {seed}): {err:?}"), + } + } + }; + let cancel_after = Duration::from_millis(random.below(4_000)); + // One third never cancel: a fixed outer bound proves they + // cannot get stranded behind a missed notification. + let kind = random.below(3); + timeout(Duration::from_secs(30), async { + match kind { + 0 => { + acquisition.await; + completed += 1; + } + 1 => { + if timeout(cancel_after, acquisition).await.is_ok() { + completed += 1; + } else { + timed_out += 1; + } + } + _ => tokio::select! { + biased; + () = acquisition => completed += 1, + () = sleep(cancel_after) => selected_out += 1, + }, + } + }) + .await + .expect("caller exceeded 30s virtual-time bound"); + } + (completed, timed_out, selected_out) + }); + } + let (mut completed, mut timed_out, mut selected_out) = (0, 0, 0); + while let Some(result) = callers.join_next().await { + let counts = result.unwrap(); + completed += counts.0; + timed_out += counts.1; + selected_out += counts.2; + } + // Drain a refresh whose last waiter cancelled, then compare all three + // views of the rotating credential: issuer, durable store and engine. + timeout(Duration::from_secs(30), async { + while strategy.shared.refresh_in_progress.load(Ordering::Acquire) { + sleep(Duration::from_millis(1)).await; + } + }) + .await + .unwrap(); + let state = strategy.shared.state.lock().await; + let installed: u64 = state + .token + .as_ref() + .unwrap() + .refresh_token() + .unwrap() + .as_str() + .parse() + .unwrap(); + let idp = idp.lock().unwrap(); + assert!(!idp.revoked, "refresh-token reuse revoked the session"); + assert_eq!(idp.saved, idp.generation, "newest rotation was not saved"); + assert_eq!( + installed, idp.generation, + "newest rotation was not installed" + ); + assert!( + idp.generation > 25 && idp.errors > 0, + "insufficient refresh/error coverage" + ); + assert!( + completed > 0 && timed_out > 0 && selected_out > 0, + "insufficient cancellation coverage" + ); + // Test-only progress counters make the long CI run's coverage reviewable. + #[allow(clippy::print_stderr)] + { + eprintln!("seed={seed}, rotations={}, injected_errors={}, completed={completed}, timeouts={timed_out}, select_cancels={selected_out}", idp.generation, idp.errors); + } +} + +#[tokio::test(start_paused = true)] +async fn cancellation_soak_short() { + soak(0x1170, 128).await; +} + +#[tokio::test(start_paused = true)] +#[ignore = "long release soak; run with test:soak:stack-auth"] +async fn cancellation_soak_long() { + for seed in [0x1170, 0xdead_beef, 0xcafe_f00d, 0x1234_5678] { + soak(seed, 4_096).await; + } +} diff --git a/packages/stack-auth/src/lib.rs b/packages/stack-auth/src/lib.rs index 6147a8d01..a5a295cb0 100644 --- a/packages/stack-auth/src/lib.rs +++ b/packages/stack-auth/src/lib.rs @@ -263,6 +263,13 @@ pub mod store { /// Only one refresh runs at a time, regardless of how many callers request a /// token concurrently. /// +/// On native targets, refreshes run in an owned task on the current Tokio +/// runtime. Cancelling `get_token` cancels only that caller's wait: the exchange, +/// persistence and token installation continue, preserving rotating device +/// sessions. Keep the runtime alive until outstanding refreshes finish; runtime +/// shutdown can still interrupt them. On wasm32 the host drives refreshes inline +/// to completion (there is no Tokio task executor). +/// /// [`OidcFederationStrategy`] runs one such engine **per provider JWT**: its /// provider is asked for the caller's JWT on every call, and the engine for /// that JWT is the one that refreshes, so users sharing a client never share a diff --git a/packages/stack-auth/src/oidc_federation_strategy.rs b/packages/stack-auth/src/oidc_federation_strategy.rs index 366ee0e44..177789e53 100644 --- a/packages/stack-auth/src/oidc_federation_strategy.rs +++ b/packages/stack-auth/src/oidc_federation_strategy.rs @@ -1660,8 +1660,9 @@ mod tests { ); assert_eq!(strategy.cached_jwts(), 0); - // The abandoned exchange's permit is still owed; the fresh call takes it. - gate.add_permits(1); + // The cancelled caller's exchange finishes in its owned task. The + // pending entry was removed, so the fresh engine also exchanges. + gate.add_permits(2); let token = (&*strategy) .get_token() .await @@ -1713,9 +1714,8 @@ mod tests { 1, "its token is cached all the same" ); - // The abandoned exchange died with the cancelled call; the survivor, - // next in line on the engine, made its own and took the permit. - assert_eq!(cts.exchanges(), 2); + // The survivor shares the exchange that outlived the first caller. + assert_eq!(cts.exchanges(), 1); } /// The remembered account refusal is forgotten after its window, so the diff --git a/packages/stack-auth/src/refresher.rs b/packages/stack-auth/src/refresher.rs index d2e2844d2..375d7aa07 100644 --- a/packages/stack-auth/src/refresher.rs +++ b/packages/stack-auth/src/refresher.rs @@ -13,9 +13,9 @@ use crate::{AuthError, Token}; /// reqwest's fetch-backed futures are not `Send` (they reference JS handles /// via `Rc>`) and edge runtimes are single-threaded anyway. #[cfg(not(target_arch = "wasm32"))] -pub(crate) trait Refresher: Send + Sync { +pub(crate) trait Refresher: Send + Sync + 'static { /// The credential extracted from the current token before a refresh attempt. - type Credential: Send; + type Credential: Send + 'static; /// Persist a token after a successful refresh. Best-effort — implementations /// should log on failure rather than returning an error. diff --git a/packages/stack-auth/src/token_store.rs b/packages/stack-auth/src/token_store.rs index a96ae33fc..99c2547df 100644 --- a/packages/stack-auth/src/token_store.rs +++ b/packages/stack-auth/src/token_store.rs @@ -70,11 +70,14 @@ use crate::Token; /// always validates freshness via [`Token::is_usable`] / [`Token::is_expired`] /// before returning a loaded token, so implementations don't need to. /// -/// On native targets the trait carries `Send + Sync` bounds so the store can -/// be shared across `tokio::spawn` background work. On wasm32 the bounds are -/// dropped — edge runtimes are single-threaded. +/// On native targets the store must be `Send + Sync + 'static`: refresh tasks +/// retain it until persistence finishes, even if their caller is cancelled. +/// Own captured state (for example with `Arc` and `move` closures) instead of +/// borrowing a request handler's stack. Save callbacks run in the refresh +/// task, so they must not depend on the initiating task's local context. +/// On wasm32 these bounds are dropped and refreshes run inline. #[cfg(not(target_arch = "wasm32"))] -pub trait TokenStore: Send + Sync { +pub trait TokenStore: Send + Sync + 'static { /// Load the most recently saved token, or `None` if none has been stored /// (or the load failed). Errors are swallowed — the calling state machine /// falls back to fresh authentication when this returns `None`. @@ -224,9 +227,9 @@ impl TokenStoreFn { #[cfg(not(target_arch = "wasm32"))] impl TokenStore for TokenStoreFn where - L: Fn() -> LF + Send + Sync, + L: Fn() -> LF + Send + Sync + 'static, LF: Future> + Send, - S: Fn(String) -> SF + Send + Sync, + S: Fn(String) -> SF + Send + Sync + 'static, SF: Future + Send, { async fn load(&self) -> Option { diff --git a/packages/stack-auth/tasks.toml b/packages/stack-auth/tasks.toml index 4c96551e5..51dd18af6 100644 --- a/packages/stack-auth/tasks.toml +++ b/packages/stack-auth/tasks.toml @@ -69,3 +69,7 @@ run = "mise x --env test -- cargo +nightly fuzz run jwt_decode --sanitizer none ["mutants:stack-auth"] description = "Full mutation-testing sweep of stack-auth (cargo-mutants; ~15 min)" run = "mise x --env test -- cargo mutants -p stack-auth --jobs 4" + +["test:soak:stack-auth"] +description = "Run the long seeded auth refresh cancellation soak (virtual time, no credentials)" +run = "mise x --env test -- cargo nextest run --locked -p stack-auth --all-features --run-ignored ignored-only -E 'test(cancellation_soak_long)' --success-output final" diff --git a/scripts/__tests__/auth-refresh-soak-ci.test.mjs b/scripts/__tests__/auth-refresh-soak-ci.test.mjs new file mode 100644 index 000000000..76edf4818 --- /dev/null +++ b/scripts/__tests__/auth-refresh-soak-ci.test.mjs @@ -0,0 +1,66 @@ +import { readFileSync } from 'node:fs' +import { join } from 'node:path' +import { describe, expect, it } from 'vitest' +import { REPO_ROOT } from './lib/repo-root.mjs' +import { readWorkflow } from './lib/workflows.mjs' + +const runner = './.github/workflows/_test-auth-refresh-soak.yml' +const prs = readWorkflow('.github/workflows/tests-auth-refresh.yml') +const release = readWorkflow('.github/workflows/release-plz.yml') +const soak = readWorkflow(runner) + +describe('auth refresh cancellation soak gates', () => { + it('runs the short soak on every PR without a path or branch filter', () => { + expect(prs.on.pull_request).toEqual({}) + expect(prs.jobs.short.uses).toBe(runner) + expect(prs.jobs.short.if).toBeUndefined() + expect(soak.on.workflow_call.inputs.long.default).toBe(false) + const short = soak.jobs.soak.steps.find( + (step) => step.name === 'Short seeded soak', + ) + expect(short.run).toContain('test(cancellation_soak_short)') + expect(short['continue-on-error']).toBeUndefined() + }) + + it('runs the long soak on Version Packages and refresher changes', () => { + expect(prs.jobs.long.uses).toBe(runner) + expect(prs.jobs.long.with.long).toBe(true) + expect(prs.jobs.long.if).toContain( + "github.head_ref == 'changeset-release/main'", + ) + expect(prs.jobs.long.if).toContain( + "needs.changes.outputs.refresh == 'true'", + ) + const paths = prs.jobs.changes.steps.find((step) => step.id === 'paths') + expect(paths.with.filters).toContain("'packages/stack-auth/**'") + }) + + it('requires a successful long soak before publishing stack-auth', () => { + const gate = release.jobs['auth-refresh-soak'] + expect( + gate.steps.find((step) => step.name === 'Long seeded soak').run, + ).toBe('mise run test:soak:stack-auth') + expect(gate.if).toBeUndefined() + expect(release.jobs['release-crates'].needs).toContain('auth-refresh-soak') + // No status function bypasses the implicit success() dependency check. + expect(release.jobs['release-crates'].if).not.toMatch( + /always\(|cancelled\(|failure\(/, + ) + }) + + it('actually selects the ignored long test and propagates failures', () => { + const long = soak.jobs.soak.steps.find( + (step) => step.name === 'Long seeded soak', + ) + expect(long.run).toBe('mise run test:soak:stack-auth') + expect(long['continue-on-error']).toBeUndefined() + const tasks = readFileSync( + join(REPO_ROOT, 'packages/stack-auth/tasks.toml'), + 'utf8', + ) + const task = tasks.split('["test:soak:stack-auth"]')[1].split('\n[')[0] + expect(task).toContain('--run-ignored ignored-only') + expect(task).toContain('test(cancellation_soak_long)') + expect(task).not.toContain('|| true') + }) +}) From 9a0707a00c49a6a67c870a6f5c05900c974f906f Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Tue, 13 Oct 2026 00:34:46 +1100 Subject: [PATCH 2/7] test(auth): run native guest exchange under Tokio Native token refresh now runs in an owned Tokio task, so the guest header test must await its exchange inside a runtime. Add Tokio only as a development dependency and retain the existing header assertions. Refs #1170. --- languages/golang/auth/guest/Cargo.lock | 1 + languages/golang/auth/guest/Cargo.toml | 2 ++ languages/golang/auth/guest/src/headers.rs | 7 +++---- 3 files changed, 6 insertions(+), 4 deletions(-) diff --git a/languages/golang/auth/guest/Cargo.lock b/languages/golang/auth/guest/Cargo.lock index 1616b6d77..9c8f0b324 100644 --- a/languages/golang/auth/guest/Cargo.lock +++ b/languages/golang/auth/guest/Cargo.lock @@ -1857,6 +1857,7 @@ dependencies = [ "stack-guest-abi", "stack-profile", "tempfile", + "tokio", "url", "vitaminc-aead-value", "vitaminc-protected", diff --git a/languages/golang/auth/guest/Cargo.toml b/languages/golang/auth/guest/Cargo.toml index d3928e995..48bef983e 100644 --- a/languages/golang/auth/guest/Cargo.toml +++ b/languages/golang/auth/guest/Cargo.toml @@ -51,6 +51,8 @@ zeroize = { version = "1", features = ["derive"] } [dev-dependencies] serde_json = "1" tempfile = "3" +# Native stack-auth refreshes run in a Tokio task, including with a mock transport. +tokio = { version = "1.47.1", default-features = false, features = ["macros", "rt"] } # `Controlled::risky_ref`, to read a codec value back in the tests. vitaminc-protected = "0.5.0" diff --git a/languages/golang/auth/guest/src/headers.rs b/languages/golang/auth/guest/src/headers.rs index 3ed454d0c..075ef3d44 100644 --- a/languages/golang/auth/guest/src/headers.rs +++ b/languages/golang/auth/guest/src/headers.rs @@ -55,7 +55,6 @@ pub fn request_headers(headers: &[(String, String)]) -> Vec { mod tests { use std::sync::{Arc, Mutex}; - use futures::executor::block_on; use stack_auth::{ AccessKeyStrategy, AuthStrategy, HttpRequest, HttpResponse, HttpTransport, RequestError, }; @@ -90,8 +89,8 @@ mod tests { /// `user-agent` is Go's default with a bare nginx 403, before CTS sees /// it. Every request this guest hands the host must name the library /// and the host, once, and keep the headers stack-auth built. - #[test] - fn every_request_identifies_itself() { + #[tokio::test] + async fn every_request_identifies_itself() { let capture = Arc::new(Capture::default()); let strategy = AccessKeyStrategy::builder( "crn:ap-southeast-2.aws:ZVATKW3VHMFG27DY".parse().unwrap(), @@ -101,7 +100,7 @@ mod tests { .transport(Arc::clone(&capture)) .build() .unwrap(); - let _ = block_on((&strategy).get_token()); + let _ = (&strategy).get_token().await; let seen = capture.0.lock().unwrap(); assert_eq!(seen.len(), 1, "one access-key exchange"); From 75e359e786408c74aa454cece24ba7a203c8b1f3 Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Tue, 13 Oct 2026 01:02:29 +1100 Subject: [PATCH 3/7] test(auth): cover token expiry boundaries in refresh Assert which token the initiating caller receives immediately before, at, and after expiry so both surviving comparison mutations fail. Keep cfg(test) separate on the cancellation and soak modules so cargo-mutants recognises the test harnesses without weakening production mutation coverage. The full PR mutation sweep passes with 5 caught and 9 unviable mutants, no survivors or timeouts. Refs #1170. --- packages/stack-auth/src/auto_refresh.rs | 57 ++++++++++++++++++++++++- 1 file changed, 55 insertions(+), 2 deletions(-) diff --git a/packages/stack-auth/src/auto_refresh.rs b/packages/stack-auth/src/auto_refresh.rs index a264dd38b..b13c2d3d4 100644 --- a/packages/stack-auth/src/auto_refresh.rs +++ b/packages/stack-auth/src/auto_refresh.rs @@ -1936,6 +1936,55 @@ mod expiry_crossing_regression { } } + #[tokio::test(start_paused = true)] + async fn initiating_caller_returns_the_old_token_only_before_its_expiry() { + for (elapsed, expected) in [ + (9, "expiring-soon"), + (10, "refreshed-token"), + (11, "refreshed-token"), + ] { + let clock = TestClock::new(1_000_000); + let started = Arc::new(Notify::new()); + let gate = Arc::new(Notify::new()); + let calls = Arc::new(AtomicUsize::new(0)); + let strategy = Arc::new(AutoRefresh::with_token_and_clock( + GatedRefresher { + started: Arc::clone(&started), + gate: Arc::clone(&gate), + calls: Arc::clone(&calls), + refreshed_expires_at: clock.now() + 3600, + }, + make_token("expiring-soon", clock.now() + 10), + clock.shared(), + )); + let caller = tokio::spawn({ + let strategy = Arc::clone(&strategy); + async move { strategy.get_token().await } + }); + tokio::time::timeout(std::time::Duration::from_secs(1), started.notified()) + .await + .expect("the usable token must start a refresh"); + + // Finish the exchange one second before, exactly at, and one + // second after the old token's expiry. Only the first case may + // return the old token to the caller that initiated the refresh. + clock.advance(elapsed); + gate.notify_one(); + let returned = tokio::time::timeout(std::time::Duration::from_secs(1), caller) + .await + .expect("the initiating caller must finish") + .unwrap() + .unwrap(); + assert_eq!(returned.as_str(), expected, "elapsed={elapsed}"); + assert_eq!( + strategy.get_token().await.unwrap().as_str(), + "refreshed-token", + "the refreshed token must be cached in every case" + ); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + } + #[tokio::test] async fn waiters_wait_for_refresh_when_token_crosses_expiry() { let clock = TestClock::new(1_000_000); @@ -2357,10 +2406,14 @@ mod expiry_crossing_regression { } } -#[cfg(all(test, feature = "http", not(target_arch = "wasm32")))] +// Keep `test` as a standalone cfg so cargo-mutants recognises these modules +// as test harnesses and mutates the refresh engine they exercise instead. +#[cfg(test)] +#[cfg(all(feature = "http", not(target_arch = "wasm32")))] #[path = "auto_refresh/tests/cancellation.rs"] mod cancellation_tests; -#[cfg(all(test, feature = "http", not(target_arch = "wasm32")))] +#[cfg(test)] +#[cfg(all(feature = "http", not(target_arch = "wasm32")))] #[path = "auto_refresh/tests/soak.rs"] mod cancellation_soak; From 530ea59d8141d25f07310005f85305d99ff4647f Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Sun, 18 Oct 2026 23:04:05 +1100 Subject: [PATCH 4/7] fix(stack-auth): give the initiating caller the token its refresh installed After a successful non-blocking refresh the engine installed the new token but handed the caller that started the refresh the old one, unless the old one had already expired. A non-blocking refresh starts only when the cached token is inside the 90-second expiry leeway, so that caller could get a token with seconds left and a 401 partway through its request. install_refreshed_token already returns the new ServiceToken; return it. That removes the expiry comparison on the success path, so the boundary test now asserts the refreshed token is returned before, at and after the old token's expiry. The failure path still returns the old token while it is unexpired, which is unchanged. The behaviour predates the cancellation work; it was raised in review of PR #1172. --- .changeset/cancelled-auth-refresh.md | 2 +- packages/stack-auth/src/auto_refresh.rs | 30 +++++++++++-------------- 2 files changed, 14 insertions(+), 18 deletions(-) diff --git a/.changeset/cancelled-auth-refresh.md b/.changeset/cancelled-auth-refresh.md index bc25e17e6..ccd0e5d70 100644 --- a/.changeset/cancelled-auth-refresh.md +++ b/.changeset/cancelled-auth-refresh.md @@ -2,4 +2,4 @@ '@cipherstash/auth': patch --- -Keep native token refreshes running when their caller is cancelled, so rotated device-session credentials are saved and subsequent requests can still authenticate. Close a refresh-waiter notification race and avoid returning a token that expired during refresh. +Keep native token refreshes running when their caller is cancelled, so rotated device-session credentials are saved and subsequent requests can still authenticate. Close a refresh-waiter notification race, and give the caller that started a refresh the new token rather than the expiring one it replaced. diff --git a/packages/stack-auth/src/auto_refresh.rs b/packages/stack-auth/src/auto_refresh.rs index b13c2d3d4..a459eb3ef 100644 --- a/packages/stack-auth/src/auto_refresh.rs +++ b/packages/stack-auth/src/auto_refresh.rs @@ -520,14 +520,13 @@ impl RefreshState { Ok(new_token) => { self.save_refreshed_token(&new_token).await; let mut state = self.state.lock().await; - let _ = self.install_refreshed_token(&mut state, new_token); + // The initiating caller gets the token it just installed. The + // old one is inside the expiry leeway by definition (that is + // why this refresh started), so handing it back could leave + // the caller only seconds to use it. + let token = self.install_refreshed_token(&mut state, new_token); guard.defuse(); - let now = self.clock.now_unix_secs(); - if now < current_expires_at { - Ok(current_service_token) - } else { - state.require_usable_token(now) - } + Ok(token) } Err(err) => { let consumed = credential_consumed(&err); @@ -1937,12 +1936,8 @@ mod expiry_crossing_regression { } #[tokio::test(start_paused = true)] - async fn initiating_caller_returns_the_old_token_only_before_its_expiry() { - for (elapsed, expected) in [ - (9, "expiring-soon"), - (10, "refreshed-token"), - (11, "refreshed-token"), - ] { + async fn initiating_caller_receives_the_refreshed_token() { + for elapsed in [0, 9, 10, 11] { let clock = TestClock::new(1_000_000); let started = Arc::new(Notify::new()); let gate = Arc::new(Notify::new()); @@ -1965,9 +1960,10 @@ mod expiry_crossing_regression { .await .expect("the usable token must start a refresh"); - // Finish the exchange one second before, exactly at, and one - // second after the old token's expiry. Only the first case may - // return the old token to the caller that initiated the refresh. + // Finish the exchange well before, one second before, exactly at, + // and one second after the old token's expiry. The caller that + // initiated the refresh gets the new token every time: the old + // one is already inside the expiry leeway. clock.advance(elapsed); gate.notify_one(); let returned = tokio::time::timeout(std::time::Duration::from_secs(1), caller) @@ -1975,7 +1971,7 @@ mod expiry_crossing_regression { .expect("the initiating caller must finish") .unwrap() .unwrap(); - assert_eq!(returned.as_str(), expected, "elapsed={elapsed}"); + assert_eq!(returned.as_str(), "refreshed-token", "elapsed={elapsed}"); assert_eq!( strategy.get_token().await.unwrap().as_str(), "refreshed-token", From 6503f060a0bdeec4ed1f89a39735015a9072b000 Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Sun, 18 Oct 2026 23:04:05 +1100 Subject: [PATCH 5/7] test(stack-auth): run the cancellation soak on a multi-thread runtime The seeded soak runs on a current-thread runtime with paused time so a seed replays exactly. That also means it never interleaves callers across worker threads, so it cannot find a bug that only parallel workers expose. Add a short run on a four-worker runtime in real time. A Pace divides every scripted delay and speeds the engine's clock by the same factor, so token lifetimes and the expiry leeway keep their proportions; at Pace(100) the run takes about four seconds. It is not replayable, so it checks the session invariants (no reuse, newest rotation saved and installed, no expired token returned, no caller stranded) and that the session rotated, but no coverage floor. The stranded-caller bound is a fixed ten seconds of real time there, so a busy CI runner cannot trip it. The seeded runs pass Pace(1) and behave exactly as before. --- .../stack-auth/src/auto_refresh/tests/soak.rs | 101 +++++++++++++----- 1 file changed, 74 insertions(+), 27 deletions(-) diff --git a/packages/stack-auth/src/auto_refresh/tests/soak.rs b/packages/stack-auth/src/auto_refresh/tests/soak.rs index 3721d2be7..5746f95df 100644 --- a/packages/stack-auth/src/auto_refresh/tests/soak.rs +++ b/packages/stack-auth/src/auto_refresh/tests/soak.rs @@ -1,4 +1,9 @@ -//! Seeded, virtual-time cancellation soak. No credentials or network needed. +//! Seeded cancellation soak. No credentials or network needed. +//! +//! The seeded runs use a current-thread runtime with paused time, so a seed +//! replays exactly. That cannot interleave callers across worker threads, so +//! a short run on a multi-thread runtime in real time, sped up by [`Pace`], +//! covers the orderings only parallel workers produce. #![allow(clippy::unwrap_used)] use super::*; @@ -23,11 +28,36 @@ impl Random { } } -struct PausedClock(Instant); +/// How many times faster than the wall clock the soak runs. Every scripted +/// delay is divided by it and the engine's clock counts that many seconds per +/// elapsed second, so token lifetimes and the expiry leeway keep their +/// proportions. The paused-time runs use 1. +#[derive(Clone, Copy)] +struct Pace(u64); -impl Clock for PausedClock { +impl Pace { + fn ms(self, millis: u64) -> Duration { + Duration::from_micros(millis * 1_000 / self.0) + } + + /// The bound within which a caller must get a token or give up. Paced + /// runs share real CPUs with the rest of the suite, so they get a fixed + /// real-time bound instead of a scaled one that a busy runner could miss. + fn stranded_after(self) -> Duration { + if self.0 == 1 { + Duration::from_secs(30) + } else { + Duration::from_secs(10) + } + } +} + +struct PacedClock(Instant, Pace); + +impl Clock for PacedClock { fn now_unix_secs(&self) -> u64 { - 1_000_000 + self.0.elapsed().as_secs() + let elapsed_ms = u64::try_from(self.0.elapsed().as_millis()).unwrap(); + 1_000_000 + elapsed_ms * self.1 .0 / 1_000 } } @@ -43,6 +73,7 @@ struct IdentityProvider { struct RotatingRefresher { idp: Arc>, clock: SharedClock, + pace: Pace, } fn token(generation: u64, expires_at: u64) -> Token { @@ -80,7 +111,7 @@ impl Refresher for RotatingRefresher { 95 + idp.random.below(30), ) }; - sleep(Duration::from_millis(delay)).await; + sleep(self.pace.ms(delay)).await; let generation = { let mut idp = self.idp.lock().unwrap(); if fail { @@ -104,7 +135,7 @@ impl Refresher for RotatingRefresher { } } -struct RotationStore(Arc>); +struct RotationStore(Arc>, Pace); impl TokenStore for RotationStore { async fn load(&self) -> Option { @@ -114,7 +145,7 @@ impl TokenStore for RotationStore { async fn save(&self, token: &Token) { let delay = self.0.lock().unwrap().random.below(2_000); // Callers also cancel after the IdP has rotated, during persistence. - sleep(Duration::from_millis(delay)).await; + sleep(self.1.ms(delay)).await; let mut idp = self.0.lock().unwrap(); let generation: u64 = token.refresh_token().unwrap().as_str().parse().unwrap(); assert_eq!(generation, idp.generation, "saved a stale rotation"); @@ -123,8 +154,8 @@ impl TokenStore for RotationStore { } } -async fn soak(seed: u64, rounds: usize) { - let clock: SharedClock = Arc::new(PausedClock(Instant::now())); +async fn soak(seed: u64, rounds: usize, pace: Pace) { + let clock: SharedClock = Arc::new(PacedClock(Instant::now(), pace)); let idp = Arc::new(StdMutex::new(IdentityProvider { random: Random(seed), generation: 0, @@ -137,8 +168,9 @@ async fn soak(seed: u64, rounds: usize) { RotatingRefresher { idp: idp.clone(), clock: clock.clone(), + pace, }, - RotationStore(idp.clone()), + RotationStore(idp.clone(), pace), Some(token(0, clock.now_unix_secs())), clock.clone(), )); @@ -150,7 +182,7 @@ async fn soak(seed: u64, rounds: usize) { let mut random = Random(seed.wrapping_add(caller)); let (mut completed, mut timed_out, mut selected_out) = (0, 0, 0); for _ in 0..rounds { - sleep(Duration::from_millis(random.below(60_000))).await; + sleep(pace.ms(random.below(60_000))).await; let acquisition = async { loop { // Read before the call: the engine promises a token @@ -170,17 +202,17 @@ async fn soak(seed: u64, rounds: usize) { // Injected transient errors may reach a caller // with no usable token. Retry within the bound. Err(AutoRefreshError::Auth(AuthError::Server(_))) => { - sleep(Duration::from_millis(10)).await; + sleep(pace.ms(10)).await; } Err(err) => panic!("session lost (seed {seed}): {err:?}"), } } }; - let cancel_after = Duration::from_millis(random.below(4_000)); + let cancel_after = pace.ms(random.below(4_000)); // One third never cancel: a fixed outer bound proves they // cannot get stranded behind a missed notification. let kind = random.below(3); - timeout(Duration::from_secs(30), async { + timeout(pace.stranded_after(), async { match kind { 0 => { acquisition.await; @@ -201,7 +233,7 @@ async fn soak(seed: u64, rounds: usize) { } }) .await - .expect("caller exceeded 30s virtual-time bound"); + .expect("caller stranded past its bound"); } (completed, timed_out, selected_out) }); @@ -215,9 +247,9 @@ async fn soak(seed: u64, rounds: usize) { } // Drain a refresh whose last waiter cancelled, then compare all three // views of the rotating credential: issuer, durable store and engine. - timeout(Duration::from_secs(30), async { + timeout(pace.stranded_after(), async { while strategy.shared.refresh_in_progress.load(Ordering::Acquire) { - sleep(Duration::from_millis(1)).await; + sleep(pace.ms(1)).await; } }) .await @@ -239,14 +271,21 @@ async fn soak(seed: u64, rounds: usize) { installed, idp.generation, "newest rotation was not installed" ); - assert!( - idp.generation > 25 && idp.errors > 0, - "insufficient refresh/error coverage" - ); - assert!( - completed > 0 && timed_out > 0 && selected_out > 0, - "insufficient cancellation coverage" - ); + // Coverage floors hold only for a replayable seed. In real time the + // scheduler decides how many callers cancel, so the multi-thread run + // checks the invariants above and that the session rotated at all. + if pace.0 == 1 { + assert!( + idp.generation > 25 && idp.errors > 0, + "insufficient refresh/error coverage" + ); + assert!( + completed > 0 && timed_out > 0 && selected_out > 0, + "insufficient cancellation coverage" + ); + } else { + assert!(idp.generation > 0, "the session never rotated"); + } // Test-only progress counters make the long CI run's coverage reviewable. #[allow(clippy::print_stderr)] { @@ -256,13 +295,21 @@ async fn soak(seed: u64, rounds: usize) { #[tokio::test(start_paused = true)] async fn cancellation_soak_short() { - soak(0x1170, 128).await; + soak(0x1170, 128, Pace(1)).await; +} + +/// Real time on four workers, 100 times faster than the seeded runs: about +/// three seconds of wall clock. Not replayable, so it asserts no coverage +/// floor; it exists to interleave callers in ways one thread cannot. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn cancellation_soak_multi_thread() { + soak(0x1170, 8, Pace(100)).await; } #[tokio::test(start_paused = true)] #[ignore = "long release soak; run with test:soak:stack-auth"] async fn cancellation_soak_long() { for seed in [0x1170, 0xdead_beef, 0xcafe_f00d, 0x1234_5678] { - soak(seed, 4_096).await; + soak(seed, 4_096, Pace(1)).await; } } From f5cd42844d86a36ba4fc9ce65ab7933f0e692889 Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Sun, 18 Oct 2026 23:04:06 +1100 Subject: [PATCH 6/7] docs(golang): say why the auth guest's block_on is wasm32-only On native targets stack-auth now spawns a token refresh onto the current Tokio runtime. futures::executor::block_on provides none, so token and refresh in the auth guest would panic on the first refresh in a native build. Today only the wasm32 abi exports reach them, where stack-auth refreshes inline and block_on is sound. A cfg(target_arch = "wasm32") gate would leave their helpers dead on the native test build, so record the constraint where block_on is imported instead, before a native test finds the panic by accident. --- languages/golang/auth/guest/src/auth.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/languages/golang/auth/guest/src/auth.rs b/languages/golang/auth/guest/src/auth.rs index e1f0154cf..d8d8acfd7 100644 --- a/languages/golang/auth/guest/src/auth.rs +++ b/languages/golang/auth/guest/src/auth.rs @@ -6,6 +6,13 @@ use std::cell::{Cell, RefCell}; use std::collections::HashMap; use cts_common::Crn; +// Sound on wasm32 only. There stack-auth runs a token refresh inline, so a +// plain executor can drive `get_token` to completion. On a native target +// stack-auth spawns the refresh onto the current Tokio runtime, and +// `block_on` provides none, so the first refresh panics (and inside a +// current-thread runtime it would block the thread the task needs). `token` +// and `refresh` below are reached only through the wasm32 `abi` exports. A +// native build, a test included, must not call them. use futures::executor::block_on; use serde::Deserialize; use stack_auth::{ From 1b270d719a33b0ca8b2d0389aa1600bdbcd0725a Mon Sep 17 00:00:00 2001 From: Dan Draper Date: Sun, 18 Oct 2026 23:04:06 +1100 Subject: [PATCH 7/7] ci(release-plz): run the auth refresh soak only on an armed crates line The long soak ran on every push matching release-plz.yml's path filter, even when the stack-* crates line is not armed and release-crates is skipped anyway. Give it the same crates-armed gate as release-crates. It still runs before release-crates' preflight asks crates.io whether any version is unpublished, so a failed soak also holds stack-kms and stack-encrypt. Moving that preflight into its own job would let the soak run only for an unpublished stack-auth, but release-crates would then need a status function to run past a skipped soak, which the soak gate test forbids on the publish job. Left as is. eql-pipeline-armed.test.mjs pins exactly which jobs the switch gates; it now lists the soak beside the publish job. --- .github/workflows/release-plz.yml | 6 ++++++ scripts/__tests__/auth-refresh-soak-ci.test.mjs | 6 +++++- scripts/__tests__/eql-pipeline-armed.test.mjs | 9 +++++++-- 3 files changed, 18 insertions(+), 3 deletions(-) diff --git a/.github/workflows/release-plz.yml b/.github/workflows/release-plz.yml index 50265415a..3e65aee13 100644 --- a/.github/workflows/release-plz.yml +++ b/.github/workflows/release-plz.yml @@ -225,8 +225,14 @@ jobs: id: armed run: node scripts/eql-pipeline-armed.mjs crates + # Gated on the switch so an unarmed line spends nothing on the soak. It + # still runs on every armed push, before release-crates' preflight decides + # whether any crate is unpublished, so a failed soak holds stack-kms and + # stack-encrypt too. auth-refresh-soak: name: Auth refresh cancellation soak before publishing + needs: [crates-armed] + if: needs.crates-armed.outputs.armed == 'true' runs-on: ubuntu-latest timeout-minutes: 60 env: diff --git a/scripts/__tests__/auth-refresh-soak-ci.test.mjs b/scripts/__tests__/auth-refresh-soak-ci.test.mjs index 76edf4818..d35763e35 100644 --- a/scripts/__tests__/auth-refresh-soak-ci.test.mjs +++ b/scripts/__tests__/auth-refresh-soak-ci.test.mjs @@ -40,7 +40,11 @@ describe('auth refresh cancellation soak gates', () => { expect( gate.steps.find((step) => step.name === 'Long seeded soak').run, ).toBe('mise run test:soak:stack-auth') - expect(gate.if).toBeUndefined() + // Only an armed line runs the soak; release-crates needs the same switch, + // so a skipped soak never stands in front of a publish. + expect(gate.needs).toEqual(['crates-armed']) + expect(gate.if).toBe("needs.crates-armed.outputs.armed == 'true'") + expect(release.jobs['release-crates'].if).toBe(gate.if) expect(release.jobs['release-crates'].needs).toContain('auth-refresh-soak') // No status function bypasses the implicit success() dependency check. expect(release.jobs['release-crates'].if).not.toMatch( diff --git a/scripts/__tests__/eql-pipeline-armed.test.mjs b/scripts/__tests__/eql-pipeline-armed.test.mjs index 0351ae0af..5895cd342 100644 --- a/scripts/__tests__/eql-pipeline-armed.test.mjs +++ b/scripts/__tests__/eql-pipeline-armed.test.mjs @@ -163,7 +163,12 @@ describe('every EQL publish job reads it', () => { */ const CRATES_PACKAGE = '@cipherstash/auth' const CRATES_WORKFLOW = '.github/workflows/release-plz.yml' -const CRATES_GATED_JOBS = [`${CRATES_WORKFLOW} / release-crates`] +// The publish job, and the auth refresh soak it needs: an unarmed line has +// nothing to publish, so it has nothing to soak either. +const CRATES_GATED_JOBS = [ + `${CRATES_WORKFLOW} / auth-refresh-soak`, + `${CRATES_WORKFLOW} / release-crates`, +] describe('the stack-* crates line', () => { it('keys on the @cipherstash/auth entry', () => { @@ -212,7 +217,7 @@ describe('the stack-* crates line', () => { } }) - it('is computed by release-plz.yml and gates exactly the crates publish job', () => { + it('is computed by release-plz.yml and gates exactly the crates publish job and its soak', () => { const job = readWorkflow(CRATES_WORKFLOW)?.jobs?.['crates-armed'] const runsIt = (job?.steps ?? []).some((step) => String(step?.run ?? '').includes('scripts/eql-pipeline-armed.mjs crates'),