Coordinate Craft App release polling across the fleet through Redis (#77)

New releases took up to ~30 min to reach \`GET /v1/craft_apps/release_info\` (ArtCraft Launcher 0.2.0 needed 17–29 min), pods disagreed with each other, and every pod polled GitHub on its own, which uses up the unauthenticated quota (60/h per IP, and 304s count) and risks backing off for up to 4 h. The pods now share their results through Redis. There's no lock: - **One Redis hash per app** (\`craftApps:releases:app:<key>\`, 7 d TTL). It holds the latest release, the ETag and \`fetched_ms\`. Before asking GitHub about an app, a pod reads its entry. If it's fresher than \`interval − 30 s\` (a clock-skew margin), the pod takes it from Redis and doesn't call GitHub. - **Last write wins, guarded by timestamp.** A Lua script never lets an entry go back in time. A 304 moves only \`fetched_ms\`, and only while the stored ETag still matches. - **A soft per-app claim** (60 s, no owner, it just lapses), so a fleet that ticks every 30 s rarely fetches the same app twice. - **Backoff shared by the whole fleet** (\`craftApps:releases:backoff\`), so a rate limit slows everyone down. Per-app failure retry is 60 s, doubling. - **The endpoint is unchanged.** It still serves each pod's in-memory copy. If Redis is down, a pod falls back to polling on its own and warns at most every 10 minutes. **The GitHub token is optional.** With \`CRAFT_APPS_RELEASE_GITHUB_TOKEN\` set, the default interval is 120 s (min 60). Without it (unset or blank), calls go out unauthenticated with no Authorization header, at 1800 s (min 300), and startup and polling are never blocked. \`CRAFT_APPS_RELEASE_COORDINATION_ENABLED=false\` switches back to the old per-pod polling. **Tests:** \`cargo test -p storyteller-web --bin storyteller-web craft\` (49 pass) and \`cargo test -p redis_schema craft\` (4 pass). The Redis tests passed against a local redis-server with \`CRAFT_APPS_TEST_REDIS_URL=… -- --include-ignored --skip live_round\` (51 pass). **Follow-ups after review:** - **The poller has its own Redis pool** (2 connections; connects within 3 s; every read and write, including the pool's `PING` check, times out after 5 s). Before this, it shared the handlers' pool, which has no read timeout, so a Redis that stopped answering could leave the poller holding every connection while handlers waited 30 s for one. A test against a fake Redis that goes silent after the handshake fails without the timeouts and passes with them. - **When Redis won't run scripts**, writes fall back to `WATCH`/`MULTI`/`EXEC`: read and decode the guarding fields, decide, and write in a transaction that Redis drops if another pod wrote in between. This switch only happens when Redis refuses the script itself. Being unreachable, busy, out of memory or read-only doesn't trigger it. It was tested against a real Redis user denied `@scripting`. - **New pods load from Redis at startup**, before the startup jitter, so they serve the fleet's releases right away without asking GitHub. - **`Cache-Control: no-store` until every app has been fetched** (`last_fetched` set), so Cloudflare never caches a cold pod's "not fetched yet" answer. The response body is unchanged. **Tests:** `cargo test -p storyteller-web --bin storyteller-web craft -- --include-ignored --skip live_round` against a local Redis 8.6: 63 pass. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 01:05:06 -04:00
1 parent 25f38b79b8
commit 854869ba53
@@ -0,0 +1,41 @@
+//! The latest release of one Craft App, shared by every storyteller-web instance so the fleet
+//! polls GitHub once per app instead of once per pod.
+//!
+//! A Redis hash, written by `poll_craft_app_releases_thread` (storyteller-web), which documents
+//! the fields. Expires if nobody refreshes it for a week, which only happens if polling stops.
+
+use chrono::Duration;
+
+pub struct CraftAppReleaseRedisKey(pub String);
+
+impl_string_key!(CraftAppReleaseRedisKey);
+
+const REDIS_KEY_TTL_DURATION: Duration = Duration::milliseconds(1000 * 60 * 60 * 24 * 7);
+
+impl CraftAppReleaseRedisKey {
+  /// `app_key` is the app's lowercase machine name (`photocraft`, `artcraft-launcher`).
+  pub fn new_for_app(app_key: &str) -> Self {
+    let key = format!("craftApps:releases:app:{}", app_key);
+    Self(key)
+  }
+
+  pub fn get_redis_ttl() -> Duration {
+    REDIS_KEY_TTL_DURATION
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn test_new_for_app() {
+    let key = CraftAppReleaseRedisKey::new_for_app("photocraft");
+    assert_eq!(key.as_str(), "craftApps:releases:app:photocraft");
+  }
+
+  #[test]
+  fn test_duration() {
+    assert_eq!(CraftAppReleaseRedisKey::get_redis_ttl().num_days(), 7);
+  }
+}
@@ -0,0 +1,39 @@
+//! The fleet-wide GitHub backoff for Craft App release polling: after GitHub rate-limits us (or
+//! every request of a round fails), no storyteller-web instance asks GitHub again until it ends.
+//!
+//! A Redis hash, written by `poll_craft_app_releases_thread` (storyteller-web), which documents
+//! the fields. It outlives the backoff itself so repeated trouble keeps growing the wait; it is
+//! deleted after a good round.
+
+use chrono::Duration;
+
+pub struct CraftAppsReleaseBackoffRedisKey(pub String);
+
+impl_string_key!(CraftAppsReleaseBackoffRedisKey);
+
+const REDIS_KEY_TTL_DURATION: Duration = Duration::milliseconds(1000 * 60 * 60 * 24);
+
+impl CraftAppsReleaseBackoffRedisKey {
+  pub fn new_for_fleet() -> Self {
+    Self("craftApps:releases:backoff".to_string())
+  }
+
+  pub fn get_redis_ttl() -> Duration {
+    REDIS_KEY_TTL_DURATION
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn test_new_for_fleet() {
+    assert_eq!(CraftAppsReleaseBackoffRedisKey::new_for_fleet().as_str(), "craftApps:releases:backoff");
+  }
+
+  #[test]
+  fn test_duration() {
+    assert_eq!(CraftAppsReleaseBackoffRedisKey::get_redis_ttl().num_hours(), 24);
+  }
+}
@@ -0,0 +1,2 @@
+pub mod craft_app_release_redis_key;
+pub mod craft_apps_release_backoff_redis_key;
@@ -1,2 +1,3 @@
+pub mod craft_apps;
 pub mod premium;
-pub mod inference_job;
+pub mod inference_job;
@@ -3,12 +3,12 @@ use std::sync::Arc;
 use actix_web::http::header::{CacheControl, CacheDirective};
 use actix_web::{web, HttpResponse};

-use artcraft_api_defs::craft_apps::release_info::{CraftAppsImportantNotices, GetCraftAppsReleaseInfoResponse};
+use artcraft_api_defs::craft_apps::release_info::{CraftAppsImportantNotices, CraftAppsReleaseInfo, GetCraftAppsReleaseInfoResponse};

 use crate::state::server_state::ServerState;

-/// Shared caches (Cloudflare) may serve the same answer for this long: the data changes at most
-/// once per polling round (30 minutes by default).
+/// Shared caches (Cloudflare) may serve the same answer for this long once every app has been
+/// fetched: the data changes at most once per polling interval.
 const CACHE_MAX_AGE_SECS: u32 = 60;

 /// The latest release of every Craft App (and ArtCraft Launcher), as storyteller-web last fetched
@@ -25,15 +25,49 @@ const CACHE_MAX_AGE_SECS: u32 = 60;
 pub async fn get_craft_apps_release_info_handler(
  server_state: web::Data<Arc<ServerState>>,
 ) -> HttpResponse {
+  let release_info = server_state.craft_apps_release_cache.snapshot();
+  let cache_control = cache_control_for(&release_info);
  let response = GetCraftAppsReleaseInfoResponse {
    success: true,
-    release_info: server_state.craft_apps_release_cache.snapshot(),
+    release_info,
    // None yet. When there are, they're listed here (newest first).
    app_renames: Vec::new(),
    repo_renames: Vec::new(),
    important_notices: CraftAppsImportantNotices::default(),
  };
  HttpResponse::Ok()
-      .insert_header(CacheControl(vec![CacheDirective::Public, CacheDirective::MaxAge(CACHE_MAX_AGE_SECS)]))
+      .insert_header(cache_control)
      .json(response)
 }
+
+/// Cacheable only once every app has been fetched (`last_fetched` is set). Until then (an
+/// instance that just started, or one whose first fetch failed) the answer is `no-store`, so a
+/// shared cache never keeps handing out "not fetched yet" after this instance has warmed up.
+fn cache_control_for(release_info: &CraftAppsReleaseInfo) -> CacheControl {
+  if release_info.last_fetched.is_some() {
+    CacheControl(vec![CacheDirective::Public, CacheDirective::MaxAge(CACHE_MAX_AGE_SECS)])
+  } else {
+    CacheControl(vec![CacheDirective::NoStore])
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use std::collections::BTreeMap;
+
+  use chrono::Utc;
+
+  use super::*;
+
+  #[test]
+  fn a_fully_fetched_answer_is_cacheable() {
+    let release_info = CraftAppsReleaseInfo { apps: BTreeMap::new(), last_fetched: Some(Utc::now()) };
+    assert_eq!(cache_control_for(&release_info).to_string(), "public, max-age=60");
+  }
+
+  #[test]
+  fn an_answer_with_unfetched_apps_is_never_stored() {
+    let release_info = CraftAppsReleaseInfo { apps: BTreeMap::new(), last_fetched: None };
+    assert_eq!(cache_control_for(&release_info).to_string(), "no-store");
+  }
+}
@@ -37,6 +37,7 @@ use errors::AnyhowResult;
 use log::{info, warn};
 use server_environment::ServerEnvironment;
 use shared_env_var_config::logging::DEFAULT_RUST_LOG;
+use shared_env_var_config::redis::env_get_redis_0_connection_string_or_default;
 use tokio::runtime::Runtime;
 use user_traits_component::traits::internal_session_cache_purge::InternalSessionCachePurge;

@@ -58,6 +59,7 @@ use crate::threads::db_health_checker_thread::db_health_checker_thread::db_healt
 use crate::threads::expire_login_challenges::expire_login_challenges;
 use crate::threads::poll_craft_app_releases_thread::config::CraftAppsPollConfig;
 use crate::threads::poll_craft_app_releases_thread::poll_craft_app_releases::poll_craft_app_releases;
+use crate::threads::poll_craft_app_releases_thread::shared_store::build_poller_redis_pool;
 use crate::threads::poll_ip_banlist_thread::poll_ip_bans;
 use crate::threads::poll_model_token_info_thread::poll_model_token_info_thread;

@@ -156,9 +158,12 @@ async fn main() -> AnyhowResult<()> {

  let craft_apps_release_cache_clone = server_state.craft_apps_release_cache.clone();
  let craft_apps_poll_config = CraftAppsPollConfig::from_env();
+  // Its own small pool with I/O timeouts, so a hung Redis can't tie up the handlers' pool.
+  let craft_apps_redis_pool = build_poller_redis_pool(&env_get_redis_0_connection_string_or_default())?;
+  let hostname_clone = server_state.hostname.clone();

  tokio_runtime.spawn(async move {
-    poll_craft_app_releases(craft_apps_release_cache_clone, craft_apps_poll_config).await;
+    poll_craft_app_releases(craft_apps_release_cache_clone, craft_apps_poll_config, craft_apps_redis_pool, hostname_clone).await;
  });

  // ==================== Metrics worker ==================== //
@@ -1,5 +1,7 @@
 //! The latest release of every Craft App, as last fetched from GitHub by
-//! `poll_craft_app_releases_thread`, served by `GET /v1/craft_apps/release_info`.
+//! `poll_craft_app_releases_thread` (directly, or from the fleet's shared copy in Redis), served
+//! by `GET /v1/craft_apps/release_info`. The endpoint always answers from this memory, so trouble
+//! with Redis or GitHub never breaks it.
 //!
 //! ## App keys
 //!
@@ -49,6 +51,14 @@ pub struct CachedCraftApp {
  /// GitHub's ETag for the last successful answer, sent back as `If-None-Match` so an unchanged
  /// release costs a `304`.
  pub etag: Option<String>,
+
+  /// When a request for this app last failed. Only meaningful when newer than
+  /// `app_last_fetched`; it delays the next attempt so a broken repo isn't retried every tick.
+  pub last_failed_at: Option<DateTime<Utc>>,
+
+  /// How many requests in a row have failed since the last success (when `last_failed_at` is
+  /// newer than `app_last_fetched`).
+  pub consecutive_failures: u32,
 }

 impl CraftAppsReleaseCache {
@@ -69,6 +79,8 @@ impl CraftAppsReleaseCache {
        latest_release: None,
        app_last_fetched: None,
        etag: None,
+        last_failed_at: None,
+        consecutive_failures: 0,
      });
    }
  }
@@ -100,6 +112,53 @@ impl CraftAppsReleaseCache {
    }
  }

+  /// A copy of what's known about the app with app key `key` (`photocraft`).
+  pub fn app(&self, key: &str) -> Option<CachedCraftApp> {
+    let map = self.inner.read().unwrap_or_else(PoisonError::into_inner);
+    map.get(key).cloned()
+  }
+
+  /// Takes a fetch made elsewhere (another instance, through Redis) for the app with app key
+  /// `key`, if it is newer than what this instance has. Returns whether it was taken. Ignored for
+  /// keys that were never registered, so the registered app list stays authoritative.
+  pub fn merge_fetched(&self, key: &str, latest_release: Option<CraftAppRelease>, etag: Option<String>, fetched_at: DateTime<Utc>) -> bool {
+    let mut map = self.inner.write().unwrap_or_else(PoisonError::into_inner);
+    let Some(app) = map.get_mut(key) else {
+      return false;
+    };
+    if app.app_last_fetched.is_some_and(|mine| mine >= fetched_at) {
+      return false;
+    }
+    app.latest_release = latest_release;
+    app.etag = etag;
+    app.app_last_fetched = Some(fetched_at);
+    true
+  }
+
+  /// Records a failed request for the app with app key `key` (`photocraft`) and returns how many
+  /// have failed in a row since its last success. Ignored (returns 0) for unregistered keys.
+  pub fn record_failure(&self, key: &str, failed_at: DateTime<Utc>) -> u32 {
+    let mut map = self.inner.write().unwrap_or_else(PoisonError::into_inner);
+    let Some(app) = map.get_mut(key) else {
+      return 0;
+    };
+    app.consecutive_failures = if app.is_failing() { app.consecutive_failures.saturating_add(1) } else { 1 };
+    app.last_failed_at = Some(failed_at);
+    app.consecutive_failures
+  }
+
+  /// Takes a failure recorded elsewhere (another instance, through Redis) for the app with app
+  /// key `key`, if it is newer than the one this instance knows.
+  pub fn merge_failure(&self, key: &str, failed_at: DateTime<Utc>, consecutive_failures: u32) {
+    let mut map = self.inner.write().unwrap_or_else(PoisonError::into_inner);
+    if let Some(app) = map.get_mut(key) {
+      if app.last_failed_at.is_none_or(|mine| mine < failed_at) {
+        app.last_failed_at = Some(failed_at);
+        app.consecutive_failures = consecutive_failures;
+      }
+    }
+  }
+
  /// Everything, as the endpoint serves it: `apps` keyed by app key.
  pub fn snapshot(&self) -> CraftAppsReleaseInfo {
    let map = self.inner.read().unwrap_or_else(PoisonError::into_inner);
@@ -119,6 +178,17 @@ impl CraftAppsReleaseCache {
  }
 }

+impl CachedCraftApp {
+  /// Whether the last request for this app failed (no success since).
+  pub fn is_failing(&self) -> bool {
+    match (self.last_failed_at, self.app_last_fetched) {
+      (Some(failed), Some(fetched)) => failed > fetched,
+      (Some(_), None) => true,
+      (None, _) => false,
+    }
+  }
+}
+
 /// The oldest `app_last_fetched`, or `None` while any app has never been fetched.
 fn least_recently_fetched(apps: &BTreeMap<String, CraftAppReleaseInfo>) -> Option<DateTime<Utc>> {
  if apps.is_empty() {
@@ -168,4 +238,42 @@ mod tests {
    cache.register_apps(&apps);
    assert_eq!(cache.snapshot().apps["photocraft"].app_last_fetched, Some(now));
  }
+
+  #[test]
+  fn merging_takes_only_newer_fetches_of_registered_apps() {
+    let cache = CraftAppsReleaseCache::new();
+    cache.register_apps(&[("photocraft".into(), "PhotoCraft".into(), "storytold/photocraft".into())]);
+    let early = Utc.with_ymd_and_hms(2026, 10, 10, 1, 0, 0).unwrap();
+    let late = Utc.with_ymd_and_hms(2026, 10, 10, 2, 0, 0).unwrap();
+
+    assert!(cache.merge_fetched("photocraft", None, Some("\"new\"".into()), late));
+    assert!(!cache.merge_fetched("photocraft", None, Some("\"old\"".into()), early), "older than ours");
+    assert!(!cache.merge_fetched("photocraft", None, Some("\"same\"".into()), late), "no newer than ours");
+    assert!(!cache.merge_fetched("unknowncraft", None, None, late), "not registered");
+
+    let snapshot = cache.snapshot();
+    assert_eq!(snapshot.apps.len(), 1);
+    assert_eq!(snapshot.apps["photocraft"].app_last_fetched, Some(late));
+    assert_eq!(cache.etag("photocraft").as_deref(), Some("\"new\""));
+  }
+
+  #[test]
+  fn failures_count_up_until_a_success() {
+    let cache = CraftAppsReleaseCache::new();
+    cache.register_apps(&[("photocraft".into(), "PhotoCraft".into(), "storytold/photocraft".into())]);
+    let t = |h| Utc.with_ymd_and_hms(2026, 10, 10, h, 0, 0).unwrap();
+
+    assert_eq!(cache.record_failure("photocraft", t(1)), 1);
+    assert_eq!(cache.record_failure("photocraft", t(2)), 2);
+    assert!(cache.app("photocraft").unwrap().is_failing());
+
+    cache.store("photocraft", None, None, t(3));
+    assert!(!cache.app("photocraft").unwrap().is_failing());
+    assert_eq!(cache.record_failure("photocraft", t(4)), 1, "counting restarts after a success");
+
+    cache.merge_failure("photocraft", t(3), 9);
+    assert_eq!(cache.app("photocraft").unwrap().consecutive_failures, 1, "older failure ignored");
+    cache.merge_failure("photocraft", t(5), 4);
+    assert_eq!(cache.app("photocraft").unwrap().consecutive_failures, 4);
+  }
 }
@@ -1,19 +1,31 @@
 //! How storyteller-web polls GitHub for Craft App releases, from the environment.
 //!
-//! | Variable                                  | Default        | Meaning                                         |
-//! |-------------------------------------------|----------------|-------------------------------------------------|
-//! | `CRAFT_APPS_RELEASE_POLLING_ENABLED`      | `true`         | Poll at all (the endpoint still answers)         |
-//! | `CRAFT_APPS_RELEASE_POLL_INTERVAL_SECS`   | `1800` (30 m)  | Time between rounds; at least 300                |
-//! | `CRAFT_APPS_RELEASE_MAX_BACKOFF_SECS`     | `14400` (4 h)  | Longest wait after repeated failures/rate limits |
-//! | `CRAFT_APPS_RELEASE_GITHUB_TOKEN`         | (none)         | Raises GitHub's limit from 60 to 5000 per hour   |
-//! | `CRAFT_APPS_RELEASE_APPS`                 | built-in list  | `key=Name=org/repo,…` to replace the app list    |
+//! | Variable                                  | Default                     | Meaning                                                       |
+//! |-------------------------------------------|-----------------------------|---------------------------------------------------------------|
+//! | `CRAFT_APPS_RELEASE_POLLING_ENABLED`      | `true`                      | Poll at all (the endpoint still answers)                      |
+//! | `CRAFT_APPS_RELEASE_COORDINATION_ENABLED` | `true`                      | Share releases and backoff through Redis (see below)          |
+//! | `CRAFT_APPS_RELEASE_POLL_INTERVAL_SECS`   | `120`; `1800` with no token | How old a release may get; at least `60`; `300` with no token |
+//! | `CRAFT_APPS_RELEASE_MAX_BACKOFF_SECS`     | `14400` (4 h)               | Longest wait after repeated failures/rate limits              |
+//! | `CRAFT_APPS_RELEASE_GITHUB_TOKEN`         | (none; optional)            | Raises GitHub's limit from 60 to 5000 per hour                |
+//! | `CRAFT_APPS_RELEASE_APPS`                 | built-in list               | `key=Name=org/repo,…` to replace the app list                 |
 //!
-//! Every instance in the fleet polls on its own, so the defaults are deliberately slow: one round
-//! is one request per app, and without a token GitHub allows 60 an hour per IP address.
+//! The GitHub token is optional. Unset, empty or blank, polling still runs: requests go out
+//! without an `Authorization` header, and the interval defaults to the slow no-token value.
+//!
+//! With coordination (the default), the fleet shares each app's release, ETag and fetch time in
+//! Redis, so the whole fleet requests an app from GitHub about once per interval instead of every
+//! pod doing so. That is what makes the 2 minute default safe with a token. Without a token
+//! GitHub allows 60 requests an hour per IP address and also counts `304`s, so the interval stays
+//! at 30 minutes (13 apps: about 26 requests an hour). While Redis is unreachable (or with
+//! coordination off) every pod polls on its own at the same interval.

 use std::time::Duration;

-const MIN_INTERVAL_SECS: u64 = 300;
+const DEFAULT_INTERVAL_SECS_WITH_TOKEN: u64 = 120;
+const MIN_INTERVAL_SECS_WITH_TOKEN: u64 = 60;
+const DEFAULT_INTERVAL_SECS_WITHOUT_TOKEN: u64 = 1800;
+const MIN_INTERVAL_SECS_WITHOUT_TOKEN: u64 = 300;
+const DEFAULT_MAX_BACKOFF_SECS: u64 = 14_400;

 /// The Craft Apps the launcher knows: `(app key, display name, GitHub repo)`. The app key is
 /// the app's lowercase machine name; it keys `CraftAppsReleaseCache` and the endpoint's
@@ -39,8 +51,13 @@ const DEFAULT_APPS: &[(&str, &str, &str)] = &[
 #[derive(Clone, Debug)]
 pub struct CraftAppsPollConfig {
  pub enabled: bool,
+  /// Share releases and backoff with the fleet through Redis; when false (or while Redis is
+  /// unreachable) this pod polls on its own.
+  pub coordination_enabled: bool,
+  /// How old an app's release may get before it is requested from GitHub again.
  pub interval: Duration,
  pub max_backoff: Duration,
+  /// Never empty or blank: those are read as no token.
  pub github_token: Option<String>,
  /// `(app key, display name, repo)`; see [`DEFAULT_APPS`].
  pub apps: Vec<(String, String, String)>,
@@ -48,17 +65,22 @@ pub struct CraftAppsPollConfig {

 impl CraftAppsPollConfig {
  pub fn from_env() -> Self {
-    let interval_secs = easyenv::get_env_num::<u64>("CRAFT_APPS_RELEASE_POLL_INTERVAL_SECS", 1800).unwrap_or(1800).max(MIN_INTERVAL_SECS);
-    let max_backoff_secs = easyenv::get_env_num::<u64>("CRAFT_APPS_RELEASE_MAX_BACKOFF_SECS", 14_400).unwrap_or(14_400).max(interval_secs);
-    let github_token = easyenv::get_env_string_optional("CRAFT_APPS_RELEASE_GITHUB_TOKEN")
-        .map(|t| t.trim().to_string())
-        .filter(|t| !t.is_empty());
+    let github_token = non_blank_token(easyenv::get_env_string_optional("CRAFT_APPS_RELEASE_GITHUB_TOKEN"));
+    let has_token = github_token.is_some();
+    let default_interval = default_interval_secs(has_token);
+    let requested_interval = easyenv::get_env_num::<u64>("CRAFT_APPS_RELEASE_POLL_INTERVAL_SECS", default_interval)
+        .unwrap_or(default_interval);
+    let interval_secs = clamp_interval_secs(requested_interval, has_token);
+    let max_backoff_secs = easyenv::get_env_num::<u64>("CRAFT_APPS_RELEASE_MAX_BACKOFF_SECS", DEFAULT_MAX_BACKOFF_SECS)
+        .unwrap_or(DEFAULT_MAX_BACKOFF_SECS)
+        .max(interval_secs);
    let apps = easyenv::get_env_string_optional("CRAFT_APPS_RELEASE_APPS")
        .map(|v| parse_app_list(&v))
        .filter(|apps| !apps.is_empty())
        .unwrap_or_else(default_apps);
    Self {
      enabled: easyenv::get_env_bool_or_default("CRAFT_APPS_RELEASE_POLLING_ENABLED", true),
+      coordination_enabled: easyenv::get_env_bool_or_default("CRAFT_APPS_RELEASE_COORDINATION_ENABLED", true),
      interval: Duration::from_secs(interval_secs),
      max_backoff: Duration::from_secs(max_backoff_secs),
      github_token,
@@ -71,6 +93,20 @@ pub fn default_apps() -> Vec<(String, String, String)> {
  DEFAULT_APPS.iter().map(|(k, n, r)| (k.to_string(), n.to_string(), r.to_string())).collect()
 }

+/// The token, trimmed; `None` when unset, empty or blank.
+fn non_blank_token(value: Option<String>) -> Option<String> {
+  value.map(|t| t.trim().to_string()).filter(|t| !t.is_empty())
+}
+
+fn default_interval_secs(has_token: bool) -> u64 {
+  if has_token { DEFAULT_INTERVAL_SECS_WITH_TOKEN } else { DEFAULT_INTERVAL_SECS_WITHOUT_TOKEN }
+}
+
+fn clamp_interval_secs(requested: u64, has_token: bool) -> u64 {
+  let min = if has_token { MIN_INTERVAL_SECS_WITH_TOKEN } else { MIN_INTERVAL_SECS_WITHOUT_TOKEN };
+  requested.max(min)
+}
+
 /// Reads `key=Name=org/repo,…` (app key, display name, repo). Keys are lowercased. Malformed
 /// entries are skipped.
 fn parse_app_list(value: &str) -> Vec<(String, String, String)> {
@@ -106,4 +142,23 @@ mod tests {
    assert_eq!(apps.len(), 13);
    assert!(apps.iter().any(|(k, _, r)| k == "artcraft-launcher" && r == "storytold/craft-launcher"));
  }
+
+  #[test]
+  fn a_missing_empty_or_blank_token_is_no_token() {
+    assert_eq!(non_blank_token(None), None);
+    assert_eq!(non_blank_token(Some(String::new())), None);
+    assert_eq!(non_blank_token(Some("  \n".to_string())), None);
+    assert_eq!(non_blank_token(Some(" ghp_x ".to_string())).as_deref(), Some("ghp_x"));
+  }
+
+  #[test]
+  fn the_interval_defaults_and_minimum_depend_on_the_token() {
+    assert_eq!(default_interval_secs(true), 120);
+    assert_eq!(default_interval_secs(false), 1800);
+    assert_eq!(clamp_interval_secs(10, true), 60);
+    assert_eq!(clamp_interval_secs(10, false), 300);
+    assert_eq!(clamp_interval_secs(90, true), 90);
+    assert_eq!(clamp_interval_secs(90, false), 300);
+    assert_eq!(clamp_interval_secs(3600, false), 3600);
+  }
 }
@@ -0,0 +1,97 @@
+//! Falling back to polling alone when Redis is unreachable, and saying so without flooding the
+//! log every tick.
+
+use std::time::{Duration, Instant};
+
+use log::{info, warn};
+
+/// While Redis keeps failing, repeat the warning at most this often.
+const WARN_EVERY: Duration = Duration::from_secs(10 * 60);
+
+/// How this tick works.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PollMode {
+  /// With the fleet: shared releases, claims and backoff in Redis.
+  Coordinated,
+  /// Alone, from this instance's memory: coordination is off or Redis is unreachable. Same
+  /// interval and backoff rules, so a Redis outage never stops releases from refreshing.
+  Standalone,
+}
+
+/// Remembers Redis trouble so it is logged when it starts, every [`WARN_EVERY`] while it lasts,
+/// and when it ends.
+#[derive(Debug, Default)]
+pub struct RedisTroubleLog {
+  failing_since: Option<Instant>,
+  last_warned: Option<Instant>,
+  suppressed: u32,
+}
+
+impl PollMode {
+  /// Coordinated only when coordination is on and the shared state could be read.
+  pub fn choose(coordination_enabled: bool, shared_state_readable: bool) -> Self {
+    if coordination_enabled && shared_state_readable { Self::Coordinated } else { Self::Standalone }
+  }
+}
+
+impl RedisTroubleLog {
+  /// Notes a Redis error; returns whether it was logged.
+  pub fn failed(&mut self, err: &anyhow::Error, now: Instant) -> bool {
+    self.failing_since.get_or_insert(now);
+    if self.last_warned.is_some_and(|last| now.duration_since(last) < WARN_EVERY) {
+      self.suppressed = self.suppressed.saturating_add(1);
+      return false;
+    }
+    warn!(
+      "Craft app releases: Redis unavailable ({:#}); polling GitHub on this instance alone until it is back ({} similar errors not logged).",
+      err, self.suppressed,
+    );
+    self.last_warned = Some(now);
+    self.suppressed = 0;
+    true
+  }
+
+  /// Notes that Redis works; logs the recovery once.
+  pub fn succeeded(&mut self, now: Instant) {
+    if let Some(since) = self.failing_since.take() {
+      info!("Craft app releases: Redis is back after {:?}; coordinating with the fleet again.", now.duration_since(since));
+      self.last_warned = None;
+      self.suppressed = 0;
+    }
+  }
+
+  pub fn is_failing(&self) -> bool {
+    self.failing_since.is_some()
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use anyhow::anyhow;
+
+  use super::*;
+
+  #[test]
+  fn coordinates_only_when_enabled_and_redis_answers() {
+    assert_eq!(PollMode::choose(true, true), PollMode::Coordinated);
+    assert_eq!(PollMode::choose(true, false), PollMode::Standalone);
+    assert_eq!(PollMode::choose(false, true), PollMode::Standalone);
+    assert_eq!(PollMode::choose(false, false), PollMode::Standalone);
+  }
+
+  #[test]
+  fn warns_once_per_window_while_redis_keeps_failing() {
+    let mut log = RedisTroubleLog::default();
+    let start = Instant::now();
+    let err = anyhow!("connection refused");
+    assert!(log.failed(&err, start));
+    assert!(!log.failed(&err, start + Duration::from_secs(30)));
+    assert!(!log.failed(&err, start + Duration::from_secs(599)));
+    assert!(log.failed(&err, start + Duration::from_secs(600)));
+    assert!(log.is_failing());
+
+    log.succeeded(start + Duration::from_secs(700));
+    assert!(!log.is_failing());
+    assert!(log.failed(&err, start + Duration::from_secs(730)), "a new outage warns right away");
+  }
+}
@@ -36,20 +36,10 @@ pub fn build_client() -> wreq::Result<wreq::Client> {
      .build()
 }

-/// `GET /repos/{repo}/releases/latest`, revalidated with `etag`.
+/// `GET /repos/{repo}/releases/latest`, revalidated with `etag`. `token` is optional: without
+/// one the request is unauthenticated.
 pub async fn fetch_latest_release(client: &wreq::Client, repo: &str, etag: Option<&str>, token: Option<&str>) -> FetchOutcome {
-  let url = format!("{API_BASE}/repos/{repo}/releases/latest");
-  let mut request = client
-      .get(&url)
-      .header(USER_AGENT, "storyteller-web (craft apps release poller)")
-      .header(ACCEPT, "application/vnd.github+json")
-      .header("X-GitHub-Api-Version", "2022-11-28");
-  if let Some(etag) = etag {
-    request = request.header(IF_NONE_MATCH, etag);
-  }
-  if let Some(token) = token {
-    request = request.header(AUTHORIZATION, format!("Bearer {token}"));
-  }
+  let request = build_request(client, repo, etag, token);
  let response = match request.send().await {
    Ok(r) => r,
    Err(err) => return FetchOutcome::Failed(format!("request failed: {err}")),
@@ -87,6 +77,24 @@ pub async fn fetch_latest_release(client: &wreq::Client, repo: &str, etag: Optio
  }
 }

+/// The request for `repo`'s latest release. Sends `Authorization: Bearer` only for a non-blank
+/// token (an authenticated `304` doesn't count against the rate limit); never an empty header.
+fn build_request(client: &wreq::Client, repo: &str, etag: Option<&str>, token: Option<&str>) -> wreq::RequestBuilder {
+  let url = format!("{API_BASE}/repos/{repo}/releases/latest");
+  let mut request = client
+      .get(&url)
+      .header(USER_AGENT, "storyteller-web (craft apps release poller)")
+      .header(ACCEPT, "application/vnd.github+json")
+      .header("X-GitHub-Api-Version", "2022-11-28");
+  if let Some(etag) = etag {
+    request = request.header(IF_NONE_MATCH, etag);
+  }
+  if let Some(token) = token.map(str::trim).filter(|t| !t.is_empty()) {
+    request = request.header(AUTHORIZATION, format!("Bearer {token}"));
+  }
+  request
+}
+
 /// The parts of GitHub's release object we use. Everything optional: GitHub's answer is input.
 #[derive(Deserialize)]
 struct GithubRelease {
@@ -231,6 +239,25 @@ mod tests {
    )
  }

+  #[test]
+  fn sends_the_token_as_a_bearer_header() {
+    let client = build_client().unwrap();
+    let request = build_request(&client, REPO, Some("\"etag\""), Some("ghp_x")).build().unwrap();
+    assert_eq!(request.headers().get(AUTHORIZATION).unwrap(), "Bearer ghp_x");
+    assert_eq!(request.headers().get(IF_NONE_MATCH).unwrap(), "\"etag\"");
+  }
+
+  #[test]
+  fn without_a_token_the_request_is_unauthenticated() {
+    let client = build_client().unwrap();
+    for token in [None, Some(""), Some("  ")] {
+      let request = build_request(&client, REPO, None, token).build().unwrap();
+      assert!(request.headers().get(AUTHORIZATION).is_none(), "{:?}", token);
+      assert!(request.headers().get(IF_NONE_MATCH).is_none());
+      assert_eq!(request.uri().to_string(), "https://api.github.com/repos/storytold/photocraft/releases/latest");
+    }
+  }
+
  #[test]
  fn reads_a_release() {
    let r = parse_release(release_json("https://github.com/storytold/photocraft/releases/download/v0.5.0/photocraft-0.5.0-windows-x64.msi").as_bytes(), REPO).unwrap();
@@ -1,3 +1,7 @@
 pub mod config;
+pub mod fallback;
 pub mod github;
 pub mod poll_craft_app_releases;
+pub mod scheduling;
+pub mod shared_entry;
+pub mod shared_store;
@@ -1,34 +1,48 @@
-//! Polls GitHub for the latest release of every Craft App and keeps it in
-//! [`CraftAppsReleaseCache`], which `GET /v1/craft_apps/release_info` serves to ArtCraft Launcher
-//! (so thousands of launchers don't each poll GitHub).
+//! Keeps [`CraftAppsReleaseCache`] holding the latest release of every Craft App, which
+//! `GET /v1/craft_apps/release_info` serves to ArtCraft Launcher (so thousands of launchers don't
+//! each poll GitHub).
+//!
+//! Every instance runs this loop, waking every [`TICK`] (30 s, with jitter). At startup it first
+//! takes the fleet's shared releases from Redis, so a new instance serves them at once without
+//! asking GitHub. With coordination (the default) the fleet shares its work through Redis
+//! (`shared_entry` documents the keys):
+//!
+//! 1. Read every app's shared entry and the fleet's backoff; take anything newer into memory.
+//!    The endpoint always serves memory, so it keeps answering whatever Redis does.
+//! 2. If the fleet is backing off (GitHub rate limited us, or a whole round failed), stop.
+//! 3. For each app that is due (its shared fetch is `interval - CLOCK_SKEW_MARGIN` old, and any
+//!    failure retry has waited), read its entry again (another instance may just have fetched
+//!    it), claim it so instances ticking at the same moment don't all ask, then ask GitHub with
+//!    the shared ETag and share the answer. Shared writes never go back in time.
+//! 4. After a round with requests, update the fleet's backoff (a good round clears it).
 //!
-//! Every instance of storyteller-web polls on its own, so it is gentle: one round at startup,
-//! then one per `CRAFT_APPS_RELEASE_POLL_INTERVAL_SECS` (30 minutes by default) with jitter, one
-//! request per app with ETag revalidation, and growing waits after rate limits or failures.
 //! Each app is all-or-nothing: either its latest release is stored with a fresh
 //! `app_last_fetched`, or nothing about it changes.
+//!
+//! If Redis is unreachable (or coordination is off) the instance does the same from its own
+//! memory, as every instance did before coordination, and returns to coordinating when Redis
+//! answers again. Nothing here panics; trouble is logged and retried.

-use std::time::Duration;
+use std::time::{Duration, Instant};

-use chrono::{DateTime, Utc};
+use chrono::Utc;
 use log::{error, info, warn};

 use crate::state::craft_apps_release_cache::CraftAppsReleaseCache;
 use crate::threads::poll_craft_app_releases_thread::config::CraftAppsPollConfig;
+use crate::threads::poll_craft_app_releases_thread::fallback::{PollMode, RedisTroubleLog};
 use crate::threads::poll_craft_app_releases_thread::github;
 use crate::threads::poll_craft_app_releases_thread::github::FetchOutcome;
+use crate::threads::poll_craft_app_releases_thread::scheduling::{backoff_after, is_due, jittered, Backoff, RoundResult, TICK};
+use crate::threads::poll_craft_app_releases_thread::shared_entry::{SharedAppEntry, SharedFailure};
+use crate::threads::poll_craft_app_releases_thread::shared_store::{PollerRedisPool, SharedStore};

-/// Spreads a fleet's first round over this window, so a deploy doesn't hit GitHub all at once.
+/// Spreads a fleet's first tick over this window, so a deploy doesn't hit GitHub all at once.
 const STARTUP_JITTER: Duration = Duration::from_secs(15);
 /// Pause between the requests of one round.
 const BETWEEN_REQUESTS: Duration = Duration::from_millis(1_000);
-/// First wait after a round where every request failed (network trouble); doubles up to the
-/// normal interval.
-const FIRST_FAILURE_RETRY: Duration = Duration::from_secs(60);
-/// Extra margin after GitHub's announced rate-limit reset.
-const RESET_MARGIN: Duration = Duration::from_secs(60);

-pub async fn poll_craft_app_releases(cache: CraftAppsReleaseCache, config: CraftAppsPollConfig) {
+pub async fn poll_craft_app_releases(cache: CraftAppsReleaseCache, config: CraftAppsPollConfig, redis_pool: PollerRedisPool, hostname: String) {
  cache.register_apps(&config.apps);
  if !config.enabled {
    info!("Craft app release polling is disabled (CRAFT_APPS_RELEASE_POLLING_ENABLED=false).");
@@ -42,136 +56,265 @@ pub async fn poll_craft_app_releases(cache: CraftAppsReleaseCache, config: Craft
    }
  };
  info!(
-    "Polling GitHub for {} craft app releases every {:?} (token: {}).",
-    config.apps.len(), config.interval, config.github_token.is_some(),
+    "Polling GitHub for {} craft app releases, each at most every {:?} (GitHub token: {}; coordinated through Redis: {}).",
+    config.apps.len(), config.interval, config.github_token.is_some(), config.coordination_enabled,
  );
+  let store = config.coordination_enabled.then(|| SharedStore::new(redis_pool, hostname));
+  let mut poller = Poller::new(client, cache, config, store);
+  // Warm memory from the fleet's shared copy right away: it costs no GitHub quota, and the
+  // jitter below only needs to spread out requests to GitHub.
+  if poller.refresh_from_shared().await {
+    info!("Craft app releases: took the fleet's shared releases from Redis at startup.");
+  }
  tokio::time::sleep(jittered(STARTUP_JITTER, 1.0)).await;
-
-  let mut consecutive_trouble = 0u32;
  loop {
-    let round = poll_round(&client, &cache, &config).await;
-    let wait = next_wait(&config, &round, &mut consecutive_trouble, Utc::now());
+    poller.tick().await;
+    tokio::time::sleep(jittered(TICK, 0.1)).await;
+  }
+}
+
+/// The polling loop's state.
+struct Poller {
+  client: wreq::Client,
+  cache: CraftAppsReleaseCache,
+  config: CraftAppsPollConfig,
+  /// `None` when coordination is off.
+  store: Option<SharedStore>,
+  /// The backoff this instance knows: the fleet's while Redis answers, its own otherwise.
+  backoff: Backoff,
+  redis_log: RedisTroubleLog,
+}
+
+/// What one tick did.
+#[derive(Debug)]
+struct TickReport {
+  mode: PollMode,
+  round: RoundResult,
+}
+
+/// Whether to ask GitHub for an app, after looking at its shared entry once more.
+enum SharedCheck {
+  Ask,
+  /// Fresh after all, or another instance is asking right now.
+  Skip,
+  /// The fleet started backing off.
+  StopRound,
+}
+
+impl Poller {
+  fn new(client: wreq::Client, cache: CraftAppsReleaseCache, config: CraftAppsPollConfig, store: Option<SharedStore>) -> Self {
+    Self { client, cache, config, store, backoff: Backoff::default(), redis_log: RedisTroubleLog::default() }
+  }
+
+  async fn tick(&mut self) -> TickReport {
+    let shared_state_readable = self.refresh_from_shared().await;
+    let mode = PollMode::choose(self.store.is_some(), shared_state_readable);
+    let mut report = TickReport { mode, round: RoundResult::default() };
+    let now = Utc::now();
+    if self.backoff.is_active(now) {
+      return report;
+    }
+    let due: Vec<(String, String)> = self
+        .config
+        .apps
+        .iter()
+        .filter(|(key, _, _)| self.cache.app(key).is_some_and(|app| is_due(&app, now, self.config.interval)))
+        .map(|(key, _, repo)| (key.clone(), repo.clone()))
+        .collect();
+    if due.is_empty() {
+      return report;
+    }
+    let (round, mode) = self.poll_round(&due, mode).await;
+    report.round = round;
+    report.mode = mode;
+    if report.round.requests() == 0 {
+      return report;
+    }
+    self.update_backoff(&report.round, mode).await;
    info!(
-      "Craft app releases: {} fetched, {} unchanged, {} failed{}; next round in {:?}.",
-      round.fetched, round.unchanged, round.failed,
-      if round.rate_limited.is_some() { ", rate limited" } else { "" },
-      wait,
+      "Craft app releases ({:?}): {} fetched, {} unchanged, {} failed{}{}.",
+      mode, report.round.fetched, report.round.unchanged, report.round.failed,
+      if report.round.rate_limited.is_some() { ", rate limited" } else { "" },
+      self.backoff.until.map(|until| format!("; backing off until {until}")).unwrap_or_default(),
    );
-    tokio::time::sleep(wait).await;
+    report
  }
-}

-/// What one round did.
-#[derive(Debug, Default)]
-struct RoundResult {
-  fetched: usize,
-  unchanged: usize,
-  failed: usize,
-  /// GitHub refused; `Some(Some(t))` when it said when to come back.
-  rate_limited: Option<Option<DateTime<Utc>>>,
-}
-
-async fn poll_round(client: &wreq::Client, cache: &CraftAppsReleaseCache, config: &CraftAppsPollConfig) -> RoundResult {
-  let mut round = RoundResult::default();
-  for (i, (key, _, repo)) in config.apps.iter().enumerate() {
-    if i > 0 {
-      tokio::time::sleep(BETWEEN_REQUESTS).await;
-    }
-    let etag = cache.etag(key);
-    match github::fetch_latest_release(client, repo, etag.as_deref(), config.github_token.as_deref()).await {
-      FetchOutcome::Fetched { release, etag } => {
-        cache.store(key, release, etag, Utc::now());
-        round.fetched += 1;
+  /// Takes the fleet's shared state into memory. False when coordination is off or Redis
+  /// couldn't be read (this tick then works alone).
+  async fn refresh_from_shared(&mut self) -> bool {
+    let Some(store) = self.store.clone() else {
+      return false;
+    };
+    let keys: Vec<String> = self.config.apps.iter().map(|(key, _, _)| key.clone()).collect();
+    match store.read(&keys).await {
+      Ok((entries, backoff)) => {
+        self.redis_log.succeeded(Instant::now());
+        for (key, entry) in keys.iter().zip(entries) {
+          if let Some(entry) = entry {
+            merge_shared_entry(&self.cache, key, entry);
+          }
+        }
+        self.backoff = backoff;
+        true
      }
-      FetchOutcome::NotModified => {
-        cache.confirm_unchanged(key, Utc::now());
-        round.unchanged += 1;
-      }
-      FetchOutcome::RateLimited { retry_at } => {
-        warn!("GitHub rate limit reached polling {repo}; retry after {:?}.", retry_at);
-        round.rate_limited = Some(retry_at);
-        // The rest of the round would be refused too.
-        break;
-      }
-      FetchOutcome::Failed(why) => {
-        warn!("Couldn't fetch the latest release of {repo}: {why}");
-        round.failed += 1;
+      Err(err) => {
+        self.redis_log.failed(&err, Instant::now());
+        false
      }
    }
  }
-  round
+
+  /// Asks GitHub about each due app. Returns what happened and the mode it ended in (a Redis
+  /// error mid-round finishes the round alone).
+  async fn poll_round(&mut self, due: &[(String, String)], mut mode: PollMode) -> (RoundResult, PollMode) {
+    let mut round = RoundResult::default();
+    for (key, repo) in due {
+      if let Some(store) = self.coordinated_store(mode) {
+        match self.check_shared(&store, key).await {
+          Ok(SharedCheck::Ask) => {}
+          Ok(SharedCheck::Skip) => continue,
+          Ok(SharedCheck::StopRound) => break,
+          Err(err) => {
+            self.redis_log.failed(&err, Instant::now());
+            mode = PollMode::Standalone;
+          }
+        }
+      }
+      if round.requests() > 0 {
+        tokio::time::sleep(BETWEEN_REQUESTS).await;
+      }
+      let etag = self.cache.etag(key);
+      let outcome = github::fetch_latest_release(&self.client, repo, etag.as_deref(), self.config.github_token.as_deref()).await;
+      let now = Utc::now();
+      match outcome {
+        FetchOutcome::Fetched { release, etag } => {
+          round.fetched += 1;
+          if let Some(store) = self.coordinated_store(mode) {
+            let shared = store.write_fetched(key, release.as_ref(), etag.as_deref(), now).await;
+            mode = self.after_shared_write(shared, mode);
+          }
+          self.cache.store(key, release, etag, now);
+        }
+        FetchOutcome::NotModified => {
+          round.unchanged += 1;
+          self.cache.confirm_unchanged(key, now);
+          if let (Some(store), Some(etag)) = (self.coordinated_store(mode), etag.as_deref()) {
+            let shared = store.write_unchanged(key, etag, now).await;
+            mode = self.after_shared_write(shared, mode);
+          }
+        }
+        FetchOutcome::RateLimited { retry_at } => {
+          warn!("GitHub rate limit reached polling {repo}; retry after {:?}.", retry_at);
+          round.rate_limited = Some(retry_at);
+          // The rest of the round would be refused too.
+          break;
+        }
+        FetchOutcome::Failed(why) => {
+          warn!("Couldn't fetch the latest release of {repo}: {why}");
+          round.failed += 1;
+          let consecutive_failures = self.cache.record_failure(key, now);
+          if let Some(store) = self.coordinated_store(mode) {
+            let failure = SharedFailure { failed_at: now, consecutive_failures };
+            let shared = store.write_failure(key, failure).await;
+            mode = self.after_shared_write(shared, mode);
+          }
+        }
+      }
+    }
+    (round, mode)
+  }
+
+  /// Right before asking GitHub about `key`: takes its latest shared entry, and claims it if it
+  /// is still due and the fleet isn't backing off.
+  async fn check_shared(&mut self, store: &SharedStore, key: &str) -> anyhow::Result<SharedCheck> {
+    let (mut entries, backoff) = store.read(&[key.to_string()]).await?;
+    if let Some(entry) = entries.pop().flatten() {
+      merge_shared_entry(&self.cache, key, entry);
+    }
+    self.backoff = backoff;
+    let now = Utc::now();
+    if backoff.is_active(now) {
+      return Ok(SharedCheck::StopRound);
+    }
+    if !self.cache.app(key).is_some_and(|app| is_due(&app, now, self.config.interval)) {
+      return Ok(SharedCheck::Skip);
+    }
+    Ok(if store.claim(key, now).await? { SharedCheck::Ask } else { SharedCheck::Skip })
+  }
+
+  /// Sets the backoff after a round with requests, and shares it if it changed.
+  async fn update_backoff(&mut self, round: &RoundResult, mode: PollMode) {
+    let previous = self.backoff;
+    self.backoff = backoff_after(&self.config, round, previous, Utc::now());
+    if self.backoff == previous {
+      return;
+    }
+    if let Some(store) = self.coordinated_store(mode) {
+      if let Err(err) = store.write_backoff(self.backoff).await {
+        self.redis_log.failed(&err, Instant::now());
+      }
+    }
+  }
+
+  /// Logs a failed shared write; the rest of the round then works alone.
+  fn after_shared_write(&mut self, result: anyhow::Result<bool>, mode: PollMode) -> PollMode {
+    match result {
+      Ok(_) => mode,
+      Err(err) => {
+        self.redis_log.failed(&err, Instant::now());
+        PollMode::Standalone
+      }
+    }
+  }
+
+  /// The shared store, in [`PollMode::Coordinated`].
+  fn coordinated_store(&self, mode: PollMode) -> Option<SharedStore> {
+    self.store.clone().filter(|_| mode == PollMode::Coordinated)
+  }
 }

-/// How long to wait before the next round. Rate limits wait for GitHub's reset and back off
-/// further each time; rounds where everything failed retry sooner, but never faster than a
-/// minute, doubling up to the normal interval.
-fn next_wait(config: &CraftAppsPollConfig, round: &RoundResult, consecutive_trouble: &mut u32, now: DateTime<Utc>) -> Duration {
-  let base = if let Some(retry_at) = round.rate_limited {
-    *consecutive_trouble = consecutive_trouble.saturating_add(1);
-    let until_reset = retry_at
-        .and_then(|t| (t - now).to_std().ok())
-        .map(|d| d + RESET_MARGIN)
-        .unwrap_or(Duration::ZERO);
-    let backoff = scaled(config.interval, *consecutive_trouble);
-    until_reset.max(backoff).min(config.max_backoff)
-  } else if round.fetched == 0 && round.unchanged == 0 && round.failed > 0 {
-    *consecutive_trouble = consecutive_trouble.saturating_add(1);
-    scaled(FIRST_FAILURE_RETRY, *consecutive_trouble).min(config.interval)
-  } else {
-    *consecutive_trouble = 0;
-    config.interval
-  };
-  jittered(base, 0.1)
-}
-
-/// `base × 2^(n-1)`, without overflow.
-fn scaled(base: Duration, n: u32) -> Duration {
-  base.saturating_mul(1u32 << n.saturating_sub(1).min(16))
-}
-
-/// `base` ± `fraction`, so a fleet doesn't poll in lockstep. With `fraction == 1.0`, anywhere in
-/// `0..=2×base`; used for the startup spread.
-fn jittered(base: Duration, fraction: f64) -> Duration {
-  let factor = 1.0 + fraction * (rand::random::<f64>() * 2.0 - 1.0);
-  base.mul_f64(factor.max(0.0))
+/// Takes a shared entry into memory (only what is newer than what this instance has).
+fn merge_shared_entry(cache: &CraftAppsReleaseCache, key: &str, entry: SharedAppEntry) {
+  if let Some(fetched) = entry.fetched {
+    cache.merge_fetched(key, fetched.latest_release, fetched.etag, fetched.fetched_at);
+  }
+  if let Some(failure) = entry.failure {
+    cache.merge_failure(key, failure.failed_at, failure.consecutive_failures);
+  }
 }

 #[cfg(test)]
 mod tests {
+  use chrono::DateTime;
+
+  use redis_schema::keys::craft_apps::craft_app_release_redis_key::CraftAppReleaseRedisKey;
+
  use crate::threads::poll_craft_app_releases_thread::config::default_apps;
+  use crate::threads::poll_craft_app_releases_thread::shared_entry::{decode_app_entry, encode_fetch};
+  use crate::threads::poll_craft_app_releases_thread::shared_store::build_poller_redis_pool;

  use super::*;

-  fn config() -> CraftAppsPollConfig {
-    CraftAppsPollConfig {
-      enabled: true,
-      interval: Duration::from_secs(1800),
-      max_backoff: Duration::from_secs(14_400),
-      github_token: None,
-      apps: default_apps(),
-    }
-  }
-
-  fn within(d: Duration, around: Duration) -> bool {
-    d >= around.mul_f64(0.89) && d <= around.mul_f64(1.11)
-  }
-
-  /// Live: one real round against GitHub (two apps), then the endpoint's response, written to
-  /// `CRAFT_APPS_SAMPLE_OUT` when set (ArtCraft Launcher's parser reads it back).
+  /// Live: real rounds against GitHub (two apps), then the endpoint's response, written to
+  /// `CRAFT_APPS_SAMPLE_OUT` when set (ArtCraft Launcher's parser reads it back). Uses
+  /// `CRAFT_APPS_RELEASE_GITHUB_TOKEN` if set, and works without it.
  /// `cargo test -p storyteller-web --bin storyteller-web live_round -- --ignored`
  #[tokio::test]
  #[ignore = "needs network access"]
  async fn live_round_fetches_real_releases() {
    use artcraft_api_defs::craft_apps::release_info::{CraftAppsImportantNotices, GetCraftAppsReleaseInfoResponse};
    let mut config = config();
+    config.github_token = std::env::var("CRAFT_APPS_RELEASE_GITHUB_TOKEN").ok().filter(|t| !t.trim().is_empty());
    config.apps.retain(|(key, _, _)| key == "photocraft" || key == "artcraft-launcher");
    let cache = CraftAppsReleaseCache::new();
    cache.register_apps(&config.apps);
-    let client = github::build_client().unwrap();
-    let round = poll_round(&client, &cache, &config).await;
+    let mut poller = Poller::new(github::build_client().unwrap(), cache.clone(), config.clone(), None);
+    let due: Vec<(String, String)> = config.apps.iter().map(|(k, _, r)| (k.clone(), r.clone())).collect();
+    let (round, _) = poller.poll_round(&due, PollMode::Standalone).await;
    assert_eq!(round.fetched + round.unchanged, 2, "{round:?}");
    // A second round revalidates with ETags.
-    let again = poll_round(&client, &cache, &config).await;
+    let (again, _) = poller.poll_round(&due, PollMode::Standalone).await;
    assert_eq!(again.fetched + again.unchanged, 2, "{again:?}");
    let info = cache.snapshot();
    assert!(info.last_fetched.is_some());
@@ -190,40 +333,137 @@ mod tests {
    }
  }

-  #[test]
-  fn a_good_round_waits_the_interval_and_resets_trouble() {
-    let mut trouble = 3;
-    let round = RoundResult { fetched: 5, unchanged: 8, ..RoundResult::default() };
-    assert!(within(next_wait(&config(), &round, &mut trouble, Utc::now()), Duration::from_secs(1800)));
-    assert_eq!(trouble, 0);
+  /// Against a real Redis (`CRAFT_APPS_TEST_REDIS_URL`, default `redis://127.0.0.1:6379/`): an
+  /// instance takes another's fresh fetches from Redis instead of asking GitHub.
+  /// `cargo test -p storyteller-web --bin storyteller-web takes_fresh -- --ignored`
+  #[tokio::test]
+  #[ignore = "needs a Redis server"]
+  async fn takes_fresh_releases_from_redis_instead_of_github() {
+    let url = std::env::var("CRAFT_APPS_TEST_REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/".to_string());
+    let pool = build_poller_redis_pool(&url).unwrap();
+    let mut config = config();
+    let prefix = format!("test-craft-{}", rand::random::<u32>());
+    config.apps = vec![(format!("{prefix}-a"), "A".into(), "storytold/a".into()), (format!("{prefix}-b"), "B".into(), "storytold/b".into())];
+    let pod_a = SharedStore::new(pool.clone(), "pod-a".into());
+    let now = Utc::now();
+    for (key, _, _) in &config.apps {
+      assert!(pod_a.write_fetched(key, None, Some("\"e\""), now).await.unwrap());
+    }
+
+    let cache = CraftAppsReleaseCache::new();
+    cache.register_apps(&config.apps);
+    let mut pod_b = Poller::new(github::build_client().unwrap(), cache.clone(), config.clone(), Some(SharedStore::new(pool.clone(), "pod-b".into())));
+    let report = pod_b.tick().await;
+    assert_eq!(report.mode, PollMode::Coordinated);
+    assert_eq!(report.round.requests(), 0, "fresh in Redis, so GitHub isn't asked");
+    for (key, _, _) in &config.apps {
+      assert_eq!(fetched_at(&cache, key).map(|t| t.timestamp_millis()), Some(now.timestamp_millis()));
+      assert_eq!(cache.etag(key).as_deref(), Some("\"e\""));
+    }
+
+    let keys: Vec<String> = config.apps.iter().map(|(k, _, _)| CraftAppReleaseRedisKey::new_for_app(k).to_string()).collect();
+    let mut conn = pool.get().unwrap();
+    redis::cmd("DEL").arg(&keys).query::<()>(&mut *conn).unwrap();
+  }
+
+  /// Against a real Redis, as above: a new instance serves the fleet's releases right after
+  /// startup, before its startup jitter ends, without asking GitHub.
+  /// `cargo test -p storyteller-web --bin storyteller-web warm -- --ignored`
+  #[tokio::test]
+  #[ignore = "needs a Redis server"]
+  async fn a_new_instance_is_warm_from_redis_at_startup() {
+    let url = std::env::var("CRAFT_APPS_TEST_REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/".to_string());
+    let pool = build_poller_redis_pool(&url).unwrap();
+    let mut config = config();
+    let prefix = format!("test-craft-{}", rand::random::<u32>());
+    config.apps = vec![(format!("{prefix}-a"), "A".into(), "storytold/a".into()), (format!("{prefix}-b"), "B".into(), "storytold/b".into())];
+    let fetched_by_another_pod = Utc::now();
+    let other_pod = SharedStore::new(pool.clone(), "other-pod".into());
+    for (key, _, _) in &config.apps {
+      assert!(other_pod.write_fetched(key, None, Some("\"e\""), fetched_by_another_pod).await.unwrap());
+    }
+
+    let cache = CraftAppsReleaseCache::new();
+    let poller = tokio::spawn(poll_craft_app_releases(cache.clone(), config.clone(), pool.clone(), "new-pod".into()));
+    let deadline = Instant::now() + Duration::from_secs(2);
+    while config.apps.iter().any(|(key, _, _)| fetched_at(&cache, key).is_none()) && Instant::now() < deadline {
+      tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    poller.abort();
+
+    for (key, _, _) in &config.apps {
+      assert_eq!(fetched_at(&cache, key).map(|t| t.timestamp_millis()), Some(fetched_by_another_pod.timestamp_millis()), "{}", key);
+    }
+    let keys: Vec<String> = config.apps.iter().map(|(k, _, _)| CraftAppReleaseRedisKey::new_for_app(k).to_string()).collect();
+    let mut conn = pool.get().unwrap();
+    redis::cmd("DEL").arg(&keys).query::<()>(&mut *conn).unwrap();
+  }
+
+  #[tokio::test]
+  async fn an_unreachable_redis_falls_back_to_polling_alone() {
+    let pool = build_poller_redis_pool("redis://127.0.0.1:1/").unwrap();
+    let mut poller = fresh_poller(Some(SharedStore::new(pool, "pod".into())));
+    let report = poller.tick().await;
+    assert_eq!(report.mode, PollMode::Standalone);
+    assert_eq!(report.round.requests(), 0, "everything was fresh");
+    assert!(poller.redis_log.is_failing());
+  }
+
+  #[tokio::test]
+  async fn with_coordination_off_redis_is_never_used() {
+    let mut poller = fresh_poller(None);
+    let report = poller.tick().await;
+    assert_eq!(report.mode, PollMode::Standalone);
+    assert!(!poller.redis_log.is_failing());
+  }
+
+  #[tokio::test]
+  async fn an_active_backoff_skips_github_even_when_apps_are_due() {
+    let mut poller = Poller::new(github::build_client().unwrap(), registered_cache(), config(), None);
+    poller.backoff = Backoff { until: Some(Utc::now() + chrono::Duration::minutes(5)), trouble: 1 };
+    let report = poller.tick().await;
+    assert_eq!(report.round.requests(), 0);
+    assert_eq!(fetched_at(&poller.cache, "photocraft"), None);
  }

  #[test]
-  fn rate_limits_wait_for_the_reset_and_back_off_further_each_time() {
-    let now = Utc::now();
-    let mut trouble = 0;
-    // GitHub says come back in 50 minutes: wait that (plus margin), more than the interval.
-    let round = RoundResult { rate_limited: Some(Some(now + chrono::Duration::minutes(50))), ..RoundResult::default() };
-    assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(51 * 60)));
-    // Again and again, with no reset time: 2×, 4×, … the interval, capped at the maximum.
-    let round = RoundResult { rate_limited: Some(None), ..RoundResult::default() };
-    assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(3600)));
-    assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(7200)));
-    for _ in 0..10 {
-      assert!(next_wait(&config(), &round, &mut trouble, now) <= Duration::from_secs(14_400).mul_f64(1.11));
+  fn shared_entries_merge_into_memory() {
+    let cache = registered_cache();
+    let at = Utc::now();
+    let fields = encode_fetch(None, Some("\"e\""), at, "pod").unwrap();
+    let map = fields.into_iter().map(|(k, v)| (k.to_string(), v)).collect();
+    merge_shared_entry(&cache, "photocraft", decode_app_entry(&map).unwrap());
+    assert_eq!(fetched_at(&cache, "photocraft").map(|t| t.timestamp_millis()), Some(at.timestamp_millis()));
+    assert_eq!(cache.etag("photocraft").as_deref(), Some("\"e\""));
+  }
+
+  fn config() -> CraftAppsPollConfig {
+    CraftAppsPollConfig {
+      enabled: true,
+      coordination_enabled: true,
+      interval: Duration::from_secs(1800),
+      max_backoff: Duration::from_secs(14_400),
+      github_token: None,
+      apps: default_apps(),
    }
  }

-  #[test]
-  fn a_round_where_everything_failed_retries_sooner_but_not_in_a_tight_loop() {
-    let now = Utc::now();
-    let mut trouble = 0;
-    let round = RoundResult { failed: 13, ..RoundResult::default() };
-    assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(60)));
-    assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(120)));
-    for _ in 0..20 {
-      let w = next_wait(&config(), &round, &mut trouble, now);
-      assert!(w >= Duration::from_secs(53) && w <= Duration::from_secs(1800).mul_f64(1.11), "{w:?}");
+  fn registered_cache() -> CraftAppsReleaseCache {
+    let cache = CraftAppsReleaseCache::new();
+    cache.register_apps(&config().apps);
+    cache
+  }
+
+  /// A poller whose apps were all just fetched, so a tick makes no GitHub requests.
+  fn fresh_poller(store: Option<SharedStore>) -> Poller {
+    let cache = registered_cache();
+    for (key, _, _) in &config().apps {
+      cache.store(key, None, None, Utc::now());
    }
+    Poller::new(github::build_client().unwrap(), cache, config(), store)
+  }
+
+  fn fetched_at(cache: &CraftAppsReleaseCache, app_key: &str) -> Option<DateTime<Utc>> {
+    cache.app(app_key).and_then(|app| app.app_last_fetched)
  }
 }
