fix(observe): a failed observation is not evidence of absence (audit theme 3)

crates/core/src/system/observer.rs

@@ -196,7 +196,17 @@ impl Observer {
                     ContainerStatus::Exited => ObservationFact::ContainerExited {
                         exit_code: s.exit_code.unwrap_or(-1),
                     },
-                    ContainerStatus::Unknown => ObservationFact::ContainerMissing,
+                    // r[impl observe.failure-not-absence] — this arm has
+                    // already proven the container exists: the inspect
+                    // returned it. Mapping either of these to
+                    // ContainerMissing recorded `container_removed`, which
+                    // the oracle reads as the transition to Unscheduled and
+                    // termination_success reads as terminal success — so a
+                    // container still draining through its stop timeout
+                    // released barriers over volumes and networks it held.
+                    ContainerStatus::Stopping | ContainerStatus::Unknown => {
+                        ObservationFact::ContainerPresentIndeterminate
+                    }
                 };
                 facts.push((lifecycle_fact, now));
 

crates/core/src/system/podman.rs

@@ -802,13 +802,32 @@ fn is_not_found(e: &podman_rest_client::Error) -> bool {
     }
 }
 
+/// Map podman's documented container states.
+///
+/// The list is exhaustive over what podman documents rather than over what
+/// this code happens to have needed: the catch-all used to swallow
+/// `stopping`, `removing` and `initialized`, and three layers away that
+/// became "the container was removed".
+///
+/// r[impl observe.failure-not-absence] — an unrecognised state is logged and
+/// mapped to `Unknown`, which means *present but indeterminate*, never
+/// absent. The next state podman invents then fails loudly in one place.
 fn parse_container_status(s: &str) -> ContainerStatus {
     match s {
-        "created" | "configured" => ContainerStatus::Created,
+        "created" | "configured" | "initialized" => ContainerStatus::Created,
         "running" => ContainerStatus::Running,
         "paused" => ContainerStatus::Paused,
         "exited" | "stopped" | "dead" => ContainerStatus::Exited,
-        _ => ContainerStatus::Unknown,
+        "stopping" | "removing" => ContainerStatus::Stopping,
+        "unknown" => ContainerStatus::Unknown,
+        other => {
+            tracing::error!(
+                state = other,
+                "podman reported a container state this build does not model; \
+                 treating the container as present but indeterminate"
+            );
+            ContainerStatus::Unknown
+        }
     }
 }
 
@@ -1036,3 +1055,61 @@ impl ContainerRuntime for PodmanRuntime {
         })
     }
 }
+
+#[cfg(test)]
+mod status_tests {
+    use super::*;
+
+    // r[verify observe.failure-not-absence]
+    // The catch-all used to swallow podman's real transitional states, and
+    // three layers away `Unknown` became "the container was removed" — so a
+    // postgres draining through a long stop_timeout_secs was recorded as
+    // removed while it still held its volumes and network.
+    #[test]
+    fn draining_states_are_not_absence() {
+        assert_eq!(
+            parse_container_status("stopping"),
+            ContainerStatus::Stopping
+        );
+        assert_eq!(
+            parse_container_status("removing"),
+            ContainerStatus::Stopping
+        );
+    }
+
+    // r[verify observe.failure-not-absence]
+    #[test]
+    fn initialized_is_created_not_unknown() {
+        assert_eq!(
+            parse_container_status("initialized"),
+            ContainerStatus::Created
+        );
+    }
+
+    // r[verify observe.failure-not-absence]
+    // An unmodelled state means "present, and nothing more can be said" —
+    // never "absent". The next state podman invents fails loudly here rather
+    // than quietly becoming a removal.
+    #[test]
+    fn an_unmodelled_state_is_present_but_indeterminate() {
+        assert_eq!(
+            parse_container_status("some-future-podman-state"),
+            ContainerStatus::Unknown
+        );
+    }
+
+    #[test]
+    fn documented_states_round_trip() {
+        for (input, expected) in [
+            ("created", ContainerStatus::Created),
+            ("configured", ContainerStatus::Created),
+            ("running", ContainerStatus::Running),
+            ("paused", ContainerStatus::Paused),
+            ("exited", ContainerStatus::Exited),
+            ("stopped", ContainerStatus::Exited),
+            ("dead", ContainerStatus::Exited),
+        ] {
+            assert_eq!(parse_container_status(input), expected, "{input}");
+        }
+    }
+}

crates/core/src/system/reconcile.rs

