feat: Docker events watcher for near-instant liveness reconcile (#1018 slice B)

crates/ruscker-admin/src/lib.rs

@@ -527,6 +527,13 @@ impl AdminServer {
             // a false positive here.
             #[allow(clippy::let_underscore_future)]
             let _ = scaler::spawn(self.state.clone(), scaler::DEFAULT_INTERVAL);
+            // Docker events watcher (#1018 slice B): reconciles within ~1 s of
+            // an external `docker rm -f` / `docker restart` instead of waiting
+            // for the periodic scaler tick. The periodic reconcile above stays
+            // as the fallback; backends without event support park on an empty
+            // stream. Detached like the scaler — every reconcile is idempotent.
+            #[allow(clippy::let_underscore_future)]
+            let _ = scaler::spawn_event_watcher(self.state.clone());
             // Session sweeper: evicts idle sessions per the
             // global `heartbeat-timeout`. `-1` (the ShinyProxy
             // idiom for "never expire") becomes a no-op loop

crates/ruscker-admin/src/scaler.rs

@@ -277,6 +277,104 @@ pub fn spawn(state: AppState, interval: Duration) -> JoinHandle<()> {
     })
 }
 
+/// Debounce window for the events watcher: a burst of related events (a
+/// `docker restart` emits die+start; a `docker rm -f` emits die+destroy)
+/// collapses into a single reconcile once the stream is quiet this long.
+const EVENT_DEBOUNCE: Duration = Duration::from_millis(400);
+/// Backoff before reopening the events stream after it ends or errors.
+const EVENT_RECONNECT_BACKOFF: Duration = Duration::from_secs(3);
+
+/// Run one bidirectional liveness reconcile pass — the #1017 core, factored out
+/// so both the periodic scaler tick and the Docker-events watcher (#1018 slice
+/// B) take the exact same path: classify known replicas with per-host authority
+/// (Missing vs Unknown), prune Missing/Stopped-past-grace, and re-adopt
+/// externally-restarted containers. Idempotent and lock-safe, so a periodic
+/// tick and an event-triggered pass may run concurrently without conflict — the
+/// registry writes are serialized and every action re-checks state under lock.
+pub(crate) async fn reconcile_liveness_once(state: &AppState) {
+    let Some(backend) = state.backend.clone() else {
+        return;
+    };
+    let specs = crate::catalog::effective_specs(state.db.as_ref(), &state.config).await;
+    let registry_snap: Vec<Replica> = {
+        let reg = state.replicas.read().await;
+        specs.iter().flat_map(|s| reg.replicas_of(&s.id).to_vec()).collect()
+    };
+    let queries: Vec<ReplicaLivenessQuery> =
+        registry_snap.iter().map(ReplicaLivenessQuery::from).collect();
+    if let Ok(report) = backend.replica_liveness(&queries).await {
+        reconcile_backend_liveness(
+            state,
+            backend.as_ref(),
+            &specs,
+            &registry_snap,
+            report,
+            chrono::Utc::now(),
+        )
+        .await;
+    }
+}
+
+/// Watch the backend's Docker event stream and run an immediate liveness
+/// reconcile within ~1 s of an external container change (`docker rm -f`,
+/// `docker restart`), instead of waiting up to a full scaler tick (#1018 slice
+/// B). The reconcile is authoritative and idempotent, so the event is only a
+/// "reconcile now" nudge — a missed or duplicated event is always safe, and the
+/// periodic tick remains the fallback.
+///
+/// Leader-gated (HA), like the tick. Reconnects with backoff when the stream
+/// drops. Backends without event support yield an empty stream, so this task
+/// simply parks on it and relies entirely on the periodic reconcile.
+pub fn spawn_event_watcher(state: AppState) -> JoinHandle<()> {
+    use futures_util::StreamExt;
+    tokio::spawn(async move {
+        if state.backend.is_none() {
+            return;
+        }
+        loop {
+            let Some(backend) = state.backend.clone() else {
+                return;
+            };
+            match backend.container_events().await {
+                Ok(mut stream) => {
+                    info!("docker events watcher connected");
+                    while let Some(first) = stream.next().await {
+                        debug!(
+                            kind = ?first.kind, container = %first.container_id,
+                            replica = ?first.replica_id, "docker event"
+                        );
+                        // Coalesce a burst: keep pulling until the stream is
+                        // quiet for EVENT_DEBOUNCE (or it ends).
+                        let mut ended = false;
+                        loop {
+                            match tokio::time::timeout(EVENT_DEBOUNCE, stream.next()).await {
+                                Ok(Some(ev)) => {
+                                    debug!(kind = ?ev.kind, container = %ev.container_id, "docker event (coalesced)");
+                                }
+                                Ok(None) => {
+                                    ended = true;
+                                    break;
+                                }
+                                Err(_) => break, // quiet window elapsed → reconcile
+                            }
+                        }
+                        // Only the leader reconciles — mirrors the tick's gate.
+                        if state.leader.is_leader().await {
+                            reconcile_liveness_once(&state).await;
+                        }
+                        if ended {
+                            break;
+                        }
+                    }
+                    warn!("docker events stream ended; reconnecting");
+                }
+                Err(e) => warn!(error = %e, "docker events connect failed; will retry"),
+            }
+            tokio::time::sleep(EVENT_RECONNECT_BACKOFF).await;
+        }
+    })
+}
+
 #[derive(Clone, Copy)]
 enum ReplicaDownReason {
     Missing,
@@ -3006,6 +3104,34 @@ proxy:
         assert_eq!(state.sessions.len(), 0, "missing cleanup releases sticky sessions");
     }
 
