Watch
0
0
Fork
You've already forked hyperhive
0

metrics auth responders keep the last good bearer while the store doesn't answer

This commit is contained in:
atlas 2026-11-04 17:21:28 +01:00
commit dff2f3ca4d
5 changed files with 325 additions and 145 deletions

2
Cargo.lock generated
View file

@ -5109,7 +5109,9 @@ version = "0.1.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"axum", "axum",
"chrono",
"clap", "clap",
"mockall",
"serde", "serde",
"subtle", "subtle",
"swarm-secret-client", "swarm-secret-client",

View file

@ -43,7 +43,7 @@ the rest of the cell says how.
| `swarm/hives/<hive>/oidc/client` | authelia, first boot, straight into the store | the hive | ❌ minted once, straight into the store; minted again only when the store holds none | ✅ re-read when authelia refuses it | | `swarm/hives/<hive>/oidc/client` | authelia, first boot, straight into the store | the hive | ❌ minted once, straight into the store; minted again only when the store holds none | ✅ re-read when authelia refuses it |
| `swarm/services/<clientId>/oidc/client` | authelia | that service | must be stated | must be stated | | `swarm/services/<clientId>/oidc/client` | authelia | that service | must be stated | must be stated |
| `swarm/services/ci-runner/registration-token` | `swarm-controller`, from the forge, at start and every 5 minutes | `swarm-bao-ci-runner` on the runner's host | ❌ the forge keeps one token until an admin resets it; the controller then writes the new one within 5 minutes | ✅ every 10 minutes; restarts the runner on a change | | `swarm/services/ci-runner/registration-token` | `swarm-controller`, from the forge, at start and every 5 minutes | `swarm-bao-ci-runner` on the runner's host | ❌ the forge keeps one token until an admin resets it; the controller then writes the new one within 5 minutes | ✅ every 10 minutes; restarts the runner on a change |
| `swarm/services/<grafana client id>/metrics/bearer` | `swarm-bao-grafana-metrics-bearer` on Grafana's host, when none is stored | `swarm-metrics-auth` and `swarm-logs-grafana-auth` | ❌ delete it and restart `swarm-bao-grafana-metrics-bearer` | ✅ the auth services re-read within a minute; ❌ Grafana restarts when its copy changes | | `swarm/services/<grafana client id>/metrics/bearer` | `swarm-bao-grafana-metrics-bearer` on Grafana's host, when none is stored | `swarm-metrics-auth` and `swarm-logs-grafana-auth` | ❌ delete it and restart `swarm-bao-grafana-metrics-bearer` | ✅ the auth services re-read within a minute, and until they can reach the store they keep accepting the value they last read; ❌ Grafana restarts when its copy changes |
| `swarm/services/<grafana client id>/secret-key` | `swarm-bao-grafana-secret-key` on Grafana's host, from the key an earlier release generated; a new one only when there is none | Grafana | ❌ minted once; a different key leaves Grafana unable to read its stored datasource secrets | ❌ Grafana reads it at start; the unit restarts Grafana when the file changes | | `swarm/services/<grafana client id>/secret-key` | `swarm-bao-grafana-secret-key` on Grafana's host, from the key an earlier release generated; a new one only when there is none | Grafana | ❌ minted once; a different key leaves Grafana unable to read its stored datasource secrets | ❌ Grafana reads it at start; the unit restarts Grafana when the file changes |
| `swarm/services/swarm-authelia/keys/<name>`, for `jwt`, `session`, `storage-encryption`, `oidc-hmac` and `oidc-issuer` | authelia's `authelia-swarm-secrets` unit in its container, under the secret publisher's identity, from the key file an earlier release generated; a new one only when there is none | authelia | ❌ minted once; a different storage-encryption key leaves authelia's database unreadable and a different issuer key invalidates every issued token | ❌ read when the unit runs; the unit restarts authelia when the file changes | | `swarm/services/swarm-authelia/keys/<name>`, for `jwt`, `session`, `storage-encryption`, `oidc-hmac` and `oidc-issuer` | authelia's `authelia-swarm-secrets` unit in its container, under the secret publisher's identity, from the key file an earlier release generated; a new one only when there is none | authelia | ❌ minted once; a different storage-encryption key leaves authelia's database unreadable and a different issuer key invalidates every issued token | ❌ read when the unit runs; the unit restarts authelia when the file changes |
| `swarm/services/review-classifier/anthropic-api-key` | you, in the store's UI, field `api_key` | `swarm-review-classifier` | ❌ write a new key | ✅ read on every run | | `swarm/services/review-classifier/anthropic-api-key` | you, in the store's UI, field `api_key` | `swarm-review-classifier` | ❌ write a new key | ✅ read on every run |

