diff --git a/crates/ruscker-admin/src/lib.rs b/crates/ruscker-admin/src/lib.rs index 134805e..84f36f4 100644 --- a/crates/ruscker-admin/src/lib.rs +++ 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 diff --git a/crates/ruscker-admin/src/scaler.rs b/crates/ruscker-admin/src/scaler.rs index 2ec8cc0..361408d 100644 --- a/crates/ruscker-admin/src/scaler.rs +++ b/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 = { + let reg = state.replicas.read().await; + specs.iter().flat_map(|s| reg.replicas_of(&s.id).to_vec()).collect() + }; + let queries: Vec = + registry_snap.iter().map(ReplicaLivenessQuery::from).collect(); + if let Ok(report) = backend.replica_liveness(&queries).await { + reconcile_backend_liveness( + state, + backend.as_ref(), + &specs, + ®istry_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"); diff --git a/crates/ruscker-core/src/lib.rs b/crates/ruscker-core/src/lib.rs index c94355d..162536f 100644 --- a/crates/ruscker-core/src/lib.rs +++ b/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 + 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, + pub replica_id: Option, +} + +/// 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 + 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 { + 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 diff --git a/crates/ruscker-docker/src/lib.rs b/crates/ruscker-docker/src/lib.rs index a67ec58..3272709 100644 --- a/crates/ruscker-docker/src/lib.rs +++ b/crates/ruscker-docker/src/lib.rs @@ -16,12 +16,14 @@ use async_trait::async_trait; use bollard::models::{ - ContainerCreateBody, ContainerSummaryStateEnum, HostConfig, NetworkCreateRequest, PortBinding, + ContainerCreateBody, ContainerSummaryStateEnum, EventMessageTypeEnum, HostConfig, + NetworkCreateRequest, PortBinding, }; use bollard::query_parameters::{ - CreateContainerOptions, CreateImageOptions, ListContainersOptions, ListImagesOptions, - ListVolumesOptions, LogsOptionsBuilder, RemoveContainerOptions, RemoveImageOptions, - RemoveVolumeOptions, StartContainerOptions, StatsOptionsBuilder, StopContainerOptions, + CreateContainerOptions, CreateImageOptions, EventsOptionsBuilder, ListContainersOptions, + ListImagesOptions, ListVolumesOptions, LogsOptionsBuilder, RemoveContainerOptions, + RemoveImageOptions, RemoveVolumeOptions, StartContainerOptions, StatsOptionsBuilder, + StopContainerOptions, }; pub mod multihost; pub use multihost::MultiHostDockerBackend; @@ -567,6 +569,44 @@ impl LocalDockerBackend { } } +/// Map a Docker `action` string to the [`ContainerEventKind`] the runtime +/// reacts to. Unknown/other actions fold to `Other` — the consumer reconciles +/// on any event, so the exact kind is for logging, not control flow. +fn event_kind_from_action(action: Option<&str>) -> ruscker_core::ContainerEventKind { + use ruscker_core::ContainerEventKind as K; + match action { + Some("start") => K::Start, + // OOM/die both mean the process is gone; treat as Die. + Some("die") | Some("oom") => K::Die, + Some("stop") | Some("kill") => K::Stop, + Some("destroy") => K::Destroy, + _ => K::Other, + } +} + +/// Translate a bollard [`EventMessage`](bollard::models::EventMessage) into a +/// backend-neutral [`ContainerEvent`]. Returns `None` for anything that isn't a +/// container event carrying our replica label (the stream is already filtered +/// server-side; this is the belt-and-braces re-check). +fn container_event_from_message( + msg: bollard::models::EventMessage, +) -> Option { + if msg.typ != Some(EventMessageTypeEnum::CONTAINER) { + return None; + } + let actor = msg.actor?; + let attrs = actor.attributes.unwrap_or_default(); + // Must be one of ours — the label filter guarantees it, but re-check so a + // stray event can never nudge a reconcile against a foreign container id. + let replica_id = attrs.get(LABEL_REPLICA_ID).cloned()?; + Some(ruscker_core::ContainerEvent { + kind: event_kind_from_action(msg.action.as_deref()), + container_id: actor.id.unwrap_or_default(), + spec_id: attrs.get(LABEL_SPEC_ID).cloned(), + replica_id: Some(replica_id), + }) +} + fn classify_local_liveness( known: &[ReplicaLivenessQuery], managed: &[ManagedContainer], @@ -900,6 +940,36 @@ impl ContainerBackend for LocalDockerBackend { wait_for_ready(self, &replica.container_id, replica.upstream).await } + async fn container_events(&self) -> CoreResult { + // Scope the stream server-side to OUR containers' lifecycle events: + // container-type, carrying the replica label, and only the actions + // that change liveness (`start`/`die`/`stop`/`kill`/`destroy`/`oom`) — + // so daemon-wide churn and noisy actions (exec/health) never reach us. + let mut filters: HashMap> = HashMap::new(); + filters.insert("type".to_string(), vec!["container".to_string()]); + filters.insert("label".to_string(), vec![LABEL_REPLICA_ID.to_string()]); + filters.insert( + "event".to_string(), + ["start", "die", "stop", "kill", "destroy", "oom"] + .iter() + .map(|s| s.to_string()) + .collect(), + ); + let opts = EventsOptionsBuilder::new().filters(&filters).build(); + let raw = self.docker.events(Some(opts)); + // End the stream on the first transport error (the daemon connection + // dropped) so the consumer reconnects; the periodic reconcile covers + // anything missed during the gap. `filter_map` then drops events we + // don't model (all remaining items are `Ok`). + let stream = raw + .take_while(|res| { + let keep = res.is_ok(); + async move { keep } + }) + .filter_map(|res| async move { container_event_from_message(res.ok()?) }); + Ok(Box::pin(stream)) + } + async fn all_container_image_refs(&self) -> CoreResult> { // ALL containers (running + stopped), no label filter. let opts = ListContainersOptions { @@ -1795,6 +1865,62 @@ fn apply_limits(host_config: &mut HostConfig, limits: &ruscker_core::ResourceLim mod tests { use super::*; + #[test] + fn event_kind_maps_docker_actions() { + use ruscker_core::ContainerEventKind as K; + assert_eq!(event_kind_from_action(Some("start")), K::Start); + assert_eq!(event_kind_from_action(Some("die")), K::Die); + assert_eq!(event_kind_from_action(Some("oom")), K::Die); + assert_eq!(event_kind_from_action(Some("stop")), K::Stop); + assert_eq!(event_kind_from_action(Some("kill")), K::Stop); + assert_eq!(event_kind_from_action(Some("destroy")), K::Destroy); + assert_eq!(event_kind_from_action(Some("health_status")), K::Other); + assert_eq!(event_kind_from_action(None), K::Other); + } + + #[test] + fn container_event_maps_labelled_container_message() { + use bollard::models::{EventActor, EventMessage, EventMessageTypeEnum}; + let mut attrs = HashMap::new(); + attrs.insert(LABEL_REPLICA_ID.to_string(), "11111111-1111-1111-1111-111111111111".to_string()); + attrs.insert(LABEL_SPEC_ID.to_string(), "app".to_string()); + let msg = EventMessage { + typ: Some(EventMessageTypeEnum::CONTAINER), + action: Some("destroy".to_string()), + actor: Some(EventActor { + id: Some("deadbeef".to_string()), + attributes: Some(attrs), + }), + ..Default::default() + }; + let ev = container_event_from_message(msg).expect("a labelled container event maps"); + assert_eq!(ev.kind, ruscker_core::ContainerEventKind::Destroy); + assert_eq!(ev.container_id, "deadbeef"); + assert_eq!(ev.spec_id.as_deref(), Some("app")); + assert_eq!(ev.replica_id.as_deref(), Some("11111111-1111-1111-1111-111111111111")); + } + + #[test] + fn container_event_ignores_non_container_and_unlabelled() { + use bollard::models::{EventActor, EventMessage, EventMessageTypeEnum}; + // Non-container type → None. + let image_evt = EventMessage { + typ: Some(EventMessageTypeEnum::IMAGE), + action: Some("pull".to_string()), + actor: Some(EventActor { id: Some("img".into()), attributes: Some(HashMap::new()) }), + ..Default::default() + }; + assert!(container_event_from_message(image_evt).is_none()); + // Container event without our replica label → None (belt-and-braces). + let foreign = EventMessage { + typ: Some(EventMessageTypeEnum::CONTAINER), + action: Some("die".to_string()), + actor: Some(EventActor { id: Some("other".into()), attributes: Some(HashMap::new()) }), + ..Default::default() + }; + assert!(container_event_from_message(foreign).is_none()); + } + #[test] fn successful_local_inventory_classifies_absent_container_as_missing() { let id = ReplicaId::new(); @@ -2192,6 +2318,74 @@ mod tests { backend.stop(&original.id).await.expect("cleanup restarted container"); } + /// Consume the event stream (skipping unrelated events) until one matches + /// `pred`, or a 15 s budget elapses. + #[cfg(feature = "docker-it")] + async fn wait_for_event( + stream: &mut ruscker_core::ContainerEventStream, + pred: impl Fn(&ruscker_core::ContainerEvent) -> bool, + ) -> bool { + use futures_util::StreamExt; + tokio::time::timeout(std::time::Duration::from_secs(15), async { + while let Some(ev) = stream.next().await { + if pred(&ev) { + return true; + } + } + false + }) + .await + .unwrap_or(false) + } + + /// #1018 slice B real smoke: the events stream reports a `Start` for a + /// freshly-spawned Ruscker container and a `die/stop/destroy` when it is + /// removed out of band — the signal the admin watcher turns into an + /// immediate reconcile. + /// + /// bollard's events stream connects lazily on first poll, so each docker + /// action is fired from a delayed task *while* the stream is already being + /// polled (`tokio::join!`) — otherwise the event races ahead of the + /// connection. This is exactly how the real watcher behaves (it polls + /// continuously); the choreography here just makes the test deterministic. + #[cfg(feature = "docker-it")] + #[tokio::test] + async fn container_events_report_start_and_external_removal() { + use ruscker_core::ContainerEventKind as K; + let image = std::env::var("RUSCKER_IT_IMAGE").unwrap_or_else(|_| "nginx:1.29-alpine".into()); + let backend = LocalDockerBackend::local().expect("connect docker"); + let mut events = backend.container_events().await.expect("open events stream"); + let spec = format!("evt-it-{}", uuid::Uuid::new_v4()); + + // Start: match by spec label (the replica id isn't known until spawn + // returns). Spawn after a short delay so the stream is polling first. + let spec_for_match = spec.clone(); + let (start_seen, spawn_res) = tokio::join!( + wait_for_event(&mut events, |ev| ev.spec_id.as_deref() == Some(&spec_for_match) + && ev.kind == K::Start), + async { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + backend.spawn_with_port(&spec, &image, 80).await + } + ); + let replica = spawn_res.expect("spawn original"); + assert!(start_seen, "expected a Start event for the spawned container"); + + // Removal: match by the now-known replica id. + let rid = replica.id.to_string(); + let container_id = replica.container_id.clone(); + let (gone_seen, rm_res) = tokio::join!( + wait_for_event(&mut events, |ev| ev.replica_id.as_deref() == Some(&rid) + && matches!(ev.kind, K::Die | K::Stop | K::Destroy)), + async { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + backend.remove_container(&container_id).await + } + ); + rm_res.expect("external remove"); + assert!(gone_seen, "expected a die/stop/destroy event after external removal"); + } + /// `container-env` / `container-cmd` smoke: spawn with both set and /// assert the real container's `Config.Env` / `Config.Cmd` carry /// them (via `docker inspect`). Uses nginx's own argv as the cmd so diff --git a/crates/ruscker-docker/src/multihost.rs b/crates/ruscker-docker/src/multihost.rs index a367795..09f9ccb 100644 --- a/crates/ruscker-docker/src/multihost.rs +++ b/crates/ruscker-docker/src/multihost.rs @@ -704,6 +704,31 @@ impl ContainerBackend for MultiHostDockerBackend { backend.wait_until_ready(replica).await } + async fn container_events(&self) -> CoreResult { + // Merge every reachable host's event stream. A host whose stream fails + // to open is skipped (logged) — its containers still recover via the + // periodic reconcile — and a host stream that later ends just drops out + // of the merge; the admin watcher's reconnect re-establishes all hosts. + // Never fail the whole call for a single host: events are a latency + // optimization, and a total failure would only fall back to periodic + // reconcile anyway. + let mut streams: Vec = Vec::new(); + for h in &self.hosts { + match h.backend.container_events().await { + Ok(s) => streams.push(Box::pin(s)), + Err(e) => { + tracing::warn!(host = %h.id, error = %e, "open docker events on host failed; skipping (periodic reconcile covers it)") + } + } + } + // `select_all` panics on an empty iterator — if no host opened a + // stream, fall back to an empty stream (periodic reconcile still runs). + if streams.is_empty() { + return Ok(Box::pin(futures_util::stream::empty())); + } + Ok(Box::pin(futures_util::stream::select_all(streams))) + } + async fn all_container_image_refs(&self) -> CoreResult> { // This feeds the disk panel's "image in use" signal, which MUST // cover every host: an image backing a container on an unreachable