@@ -440,6 +440,11 @@ pub struct Reconciler {
     /// unresolved.
     // r[impl service.site.address]
     site_resolver: Option<Arc<SiteServiceResolver>>,
+    /// Consecutive failed observations per instance, so a single blip logs an
+    /// error without minting an operator-visible fault. In memory beside the
+    /// reconciler's other per-tick state; a restart re-observes anyway.
+    // r[impl observe.failure-not-absence]
+    observe_failure_streaks: HashMap<InstanceId, u32>,
 }
 
 impl Reconciler {
@@ -511,6 +516,7 @@ impl Reconciler {
             tls_coordinator,
             resolver_health_fail_count: std::sync::atomic::AtomicU32::new(0),
             site_resolver,
+            observe_failure_streaks: HashMap::new(),
         }
     }
 

crates/core/src/system/reconcile/faults.rs

@@ -445,8 +445,43 @@ impl Reconciler {
         });
     }
 
+    /// How many consecutive ticks an instance's observation must fail before
+    /// it becomes an operator-visible fault.
+    ///
+    /// A single failed probe is a blip — podman or systemd momentarily busy —
+    /// and minting a fault for it trains operators to ignore the fault list.
+    /// The failure is logged on the first tick regardless; the threshold gates
+    /// only the fault.
+    // r[impl observe.failure-not-absence]
+    const OBSERVE_FAULT_AFTER_FAILURES: u32 = 3;
+
+    /// Advance the consecutive-observation-failure counters and return the
+    /// instances that have now failed often enough to fault.
+    // r[impl observe.failure-not-absence]
+    fn escalate_observe_failures(
+        &mut self,
+        failures: &[(ResourceInstance, String)],
+    ) -> Vec<(ResourceInstance, String)> {
+        let failing: std::collections::HashSet<crate::runtime::identity::InstanceId> =
+            failures.iter().map(|(inst, _)| inst.id).collect();
+        // Any instance not failing this tick observed successfully, so its
+        // streak resets — a flapping probe never accumulates to a fault.
+        self.observe_failure_streaks
+            .retain(|id, _| failing.contains(id));
+
+        let mut escalated = Vec::new();
+        for (instance, error) in failures {
+            let streak = self.observe_failure_streaks.entry(instance.id).or_insert(0);
+            *streak += 1;
+            if *streak >= Self::OBSERVE_FAULT_AFTER_FAILURES {
+                escalated.push((instance.clone(), error.clone()));
+            }
+        }
+        escalated
+    }
+
     pub(super) fn file_pod_actuation_faults(
-        &self,
+        &mut self,
         app: &AppName,
         update: &pods::PodActuationUpdate,
     ) {
@@ -461,11 +496,7 @@ impl Reconciler {
             .iter()
             .map(|(i, s)| (i.clone(), s.clone()))
             .collect();
-        let observe_failures: Vec<(ResourceInstance, String)> = update
-            .observe_failures
-            .iter()
-            .map(|(i, s)| (i.clone(), s.clone()))
-            .collect();
+        let observe_failures = self.escalate_observe_failures(&update.observe_failures);
         // r[impl fault.lifecycle] — the file set and the clear set must be
         // disjoint per instance. `stop_sent` is recorded before the stop is
         // attempted, so an instance whose stop just failed appears in both:

crates/core/src/system/reconcile/pods.rs

@@ -146,14 +146,42 @@ struct ObservedInstance<'a> {
     result: PodInstanceResult,
 }
 
+/// The outcome of trying to observe one instance this tick.
+///
+/// The observation used to be reported as an `ObservedInstance` with every
+/// flag false, which is byte-for-byte what "confirmed absent" looks like — so
+/// a single failed podman or systemd probe read as "the container is gone".
+/// `actuate_one_pod` then ran its Job terminal-detection predicate
+/// (`!container_exists && !is_running && previously_ran`), stopped the Job and
+/// recorded it in `completed_jobs`, from which `job-terminal.defense`
+/// guarantees it is killed again if it ever reappears. One transient hiccup
+/// permanently destroyed an in-flight batch workload.
+///
+/// Splitting the two makes that coercion unrepresentable: `actuate_one_pod`
+/// takes an `ObservedInstance`, so the compiler forces every caller to route
+/// `Failed` somewhere explicit. Note that *absence* stays inside `Observed` —
+/// a `ContainerMissing` from a successful query is a real observation and must
+/// keep driving teardown and job-terminal detection.
+// r[impl observe.failure-not-absence]
+enum PodObservation<'a> {
+    /// Every probe succeeded; the flags are evidence.
+    Observed(Box<ObservedInstance<'a>>),
+    /// At least one probe failed. Nothing is known about this instance this
+    /// tick — which is not the same as knowing it is gone.
+    Failed {
+        dr: &'a DesiredResource,
+        result: Box<PodInstanceResult>,
+    },
+}
+
 // r[observe.deployment]
 async fn observe_one_pod<'a>(
     observer: &Observer,
     actuator: &Actuator,
     driver: &Arc<System>,
     dr: &'a DesiredResource,
     node_prefix: &Ipv6Net,
-) -> Option<ObservedInstance<'a>> {
+) -> PodObservation<'a> {
     let mut result = PodInstanceResult {
         running: None,
         observations: Vec::new(),
@@ -183,20 +211,14 @@ async fn observe_one_pod<'a>(
                 "pods: observe failed, skipping instance"
             );
             result.observe_failure = Some((dr.instance.clone(), e.to_string()));