View file

@ -266,8 +266,15 @@ Where the secret store runs on the same host, these default to the leaf it
mints. On any other host you place a leaf whose CN is mints. On any other host you place a leaf whose CN is
`deploy.bao.metricsAuthCommonName`, and evaluation refuses the config until `deploy.bao.metricsAuthCommonName`, and evaluation refuses the config until
you set both. `swarm-metrics-auth` reads the stored bearer at most once a you set both. `swarm-metrics-auth` reads the stored bearer at most once a
minute, so a replaced bearer starts working up to a minute later. After a minute, so a replaced bearer starts working up to a minute later. When the
failed or timed-out read it denies for three seconds without trying again. store doesn't answer within five seconds, it checks requests against the
last value it read instead of denying, and still refuses a wrong bearer;
it tries the store again after three seconds. A revoked or rotated bearer
stays accepted for as long as the store is down. The store's next answer
replaces the value, and a process that has read none yet (it started while
the store was down) denies. `swarm-logs-grafana-auth`
is the same binary and does the same.
`swarm-metrics-agent-auth` is the log store's `swarm-logs-auth` binary, with `swarm-metrics-agent-auth` is the log store's `swarm-logs-auth` binary, with
the same grant TTL and rate limit, under `deploy.victoriametrics.agentAuth*` the same grant TTL and rate limit, under `deploy.victoriametrics.agentAuth*`
(table above). (table above).

View file

@ -10,6 +10,8 @@ path = "src/main.rs"
[dependencies] [dependencies]
anyhow.workspace = true anyhow.workspace = true
axum.workspace = true axum.workspace = true
# The read time in the fallback log line.
chrono.workspace = true
clap.workspace = true clap.workspace = true
serde.workspace = true serde.workspace = true
# Comparing the presented bearer with the stored one. # Comparing the presented bearer with the stored one.
@ -25,4 +27,5 @@ workspace = true
# `test-util` for `#[tokio::test(start_paused = true)]`: the lookup-timeout # `test-util` for `#[tokio::test(start_paused = true)]`: the lookup-timeout
# test waits out `LOOKUP_TIMEOUT` on paused time. # test waits out `LOOKUP_TIMEOUT` on paused time.
[dev-dependencies] [dev-dependencies]
mockall.workspace = true
tokio = { workspace = true, features = ["test-util"] } tokio = { workspace = true, features = ["test-util"] }

View file

