Chaos scenarios, retry jitter validation, timeout convergence tests, consensus reconciliation

backend/src/bulkhead.rs

@@ -162,6 +162,20 @@ impl BulkheadRegistry {
             )
         };
 
+        // Fast path: a concurrency slot is immediately available, so this
+        // request never actually has to wait — it must not be gated by
+        // `max_queue_size`, which bounds the *wait queue*, not total
+        // concurrency. Without this, `max_queue_size: 0` would reject every
+        // request outright even with free concurrent slots.
+        if let Ok(permit) = Arc::clone(&semaphore).try_acquire_owned() {
+            metrics.active.fetch_add(1, Ordering::SeqCst);
+            return Ok(BulkheadPermit {
+                _permit: permit,
+                metrics,
+            });
+        }
+
+        // No slot free: this request must wait, gated by the queue budget.
         let queued_now = metrics.queued.fetch_add(1, Ordering::SeqCst) + 1;
         if queued_now > max_queue_size {
             metrics.queued.fetch_sub(1, Ordering::SeqCst);

backend/src/chaos.rs

@@ -542,6 +542,158 @@ mod tests {
         assert!(!result.passed());
     }
 
+    /// #370: cache unavailability + one replica down simultaneously. Rather
+    /// than erroring outright, a caller using `fallback::cascade` should
+    /// route around both down targets to the next healthy one.
+    #[test]
+    fn cascading_cache_and_replica_failure_falls_back_gracefully() {
+        use crate::fallback::{cascade, FallbackChain, FallbackTarget};
+
+        let sim = NetworkPartitionSimulator::new();
+        sim.partition("cache");
+        sim.partition("replica-b");
+        // "primary-db" is left healthy.
+
+        let chain = FallbackChain {
+            id: "test-chain".to_string(),
+            name: "vault-read".to_string(),
+            resource: "vault:read".to_string(),
+            targets: vec![
+                FallbackTarget {
+                    name: "cache".to_string(),
+                    endpoint: "cache://local".to_string(),
+                    priority: 0,
+                },
+                FallbackTarget {
+                    name: "replica-b".to_string(),
+                    endpoint: "replica://b".to_string(),
+                    priority: 1,
+                },
+                FallbackTarget {
+                    name: "primary-db".to_string(),
+                    endpoint: "db://primary".to_string(),
+                    priority: 2,
+                },
+            ],
+            created_at: chrono::Utc::now(),
+            active: true,
+        };
+
+        let result = cascade(&chain, |target| sim.call("client", &target.name));
+
+        // Falls back instead of erroring: a resolved target is found even
+        // though the two highest-priority targets are both down.
+        assert_eq!(result.resolved_target, Some("db://primary".to_string()));
+        assert!(
+            result.degraded,
+            "should be reported as degraded since it didn't resolve on the first target"
+        );
+        assert_eq!(result.attempts.len(), 3);
+        assert!(!result.attempts[0].succeeded, "cache should be down");
+        assert!(!result.attempts[1].succeeded, "replica-b should be down");
+        assert!(result.attempts[2].succeeded, "primary-db should be reachable");
+
+        // Recovery: healing cache should let the chain resolve on the
+        // highest-priority target again, no longer degraded.
+        sim.heal("cache");
+        let recovered = cascade(&chain, |target| sim.call("client", &target.name));
+        assert_eq!(recovered.resolved_target, Some("cache://local".to_string()));
+        assert!(!recovered.degraded);
+    }
+
+    /// #370: circuit breaker + bulkhead interacting under concurrent load.
+    /// The bulkhead bounds concurrency/queueing to a downstream dependency;
+    /// the circuit breaker should additionally fast-reject calls once it
+    /// trips, without either mechanism panicking or deadlocking under load.
+    #[tokio::test]
+    async fn circuit_breaker_and_bulkhead_interact_under_load() {
+        use crate::bulkhead::{BulkheadConfig, BulkheadRegistry};
+        use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitState};
+        use std::sync::Arc;
+
+        let registry = Arc::new(BulkheadRegistry::new(BulkheadConfig {
+            max_concurrent: 3,
+            max_queue_size: 4,
+        }));
+        let breaker = Arc::new(CircuitBreaker::new(
+            "downstream-dep",
+            CircuitBreakerConfig {
+                failure_threshold: 4,
+                success_threshold: 2,
+                open_duration: Duration::from_secs(30),
+            },
+        ));
+
+        let bulkhead_permits_acquired = Arc::new(AtomicUsize::new(0));
+        let operation_invocations = Arc::new(AtomicUsize::new(0));
+
+        let mut handles = Vec::new();
+        for _ in 0..20 {
+            let registry = Arc::clone(&registry);
+            let breaker = Arc::clone(&breaker);
+            let bulkhead_permits_acquired = Arc::clone(&bulkhead_permits_acquired);
+            let operation_invocations = Arc::clone(&operation_invocations);
+            handles.push(tokio::spawn(async move {
+                match registry.acquire("/api/vault-dependency").await {
+                    Ok(_permit) => {
+                        bulkhead_permits_acquired.fetch_add(1, Ordering::SeqCst);
+                        // Simulate call latency while holding the slot.
+                        tokio::time::sleep(Duration::from_millis(5)).await;
+                        breaker
+                            .call(|| {
+                                operation_invocations.fetch_add(1, Ordering::SeqCst);
+                                Err::<(), &str>("simulated downstream failure")
+                            })
+                            .map_err(|_| ())
+                    }
+                    Err(_queue_full) => Err(()),
+                }
+            }));
+        }
+
+        let mut rejected_or_failed = 0;
+        for handle in handles {
+            // A panic in the spawned task surfaces here as an `Err` from
+            // `join` — exactly the "did the system crash under combined
+            // load" signal a chaos scenario needs to catch.
+            let outcome = handle.await.expect("task panicked under concurrent load");
+            if outcome.is_err() {
+                rejected_or_failed += 1;
+            }
+        }
+
+        assert_eq!(
+            rejected_or_failed, 20,
+            "every call should have failed or been rejected (operation always fails)"
+        );
+        assert_eq!(
+            breaker.state(),
+            CircuitState::Open,
+            "breaker should trip open under sustained failures"
+        );
+
+        let acquired = bulkhead_permits_acquired.load(Ordering::SeqCst);
+        let invoked = operation_invocations.load(Ordering::SeqCst);
+        assert!(
+            acquired >= 4,
+            "test needs at least failure_threshold calls to reach the breaker to prove it trips, got {acquired}"
+        );
+        assert!(
+            invoked < acquired,
+            "circuit breaker should have fast-rejected at least one call that made it past the \
+             bulkhead instead of invoking the operation every time (invoked={invoked}, acquired={acquired})"
+        );
+
+        // No permits leaked: bulkhead accounting should be back to zero
+        // active once every task has completed.
+        let snapshot = registry.metrics_snapshot();
+        let bulkhead_stats = snapshot
+            .iter()
+            .find(|s| s.endpoint == "/api/vault-dependency")
+            .expect("endpoint should have been recorded");
+        assert_eq!(bulkhead_stats.active, 0, "all bulkhead permits should have been released");
+    }
+
     #[test]
     fn chaos_report_aggregates_multiple_scenarios() {
         let mut report = ChaosReport::new();

backend/src/db.rs

@@ -86,6 +86,10 @@ pub struct AppState {
     pub query_cache: Arc<crate::query_cache::QueryCache>,
     /// Distributed-lock deadlock detector stats (#82).
     pub deadlock_detector: Arc<crate::deadlock::DeadlockDetector>,
+    /// Incident tracking: shared by the incident HTTP API and the scheduled
+    /// consensus reconciliation job (#373), so a conflict opens the same
+    /// kind of record a manually-filed incident would.
+    pub incident_state: Arc<crate::incidents::IncidentState>,
 }
 
 impl axum::extract::FromRef<AppState> for Arc<Db> {
@@ -124,6 +128,12 @@ impl axum::extract::FromRef<AppState> for Arc<crate::feature_flags::FlagState> {
     }
 }
 
+impl axum::extract::FromRef<AppState> for Arc<crate::incidents::IncidentState> {
+    fn from_ref(state: &AppState) -> Arc<crate::incidents::IncidentState> {
+        Arc::clone(&state.incident_state)
+    }
+}
+
 // NOTE: The following FromRef implementations reference fields that are not currently
 // in AppState. When these features are properly implemented, uncomment and add the
 // corresponding fields to AppState.

backend/src/incidents.rs

@@ -155,20 +155,29 @@ fn timeline_entry(actor: impl Into<String>, note: impl Into<String>) -> Timeline
     }
 }
 