-            return Some(ObservedInstance {
+            // The per-instance observation is atomic: all three probes or
+            // nothing. Reporting the ones that did succeed would resurrect
+            // the bug in a subtler form — `container_exists: false` because
+            // the inspect failed, while the network probe succeeded.
+            return PodObservation::Failed {
                 dr,
-                is_running: false,
-                spec_stale: false,
-                unit_failed: false,
-                unit_active: false,
-                unit_start_limit_hit: false,
-                container_exists: false,
-                has_exited: false,
-                network_exists: false,
-                observed_unhealthy: false,
-                observed_healthy: false,
-                result,
-            });
+                result: Box::new(result),
+            };
         }
     };
 
@@ -263,6 +285,9 @@ async fn observe_one_pod<'a>(
             ObservationFact::ContainerCreated
                 | ObservationFact::ContainerRunning { .. }
                 | ObservationFact::ContainerExited { .. }
+                // r[impl observe.failure-not-absence] — a draining container
+                // is present. It still holds its volumes and its network.
+                | ObservationFact::ContainerPresentIndeterminate
         )
     });
     let has_exited = facts
@@ -322,7 +347,7 @@ async fn observe_one_pod<'a>(
         }
     }
 
-    Some(ObservedInstance {
+    PodObservation::Observed(Box::new(ObservedInstance {
         dr,
         is_running,
         spec_stale,
@@ -335,7 +360,7 @@ async fn observe_one_pod<'a>(
         observed_unhealthy,
         observed_healthy,
         result,
-    })
+    }))
 }
 
 // r[actuate.deployment.start]
@@ -765,11 +790,26 @@ pub(super) async fn observe_and_actuate(
         .map(|dr| observe_one_pod(observer, actuator, driver, dr, node_prefix))
         .collect();
 
-    let observed: Vec<ObservedInstance<'_>> = join_all(observe_futures)
-        .await
-        .into_iter()
-        .flatten()
-        .collect();
+    // r[impl observe.failure-not-absence] — instances whose observation
+    // failed are routed straight to the results, never into the actuation
+    // phase: no start, no stop, no job-terminal detection. Their
+    // `observe_failure` still reaches the fault path, and because they
+    // emitted no facts, the lifecycle oracle keeps the last state it derived
+    // rather than being told the instance is gone.
+    let mut observed: Vec<ObservedInstance<'_>> = Vec::with_capacity(pod_resources.len());
+    let mut failed_results: Vec<PodInstanceResult> = Vec::new();
+    for observation in join_all(observe_futures).await {
+        match observation {
+            PodObservation::Observed(obs) => observed.push(*obs),
+            PodObservation::Failed { dr, result } => {
+                tracing::warn!(
+                    instance = %dr.instance.display_name,
+                    "pods: observation failed; taking no action on this instance this tick"
+                );
+                failed_results.push(*result);
+            }
+        }
+    }
 
     // Phase 2: group deployments and compute stop inhibitions.
     let mut deployment_groups: HashMap<String, Vec<usize>> = HashMap::new();
@@ -830,6 +870,11 @@ pub(super) async fn observe_and_actuate(
     }
 
     let results = join_all(actuate_futures).await;
+    let results: Vec<PodInstanceResult> = results
+        .into_iter()
+        .flatten()
+        .chain(failed_results)
+        .collect();
 
     let mut update = PodActuationUpdate {
         running: Vec::new(),
@@ -852,7 +897,7 @@ pub(super) async fn observe_and_actuate(
         completed_job_instances: Vec::new(),
     };
 
-    for result in results.into_iter().flatten() {
+    for result in results {
         if let Some(rp) = result.running {
             update.running.push(rp);
         }