@@ -0,0 +1,286 @@
+//! When to ask GitHub: per-app freshness, retries after failures, and the backoff after rate
+//! limits. Pure functions of what's known, so the same rules apply whether this instance works
+//! from the fleet's shared state in Redis or (as a fallback) only from its own memory.
+
+use std::time::Duration;
+
+use chrono::{DateTime, Utc};
+
+use crate::state::craft_apps_release_cache::CachedCraftApp;
+use crate::threads::poll_craft_app_releases_thread::config::CraftAppsPollConfig;
+
+/// How often every instance wakes up to look at the shared state (with jitter).
+pub const TICK: Duration = Duration::from_secs(30);
+/// Allowance for clocks that differ between instances: an app fetched by another instance
+/// counts as fresh until `interval - CLOCK_SKEW_MARGIN` has passed.
+pub const CLOCK_SKEW_MARGIN: Duration = Duration::from_secs(30);
+/// First wait after a failed request (or a round where every request failed); doubles up to the
+/// normal interval.
+const FIRST_FAILURE_RETRY: Duration = Duration::from_secs(60);
+/// Extra margin after GitHub's announced rate-limit reset.
+const RESET_MARGIN: Duration = Duration::from_secs(60);
+
+/// What one round did.
+#[derive(Debug, Default)]
+pub struct RoundResult {
+  pub fetched: usize,
+  pub unchanged: usize,
+  pub failed: usize,
+  /// GitHub refused; `Some(Some(t))` when it said when to come back.
+  pub rate_limited: Option<Option<DateTime<Utc>>>,
+}
+
+/// Don't ask GitHub before `until`. `trouble` counts the rounds in a row that were rate limited
+/// or failed entirely; each one makes the next wait longer.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct Backoff {
+  pub until: Option<DateTime<Utc>>,
+  pub trouble: u32,
+}
+
+impl RoundResult {
+  /// How many requests reached a conclusion (a rate limit counts as one).
+  pub fn requests(&self) -> usize {
+    self.fetched + self.unchanged + self.failed + usize::from(self.rate_limited.is_some())
+  }
+}
+
+impl Backoff {
+  pub fn is_active(&self, now: DateTime<Utc>) -> bool {
+    self.until.is_some_and(|until| until > now)
+  }
+}
+
+/// Whether `app` should be requested from GitHub now. An app is due once its last fetch (by any
+/// instance) is `interval - CLOCK_SKEW_MARGIN` old. After a failure it waits a minute, then two,
+/// four, … up to the interval, so one broken repository is never retried every tick.
+pub fn is_due(app: &CachedCraftApp, now: DateTime<Utc>, interval: Duration) -> bool {
+  if let Some(fetched) = app.app_last_fetched {
+    if age(fetched, now) < interval.saturating_sub(CLOCK_SKEW_MARGIN) {
+      return false;
+    }
+  }
+  if app.is_failing() {
+    if let Some(failed) = app.last_failed_at {
+      let retry = scaled(FIRST_FAILURE_RETRY, app.consecutive_failures).min(interval);
+      if age(failed, now) < retry.saturating_sub(CLOCK_SKEW_MARGIN) {
+        return false;
+      }
+    }
+  }
+  true
+}
+
+/// The backoff after a round that made at least one request, starting from `previous`. A good
+/// round clears it.
+pub fn backoff_after(config: &CraftAppsPollConfig, round: &RoundResult, previous: Backoff, now: DateTime<Utc>) -> Backoff {
+  let mut trouble = previous.trouble;
+  let wait = next_wait(config, round, &mut trouble, now);
+  if trouble == 0 {
+    return Backoff::default();
+  }
+  let wait = chrono::Duration::from_std(wait).unwrap_or(chrono::Duration::MAX);
+  Backoff { until: now.checked_add_signed(wait), trouble }
+}
+
+/// How long to wait before the next round. Rate limits wait for GitHub's reset and back off
+/// further each time; rounds where everything failed retry sooner, but never faster than a
+/// minute, doubling up to the normal interval.
+fn next_wait(config: &CraftAppsPollConfig, round: &RoundResult, consecutive_trouble: &mut u32, now: DateTime<Utc>) -> Duration {
+  let base = if let Some(retry_at) = round.rate_limited {
+    *consecutive_trouble = consecutive_trouble.saturating_add(1);
+    let until_reset = retry_at
+        .and_then(|t| (t - now).to_std().ok())
+        .map(|d| d + RESET_MARGIN)
+        .unwrap_or(Duration::ZERO);
+    let backoff = scaled(config.interval, *consecutive_trouble);
+    until_reset.max(backoff).min(config.max_backoff)
+  } else if round.fetched == 0 && round.unchanged == 0 && round.failed > 0 {
+    *consecutive_trouble = consecutive_trouble.saturating_add(1);
+    scaled(FIRST_FAILURE_RETRY, *consecutive_trouble).min(config.interval)
+  } else {
+    *consecutive_trouble = 0;
+    config.interval
+  };
+  jittered(base, 0.1)
+}
+
+/// `base` ± `fraction`, so a fleet doesn't poll in lockstep. With `fraction == 1.0`, anywhere in
+/// `0..=2×base`; used for the startup spread.
+pub fn jittered(base: Duration, fraction: f64) -> Duration {
+  let factor = 1.0 + fraction * (rand::random::<f64>() * 2.0 - 1.0);
+  base.mul_f64(factor.max(0.0))
+}
+
+/// `base × 2^(n-1)`, without overflow.
+fn scaled(base: Duration, n: u32) -> Duration {
+  base.saturating_mul(1u32 << n.saturating_sub(1).min(16))
+}
+
+/// How long ago `then` was; zero if it is in the future (another instance's clock runs ahead).
+fn age(then: DateTime<Utc>, now: DateTime<Utc>) -> Duration {
+  (now - then).to_std().unwrap_or(Duration::ZERO)
+}
+
+#[cfg(test)]
+mod tests {
+  use crate::threads::poll_craft_app_releases_thread::config::default_apps;
+
+  use super::*;
+
+  mod freshness {
+    use super::*;
+
+    #[test]
+    fn an_app_never_fetched_is_due() {
+      assert!(is_due(&app(None, None, 0), Utc::now(), Duration::from_secs(120)));
+    }
+
+    #[test]
+    fn a_fetch_counts_as_fresh_until_the_interval_minus_the_skew_margin() {
+      let now = Utc::now();
+      let interval = Duration::from_secs(120);
+      assert!(!is_due(&app(Some(now - secs(89)), None, 0), now, interval));
+      assert!(is_due(&app(Some(now - secs(90)), None, 0), now, interval));
+      assert!(is_due(&app(Some(now - secs(600)), None, 0), now, interval));
+    }
+
+    #[test]
+    fn a_fetch_from_a_clock_ahead_of_ours_is_fresh() {
+      let now = Utc::now();
+      assert!(!is_due(&app(Some(now + secs(20)), None, 0), now, Duration::from_secs(120)));
+    }
+
+    #[test]
+    fn a_failing_app_waits_longer_after_each_failure_up_to_the_interval() {
+      let now = Utc::now();
+      let interval = Duration::from_secs(1800);
+      // 1st failure: retry after 60 s (less the margin); 3rd: 240 s.
+      assert!(!is_due(&app(None, Some(now - secs(29)), 1), now, interval));
+      assert!(is_due(&app(None, Some(now - secs(30)), 1), now, interval));
+      assert!(!is_due(&app(None, Some(now - secs(200)), 3), now, interval));
+      assert!(is_due(&app(None, Some(now - secs(210)), 3), now, interval));
+      // Many failures: never longer than the interval.
+      assert!(is_due(&app(None, Some(now - secs(1770)), 30), now, interval));
+    }
+
+    #[test]
+    fn a_failure_older_than_the_last_fetch_is_ignored() {
+      let now = Utc::now();
+      let app = app(Some(now - secs(200)), Some(now - secs(300)), 5);
+      assert!(is_due(&app, now, Duration::from_secs(120)));
+    }
+  }
+
+  mod backoff {
+    use super::*;
+
+    #[test]
+    fn a_good_round_clears_the_backoff() {
+      let now = Utc::now();
+      let round = RoundResult { fetched: 1, unchanged: 3, failed: 1, ..RoundResult::default() };
+      let previous = Backoff { until: Some(now), trouble: 3 };
+      assert_eq!(backoff_after(&config(), &round, previous, now), Backoff::default());
+    }
+
+    #[test]
+    fn a_rate_limit_backs_off_the_fleet_until_after_the_reset() {
+      let now = Utc::now();
+      let round = RoundResult { rate_limited: Some(Some(now + chrono::Duration::minutes(50))), ..RoundResult::default() };
+      let b = backoff_after(&config(), &round, Backoff::default(), now);
+      assert_eq!(b.trouble, 1);
+      assert!(b.is_active(now + chrono::Duration::minutes(45)));
+      assert!(!b.is_active(now + chrono::Duration::minutes(57)));
+    }
+
+    #[test]
+    fn trouble_carries_over_between_rounds_and_instances() {
+      let now = Utc::now();
+      let round = RoundResult { failed: 2, ..RoundResult::default() };
+      let first = backoff_after(&config(), &round, Backoff::default(), now);
+      let second = backoff_after(&config(), &round, first, now);
+      assert_eq!((first.trouble, second.trouble), (1, 2));
+      assert!(second.until.unwrap() - now >= chrono::Duration::seconds(107), "{:?}", second);
+    }
+
+    #[test]
+    fn an_expired_backoff_is_not_active() {
+      let now = Utc::now();
+      assert!(!Backoff::default().is_active(now));
+      assert!(!Backoff { until: Some(now - secs(1)), trouble: 4 }.is_active(now));
+      assert!(Backoff { until: Some(now + secs(1)), trouble: 4 }.is_active(now));
+    }
+  }
+
+  mod waits {
+    use super::*;
+
+    #[test]
+    fn a_good_round_waits_the_interval_and_resets_trouble() {
+      let mut trouble = 3;
+      let round = RoundResult { fetched: 5, unchanged: 8, ..RoundResult::default() };
+      assert!(within(next_wait(&config(), &round, &mut trouble, Utc::now()), Duration::from_secs(1800)));
+      assert_eq!(trouble, 0);
+    }
+
+    #[test]
+    fn rate_limits_wait_for_the_reset_and_back_off_further_each_time() {
+      let now = Utc::now();
+      let mut trouble = 0;
+      // GitHub says come back in 50 minutes: wait that (plus margin), more than the interval.
+      let round = RoundResult { rate_limited: Some(Some(now + chrono::Duration::minutes(50))), ..RoundResult::default() };
+      assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(51 * 60)));
+      // Again and again, with no reset time: 2×, 4×, … the interval, capped at the maximum.
+      let round = RoundResult { rate_limited: Some(None), ..RoundResult::default() };
+      assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(3600)));
+      assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(7200)));
+      for _ in 0..10 {
+        assert!(next_wait(&config(), &round, &mut trouble, now) <= Duration::from_secs(14_400).mul_f64(1.11));
+      }
+    }
+
+    #[test]
+    fn a_round_where_everything_failed_retries_sooner_but_not_in_a_tight_loop() {
+      let now = Utc::now();
+      let mut trouble = 0;
+      let round = RoundResult { failed: 13, ..RoundResult::default() };
+      assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(60)));
+      assert!(within(next_wait(&config(), &round, &mut trouble, now), Duration::from_secs(120)));
+      for _ in 0..20 {
+        let w = next_wait(&config(), &round, &mut trouble, now);
+        assert!(w >= Duration::from_secs(53) && w <= Duration::from_secs(1800).mul_f64(1.11), "{:?}", w);
+      }
+    }
+  }
+
+  fn config() -> CraftAppsPollConfig {
+    CraftAppsPollConfig {
+      enabled: true,
+      coordination_enabled: true,
+      interval: Duration::from_secs(1800),
+      max_backoff: Duration::from_secs(14_400),
+      github_token: None,
+      apps: default_apps(),
+    }
+  }
+
+  fn app(fetched: Option<DateTime<Utc>>, failed: Option<DateTime<Utc>>, failures: u32) -> CachedCraftApp {
+    CachedCraftApp {
+      name: "PhotoCraft".into(),
+      repo: "storytold/photocraft".into(),
+      latest_release: None,
+      app_last_fetched: fetched,
+      etag: None,
+      last_failed_at: failed,
+      consecutive_failures: failures,
+    }
+  }
+
+  fn secs(n: i64) -> chrono::Duration {
+    chrono::Duration::seconds(n)
+  }
+
+  fn within(d: Duration, around: Duration) -> bool {
+    d >= around.mul_f64(0.89) && d <= around.mul_f64(1.11)
+  }
+}
@@ -0,0 +1,267 @@
+//! What the fleet shares in Redis about Craft App releases, and how it is encoded. Pure: the
+//! Redis I/O is in `shared_store`.
+//!
+//! ## `craftApps:releases:app:<app key>` (hash, 7 day TTL, refreshed on every write)
+//!
+//! | Field        | Meaning                                                                        |
+//! |--------------|--------------------------------------------------------------------------------|
+//! | `v`          | Format version, `1`. Entries with another version are ignored (and replaced)   |
+//! | `fetched_ms` | When the release was last fetched or confirmed by GitHub (Unix ms)             |
+//! | `etag`       | GitHub's ETag for that answer; empty for none                                  |
+//! | `release`    | The latest release as JSON (`CraftAppRelease`), `null` if the repo has none    |
+//! | `failed_ms`  | When a request for the app last failed (Unix ms); stale once `fetched_ms` newer |
+//! | `failures`   | Failed requests in a row as of `failed_ms`                                     |
+//! | `claimed_ms` | When an instance last started a request for the app (Unix ms), to skip doubles |
+//! | `by`         | Hostname of the instance that wrote last, for debugging                        |
+//!
+//! ## `craftApps:releases:backoff` (hash, 1 day TTL; deleted after a good round)
+//!
+//! | Field      | Meaning                                                         |
+//! |------------|-----------------------------------------------------------------|
+//! | `v`        | Format version, `1`                                             |
+//! | `until_ms` | No instance asks GitHub before this time (Unix ms)              |
+//! | `trouble`  | Rounds in a row that were rate limited or failed entirely       |
+//! | `by`       | Hostname of the instance that wrote last, for debugging         |
+//!
+//! Anything unreadable (wrong version, bad numbers, bad JSON) reads as absent, so it is
+//! refetched and overwritten rather than trusted.
+
+use std::collections::HashMap;
+
+use chrono::{DateTime, Utc};
+
+use artcraft_api_defs::craft_apps::release_info::CraftAppRelease;
+
+use crate::threads::poll_craft_app_releases_thread::scheduling::Backoff;
+
+pub const FORMAT_VERSION: &str = "1";
+
+pub const FIELD_VERSION: &str = "v";
+pub const FIELD_FETCHED_MS: &str = "fetched_ms";
+pub const FIELD_ETAG: &str = "etag";
+pub const FIELD_RELEASE: &str = "release";
+pub const FIELD_FAILED_MS: &str = "failed_ms";
+pub const FIELD_FAILURES: &str = "failures";
+pub const FIELD_CLAIMED_MS: &str = "claimed_ms";
+pub const FIELD_BY: &str = "by";
+pub const FIELD_UNTIL_MS: &str = "until_ms";
+pub const FIELD_TROUBLE: &str = "trouble";
+
+/// One app's shared entry, as read from Redis.
+#[derive(Clone, Debug, Default)]
+pub struct SharedAppEntry {
+  pub fetched: Option<SharedFetch>,
+  pub failure: Option<SharedFailure>,
+}
+
+/// A successful fetch (or `304` confirmation) by some instance.
+#[derive(Clone, Debug)]
+pub struct SharedFetch {
+  pub latest_release: Option<CraftAppRelease>,
+  pub etag: Option<String>,
+  pub fetched_at: DateTime<Utc>,
+}
+
+/// The last failed request by some instance.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct SharedFailure {
+  pub failed_at: DateTime<Utc>,
+  pub consecutive_failures: u32,
+}
+
+/// Reads an app's hash. `None` for an empty hash or another format version.
+pub fn decode_app_entry(fields: &HashMap<String, String>) -> Option<SharedAppEntry> {
+  if fields.get(FIELD_VERSION).map(String::as_str) != Some(FORMAT_VERSION) {
+    return None;
+  }
+  let fetched = decode_fetch(fields);
+  let failure = time_field(fields, FIELD_FAILED_MS).map(|failed_at| SharedFailure {
+    failed_at,
+    consecutive_failures: fields.get(FIELD_FAILURES).and_then(|f| f.parse().ok()).unwrap_or(1).max(1),
+  });
+  Some(SharedAppEntry { fetched, failure })
+}
+
+/// The fields written for a successful fetch.
+pub fn encode_fetch(latest_release: Option<&CraftAppRelease>, etag: Option<&str>, fetched_at: DateTime<Utc>, by: &str) -> Result<Vec<(&'static str, String)>, serde_json::Error> {
+  Ok(vec![
+    (FIELD_VERSION, FORMAT_VERSION.to_string()),
+    (FIELD_FETCHED_MS, fetched_at.timestamp_millis().to_string()),
+    (FIELD_ETAG, etag.unwrap_or_default().to_string()),
+    (FIELD_RELEASE, serde_json::to_string(&latest_release)?),
+    (FIELD_BY, by.to_string()),
+  ])
+}
+
+/// The fields written for a `304`: only the time moves.
+pub fn encode_unchanged(fetched_at: DateTime<Utc>, by: &str) -> Vec<(&'static str, String)> {
+  vec![
+    (FIELD_VERSION, FORMAT_VERSION.to_string()),
+    (FIELD_FETCHED_MS, fetched_at.timestamp_millis().to_string()),
+    (FIELD_BY, by.to_string()),
+  ]
+}
+
+/// The fields written for a failed request.
+pub fn encode_failure(failure: SharedFailure, by: &str) -> Vec<(&'static str, String)> {
+  vec![
+    (FIELD_VERSION, FORMAT_VERSION.to_string()),
+    (FIELD_FAILED_MS, failure.failed_at.timestamp_millis().to_string()),
+    (FIELD_FAILURES, failure.consecutive_failures.to_string()),
+    (FIELD_BY, by.to_string()),
+  ]
+}
+
+/// Reads the backoff hash; anything absent or unreadable is no backoff.
+pub fn decode_backoff(fields: &HashMap<String, String>) -> Backoff {
+  if fields.get(FIELD_VERSION).map(String::as_str) != Some(FORMAT_VERSION) {
+    return Backoff::default();
+  }
+  Backoff {
+    until: time_field(fields, FIELD_UNTIL_MS),
+    trouble: fields.get(FIELD_TROUBLE).and_then(|t| t.parse().ok()).unwrap_or(0),
+  }
+}
+
+pub fn encode_backoff(backoff: Backoff, by: &str) -> Vec<(&'static str, String)> {
+  vec![
+    (FIELD_VERSION, FORMAT_VERSION.to_string()),
+    (FIELD_UNTIL_MS, backoff.until.map(|u| u.timestamp_millis()).unwrap_or(0).to_string()),
+    (FIELD_TROUBLE, backoff.trouble.to_string()),
+    (FIELD_BY, by.to_string()),
+  ]
+}
+
+/// The fetch part of an app's hash; `None` unless the time, the ETag and the release all read.
+fn decode_fetch(fields: &HashMap<String, String>) -> Option<SharedFetch> {
+  let fetched_at = time_field(fields, FIELD_FETCHED_MS)?;
+  let latest_release: Option<CraftAppRelease> = serde_json::from_str(fields.get(FIELD_RELEASE)?).ok()?;
+  let etag = fields.get(FIELD_ETAG)?;
+  Some(SharedFetch {
+    latest_release,
+    etag: (!etag.is_empty()).then(|| etag.clone()),
+    fetched_at,
+  })
+}
+
+/// A Unix-milliseconds field; `None` when absent, unreadable or zero.
+fn time_field(fields: &HashMap<String, String>, name: &str) -> Option<DateTime<Utc>> {
+  let ms = fields.get(name)?.parse::<i64>().ok().filter(|ms| *ms > 0)?;
+  DateTime::from_timestamp_millis(ms)
+}
+
+#[cfg(test)]
+mod tests {
+  use chrono::TimeZone;
+
+  use super::*;
+
+  const BY: &str = "storyteller-web-abc";
+
+  mod app_entries {
+    use super::*;
+
+    #[test]
+    fn a_fetch_round_trips() {
+      let release = release();
+      let at = time();
+      let fields = to_map(encode_fetch(Some(&release), Some("\"e1\""), at, BY).unwrap());
+      let entry = decode_app_entry(&fields).unwrap();
+      let fetched = entry.fetched.unwrap();
+      assert_eq!((fetched.etag.as_deref(), fetched.fetched_at), (Some("\"e1\""), at));
+      assert_eq!(json(&fetched.latest_release), json(&Some(release)));
+      assert_eq!(entry.failure, None);
+    }
+
+    #[test]
+    fn a_repo_without_release_or_etag_round_trips() {
+      let fields = to_map(encode_fetch(None, None, time(), BY).unwrap());
+      let fetched = decode_app_entry(&fields).unwrap().fetched.unwrap();
+      assert!(fetched.latest_release.is_none());
+      assert_eq!(fetched.etag, None);
+    }
+
+    #[test]
+    fn a_304_moves_only_the_time() {
+      let mut fields = to_map(encode_fetch(Some(&release()), Some("\"e1\""), time(), BY).unwrap());
+      let later = time() + chrono::Duration::minutes(2);
+      fields.extend(to_map(encode_unchanged(later, BY)));
+      let fetched = decode_app_entry(&fields).unwrap().fetched.unwrap();
+      assert_eq!((fetched.fetched_at, fetched.etag.as_deref()), (later, Some("\"e1\"")));
+      assert_eq!(json(&fetched.latest_release), json(&Some(release())));
+    }
+
+    #[test]
+    fn a_failure_round_trips_alongside_a_fetch() {
+      let mut fields = to_map(encode_fetch(None, None, time(), BY).unwrap());
+      let failure = SharedFailure { failed_at: time() + chrono::Duration::minutes(1), consecutive_failures: 3 };
+      fields.extend(to_map(encode_failure(failure, BY)));
+      let entry = decode_app_entry(&fields).unwrap();
+      assert_eq!(entry.failure, Some(failure));
+      assert!(entry.fetched.is_some());
+    }
+
+    #[test]
+    fn garbage_reads_as_absent() {
+      assert!(decode_app_entry(&HashMap::new()).is_none());
+      let good = to_map(encode_fetch(Some(&release()), Some("\"e1\""), time(), BY).unwrap());
+      let with = |field: &str, value: &str| {
+        let mut fields = good.clone();
+        fields.insert(field.to_string(), value.to_string());
+        fields
+      };
+      assert!(decode_app_entry(&with(FIELD_VERSION, "2")).is_none(), "unknown version");
+      assert!(decode_app_entry(&with(FIELD_RELEASE, "{not json")).unwrap().fetched.is_none());
+      assert!(decode_app_entry(&with(FIELD_RELEASE, "{\"tag\":1}")).unwrap().fetched.is_none());
+      assert!(decode_app_entry(&with(FIELD_FETCHED_MS, "yesterday")).unwrap().fetched.is_none());
+      assert!(decode_app_entry(&with(FIELD_FAILED_MS, "-5")).unwrap().failure.is_none());
+    }
+  }
+
+  mod backoffs {
+    use super::*;
+
+    #[test]
+    fn a_backoff_round_trips() {
+      let backoff = Backoff { until: Some(time()), trouble: 4 };
+      assert_eq!(decode_backoff(&to_map(encode_backoff(backoff, BY))), backoff);
+    }
+
+    #[test]
+    fn garbage_reads_as_no_backoff() {
+      assert_eq!(decode_backoff(&HashMap::new()), Backoff::default());
+      let mut fields = to_map(encode_backoff(Backoff { until: Some(time()), trouble: 4 }, BY));
+      fields.insert(FIELD_VERSION.into(), "zz".into());
+      assert_eq!(decode_backoff(&fields), Backoff::default());
+      fields.insert(FIELD_VERSION.into(), FORMAT_VERSION.into());
+      fields.insert(FIELD_UNTIL_MS.into(), "soon".into());
+      fields.insert(FIELD_TROUBLE.into(), "-1".into());
+      assert_eq!(decode_backoff(&fields), Backoff::default());
+    }
+  }
+
+  fn time() -> DateTime<Utc> {
+    Utc.with_ymd_and_hms(2026, 10, 10, 12, 0, 0).unwrap()
+  }
+
+  fn release() -> CraftAppRelease {
+    CraftAppRelease {
+      tag: "v0.5.0".into(),
+      version: "0.5.0".into(),
+      name: "PhotoCraft v0.5.0".into(),
+      published_at: Some(time()),
+      html_url: "https://github.com/storytold/photocraft/releases/tag/v0.5.0".into(),
+      prerelease: false,
+      notes_markdown: "Notes".into(),
+      assets: Vec::new(),
+    }
+  }
+
+  fn json(release: &Option<CraftAppRelease>) -> String {
+    serde_json::to_string(release).unwrap()
+  }
+
+  fn to_map(fields: Vec<(&'static str, String)>) -> HashMap<String, String> {
+    fields.into_iter().map(|(k, v)| (k.to_string(), v)).collect()
+  }
+}
@@ -0,0 +1,644 @@
+//! Reads and writes the fleet's shared Craft App release state in Redis (keys and fields are
+//! documented in `shared_entry`).
+//!
+//! Writes are last-write-wins but guarded by time: a Lua script compares a timestamp field
+//! before writing, so a slow instance never replaces a newer fetch with an older one. A `304`
+//! moves only `fetched_ms`, and only if the stored ETag is still the one that was revalidated.
+//!
+//! Some Redis setups refuse scripts (scripting disabled, an ACL without `EVAL`, a proxy). The
+//! first time Redis refuses the script, this store switches to applying the same rule itself:
+//! it `WATCH`es the hash, reads and decodes the guarding fields, decides, and writes in a
+//! `MULTI`/`EXEC` that Redis drops if another instance wrote in between (then it decides again).
+//!
+//! The pool is r2d2 (blocking), so every call runs on tokio's blocking pool with a timeout:
+//! Redis trouble costs the polling loop a few seconds at most, never a stuck thread of the
+//! async runtime. The poller gets its own small pool ([`build_poller_redis_pool`]) whose
+//! connections time out reads and writes, so a hung Redis can neither strand blocking threads
+//! nor take connections from the request handlers' pool.
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, LazyLock};
+use std::time::Duration;
+
+use anyhow::anyhow;
+use chrono::{DateTime, Utc};
+use log::warn;
+use r2d2::{ManageConnection, Pool};
+use redis::{Client, Connection, ConnectionLike, ErrorKind, RedisError, RedisResult, Script, ServerErrorKind, Value};
+
+use artcraft_api_defs::craft_apps::release_info::CraftAppRelease;
+use redis_schema::keys::craft_apps::craft_app_release_redis_key::CraftAppReleaseRedisKey;
+use redis_schema::keys::craft_apps::craft_apps_release_backoff_redis_key::CraftAppsReleaseBackoffRedisKey;
+
+use crate::threads::poll_craft_app_releases_thread::scheduling::Backoff;
+use crate::threads::poll_craft_app_releases_thread::shared_entry;
+use crate::threads::poll_craft_app_releases_thread::shared_entry::{SharedAppEntry, SharedFailure};
+
+/// Connections in the poller's pool: one per instance is all the loop uses at a time.
+const POOL_MAX_SIZE: u32 = 2;
+/// Longest wait for a pooled connection.
+const CONNECTION_TIMEOUT: Duration = Duration::from_secs(3);
+/// Longest wait for one read or write on a connection; after it, the connection gives up
+/// instead of holding its blocking thread until the operating system notices a dead peer.
+const IO_TIMEOUT: Duration = Duration::from_secs(5);
+/// Longest wait for a whole call, connection included.
+const CALL_TIMEOUT: Duration = Duration::from_secs(10);
+/// How long a claim on an app keeps other instances from requesting it too: longer than one
+/// request can take (10 s to connect, 30 s for the answer), plus a margin.
+pub const CLAIM_WINDOW: Duration = Duration::from_secs(60);
+/// Without scripts: how many times a guarded write decides again after another instance wrote
+/// between its read and its write, before giving up (as if it lost the race).
+const WATCH_ATTEMPTS: usize = 3;
+
+/// Writes `ARGV[6..]` (field/value pairs) to the hash `KEYS[1]` and sets its TTL to `ARGV[4]` ms,
+/// unless the hash's time field `ARGV[1]` already holds a value greater than
+/// `ARGV[2] - ARGV[3]` (new time minus a minimum gap; a gap of 0 means "never go back in time"),
+/// or `ARGV[5]` is non-empty and differs from the hash's `etag`. Returns 1 if written, else 0.
+/// Unreadable stored times don't block the write, so garbage gets overwritten.
+const GUARDED_HSET_LUA: &str = r#"
+local current = tonumber(redis.call('HGET', KEYS[1], ARGV[1]))
+if current and current + tonumber(ARGV[3]) > tonumber(ARGV[2]) then
+  return 0
+end
+if ARGV[5] ~= '' and redis.call('HGET', KEYS[1], 'etag') ~= ARGV[5] then
+  return 0
+end
+redis.call('HSET', KEYS[1], unpack(ARGV, 6))
+redis.call('PEXPIRE', KEYS[1], ARGV[4])
+return 1
+"#;
+
+static GUARDED_HSET: LazyLock<Script> = LazyLock::new(|| Script::new(GUARDED_HSET_LUA));
+
+/// Handle on the shared state. Cheap to clone.
+#[derive(Clone)]
+pub struct SharedStore {
+  pool: PollerRedisPool,
+  /// Written as `by`, for debugging.
+  hostname: String,
+  /// Set once Redis refuses the guarded-write script; from then on guarded writes use `WATCH`.
+  scripts_refused: Arc<AtomicBool>,
+}
+
+/// The poller's Redis pool; see [`build_poller_redis_pool`].
+pub type PollerRedisPool = Pool<PollerRedisConnections>;
+
+/// Makes the poller's Redis connections, every one of them with I/O timeouts.
+pub struct PollerRedisConnections {
+  client: Client,
+}
+
+/// A guarded write: which time field guards it, and what to write.
+struct GuardedWrite {
+  key: String,
+  time_field: &'static str,
+  time: DateTime<Utc>,
+  min_gap: Duration,
+  required_etag: Option<String>,
+  fields: Vec<(&'static str, String)>,
+}
+
+impl SharedStore {
+  pub fn new(pool: PollerRedisPool, hostname: String) -> Self {
+    Self { pool, hostname, scripts_refused: Arc::new(AtomicBool::new(false)) }
+  }
+
+  /// The shared entries of `app_keys` (in order; `None` for absent or unreadable ones) and the
+  /// fleet's backoff, in one round trip.
+  pub async fn read(&self, app_keys: &[String]) -> anyhow::Result<(Vec<Option<SharedAppEntry>>, Backoff)> {
+    let keys: Vec<String> = app_keys.iter().map(|k| CraftAppReleaseRedisKey::new_for_app(k).to_string()).collect();
+    let mut hashes: Vec<HashMap<String, String>> = self
+        .run(move |conn| {
+          let mut pipe = redis::pipe();
+          for key in &keys {
+            pipe.hgetall(key);
+          }
+          pipe.hgetall(CraftAppsReleaseBackoffRedisKey::new_for_fleet().as_str());
+          pipe.query(conn)
+        })
+        .await?;
+    let backoff = hashes.pop().map(|fields| shared_entry::decode_backoff(&fields)).unwrap_or_default();
+    let entries = hashes.iter().map(shared_entry::decode_app_entry).collect();
+    Ok((entries, backoff))
+  }
+
+  /// Claims the next request for `app_key` at `now`. False when another instance claimed it
+  /// within [`CLAIM_WINDOW`]: it is asking GitHub already, so this one doesn't. Not a lock: it
+  /// is never released and simply lapses, so a crashed instance can't block anyone.
+  pub async fn claim(&self, app_key: &str, now: DateTime<Utc>) -> anyhow::Result<bool> {
+    self.guarded_write(GuardedWrite {
+      key: CraftAppReleaseRedisKey::new_for_app(app_key).to_string(),
+      time_field: shared_entry::FIELD_CLAIMED_MS,
+      time: now,
+      min_gap: CLAIM_WINDOW,
+      required_etag: None,
+      fields: vec![
+        (shared_entry::FIELD_VERSION, shared_entry::FORMAT_VERSION.to_string()),
+        (shared_entry::FIELD_CLAIMED_MS, now.timestamp_millis().to_string()),
+        (shared_entry::FIELD_BY, self.hostname.clone()),
+      ],
+    }).await
+  }
+
+  /// Shares a successful fetch, unless a newer one is already shared. Returns whether written.
+  pub async fn write_fetched(&self, app_key: &str, latest_release: Option<&CraftAppRelease>, etag: Option<&str>, fetched_at: DateTime<Utc>) -> anyhow::Result<bool> {
+    self.guarded_write(GuardedWrite {
+      key: CraftAppReleaseRedisKey::new_for_app(app_key).to_string(),
+      time_field: shared_entry::FIELD_FETCHED_MS,
+      time: fetched_at,
+      min_gap: Duration::ZERO,
+      required_etag: None,
+      fields: shared_entry::encode_fetch(latest_release, etag, fetched_at, &self.hostname)?,
+    }).await
+  }
+
+  /// Shares a `304` for the release with ETag `etag`: moves only the fetch time, and only if
+  /// that release is still the shared one and nothing newer was shared. Returns whether written.
+  pub async fn write_unchanged(&self, app_key: &str, etag: &str, fetched_at: DateTime<Utc>) -> anyhow::Result<bool> {
+    if etag.is_empty() {
+      return Ok(false);
+    }
+    self.guarded_write(GuardedWrite {
+      key: CraftAppReleaseRedisKey::new_for_app(app_key).to_string(),
+      time_field: shared_entry::FIELD_FETCHED_MS,
+      time: fetched_at,
+      min_gap: Duration::ZERO,
+      required_etag: Some(etag.to_string()),
+      fields: shared_entry::encode_unchanged(fetched_at, &self.hostname),
+    }).await
+  }
+
+  /// Shares a failed request, unless a newer failure is already shared. Returns whether written.
+  pub async fn write_failure(&self, app_key: &str, failure: SharedFailure) -> anyhow::Result<bool> {
+    self.guarded_write(GuardedWrite {
+      key: CraftAppReleaseRedisKey::new_for_app(app_key).to_string(),
+      time_field: shared_entry::FIELD_FAILED_MS,
+      time: failure.failed_at,
+      min_gap: Duration::ZERO,
+      required_etag: None,
+      fields: shared_entry::encode_failure(failure, &self.hostname),
+    }).await
+  }
+
+  /// Shares the fleet's backoff; no trouble deletes it.
+  pub async fn write_backoff(&self, backoff: Backoff) -> anyhow::Result<()> {
+    let key = CraftAppsReleaseBackoffRedisKey::new_for_fleet().to_string();
+    if backoff.trouble == 0 {
+      return self.run(move |conn| redis::cmd("DEL").arg(&key).query::<()>(conn)).await;
+    }
+    let fields = shared_entry::encode_backoff(backoff, &self.hostname);
+    let ttl_ms = CraftAppsReleaseBackoffRedisKey::get_redis_ttl().num_milliseconds();
+    self
+        .run(move |conn| {
+          redis::pipe()
+              .hset_multiple(&key, &fields).ignore()
+              .pexpire(&key, ttl_ms).ignore()
+              .query::<()>(conn)
+        })
+        .await
+  }
+
+  /// Applies `write` with the script, or (once Redis has refused the script) with `WATCH`.
+  async fn guarded_write(&self, write: GuardedWrite) -> anyhow::Result<bool> {
+    let ttl_ms = CraftAppReleaseRedisKey::get_redis_ttl().num_milliseconds();
+    let scripts_refused = self.scripts_refused.clone();
+    self
+        .run(move |conn| {
+          if !scripts_refused.load(Ordering::Relaxed) {
+            match guarded_write_with_script(conn, &write, ttl_ms) {
+              Err(err) if is_script_refusal(&err) => {
+                if !scripts_refused.swap(true, Ordering::Relaxed) {
+                  warn!("Craft app releases: Redis refused the guarded-write script ({err}); writing with WATCH/MULTI/EXEC instead.");
+                }
+              }
+              result => return result,
+            }
+          }
+          guarded_write_with_watch(conn, &write, ttl_ms)
+        })
+        .await
+  }
+
+  /// Runs `call` on a pooled connection, off the async runtime, within [`CALL_TIMEOUT`].
+  async fn run<T, F>(&self, call: F) -> anyhow::Result<T>
+  where
+    T: Send + 'static,
+    F: FnOnce(&mut Connection) -> RedisResult<T> + Send + 'static,
+  {
+    let pool = self.pool.clone();
+    let task = tokio::task::spawn_blocking(move || -> anyhow::Result<T> {
+      let mut conn = pool.get_timeout(CONNECTION_TIMEOUT)?;
+      Ok(call(&mut conn)?)
+    });
+    match tokio::time::timeout(CALL_TIMEOUT, task).await {
+      Ok(Ok(result)) => result,
+      Ok(Err(join_err)) => Err(anyhow!("redis task failed: {join_err}")),
+      Err(_) => Err(anyhow!("redis call timed out after {CALL_TIMEOUT:?}")),
+    }
+  }
+}
+
+/// The poller's own Redis pool, apart from the one the request handlers use. Built without
+/// connecting, so an unreachable Redis at startup doesn't stop the server: the poller then
+/// works alone until Redis answers.
+pub fn build_poller_redis_pool(redis_url: &str) -> RedisResult<PollerRedisPool> {
+  let manager = PollerRedisConnections { client: Client::open(redis_url)? };
+  Ok(Pool::builder()
+      .max_size(POOL_MAX_SIZE)
+      .min_idle(Some(0))
+      .connection_timeout(CONNECTION_TIMEOUT)
+      .build_unchecked(manager))
+}
+
+impl ManageConnection for PollerRedisConnections {
+  type Connection = Connection;
+  type Error = RedisError;
+
+  /// Connects (handshake included) within [`CONNECTION_TIMEOUT`], then makes every later read
+  /// and write give up after [`IO_TIMEOUT`], before the connection is ever used.
+  fn connect(&self) -> RedisResult<Connection> {
+    let conn = self.client.get_connection_with_timeout(CONNECTION_TIMEOUT)?;
+    conn.set_read_timeout(Some(IO_TIMEOUT))?;
+    conn.set_write_timeout(Some(IO_TIMEOUT))?;
+    Ok(conn)
+  }
+
+  /// The pool's check before lending a connection; bounded by [`IO_TIMEOUT`] like any command.
+  fn is_valid(&self, conn: &mut Connection) -> RedisResult<()> {
+    redis::cmd("PING").query::<()>(conn)
+  }
+
+  fn has_broken(&self, conn: &mut Connection) -> bool {
+    !conn.is_open()
+  }
+}
+
+/// [`GUARDED_HSET_LUA`] decides and writes on the server, in one step.
+fn guarded_write_with_script(conn: &mut Connection, write: &GuardedWrite, ttl_ms: i64) -> RedisResult<bool> {
+  let mut invocation = GUARDED_HSET.key(&write.key);
+  invocation
+      .arg(write.time_field)
+      .arg(write.time.timestamp_millis())
+      .arg(write.min_gap.as_millis() as u64)
+      .arg(ttl_ms)
+      .arg(write.required_etag.as_deref().unwrap_or_default());
+  for (field, value) in &write.fields {
+    invocation.arg(*field).arg(value);
+  }
+  let written: i64 = invocation.invoke(conn)?;
+  Ok(written == 1)
+}
+
+/// The same rule without scripts: `WATCH` the hash, read and decode its guarding fields, decide
+/// with [`write_wins`], then write in a transaction. If another instance wrote the hash in
+/// between, Redis drops the transaction and this decides again with what it wrote.
+fn guarded_write_with_watch(conn: &mut Connection, write: &GuardedWrite, ttl_ms: i64) -> RedisResult<bool> {
+  for _ in 0..WATCH_ATTEMPTS {
+    redis::cmd("WATCH").arg(&write.key).query::<()>(conn)?;
+    let stored: RedisResult<(Option<String>, Option<String>)> = redis::cmd("HMGET")
+        .arg(&write.key)
+        .arg(write.time_field)
+        .arg(shared_entry::FIELD_ETAG)
+        .query(conn);
+    let (stored_time, stored_etag) = match stored {
+      Ok(stored) => stored,
+      Err(err) => {
+        // The connection goes back to the shared pool; don't leave it watching.
+        let _ = redis::cmd("UNWATCH").query::<()>(conn);
+        return Err(err);
+      }
+    };
+    if !write_wins(write, stored_time.as_deref(), stored_etag.as_deref()) {
+      redis::cmd("UNWATCH").query::<()>(conn)?;
+      return Ok(false);
+    }
+    // `EXEC` ends the watch either way; `None` means another write got in first.
+    let committed: Option<Value> = redis::pipe()
+        .atomic()
+        .hset_multiple(&write.key, &write.fields)
+        .pexpire(&write.key, ttl_ms)
+        .query(conn)?;
+    if committed.is_some() {
+      return Ok(true);
+    }
+  }
+  Ok(false)
+}
+
+/// Whether `write` may replace what the hash holds, by the script's rule: not if the stored time
+/// plus the minimum gap is later than the new time, nor if an ETag is required and the stored
+/// one differs. An unreadable stored time doesn't block the write, so garbage gets overwritten.
+fn write_wins(write: &GuardedWrite, stored_time: Option<&str>, stored_etag: Option<&str>) -> bool {
+  let new_ms = write.time.timestamp_millis();
+  let gap_ms = write.min_gap.as_millis().min(i64::MAX as u128) as i64;
+  if let Some(current_ms) = stored_time.and_then(|t| t.parse::<i64>().ok()) {
+    if current_ms.saturating_add(gap_ms) > new_ms {
+      return false;
+    }
+  }
+  match write.required_etag.as_deref().filter(|etag| !etag.is_empty()) {
+    Some(required) => stored_etag == Some(required),
+    None => true,
+  }
+}
+
+/// Whether Redis answered that it won't run the script (as opposed to being unreachable, busy,
+/// out of memory or read-only, which the `WATCH` path wouldn't get past either): an ACL
+/// refusal, an unknown `EVALSHA`/`EVAL`/`SCRIPT` command, or scripting switched off.
+fn is_script_refusal(err: &RedisError) -> bool {
+  if err.is_io_error() || err.is_timeout() || err.is_connection_dropped() || err.is_connection_refusal() {
+    return false;
+  }
+  match err.kind() {
+    ErrorKind::Server(ServerErrorKind::NoPerm) => true,
+    ErrorKind::Server(ServerErrorKind::ResponseError) | ErrorKind::Extension => {
+      if err.code() == Some("BUSY") {
+        return false;
+      }
+      let detail = err.detail().unwrap_or_default().to_ascii_lowercase();
+      detail.contains("unknown command") || detail.contains("eval") || detail.contains("script")
+    }
+    _ => false,
+  }
+}
+
+
+#[cfg(test)]
+mod tests {
+  use std::io::{Read, Write};
+  use std::net::{TcpListener, TcpStream};
+
+  use chrono::TimeZone;
+  use url::Url;
+
+  use super::*;
+
+  const DEFAULT_TEST_REDIS_URL: &str = "redis://127.0.0.1:6379/";
+
+  /// Against a real Redis (`CRAFT_APPS_TEST_REDIS_URL`, default `redis://127.0.0.1:6379/`). Each
+  /// test uses app keys of its own, deletes them afterwards, and leaves the fleet backoff key
+  /// alone unless it was absent.
+  /// `cargo test -p storyteller-web --bin storyteller-web shared_store -- --ignored`
+  mod live_redis_tests {
+    use super::*;
+
+    #[tokio::test]
+    #[ignore = "needs a Redis server"]
+    async fn guarded_writes_with_the_script() {
+      let pool = test_pool(&test_redis_url());
+      let (pod_a, pod_b) = (SharedStore::new(pool.clone(), "pod-a".into()), SharedStore::new(pool, "pod-b".into()));
+      exercise_guarded_writes(&pod_a, &pod_b).await;
+      assert!(!pod_a.scripts_refused.load(Ordering::Relaxed));
+    }
+
+    #[tokio::test]
+    #[ignore = "needs a Redis server"]
+    async fn guarded_writes_with_watch() {
+      let pool = test_pool(&test_redis_url());
+      let (pod_a, pod_b) = (SharedStore::new(pool.clone(), "pod-a".into()), SharedStore::new(pool, "pod-b".into()));
+      pod_a.scripts_refused.store(true, Ordering::Relaxed);
+      pod_b.scripts_refused.store(true, Ordering::Relaxed);
+      exercise_guarded_writes(&pod_a, &pod_b).await;
+    }
+
+    /// A Redis user without scripting permissions: the first guarded write is refused, and the
+    /// store switches to `WATCH` and still applies every rule.
+    #[tokio::test]
+    #[ignore = "needs a Redis server that supports ACLs"]
+    async fn a_refused_script_falls_back_to_watch() {
+      let admin_url = test_redis_url();
+      let admin = test_pool(&admin_url);
+      let user = format!("craft-test-{}", rand::random::<u32>());
+      let password = format!("pw-{}", rand::random::<u64>());
+      let mut conn = admin.get().unwrap();
+      redis::cmd("ACL").arg("SETUSER").arg(&user).arg("reset").arg("on").arg(format!(">{password}")).arg("~*").arg("&*").arg("+@all").arg("-@scripting")
+          .query::<()>(&mut *conn).unwrap();
+
+      let mut url = Url::parse(&admin_url).unwrap();
+      url.set_username(&user).unwrap();
+      url.set_password(Some(&password)).unwrap();
+      let pool = test_pool(url.as_str());
+      let (pod_a, pod_b) = (SharedStore::new(pool.clone(), "pod-a".into()), SharedStore::new(pool, "pod-b".into()));
+      exercise_guarded_writes(&pod_a, &pod_b).await;
+
+      redis::cmd("ACL").arg("DELUSER").arg(&user).query::<()>(&mut *conn).unwrap();
+      assert!(pod_a.scripts_refused.load(Ordering::Relaxed), "the script should have been refused");
+      assert!(pod_b.scripts_refused.load(Ordering::Relaxed), "the script should have been refused");
+    }
+  }
+
+  /// An unreachable Redis fails fast, so the loop can fall back to polling on its own.
+  #[tokio::test]
+  async fn an_unreachable_redis_fails_fast() {
+    let pool = build_poller_redis_pool("redis://127.0.0.1:1/").unwrap();
+    let store = SharedStore::new(pool, "pod".into());
+    let started = std::time::Instant::now();
+    assert!(store.read(&["photocraft".to_string()]).await.is_err());
+    assert!(started.elapsed() < CALL_TIMEOUT + Duration::from_secs(1));
+  }
+
+  /// A Redis that completes the connection handshake and then stops answering (a wedged server,
+  /// a half-open connection): the call fails by itself, before [`CALL_TIMEOUT`] gives up on it,
+  /// so its blocking thread and connection are released rather than stranded until the
+  /// operating system notices. Without I/O timeouts the pool's `PING` would hang here forever.
+  #[tokio::test]
+  async fn a_redis_that_stops_answering_releases_the_blocking_thread() {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let url = format!("redis://{}/", listener.local_addr().unwrap());
+    std::thread::spawn(move || {
+      for stream in listener.incoming().flatten() {
+        std::thread::spawn(move || answer_handshake_then_go_silent(stream));
+      }
+    });
+    let store = SharedStore::new(build_poller_redis_pool(&url).unwrap(), "pod".into());
+    let started = std::time::Instant::now();
+    let err = store.read(&["photocraft".to_string()]).await.unwrap_err();
+    assert!(started.elapsed() < CALL_TIMEOUT, "{:?} after {:?}", err, started.elapsed());
+  }
+
+  /// [`write_wins`] must decide exactly as [`GUARDED_HSET_LUA`] does.
+  mod write_rule_tests {
+    use super::*;
+
+    #[test]
+    fn a_write_never_goes_back_in_time() {
+      let write = guarded_write_at(t(2), Duration::ZERO, None);
+      assert!(write_wins(&write, None, None), "nothing stored");
+      assert!(write_wins(&write, Some(&ms(t(1))), None), "newer than stored");
+      assert!(write_wins(&write, Some(&ms(t(2))), None), "same time, as the script allows");
+      assert!(!write_wins(&write, Some(&ms(t(3))), None), "older than stored");
+    }
+
+    #[test]
+    fn a_claim_waits_for_the_window() {
+      let claim = guarded_write_at(t(1), CLAIM_WINDOW, None);
+      assert!(!write_wins(&claim, Some(&ms(t(0) + chrono::Duration::seconds(30))), None));
+      assert!(write_wins(&claim, Some(&ms(t(0))), None), "exactly one window later");
+    }
+
+    #[test]
+    fn a_required_etag_must_match() {
+      let write = guarded_write_at(t(2), Duration::ZERO, Some("\"e2\""));
+      assert!(write_wins(&write, Some(&ms(t(1))), Some("\"e2\"")));
+      assert!(!write_wins(&write, Some(&ms(t(1))), Some("\"e1\"")), "another release is shared");
+      assert!(!write_wins(&write, Some(&ms(t(1))), None), "no release is shared");
+      assert!(write_wins(&guarded_write_at(t(2), Duration::ZERO, Some("")), None, None), "empty means none required");
+    }
+
+    #[test]
+    fn an_unreadable_stored_time_is_overwritten() {
+      let write = guarded_write_at(t(2), Duration::ZERO, None);
+      assert!(write_wins(&write, Some("soon"), None));
+      assert!(write_wins(&write, Some(""), None));
+    }
+  }
+
+  mod script_refusal_tests {
+    use redis::make_extension_error;
+
+    use super::*;
+
+    #[test]
+    fn redis_refusing_scripts_is_a_refusal() {
+      let refusals = [
+        server_error(ServerErrorKind::NoPerm, "this user has no permissions to run the 'evalsha' command"),
+        server_error(ServerErrorKind::ResponseError, "unknown command 'EVALSHA', with args beginning with: "),
+        server_error(ServerErrorKind::ResponseError, "This Redis instance has scripting disabled"),
+        make_extension_error("ERR".to_string(), Some("unknown command 'SCRIPT'".to_string())),
+      ];
+      for err in refusals {
+        assert!(is_script_refusal(&err), "{:?}", err);
+      }
+    }
+
+    #[test]
+    fn redis_being_unavailable_is_not_a_refusal() {
+      let unavailable = [
+        RedisError::from(std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "refused")),
+        RedisError::from(std::io::Error::new(std::io::ErrorKind::TimedOut, "timed out")),
+        make_extension_error("OOM".to_string(), Some("command not allowed when used memory > 'maxmemory'.".to_string())),
+        make_extension_error("BUSY".to_string(), Some("Redis is busy running a script.".to_string())),
+        server_error(ServerErrorKind::ReadOnly, "You can't write against a read only replica."),
+        server_error(ServerErrorKind::ResponseError, "wrong number of arguments for 'hset' command"),
+      ];
+      for err in unavailable {
+        assert!(!is_script_refusal(&err), "{:?}", err);
+      }
+    }
+
+    fn server_error(kind: ServerErrorKind, detail: &str) -> RedisError {
+      RedisError::from((ErrorKind::Server(kind), "server error", detail.to_string()))
+    }
+  }
+
+  /// Every guarded-write rule, as two instances sharing one Redis see it.
+  async fn exercise_guarded_writes(pod_a: &SharedStore, pod_b: &SharedStore) {
+    let app = format!("test-craft-{}", rand::random::<u32>());
+    let other = format!("{app}-other");
+    let keys = vec![app.clone(), other.clone()];
+    let read_app = |store: SharedStore, keys: Vec<String>| async move { store.read(&keys).await.unwrap().0.remove(0) };
+
+    // Nothing yet.
+    let (entries, _) = pod_a.read(&keys).await.unwrap();
+    assert!(entries.iter().all(Option::is_none));
+
+    // Two instances tick together: only one gets to ask GitHub; the claim lapses after the window.
+    assert!(pod_a.claim(&app, t(0)).await.unwrap());
+    assert!(!pod_b.claim(&app, t(0) + chrono::Duration::seconds(5)).await.unwrap());
+    assert!(pod_b.claim(&app, t(2)).await.unwrap());
+
+    // A newer fetch is never replaced by an older one.
+    assert!(pod_a.write_fetched(&app, None, Some("\"e2\""), t(2)).await.unwrap());
+    assert!(!pod_b.write_fetched(&app, None, Some("\"e1\""), t(1)).await.unwrap());
+    let fetched = read_app(pod_a.clone(), keys.clone()).await.unwrap().fetched.unwrap();
+    assert_eq!((fetched.etag.as_deref(), fetched.fetched_at), (Some("\"e2\""), t(2)));
+
+    // A 304 moves the time only, and only for the release it revalidated.
+    assert!(!pod_b.write_unchanged(&app, "\"e1\"", t(3)).await.unwrap(), "stale ETag");
+    assert!(!pod_b.write_unchanged(&app, "\"e2\"", t(1)).await.unwrap(), "older than the shared fetch");
+    assert!(pod_b.write_unchanged(&app, "\"e2\"", t(3)).await.unwrap());
+    let fetched = read_app(pod_a.clone(), keys.clone()).await.unwrap().fetched.unwrap();
+    assert_eq!((fetched.etag.as_deref(), fetched.fetched_at), (Some("\"e2\""), t(3)));
+
+    // Failures: newest wins.
+    let failure = SharedFailure { failed_at: t(5), consecutive_failures: 2 };
+    assert!(pod_a.write_failure(&app, failure).await.unwrap());
+    assert!(!pod_b.write_failure(&app, SharedFailure { failed_at: t(4), consecutive_failures: 1 }).await.unwrap());
+    assert_eq!(read_app(pod_a.clone(), keys.clone()).await.unwrap().failure, Some(failure));
+
+    // Garbage in an entry is overwritten, not trusted.
+    let key = CraftAppReleaseRedisKey::new_for_app(&other).to_string();
+    let garbage_key = key.clone();
+    pod_a.run(move |conn| redis::cmd("HSET").arg(&garbage_key).arg("v").arg("1").arg("fetched_ms").arg("soon").arg("release").arg("{").query::<()>(conn)).await.unwrap();
+    assert!(pod_a.read(&keys).await.unwrap().0[1].as_ref().unwrap().fetched.is_none());
+    assert!(pod_a.write_fetched(&other, None, None, t(1)).await.unwrap());
+    assert!(pod_a.read(&keys).await.unwrap().0[1].as_ref().unwrap().fetched.is_some());
+
+    // Entries expire if nobody refreshes them.
+    let ttl_key = key.clone();
+    let ttl: i64 = pod_a.run(move |conn| redis::cmd("PTTL").arg(&ttl_key).query(conn)).await.unwrap();
+    assert!(ttl > 6 * 24 * 3600 * 1000, "{}", ttl);
+
+    // The fleet backoff round-trips (only touched when absent, to leave a real one alone).
+    let (_, existing) = pod_a.read(&keys).await.unwrap();
+    if existing == Backoff::default() {
+      let backoff = Backoff { until: Some(t(30)), trouble: 2 };
+      pod_a.write_backoff(backoff).await.unwrap();
+      assert_eq!(pod_b.read(&keys).await.unwrap().1, backoff);
+      pod_b.write_backoff(Backoff::default()).await.unwrap();
+      assert_eq!(pod_a.read(&keys).await.unwrap().1, Backoff::default());
+    }
+
+    let cleanup: Vec<String> = keys.iter().map(|k| CraftAppReleaseRedisKey::new_for_app(k).to_string()).collect();
+    pod_a.run(move |conn| redis::cmd("DEL").arg(&cleanup).query::<()>(conn)).await.unwrap();
+  }
+
+  fn guarded_write_at(time: DateTime<Utc>, min_gap: Duration, required_etag: Option<&str>) -> GuardedWrite {
+    GuardedWrite {
+      key: "craftApps:releases:app:test".to_string(),
+      time_field: shared_entry::FIELD_FETCHED_MS,
+      time,
+      min_gap,
+      required_etag: required_etag.map(str::to_string),
+      fields: Vec::new(),
+    }
+  }
+
+  /// Answers `+OK` to connection setup (`CLIENT SETINFO`, `SELECT`, `HELLO`), then keeps the
+  /// connection open without another word once a real command (`PING`, `HGETALL`, ...) arrives.
+  fn answer_handshake_then_go_silent(mut stream: TcpStream) {
+    let mut buf = [0u8; 4096];
+    loop {
+      let n = match stream.read(&mut buf) {
+        Ok(0) | Err(_) => return,
+        Ok(n) => n,
+      };
+      let chunk = String::from_utf8_lossy(&buf[..n]).to_ascii_uppercase();
+      if !(chunk.contains("CLIENT") || chunk.contains("SELECT") || chunk.contains("HELLO")) {
+        std::thread::sleep(Duration::from_secs(60));
+        return;
+      }
+      let commands = chunk.lines().filter(|line| line.starts_with('*')).count();
+      if stream.write_all("+OK\r\n".repeat(commands).as_bytes()).is_err() {
+        return;
+      }
+    }
+  }
+
+  fn test_redis_url() -> String {
+    std::env::var("CRAFT_APPS_TEST_REDIS_URL").unwrap_or_else(|_| DEFAULT_TEST_REDIS_URL.to_string())
+  }
+
+  fn test_pool(url: &str) -> PollerRedisPool {
+    build_poller_redis_pool(url).unwrap()
+  }
+
+  fn t(minute: u32) -> DateTime<Utc> {
+    Utc.with_ymd_and_hms(2026, 10, 10, 12, minute, 0).unwrap()
+  }
+
+  fn ms(time: DateTime<Utc>) -> String {
+    time.timestamp_millis().to_string()
+  }
+}