diff --git a/Cargo.lock b/Cargo.lock index f64586e0..a32c8981 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5109,7 +5109,9 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", + "chrono", "clap", + "mockall", "serde", "subtle", "swarm-secret-client", diff --git a/docs/swarm/credentials-reference.md b/docs/swarm/credentials-reference.md index a2a12ee9..37ab0b8c 100644 --- a/docs/swarm/credentials-reference.md +++ b/docs/swarm/credentials-reference.md @@ -43,7 +43,7 @@ the rest of the cell says how. | `swarm/hives//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//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//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//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//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/`, 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 | diff --git a/docs/swarm/services.md b/docs/swarm/services.md index 76524c38..1a1e646d 100644 --- a/docs/swarm/services.md +++ b/docs/swarm/services.md @@ -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 `deploy.bao.metricsAuthCommonName`, and evaluation refuses the config until 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 -failed or timed-out read it denies for three seconds without trying again. +minute, so a replaced bearer starts working up to a minute later. When the +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 the same grant TTL and rate limit, under `deploy.victoriametrics.agentAuth*` (table above). diff --git a/swarm-metrics-auth/Cargo.toml b/swarm-metrics-auth/Cargo.toml index 29476439..1e5a5264 100644 --- a/swarm-metrics-auth/Cargo.toml +++ b/swarm-metrics-auth/Cargo.toml @@ -10,6 +10,8 @@ path = "src/main.rs" [dependencies] anyhow.workspace = true axum.workspace = true +# The read time in the fallback log line. +chrono.workspace = true clap.workspace = true serde.workspace = true # 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 waits out `LOOKUP_TIMEOUT` on paused time. [dev-dependencies] +mockall.workspace = true tokio = { workspace = true, features = ["test-util"] } diff --git a/swarm-metrics-auth/src/main.rs b/swarm-metrics-auth/src/main.rs index cb7bfc77..d03863b4 100644 --- a/swarm-metrics-auth/src/main.rs +++ b/swarm-metrics-auth/src/main.rs @@ -3,8 +3,9 @@ //! 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 //! the value stored at `--store-path`, which Grafana's host creates and its -//! datasource presents, and 401 otherwise, store failures included: -//! `auth_request` turns any other status into a 500 for the caller. +//! datasource presents, and 401 otherwise: `auth_request` turns any other +//! 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 //! 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 //! burst of requests against an expired copy costs one lookup, success or //! 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::os::unix::fs::PermissionsExt as _; @@ -26,18 +34,19 @@ use std::time::{Duration, Instant}; use anyhow::Context as _; use axum::extract::State; use axum::http::{HeaderMap, StatusCode, header}; +use chrono::{DateTime, SecondsFormat, Utc}; use clap::Parser; use subtle::ConstantTimeEq as _; use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings}; -/// How long one store lookup may take before the request is denied. Well -/// under nginx's 60s `proxy_read_timeout`, so a slow store reads as a 401 -/// rather than a gateway timeout. +/// How long one store lookup may take before the store counts as not +/// answering. Well under nginx's 60s `proxy_read_timeout`, so a slow store +/// reads as a 401 or a fallback answer rather than a gateway timeout. const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5); -/// How long requests are denied without a lookup after one failed or timed -/// out. Each request queued behind a failing lookup would otherwise run its -/// own, and the k-th would wait k × `LOOKUP_TIMEOUT`, past nginx's 60s read +/// How long requests skip the lookup after one failed or timed out. Each +/// request queued behind a failing lookup would otherwise run its own, and +/// 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 /// within seconds. const LOOKUP_FAILURE_WINDOW: Duration = Duration::from_secs(3); @@ -63,6 +72,7 @@ struct Args { } /// Where the stored bearer comes from. +#[cfg_attr(test, mockall::automock)] trait BearerSource { /// The stored bearer, `None` when nothing is stored, or an error when the /// store could not answer. @@ -102,18 +112,42 @@ impl BearerSource for Store { } } +/// The last value the store returned. struct Cached { value: String, + /// When it was read, on the clock the TTL is measured on. read_at: Instant, + /// The same moment, for the log. + read_at_wall: DateTime, } #[derive(Default)] 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, - /// Until when requests are denied without a lookup, after one failed or - /// timed out. Stamped when that lookup ends: the requests queued behind - /// it carry the time they arrived, which is earlier. + /// Until when requests skip the lookup, after one failed or timed out. + /// Stamped when that lookup ends: the requests queued behind it carry the + /// time they arrived, which is earlier. failed_until: Option, + /// 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 { + 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 { @@ -122,6 +156,51 @@ struct Responder { cached: tokio::sync::Mutex, } +/// 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`. async fn decide( responder: &Responder, @@ -140,31 +219,17 @@ async fn decide( .as_ref() .is_some_and(|c| now.saturating_duration_since(c.read_at) < responder.ttl); if !fresh { - if cache.failed_until.is_some_and(|until| now < until) { - tracing::warn!("bearer lookup failed a moment ago; denying without another"); - return StatusCode::UNAUTHORIZED; - } - match tokio::time::timeout(LOOKUP_TIMEOUT, responder.source.lookup()).await { - Ok(Ok(Some(value))) if !value.is_empty() => { - cache.stored = Some(Cached { - value, - read_at: now, - }); - 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; + match refresh(&responder.source, &mut cache, now).await { + Refresh::Replaced => {} + Refresh::Empty => return StatusCode::UNAUTHORIZED, + Refresh::Unanswered => { + if cache.stored.is_none() { + tracing::warn!("the store did not answer and no bearer was read yet; denying"); + return StatusCode::UNAUTHORIZED; + } + if let Some(line) = cache.fallback_notice() { + tracing::warn!("{line}"); + } } } } @@ -236,85 +301,82 @@ async fn main() -> anyhow::Result<()> { #[cfg(test)] mod tests { use super::*; - use std::sync::Mutex; - use std::sync::atomic::{AtomicUsize, Ordering}; + use std::pin::Pin; - /// A store answering `stored` (or failing when it is `Err`) after `delay`, - /// counting lookups. - struct Fake { - stored: Mutex, ()>>, - delay: Duration, - lookups: AtomicUsize, + type Lookup = Pin>> + Send>>; + type Answer = fn() -> Lookup; + + fn holds(value: &str) -> Lookup { + let value = value.to_owned(); + Box::pin(async move { Ok(Some(value)) }) } - impl Fake { - fn holding(value: &str) -> Self { - 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, ()>) { - *self.stored.lock().unwrap() = stored.map(|v| v.map(str::to_owned)); - } - - fn lookups(&self) -> usize { - self.lookups.load(Ordering::SeqCst) - } + fn holds_nothing() -> Lookup { + Box::pin(async { Ok(None) }) } - impl BearerSource for &Fake { - async fn lookup(&self) -> anyhow::Result> { - self.lookups.fetch_add(1, Ordering::SeqCst); - tokio::time::sleep(self.delay).await; - self.stored - .lock() - .unwrap() - .clone() - .map_err(|()| anyhow::anyhow!("store unreachable")) - } + fn fails() -> Lookup { + Box::pin(async { Err(anyhow::anyhow!("store unreachable")) }) } - impl BearerSource for Arc { - async fn lookup(&self) -> anyhow::Result> { - (&**self).lookup().await + /// Accepts the lookup and never answers. + fn hangs() -> Lookup { + 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 SECRET: &str = "0f1e2d3c4b5a"; const GOOD: &str = "Bearer 0f1e2d3c4b5a"; + const REPLACED: &str = "Bearer replaced"; - fn responder(store: &Fake) -> Responder<&Fake> { + fn responder(source: MockBearerSource) -> Responder { Responder { - source: store, + source, ttl: TTL, 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] async fn the_stored_bearer_is_200() { - let store = Fake::holding(SECRET); - assert_eq!( - decide(&responder(&store), Some(GOOD), Instant::now()).await, - StatusCode::OK - ); + let r = responder(store(&[secret])); + assert_eq!(decide(&r, Some(GOOD), Instant::now()).await, StatusCode::OK); } #[tokio::test] async fn everything_else_is_401() { - let store = Fake::holding(SECRET); - let r = responder(&store); + let r = responder(store(&[secret])); for authorization in [ None, Some(SECRET), @@ -333,52 +395,41 @@ mod tests { } /// 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] async fn no_usable_stored_value_is_401() { - let store = Fake::holding(SECRET); - let r = responder(&store); - for stored in [Ok(None), Ok(Some("")), Err(())] { - store.set(stored); - assert_eq!( - decide(&r, Some(GOOD), Instant::now()).await, - StatusCode::UNAUTHORIZED, - "{stored:?}" - ); - assert_eq!( - decide(&r, Some("Bearer "), Instant::now()).await, - StatusCode::UNAUTHORIZED, - "{stored:?}" - ); + for answer in [holds_nothing, || holds(""), fails] { + let mut source = MockBearerSource::new(); + source.expect_lookup().returning(answer); + let r = responder(source); + for authorization in [GOOD, "Bearer "] { + assert_eq!( + decide(&r, Some(authorization), Instant::now()).await, + StatusCode::UNAUTHORIZED + ); + } } } /// Within the TTL no request reaches the store, a wrong bearer included, /// 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] async fn the_stored_value_is_reused_until_the_ttl() { - let store = Fake::holding(SECRET); - let r = responder(&store); + let r = responder(store(&[secret, replacement])); let t0 = Instant::now(); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK); - store.set(Ok(Some("replaced"))); let within = t0 + Duration::from_secs(59); assert_eq!(decide(&r, Some(GOOD), within).await, StatusCode::OK); assert_eq!( - decide(&r, Some("Bearer replaced"), within).await, + decide(&r, Some(REPLACED), within).await, StatusCode::UNAUTHORIZED ); - assert_eq!(store.lookups(), 1); assert_eq!( decide(&r, Some(GOOD), t0 + TTL).await, StatusCode::UNAUTHORIZED ); - assert_eq!( - decide(&r, Some("Bearer replaced"), t0 + TTL).await, - StatusCode::OK - ); - assert_eq!(store.lookups(), 2); + assert_eq!(decide(&r, Some(REPLACED), t0 + TTL).await, StatusCode::OK); } /// A failed read is remembered for `LOOKUP_FAILURE_WINDOW`, not for a TTL: @@ -386,46 +437,163 @@ mod tests { /// reads the store again. #[tokio::test] async fn a_failed_lookup_is_retried_after_the_failure_window() { - let store = Fake::holding(SECRET); - store.set(Err(())); - let r = responder(&store); + let r = responder(store(&[fails, secret])); let t0 = Instant::now(); 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!(store.lookups(), 1); let after = t0 + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1); 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 - /// `LOOKUP_TIMEOUT`, costs one lookup, and every request is denied - /// within that one timeout rather than queueing behind one timeout each. + /// `LOOKUP_TIMEOUT`, costs one lookup, and every request is answered + /// 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)] - 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; - let erroring = Fake::holding(SECRET); - erroring.set(Err(())); - for (name, store) in [("erroring", erroring), ("hanging", Fake::hanging())] { - let store = Arc::new(store); - let r = Arc::new(Responder { - source: Arc::clone(&store), - ttl: TTL, - cached: tokio::sync::Mutex::new(Cache::default()), - }); - let started = tokio::time::Instant::now(); - 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 }); + for (name, outage) in [("erroring", fails as Answer), ("hanging", hangs)] { + for (value_read, expected) in + [(false, StatusCode::UNAUTHORIZED), (true, StatusCode::OK)] + { + let answers: Vec = if value_read { + vec![secret, outage] + } else { + vec![outage] + }; + let r = Arc::new(responder(store(&answers))); + if value_read { + assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK); + } + let started = tokio::time::Instant::now(); + 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); + } }