fix(schedule): bound preload startup wait

.github/workflows/ci-backend.yml

@@ -9,6 +9,10 @@ on:
     branches:
       - main
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 permissions:
   contents: read
 

docs/development/recovery-internals.md

@@ -18,6 +18,14 @@ Recovery behavior is designed to preserve committed durability state and reject
 
 `/targetz` is intentionally weaker than data-plane readiness. It can return `200` once the HTTP listener is usable and the process is not draining, even while storage or domain preload is still pending. It is only for a separate orchestration path; a customer-facing ALB target group must use `/healthz`. WebSocket upgrades and TCP sessions still reject data-plane traffic until the strict readiness gate passes.
 
+Schedule preload waits for the actor-owned preload result within the
+`FITZ_SCHEDULE_PRELOAD_TIMEOUT_SECS` startup watchdog. The default 120-second
+deadline replaces the former one-second actor reply deadline while preserving a
+bounded, diagnosable startup failure. Preload logs its configured deadline,
+discovered family count, per-family progress at debug level, elapsed completion
+time, and timeout. Actor failure also disconnects the reply channel so boot
+fails closed before the watchdog expires.
+
 Live session state is never recovered during startup. Notice subscriptions, Stream live subscriptions and append sessions, KV open transactions, Queue inflight ownership tokens, RPC worker registrations and pending calls, Lease ownership, and Schedule subscriptions are rebuilt only by reconnecting clients when their domain contract permits it.
 
 ## Persistent Domain Partial-State Policy

docs/user-guides/vars.md

@@ -131,6 +131,7 @@ For the auth and browser-perimeter checklist, see
 | FITZ_STORAGE_CACHE_PATH | Filesystem path | ./.fitz-cloud-cache | Local cache path for cloud-backed storage mode. |
 | FITZ_STORAGE_CLOUD_DURABILITY | background or strict | background | Cloud policy for broker-selected durable writes and client sync intent. `background` completes at Midge's local cloud commit barrier and uploads asynchronously; `strict` waits for provider acknowledgement. |
 | FITZ_STORAGE_MEMTABLE_BYTES | Unsigned integer byte count | Auto | Optional explicit memtable size override for embedded engine. |
+| FITZ_SCHEDULE_PRELOAD_TIMEOUT_SECS | Positive integer second count | 120 | Maximum aggregate wait for required Schedule actor preload during startup. Expiry fails startup with an explicit timeout rather than leaving the broker wedged indefinitely. |
 | FITZ_QUEUE_WRITE_POLICY | fast, buffered, or strict | fast | Queue mutation write policy. `fast` skips WAL and flushes in the background; `buffered` uses local buffered WAL or cloud asynchronous durability; `strict` waits for local sync or cloud provider acknowledgement. |
 | FITZ_QUEUE_LOSS_WINDOW_MS | Positive integer millisecond count | 100 | Target background flush interval for fast queue writes. Accepted recent queue mutations can be lost before this window closes. |
 | FITZ_KV_IDLE_TRANSACTION_TTL_SECS | Positive integer second count | 300 | Maximum inactivity for an open KV transaction before the broker force-rolls it back and releases its broker-local resource lock. |

src/api/admin/auth.rs

@@ -465,8 +465,7 @@ impl AdminAuth {
             AdminRouteFamilyAccess::Explicit(values) => values.iter().all(|value| {
                 value
                     .parse::<u32>()
-                    .ok()
-                    .is_some_and(|family| provisioned.contains(&family))
+                    .is_ok_and(|family| provisioned.contains(&family))
             }),
         }
     }

src/boot/domains.rs