+    /// #1018 slice B: the events watcher runs `reconcile_liveness_once`, which
+    /// must take the same authoritative path as the tick — a Missing replica is
+    /// pruned. This is the entry the watcher nudges on a `docker rm -f` event.
+    #[tokio::test]
+    async fn reconcile_liveness_once_prunes_missing_replica() {
+        let replica = Replica {
+            container_id: "gone".into(),
+            ..fake_replica("app")
+        };
+        let backend = Arc::new(ScriptedLivenessBackend::new(vec![one_liveness(
+            &replica.id,
+            ReplicaLiveness::Missing,
+            Vec::new(),
+        )]));
+        let state = state_with_yaml(
+            "proxy:\n  specs:\n    - id: app\n      container-image: t:1\n      min-replicas: 0\n",
+            backend,
+        );
+        state.replicas.write().await.add(replica.clone());
+
+        reconcile_liveness_once(&state).await;
+
+        assert!(
+            state.replicas.read().await.replicas_of("app").is_empty(),
+            "event-triggered reconcile prunes a Missing replica just like the tick"
+        );
+    }
+
     #[tokio::test]
     async fn unknown_keeps_replica_sessions_and_blocks_duplicate_spawn() {
         let replica = fake_replica("app");

crates/ruscker-core/src/lib.rs

@@ -35,6 +35,41 @@ use thiserror::Error;
 /// `dyn ContainerBackend` trait object the proxy/admin hold.
 pub type LogStream = Pin<Box<dyn Stream<Item = String> + Send>>;
 
+/// The lifecycle transition a [`ContainerEvent`] reports. Only the
+/// transitions the runtime reacts to are named; everything else is
+/// [`Other`](ContainerEventKind::Other) and simply nudges a reconcile.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ContainerEventKind {
+    /// The container started (e.g. the second half of a `docker restart`).
+    Start,
+    /// The container's main process exited.
+    Die,
+    /// The container was stopped.
+    Stop,
+    /// The container was removed (`docker rm`).
+    Destroy,
+    /// Any other container action — still worth a reconcile, but not
+    /// individually interesting.
+    Other,
+}
+
+/// A single Docker lifecycle event for a Ruscker-managed container. The
+/// consumer (the admin events watcher, #1018 slice B) treats it mainly as a
+/// "something changed, reconcile now" nudge and lets the authoritative
+/// [`ContainerBackend::replica_liveness`] pass decide the actual action, so a
+/// missed or duplicated event is always safe.
+#[derive(Debug, Clone)]
+pub struct ContainerEvent {
+    pub kind: ContainerEventKind,
+    pub container_id: String,
+    pub spec_id: Option<String>,
+    pub replica_id: Option<String>,
+}
+
+/// A boxed, owned stream of [`ContainerEvent`]s. Boxed for the same reason as
+/// [`LogStream`] — it travels through the `dyn ContainerBackend` trait object.
+pub type ContainerEventStream = Pin<Box<dyn Stream<Item = ContainerEvent> + Send>>;
+
 #[derive(Debug, Error)]
 pub enum CoreError {
     #[error("spec {0} not found in registry")]
@@ -366,6 +401,20 @@ pub trait ContainerBackend: Send + Sync {
         Ok(Box::pin(futures_util::stream::empty()))
     }
 
+    /// A live stream of Docker lifecycle events (start/die/stop/destroy) for
+    /// Ruscker-managed containers, so the runtime can reconcile within ~1 s of
+    /// an external `docker rm -f` / `docker restart` instead of waiting out the
+    /// periodic scaler tick (#1018 slice B).
+    ///
+    /// Default returns an empty stream: a backend without event support (mocks,
+    /// future backends) simply relies on the periodic reconcile — events are a
+    /// latency optimization, never the source of truth. The stream ends when
+    /// the daemon connection drops; the consumer reopens it, and the periodic
+    /// reconcile remains the fallback for anything missed during a gap.
+    async fn container_events(&self) -> CoreResult<ContainerEventStream> {
+        Ok(Box::pin(futures_util::stream::empty()))
+    }
+
     /// Creation/publish timestamp of an image as an RFC3339 string, if
     /// the backend can read it (Docker: `inspect_image().created`).
     /// Used to stamp a card's "updated" date from the image it runs