@ -3,8 +3,9 @@
//! nginx sends it every request to the store's query API that carries a //! nginx sends it every request to the store's query API that carries a
//! `Bearer` (`swarm-victoriametrics.nix`). It answers 200 when the bearer is //! `Bearer` (`swarm-victoriametrics.nix`). It answers 200 when the bearer is
//! the value stored at `--store-path`, which Grafana's host creates and its //! the value stored at `--store-path`, which Grafana's host creates and its
//! datasource presents, and 401 otherwise, store failures included: //! datasource presents, and 401 otherwise: `auth_request` turns any other
//! `auth_request` turns any other status into a 500 for the caller. //! status into a 500 for the caller. A store that does not answer is a 401
//! only until a value has been read; see below.
//! //!
//! It reads the store under its own certificate, whose policy reaches that one //! It reads the store under its own certificate, whose policy reaches that one
//! path. It serves a unix socket whose directory only the gateway's nginx may //! path. It serves a unix socket whose directory only the gateway's nginx may
@ -16,6 +17,13 @@
//! failed or timed-out one is remembered for `LOOKUP_FAILURE_WINDOW`, so a //! failed or timed-out one is remembered for `LOOKUP_FAILURE_WINDOW`, so a
//! burst of requests against an expired copy costs one lookup, success or //! burst of requests against an expired copy costs one lookup, success or
//! failure. //! failure.
//!
//! When the store does not answer, requests are checked against the last
//! value it did return, however old, and a wrong bearer is still refused. The
//! first such request of each failure streak logs it. A revoked or rotated
//! bearer therefore stays accepted for as long as the store cannot be reached;
//! the first answer replaces the value, or drops it when the store holds none.
//! A process that has not yet read the store has no value and denies.
use std::future::Future; use std::future::Future;
use std::os::unix::fs::PermissionsExt as _; use std::os::unix::fs::PermissionsExt as _;
@ -26,18 +34,19 @@ use std::time::{Duration, Instant};
use anyhow::Context as _; use anyhow::Context as _;
use axum::extract::State; use axum::extract::State;
use axum::http::{HeaderMap, StatusCode, header}; use axum::http::{HeaderMap, StatusCode, header};
use chrono::{DateTime, SecondsFormat, Utc};
use clap::Parser; use clap::Parser;
use subtle::ConstantTimeEq as _; use subtle::ConstantTimeEq as _;
use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings}; use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings};
/// How long one store lookup may take before the request is denied. Well /// How long one store lookup may take before the store counts as not
/// under nginx's 60s `proxy_read_timeout`, so a slow store reads as a 401 /// answering. Well under nginx's 60s `proxy_read_timeout`, so a slow store
/// rather than a gateway timeout. /// reads as a 401 or a fallback answer rather than a gateway timeout.
const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5); const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5);
/// How long requests are denied without a lookup after one failed or timed /// How long requests skip the lookup after one failed or timed out. Each
/// out. Each request queued behind a failing lookup would otherwise run its /// request queued behind a failing lookup would otherwise run its own, and
/// own, and the k-th would wait k × `LOOKUP_TIMEOUT`, past nginx's 60s read /// the k-th would wait k × `LOOKUP_TIMEOUT`, past nginx's 60s read
/// timeout. Short against `--cache-ttl`, so a recovered store is used again /// timeout. Short against `--cache-ttl`, so a recovered store is used again
/// within seconds. /// within seconds.
const LOOKUP_FAILURE_WINDOW: Duration = Duration::from_secs(3); const LOOKUP_FAILURE_WINDOW: Duration = Duration::from_secs(3);
@ -63,6 +72,7 @@ struct Args {
} }
/// Where the stored bearer comes from. /// Where the stored bearer comes from.
#[cfg_attr(test, mockall::automock)]
trait BearerSource { trait BearerSource {
/// The stored bearer, `None` when nothing is stored, or an error when the /// The stored bearer, `None` when nothing is stored, or an error when the
/// store could not answer. /// store could not answer.
@ -102,18 +112,42 @@ impl BearerSource for Store {
} }
} }
/// The last value the store returned.
struct Cached { struct Cached {
value: String, value: String,
/// When it was read, on the clock the TTL is measured on.
read_at: Instant, read_at: Instant,
/// The same moment, for the log.
read_at_wall: DateTime<Utc>,
} }
#[derive(Default)] #[derive(Default)]
struct Cache { struct Cache {
/// Kept past `--cache-ttl`: an expired value is replaced by the next
/// answer from the store, and checked against while there is none.
stored: Option<Cached>, stored: Option<Cached>,
/// Until when requests are denied without a lookup, after one failed or /// Until when requests skip the lookup, after one failed or timed out.
/// timed out. Stamped when that lookup ends: the requests queued behind /// Stamped when that lookup ends: the requests queued behind it carry the
/// it carry the time they arrived, which is earlier. /// time they arrived, which is earlier.
failed_until: Option<Instant>, failed_until: Option<Instant>,
/// Whether this failure streak has logged that it checks against `stored`.
fallback_logged: bool,
}
impl Cache {
/// The line to log when a request is checked against `stored` because the
/// store did not answer: `Some` for the first such request of a failure
/// streak, `None` for the rest.
fn fallback_notice(&mut self) -> Option<String> {
let read_at = self.stored.as_ref()?.read_at_wall;
if std::mem::replace(&mut self.fallback_logged, true) {
return None;
}
Some(format!(
"the store did not answer; checking against the bearer read at {}",
read_at.to_rfc3339_opts(SecondsFormat::Secs, true)
))
}
} }
struct Responder<S> { struct Responder<S> {
@ -122,6 +156,51 @@ struct Responder<S> {
cached: tokio::sync::Mutex<Cache>, cached: tokio::sync::Mutex<Cache>,
} }
/// What reading the store for an expired copy came to.
enum Refresh {
/// The store returned a bearer, now in the cache.
Replaced,
/// The store answered that it holds no usable bearer; the cache is empty.
Empty,
/// The store failed or timed out, now or a moment ago.
Unanswered,
}
/// Reads the store into `cache`, unless a lookup failed less than
/// `LOOKUP_FAILURE_WINDOW` ago.
async fn refresh(source: &impl BearerSource, cache: &mut Cache, now: Instant) -> Refresh {
if cache.failed_until.is_some_and(|until| now < until) {
return Refresh::Unanswered;
}
match tokio::time::timeout(LOOKUP_TIMEOUT, source.lookup()).await {
Ok(Ok(Some(value))) if !value.is_empty() => {
cache.stored = Some(Cached {
value,
read_at: now,
read_at_wall: Utc::now(),
});
cache.failed_until = None;
cache.fallback_logged = false;
Refresh::Replaced
}
Ok(Ok(_)) => {
tracing::warn!("no bearer is stored; denying");
*cache = Cache::default();
Refresh::Empty
}
Ok(Err(e)) => {
tracing::warn!(error = format!("{e:#}"), "bearer lookup failed");
cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW);
Refresh::Unanswered
}
Err(_) => {
tracing::warn!("bearer lookup timed out");
cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW);
Refresh::Unanswered
}
}
}
/// The answer for one subrequest's `Authorization` header, at `now`. /// The answer for one subrequest's `Authorization` header, at `now`.
async fn decide( async fn decide(
responder: &Responder<impl BearerSource>, responder: &Responder<impl BearerSource>,
@ -140,31 +219,17 @@ async fn decide(
.as_ref() .as_ref()
.is_some_and(|c| now.saturating_duration_since(c.read_at) < responder.ttl); .is_some_and(|c| now.saturating_duration_since(c.read_at) < responder.ttl);
if !fresh { if !fresh {
if cache.failed_until.is_some_and(|until| now < until) { match refresh(&responder.source, &mut cache, now).await {
tracing::warn!("bearer lookup failed a moment ago; denying without another"); Refresh::Replaced => {}
return StatusCode::UNAUTHORIZED; Refresh::Empty => return StatusCode::UNAUTHORIZED,
} Refresh::Unanswered => {
match tokio::time::timeout(LOOKUP_TIMEOUT, responder.source.lookup()).await { if cache.stored.is_none() {
Ok(Ok(Some(value))) if !value.is_empty() => { tracing::warn!("the store did not answer and no bearer was read yet; denying");
cache.stored = Some(Cached { return StatusCode::UNAUTHORIZED;
value, }
read_at: now, if let Some(line) = cache.fallback_notice() {
}); tracing::warn!("{line}");
cache.failed_until = None; }
}
Ok(Ok(_)) => {
tracing::warn!("no bearer is stored; denying");
return StatusCode::UNAUTHORIZED;
}
Ok(Err(e)) => {
tracing::warn!(error = format!("{e:#}"), "bearer lookup failed; denying");
cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW);
return StatusCode::UNAUTHORIZED;
}
Err(_) => {
tracing::warn!("bearer lookup timed out; denying");
cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW);
return StatusCode::UNAUTHORIZED;
} }
} }
} }
@ -236,85 +301,82 @@ async fn main() -> anyhow::Result<()> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use std::sync::Mutex; use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
/// A store answering `stored` (or failing when it is `Err`) after `delay`, type Lookup = Pin<Box<dyn Future<Output = anyhow::Result<Option<String>>> + Send>>;
/// counting lookups. type Answer = fn() -> Lookup;
struct Fake {
stored: Mutex<Result<Option<String>, ()>>, fn holds(value: &str) -> Lookup {
delay: Duration, let value = value.to_owned();
lookups: AtomicUsize, Box::pin(async move { Ok(Some(value)) })
} }
impl Fake { fn holds_nothing() -> Lookup {
fn holding(value: &str) -> Self { Box::pin(async { Ok(None) })
Self {
stored: Mutex::new(Ok(Some(value.to_owned()))),
delay: Duration::ZERO,
lookups: AtomicUsize::new(0),
}
}
fn hanging() -> Self {
Self {
delay: Duration::from_hours(1),
..Self::holding(SECRET)
}
}
fn set(&self, stored: Result<Option<&str>, ()>) {
*self.stored.lock().unwrap() = stored.map(|v| v.map(str::to_owned));
}
fn lookups(&self) -> usize {
self.lookups.load(Ordering::SeqCst)
}
} }
impl BearerSource for &Fake { fn fails() -> Lookup {
async fn lookup(&self) -> anyhow::Result<Option<String>> { Box::pin(async { Err(anyhow::anyhow!("store unreachable")) })
self.lookups.fetch_add(1, Ordering::SeqCst);
tokio::time::sleep(self.delay).await;
self.stored
.lock()
.unwrap()
.clone()
.map_err(|()| anyhow::anyhow!("store unreachable"))
}
} }
impl BearerSource for Arc<Fake> { /// Accepts the lookup and never answers.
async fn lookup(&self) -> anyhow::Result<Option<String>> { fn hangs() -> Lookup {
(&**self).lookup().await Box::pin(std::future::pending())
}
/// A store that answers each lookup in turn with the next of `answers`,
/// and fails the test on one more.
fn store(answers: &[Answer]) -> MockBearerSource {
let mut store = MockBearerSource::new();
let mut order = mockall::Sequence::new();
for &answer in answers {
store
.expect_lookup()
.times(1)
.in_sequence(&mut order)
.returning(answer);
} }
store
}
fn secret() -> Lookup {
holds(SECRET)
}
fn replacement() -> Lookup {
holds("replaced")
} }
const TTL: Duration = Duration::from_mins(1); const TTL: Duration = Duration::from_mins(1);
const SECRET: &str = "0f1e2d3c4b5a"; const SECRET: &str = "0f1e2d3c4b5a";
const GOOD: &str = "Bearer 0f1e2d3c4b5a"; const GOOD: &str = "Bearer 0f1e2d3c4b5a";
const REPLACED: &str = "Bearer replaced";
fn responder(store: &Fake) -> Responder<&Fake> { fn responder(source: MockBearerSource) -> Responder<MockBearerSource> {
Responder { Responder {
source: store, source,
ttl: TTL, ttl: TTL,
cached: tokio::sync::Mutex::new(Cache::default()), cached: tokio::sync::Mutex::new(Cache::default()),
} }
} }
/// A request time far enough back that a value read then has expired, and
/// that is still before the failure window a lookup now opens.
fn long_ago() -> Instant {
Instant::now()
.checked_sub(TTL * 2)
.expect("the monotonic clock is older than two minutes")
}
#[tokio::test] #[tokio::test]
async fn the_stored_bearer_is_200() { async fn the_stored_bearer_is_200() {
let store = Fake::holding(SECRET); let r = responder(store(&[secret]));
assert_eq!( assert_eq!(decide(&r, Some(GOOD), Instant::now()).await, StatusCode::OK);
decide(&responder(&store), Some(GOOD), Instant::now()).await,
StatusCode::OK
);
} }
#[tokio::test] #[tokio::test]
async fn everything_else_is_401() { async fn everything_else_is_401() {
let store = Fake::holding(SECRET); let r = responder(store(&[secret]));
let r = responder(&store);
for authorization in [ for authorization in [
None, None,
Some(SECRET), Some(SECRET),
@ -333,52 +395,41 @@ mod tests {
} }
/// Nothing stored, an empty value, and a store that cannot answer are /// Nothing stored, an empty value, and a store that cannot answer are
/// all denials, whatever the bearer. /// all denials, whatever the bearer, when no value was ever read.
#[tokio::test] #[tokio::test]
async fn no_usable_stored_value_is_401() { async fn no_usable_stored_value_is_401() {
let store = Fake::holding(SECRET); for answer in [holds_nothing, || holds(""), fails] {
let r = responder(&store); let mut source = MockBearerSource::new();
for stored in [Ok(None), Ok(Some("")), Err(())] { source.expect_lookup().returning(answer);
store.set(stored); let r = responder(source);
assert_eq!( for authorization in [GOOD, "Bearer "] {
decide(&r, Some(GOOD), Instant::now()).await, assert_eq!(
StatusCode::UNAUTHORIZED, decide(&r, Some(authorization), Instant::now()).await,
"{stored:?}" StatusCode::UNAUTHORIZED
); );
assert_eq!( }
decide(&r, Some("Bearer "), Instant::now()).await,
StatusCode::UNAUTHORIZED,
"{stored:?}"
);
} }
} }
/// Within the TTL no request reaches the store, a wrong bearer included, /// Within the TTL no request reaches the store, a wrong bearer included,
/// and a replacement the store already holds goes unseen; after it the /// and a replacement the store already holds goes unseen; after it the
/// store is read again. /// store is read again, and its answer replaces the old value at once.
#[tokio::test] #[tokio::test]
async fn the_stored_value_is_reused_until_the_ttl() { async fn the_stored_value_is_reused_until_the_ttl() {
let store = Fake::holding(SECRET); let r = responder(store(&[secret, replacement]));
let r = responder(&store);
let t0 = Instant::now(); let t0 = Instant::now();
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK);
store.set(Ok(Some("replaced")));
let within = t0 + Duration::from_secs(59); let within = t0 + Duration::from_secs(59);
assert_eq!(decide(&r, Some(GOOD), within).await, StatusCode::OK); assert_eq!(decide(&r, Some(GOOD), within).await, StatusCode::OK);
assert_eq!( assert_eq!(
decide(&r, Some("Bearer replaced"), within).await, decide(&r, Some(REPLACED), within).await,
StatusCode::UNAUTHORIZED StatusCode::UNAUTHORIZED
); );
assert_eq!(store.lookups(), 1);
assert_eq!( assert_eq!(
decide(&r, Some(GOOD), t0 + TTL).await, decide(&r, Some(GOOD), t0 + TTL).await,
StatusCode::UNAUTHORIZED StatusCode::UNAUTHORIZED
); );
assert_eq!( assert_eq!(decide(&r, Some(REPLACED), t0 + TTL).await, StatusCode::OK);
decide(&r, Some("Bearer replaced"), t0 + TTL).await,
StatusCode::OK
);
assert_eq!(store.lookups(), 2);
} }
/// A failed read is remembered for `LOOKUP_FAILURE_WINDOW`, not for a TTL: /// A failed read is remembered for `LOOKUP_FAILURE_WINDOW`, not for a TTL:
@ -386,46 +437,163 @@ mod tests {
/// reads the store again. /// reads the store again.
#[tokio::test] #[tokio::test]
async fn a_failed_lookup_is_retried_after_the_failure_window() { async fn a_failed_lookup_is_retried_after_the_failure_window() {
let store = Fake::holding(SECRET); let r = responder(store(&[fails, secret]));
store.set(Err(()));
let r = responder(&store);
let t0 = Instant::now(); let t0 = Instant::now();
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED);
store.set(Ok(Some(SECRET)));
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED);
assert_eq!(store.lookups(), 1);
let after = t0 + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1); let after = t0 + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1);
assert_eq!(decide(&r, Some(GOOD), after).await, StatusCode::OK); assert_eq!(decide(&r, Some(GOOD), after).await, StatusCode::OK);
assert_eq!(store.lookups(), 2); }
/// With a value read before, a store that errors or sleeps past
/// `LOOKUP_TIMEOUT` leaves the requests checked against that value: the
/// matching bearer is accepted, a wrong one refused, and the store is not
/// asked again inside the failure window.
#[tokio::test(start_paused = true)]
async fn an_unanswering_store_leaves_the_last_good_bearer_in_force() {
for (name, outage) in [("erroring", fails as Answer), ("hanging", hangs)] {
let r = responder(store(&[secret, outage]));
assert_eq!(
decide(&r, Some(GOOD), long_ago()).await,
StatusCode::OK,
"{name}"
);
for _ in 0..3 {
assert_eq!(
decide(&r, Some(GOOD), Instant::now()).await,
StatusCode::OK,
"{name}"
);
assert_eq!(
decide(&r, Some("Bearer 0f1e2d3c4b5b"), Instant::now()).await,
StatusCode::UNAUTHORIZED,
"{name}"
);
}
}
}
/// With no value ever read, a store that sleeps past `LOOKUP_TIMEOUT`
/// denies the bearer the last test accepts.
#[tokio::test(start_paused = true)]
async fn a_timeout_without_a_last_good_bearer_is_401() {
let r = responder(store(&[hangs]));
for _ in 0..3 {
assert_eq!(
decide(&r, Some(GOOD), Instant::now()).await,
StatusCode::UNAUTHORIZED
);
}
}
/// The store's next answer replaces the value kept through the outage.
#[tokio::test(start_paused = true)]
async fn the_first_answer_after_an_outage_replaces_the_value() {
let r = responder(store(&[secret, hangs, replacement]));
assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK);
assert_eq!(decide(&r, Some(GOOD), Instant::now()).await, StatusCode::OK);
let recovered = Instant::now() + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1);
assert_eq!(
decide(&r, Some(GOOD), recovered).await,
StatusCode::UNAUTHORIZED
);
assert_eq!(decide(&r, Some(REPLACED), recovered).await, StatusCode::OK);
}
/// A store that answers it holds no bearer drops the value: a later
/// outage does not bring it back.
#[tokio::test]
async fn an_empty_answer_drops_the_last_good_bearer() {
let r = responder(store(&[secret, holds_nothing, fails]));
assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK);
assert_eq!(
decide(&r, Some(GOOD), Instant::now()).await,
StatusCode::UNAUTHORIZED
);
assert_eq!(
decide(&r, Some(GOOD), Instant::now()).await,
StatusCode::UNAUTHORIZED
);
} }
/// A burst against a store that errors, or that sleeps past /// A burst against a store that errors, or that sleeps past
/// `LOOKUP_TIMEOUT`, costs one lookup, and every request is denied /// `LOOKUP_TIMEOUT`, costs one lookup, and every request is answered
/// within that one timeout rather than queueing behind one timeout each. /// within that one timeout rather than queueing behind one timeout each:
/// denied with no value read before, checked against it with one.
#[tokio::test(start_paused = true)] #[tokio::test(start_paused = true)]
async fn a_burst_against_a_failing_store_costs_one_lookup() { async fn a_burst_against_an_unanswering_store_costs_one_lookup() {
const BURST: usize = 20; const BURST: usize = 20;
let erroring = Fake::holding(SECRET); for (name, outage) in [("erroring", fails as Answer), ("hanging", hangs)] {
erroring.set(Err(())); for (value_read, expected) in
for (name, store) in [("erroring", erroring), ("hanging", Fake::hanging())] { [(false, StatusCode::UNAUTHORIZED), (true, StatusCode::OK)]
let store = Arc::new(store); {
let r = Arc::new(Responder { let answers: Vec<Answer> = if value_read {
source: Arc::clone(&store), vec![secret, outage]
ttl: TTL, } else {
cached: tokio::sync::Mutex::new(Cache::default()), vec![outage]
}); };
let started = tokio::time::Instant::now(); let r = Arc::new(responder(store(&answers)));
let now = Instant::now(); if value_read {
let mut burst = tokio::task::JoinSet::new(); assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK);
for _ in 0..BURST { }
let r = Arc::clone(&r); let started = tokio::time::Instant::now();
burst.spawn(async move { decide(&r, Some(GOOD), now).await }); let now = Instant::now();
let mut burst = tokio::task::JoinSet::new();
for _ in 0..BURST {
let r = Arc::clone(&r);
burst.spawn(async move { decide(&r, Some(GOOD), now).await });
}
while let Some(status) = burst.join_next().await {
assert_eq!(
status.unwrap(),
expected,
"{name}, value read: {value_read}"
);
}
assert!(started.elapsed() <= LOOKUP_TIMEOUT, "{name}");
} }
while let Some(status) = burst.join_next().await {
assert_eq!(status.unwrap(), StatusCode::UNAUTHORIZED, "{name}");
}
assert_eq!(store.lookups(), 1, "{name}");
assert!(started.elapsed() <= LOOKUP_TIMEOUT, "{name}");
} }
} }
/// Each failure streak yields the fallback line once, naming when the
/// value was read; a streak that ends in an answer, and a cache holding no
/// value, yield none.
#[test]
fn the_fallback_is_noticed_once_per_failure_streak() {
let mut cache = Cache::default();
assert_eq!(cache.fallback_notice(), None);
cache.stored = Some(Cached {
value: SECRET.to_owned(),
read_at: Instant::now(),
read_at_wall: "2026-11-04T07:12:11Z".parse().unwrap(),
});
assert_eq!(
cache.fallback_notice().as_deref(),
Some(
"the store did not answer; checking against the bearer read at \
2026-11-04T07:12:11Z"
)
);
for _ in 0..3 {
assert_eq!(cache.fallback_notice(), None);
}
cache.fallback_logged = false;
assert!(cache.fallback_notice().is_some());
}
/// `decide` marks the streak as noticed on the first request checked
/// against the kept value, and an answer from the store ends the streak.
#[tokio::test(start_paused = true)]
async fn an_answer_from_the_store_ends_the_failure_streak() {
let r = responder(store(&[secret, hangs, replacement]));
assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK);
assert!(!r.cached.lock().await.fallback_logged);
for _ in 0..3 {
decide(&r, Some(GOOD), Instant::now()).await;
assert!(r.cached.lock().await.fallback_logged);
}
let recovered = Instant::now() + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1);
assert_eq!(decide(&r, Some(REPLACED), recovered).await, StatusCode::OK);
assert!(!r.cached.lock().await.fallback_logged);
}
} }