@@ -421,6 +421,7 @@ pub struct DomainSetupOptions {
     pub rpc_request_timeout: Option<std::time::Duration>,
     pub stream_storage_layout: crate::domains::stream::StreamStorageLayout,
     pub kv_idle_transaction_ttl: std::time::Duration,
+    pub schedule_preload_timeout: std::time::Duration,
 }
 
 fn provisioned_route_families(options: &DomainSetupOptions) -> Vec<RouteFamily> {
@@ -546,7 +547,7 @@ pub fn setup(
     );
     register_domain_sink(DomainKind::Schedule, router, schedule_sink.clone());
     schedule_sink
-        .preload_persisted_families()
+        .preload_persisted_families_with_timeout(options.schedule_preload_timeout)
         .map_err(|error| format!("schedule preload failed: {error}"))?;
     tracing::info!(
         "All {} domain sinks registered with router",
@@ -595,6 +596,8 @@ mod tests {
             rpc_request_timeout: None,
             stream_storage_layout: crate::domains::stream::StreamStorageLayout::default(),
             kv_idle_transaction_ttl: std::time::Duration::from_mins(5),
+            schedule_preload_timeout:
+                crate::domains::schedule::sink::DEFAULT_SCHEDULE_PRELOAD_TIMEOUT,
         }
     }
 
@@ -611,6 +614,8 @@ mod tests {
             rpc_request_timeout: None,
             stream_storage_layout: crate::domains::stream::StreamStorageLayout::default(),
             kv_idle_transaction_ttl: std::time::Duration::from_mins(5),
+            schedule_preload_timeout:
+                crate::domains::schedule::sink::DEFAULT_SCHEDULE_PRELOAD_TIMEOUT,
         }
     }
 

src/boot/mod.rs

@@ -211,6 +211,7 @@ fn register_domains_stage(
         rpc_request_timeout: None,
         stream_storage_layout: config.stream_storage_layout,
         kv_idle_transaction_ttl: Duration::from_secs(config.kv_idle_transaction_ttl_seconds),
+        schedule_preload_timeout: config.schedule_preload_timeout(),
     };
     match domains::setup(router, store, &runtime.admin_read_model(), &options) {
         Ok(handles) => BootStage::Continue(handles),

src/boot/runtime/config.rs

@@ -15,6 +15,7 @@ const ENV_STORAGE_MEMTABLE_BYTES: &str = "FITZ_STORAGE_MEMTABLE_BYTES";
 const ENV_QUEUE_WRITE_POLICY: &str = "FITZ_QUEUE_WRITE_POLICY";
 const ENV_QUEUE_LOSS_WINDOW_MS: &str = "FITZ_QUEUE_LOSS_WINDOW_MS";
 const ENV_KV_IDLE_TRANSACTION_TTL_SECS: &str = "FITZ_KV_IDLE_TRANSACTION_TTL_SECS";
+const ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS: &str = "FITZ_SCHEDULE_PRELOAD_TIMEOUT_SECS";
 const ENV_DRAIN_GRACE_SECONDS: &str = "FITZ_DRAIN_GRACE_SECONDS";
 const ENV_DRAIN_CLOSE_REASON: &str = "FITZ_DRAIN_CLOSE_REASON";
 const DEFAULT_QUEUE_LOSS_WINDOW_MS: u64 = 100;
@@ -327,6 +328,7 @@ mod env;
 use env::{
     drain_close_reason_from_env, drain_grace_seconds_from_env, env_non_empty,
     kv_idle_transaction_ttl_seconds_from_env, queue_loss_window_ms_from_env, required_env,
+    schedule_preload_timeout_seconds_from_env,
 };
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -434,6 +436,14 @@ impl<'a> StorageConfig<'a> {
                 format!("{ENV_KV_IDLE_TRANSACTION_TTL_SECS} must be greater than 0").into(),
             );
         }
+        if let Some(error) = &config.schedule_preload_timeout_error {
+            return Err(error.clone().into());
+        }
+        if config.schedule_preload_timeout_seconds == 0 {
+            return Err(
+                format!("{ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS} must be greater than 0").into(),
+            );
+        }
         config
             .storage_memtable
             .validate()