-/// `POST /incidents` — open a new incident with severity classification.
-pub async fn create_incident(
-    State(state): State<Arc<IncidentState>>,
-    Json(body): Json<CreateIncidentRequest>,
-) -> (StatusCode, Json<Incident>) {
+/// Open a new incident directly against `store`, bypassing the HTTP layer.
+///
+/// This is the synchronous entry point used by background jobs (e.g. the
+/// scheduled consensus reconciliation and backup checksum verification
+/// jobs) that detect a problem outside of a request/response cycle and need
+/// to surface it to operators the same way a manually-filed incident would
+/// be tracked. `create_incident` (the `POST /incidents` handler) builds on
+/// top of this so both paths produce identical `Incident` records.
+pub fn open_incident(
+    store: &IncidentStore,
+    title: impl Into<String>,
+    description: impl Into<String>,
+    severity: IncidentSeverity,
+) -> Incident {
     let now = Utc::now();
     let incident = Incident {
         id: Uuid::new_v4().to_string(),
-        title: body.title,
-        description: body.description,
-        severity: body.severity,
+        title: title.into(),
+        description: description.into(),
+        severity,
         status: IncidentStatus::Open,
         escalation_level: 0,
-        assigned_to: body.assigned_to,
+        assigned_to: None,
         timeline: vec![timeline_entry("system", "incident opened")],
         created_at: now,
         updated_at: now,
@@ -177,11 +186,30 @@ pub async fn create_incident(
     tracing::warn!(
         incident_id = %incident.id,
         severity = ?incident.severity,
+        title = %incident.title,
         "incident opened"
     );
 
-    let mut store = state.store.lock().unwrap();
-    store.insert(incident.id.clone(), incident.clone());
+    store
+        .lock()
+        .unwrap()
+        .insert(incident.id.clone(), incident.clone());
+
+    incident
+}
+
+/// `POST /incidents` — open a new incident with severity classification.
+pub async fn create_incident(
+    State(state): State<Arc<IncidentState>>,
+    Json(body): Json<CreateIncidentRequest>,
+) -> (StatusCode, Json<Incident>) {
+    let mut incident = open_incident(&state.store, body.title, body.description, body.severity);
+    incident.assigned_to = body.assigned_to.clone();
+    state
+        .store
+        .lock()
+        .unwrap()
+        .insert(incident.id.clone(), incident.clone());
 
     (StatusCode::CREATED, Json(incident))
 }
@@ -268,3 +296,29 @@ pub fn is_past_escalation_sla(incident: &Incident) -> bool {
     let elapsed = Utc::now() - incident.created_at;
     elapsed.num_minutes() > incident.severity.escalation_sla_minutes()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn open_incident_inserts_into_store_as_open() {
+        let store = create_incident_store();
+        let incident = open_incident(&store, "cache drift detected", "3 conflicting keys", IncidentSeverity::Sev2);
+
+        assert_eq!(incident.status, IncidentStatus::Open);
+        assert_eq!(incident.escalation_level, 0);
+        assert_eq!(incident.timeline.len(), 1);
+
+        let stored = store.lock().unwrap().get(&incident.id).cloned();
+        assert!(stored.is_some());
+        assert_eq!(stored.unwrap().title, "cache drift detected");
+    }
+
+    #[test]
+    fn fresh_incident_is_not_past_sla() {
+        let store = create_incident_store();
+        let incident = open_incident(&store, "t", "d", IncidentSeverity::Sev1);
+        assert!(!is_past_escalation_sla(&incident));
+    }
+}

backend/src/lib.rs

@@ -3,6 +3,7 @@ pub mod anomaly_detection;
 pub mod audit;
 pub mod backup_validation;
 pub mod batching;
+pub mod bulkhead;
 pub mod cache;
 pub mod chaos;
 pub mod circuit_breaker;
@@ -18,10 +19,12 @@ pub mod dlq;
 pub mod error;
 pub mod error_context;
 pub mod event_sourcing;
+pub mod fallback;
 pub mod feature_flags;
 pub mod graphql;
 pub mod handlers;
 pub mod health_routing;
+pub mod incidents;
 pub mod load_shedding;
 pub mod message_queue;
 pub mod metrics;

backend/src/main.rs

@@ -32,6 +32,10 @@ use ethos_protocol_backend::{
     event_sourcing::EventSourcingState,
     feature_flags::{evaluate_flag_handler, get_flag, list_flags, upsert_flag, FlagState},
     graphql::{build_schema, graphql_handler, graphql_playground},
+    incidents::{
+        add_timeline_entry, create_incident, escalate_incident, get_incident, list_incidents,
+        update_incident_status, IncidentState,
+    },
     load_shedding::{admission_middleware, LoadMonitor, LoadShedder, SheddingConfig},
     message_queue::MessageQueueState,
     metrics::Metrics,
@@ -42,6 +46,7 @@ use ethos_protocol_backend::{
     routes,
     rpc_pool::{RpcPool, RpcPoolConfig},
     scheduler,
+    scheduler::SchedulerContext,
     streaming::{stream_events, stream_vaults},
     timeout_policy::TimeoutState,
     tracing_sampling::TraceSampler,
@@ -187,6 +192,12 @@ pub fn build_router(state: AppState) -> Router {
         .route("/webhooks", post(register_webhook).get(list_webhooks))
         .route("/webhooks/:id", delete(delete_webhook))
         .route("/webhooks/verify", post(verify_webhook))
+        // ── Incident tracking routes (#373) ───────────────────────────────────
+        .route("/incidents", post(create_incident).get(list_incidents))
+        .route("/incidents/:id", get(get_incident))
+        .route("/incidents/:id/timeline", post(add_timeline_entry))
+        .route("/incidents/:id/status", post(update_incident_status))
+        .route("/incidents/:id/escalate", post(escalate_incident))
         // ── GraphQL routes (#66) ─────────────────────────────────────────────
         .route("/graphql", post(graphql_handler))
         .route("/graphql/playground", get(graphql_playground))
@@ -284,11 +295,6 @@ async fn main() {
         "consensus cache initialized"
     );
 
-    let scheduler_db = Arc::clone(&db);
-    tokio::spawn(async move {
-        scheduler::run(scheduler_db).await;
-    });
-
     let vault_store = create_vault_store();
     let event_store = create_event_store();
     let graphql_schema = build_schema(Arc::clone(&vault_store), Arc::clone(&event_store));
@@ -322,6 +328,20 @@ async fn main() {
 
     let flag_state = Arc::new(FlagState::new(Arc::clone(&db)));
 
+    let incident_state = Arc::new(IncidentState::new());
+
+    // ── Background scheduler (reminders, TTL insurance, retention, secret
+    // rotation, consensus reconciliation (#373)) ──────────────────────────
+    let scheduler_ctx = SchedulerContext {
+        db: Arc::clone(&db),
+        consensus: Arc::clone(&consensus),
+        metrics: Arc::clone(&metrics),
+        incident_state: Arc::clone(&incident_state),
+    };
+    tokio::spawn(async move {
+        scheduler::run(scheduler_ctx).await;
+    });
+
     let state = AppState {
         db: Arc::clone(&db),
         vault_store,
@@ -345,6 +365,7 @@ async fn main() {
         flag_state,
         query_cache: Arc::new(ethos_protocol_backend::query_cache::QueryCache::new()),
         deadlock_detector: Arc::new(ethos_protocol_backend::deadlock::DeadlockDetector::new()),
+        incident_state,
     };
 
     // ── Dynamic ACL admin routes ─────────────────────────────────────────