@@ -519,6 +529,9 @@ pub struct BootConfig {
     /// Maximum inactivity before an open KV transaction is force-rolled back.
     pub kv_idle_transaction_ttl_seconds: u64,
     pub(crate) kv_idle_transaction_ttl_error: Option<String>,
+    /// Maximum wait for required Schedule preload during broker startup.
+    pub schedule_preload_timeout_seconds: u64,
+    pub(crate) schedule_preload_timeout_error: Option<String>,
     /// Whether an external TLS terminator is explicitly protecting public listeners.
     pub assume_external_tls: bool,
     pub(crate) local_listener_exposure: LocalListenerExposure,
@@ -599,6 +612,11 @@ impl BootConfig {
             .then(|| Duration::from_millis(self.queue_loss_window_ms))
     }
 
+    #[must_use]
+    pub fn schedule_preload_timeout(&self) -> Duration {
+        Duration::from_secs(self.schedule_preload_timeout_seconds)
+    }
+
     #[must_use]
     pub fn queue_write_policy_defaulted_fast(&self) -> bool {
         self.queue_write_policy_source.is_defaulted()
@@ -656,6 +674,8 @@ impl Default for BootConfig {
         let (queue_loss_window_ms, queue_loss_window_error) = queue_loss_window_ms_from_env();
         let (kv_idle_transaction_ttl_seconds, kv_idle_transaction_ttl_error) =
             kv_idle_transaction_ttl_seconds_from_env();
+        let (schedule_preload_timeout_seconds, schedule_preload_timeout_error) =
+            schedule_preload_timeout_seconds_from_env();
         let (queue_write_policy, queue_write_policy_source) =
             QueueWritePolicy::from_env_with_source();
         let drain_close_reason = drain_close_reason_from_env();
@@ -690,6 +710,8 @@ impl Default for BootConfig {
             queue_loss_window_error,
             kv_idle_transaction_ttl_seconds,
             kv_idle_transaction_ttl_error,
+            schedule_preload_timeout_seconds,
+            schedule_preload_timeout_error,
             assume_external_tls,
             local_listener_exposure,
             ws_allowed_origins,
@@ -800,6 +822,13 @@ impl BootConfig {
         self
     }
 
+    #[must_use]
+    pub fn with_schedule_preload_timeout_seconds(mut self, seconds: u64) -> Self {
+        self.schedule_preload_timeout_seconds = seconds;
+        self.schedule_preload_timeout_error = None;
+        self
+    }
+
     #[must_use]
     pub fn with_drain_close_reason(mut self, reason: impl Into<String>) -> Self {
         self.drain_close_reason = reason.into();

src/boot/runtime/config/env.rs

@@ -1,7 +1,7 @@
 use super::{
     DEFAULT_DRAIN_CLOSE_REASON, DEFAULT_DRAIN_GRACE_SECONDS, DEFAULT_KV_IDLE_TRANSACTION_TTL_SECS,
     DEFAULT_QUEUE_LOSS_WINDOW_MS, ENV_DRAIN_CLOSE_REASON, ENV_DRAIN_GRACE_SECONDS,
-    ENV_KV_IDLE_TRANSACTION_TTL_SECS, ENV_QUEUE_LOSS_WINDOW_MS,
+    ENV_KV_IDLE_TRANSACTION_TTL_SECS, ENV_QUEUE_LOSS_WINDOW_MS, ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS,
 };
 
 pub(super) fn env_non_empty(key: &str) -> Option<String> {
@@ -84,6 +84,14 @@ pub(super) fn kv_idle_transaction_ttl_seconds_from_env() -> (u64, Option<String>
     }
 }
 
+pub(super) fn schedule_preload_timeout_seconds_from_env() -> (u64, Option<String>) {
+    positive_u64_from_env(
+        ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS,
+        crate::domains::schedule::sink::DEFAULT_SCHEDULE_PRELOAD_TIMEOUT.as_secs(),
+        "second count",
+    )
+}
+
 pub(super) fn drain_close_reason_from_env() -> String {
     env_non_empty(ENV_DRAIN_CLOSE_REASON).unwrap_or_else(|| DEFAULT_DRAIN_CLOSE_REASON.to_string())
 }

src/boot/runtime/config/tests/base_auth_and_network.rs

@@ -47,6 +47,7 @@ pub(super) fn with_auth_env<T>(values: &[(&str, &str)], test: impl FnOnce() -> T
         ENV_QUEUE_WRITE_POLICY,
         ENV_QUEUE_LOSS_WINDOW_MS,
         ENV_KV_IDLE_TRANSACTION_TTL_SECS,
+        ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS,
         ENV_DRAIN_GRACE_SECONDS,
         ENV_DRAIN_CLOSE_REASON,
     ];
@@ -97,6 +98,7 @@ pub(super) fn with_storage_env<T>(values: &[(&str, &str)], test: impl FnOnce() -
         ENV_QUEUE_WRITE_POLICY,
         ENV_QUEUE_LOSS_WINDOW_MS,
         ENV_KV_IDLE_TRANSACTION_TTL_SECS,
+        ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS,
         "AWS_REGION",
         "AWS_DEFAULT_REGION",
         "AZURE_STORAGE_ACCOUNT_NAME",
@@ -174,6 +176,10 @@ pub(super) fn should_create_default_boot_config() {
     );
     assert!(config.queue_write_policy_defaulted_fast());
     assert_eq!(config.queue_loss_window_ms, DEFAULT_QUEUE_LOSS_WINDOW_MS);
+    assert_eq!(
+        config.schedule_preload_timeout(),
+        crate::domains::schedule::sink::DEFAULT_SCHEDULE_PRELOAD_TIMEOUT
+    );
     assert!(config.queue_write_options().is_best_effort());
     assert_eq!(config.drain_grace_seconds, DEFAULT_DRAIN_GRACE_SECONDS);
     assert_eq!(config.drain_close_reason, DEFAULT_DRAIN_CLOSE_REASON);
@@ -278,6 +284,27 @@ pub(super) fn should_read_drain_config_from_environment() {
     );
 }
 
+#[test]
+#[serial]
+pub(super) fn should_read_schedule_preload_timeout_from_environment() {
+    with_auth_env(
+        &[
+            ("FITZ_AUTH_REQUIRED", "false"),
+            (ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS, "75"),
+        ],
+        || {
+            // Arrange
+
+            // Act
+            let config = BootConfig::default();
+
+            // Assert
+            assert_eq!(config.schedule_preload_timeout(), Duration::from_secs(75));
+            assert!(config.validate().is_ok());
+        },
+    );
+}
+
 #[test]
 #[serial]
 pub(super) fn should_reject_invalid_drain_grace_from_environment() {

src/boot/runtime/config/tests/cloud_storage.rs

@@ -261,6 +261,24 @@ fn should_reject_invalid_kv_idle_transaction_ttl() {
     });
 }
 
+#[test]
+#[serial]
+fn should_reject_invalid_schedule_preload_timeout() {
+    with_storage_env(&[(ENV_SCHEDULE_PRELOAD_TIMEOUT_SECS, "0")], || {
+        // Arrange
+
+        // Act
+        let result = BootConfig::new().validate();
+
+        // Assert
+        assert!(result.is_err());
+        assert!(result
+            .unwrap_err()
+            .to_string()
+            .contains("FITZ_SCHEDULE_PRELOAD_TIMEOUT_SECS must be greater than 0"));
+    });
+}
+
 #[test]
 fn should_keep_non_cloud_sync_write_options_local() {
     // Arrange

src/boot/stats/core.rs

@@ -166,6 +166,8 @@ impl Runtime {
                 rpc_request_timeout: None,
                 stream_storage_layout: crate::domains::stream::StreamStorageLayout::default(),
                 kv_idle_transaction_ttl: Duration::from_mins(5),
+                schedule_preload_timeout:
+                    crate::domains::schedule::sink::DEFAULT_SCHEDULE_PRELOAD_TIMEOUT,
             },
         )
         .expect("setup domains");