From 9ce44fa07dc2e5a0c18061c0e3b58c283f7c5fcc Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 20 Oct 2026 17:30:39 -0400 Subject: [PATCH] test(agents): cover the sandbox reaper pass against a fake apiserver Add an in-memory fake Kubernetes apiserver (Sandboxes, Pods, a Pod watch, Events, merge patches with resourceVersion and UID preconditions, concurrent writers, Date header) and reconcile scenario tests over it: tombstoning with verified, listed, missing, and inherited checkpoints, the harness receipt wire format, the 24-hour hold from first observation, lost-Pod grace, a restarted reaper without receipts, invalid generations, CAS conflicts and their retry bound, races against tombstone deletion and bind-abandoned tombstones, expired tombstone deletion, foreign owners, dry run, and the namespace loop with its watch and health. A property test runs two passes with the clock advanced between them and checks that no pass resurrects a session, tombstones a live one (active, still binding, or draining), or deletes anything but an expired tombstone; its proptest regression seeds are checked in. Signed-off-by: Tee8z --- .../proptest-regressions/reconcile_tests.txt | 8 + .../buzz-sandbox-reaper/src/fake_apiserver.rs | 270 ++++ crates/buzz-sandbox-reaper/src/reconcile.rs | 4 + .../src/reconcile_tests.rs | 1122 +++++++++++++++++ 4 files changed, 1404 insertions(+) create mode 100644 crates/buzz-sandbox-reaper/proptest-regressions/reconcile_tests.txt create mode 100644 crates/buzz-sandbox-reaper/src/fake_apiserver.rs create mode 100644 crates/buzz-sandbox-reaper/src/reconcile_tests.rs diff --git a/crates/buzz-sandbox-reaper/proptest-regressions/reconcile_tests.txt b/crates/buzz-sandbox-reaper/proptest-regressions/reconcile_tests.txt new file mode 100644 index 00000000000..8688fde2168 --- /dev/null +++ b/crates/buzz-sandbox-reaper/proptest-regressions/reconcile_tests.txt @@ -0,0 +1,8 @@ +# Seeds for failure cases proptest has generated in the past. It is +# automatically read and these particular cases re-run before any +# novel cases are generated. +# +# It is recommended to check this file in to source control so that +# everyone who runs the test benefits from these saved cases. +cc 39b358909b33e906e9d30b9f9685a9607b75646de7a4ac32f7430253618d03dc # shrinks to sessions = [Session { owner_matches: false, mode: "Running", lifecycle: None, age: 0, ended_ago: None, bound: false, pod: None, checkpointing: false, receipt: false }], verified = false +cc de5db39b9b6a9f2b561ccb02a7d68e1e1aedc7fd2f35060480fcb34d9587b05b # shrinks to sessions = [Session { owner_matches: true, mode: "Running", lifecycle: Some("ending"), age: 0, ended_ago: None, bound: false, pod: None, checkpointing: false, receipt: false }], verified = false diff --git a/crates/buzz-sandbox-reaper/src/fake_apiserver.rs b/crates/buzz-sandbox-reaper/src/fake_apiserver.rs new file mode 100644 index 00000000000..829a5768592 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/fake_apiserver.rs @@ -0,0 +1,270 @@ +//! A minimal in-memory apiserver for one namespace, driving the real +//! `kube::Client` request path (`tower::service_fn`, as the provider's +//! `cluster.rs` tests do). It models only what the reconciler touches and +//! records every request so tests can assert what was (never) written. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use chrono::{DateTime, Utc}; +use http::{Method, Request, Response}; +use kube::client::Body; +use kube::Client; +use serde_json::{json, Value}; + +/// One recorded request. +#[derive(Debug, Clone)] +pub struct Call { + pub method: Method, + pub path: String, + pub body: Value, +} + +pub struct State { + /// Apiserver clock, served as the `Date` header. + pub now: DateTime, + /// Omit the `Date` header. + pub no_date: bool, + pub reject_writes: bool, + pub sandboxes: BTreeMap, + pub pods: BTreeMap, + pub rv: u64, + /// Applied to the named Sandbox (bumping its resourceVersion) just + /// before the next PATCH or DELETE is evaluated: a concurrent writer. + pub race: Option<(String, Value)>, + /// Re-apply `race` before every write instead of once. + pub sticky_race: bool, + /// Pod watch events served (once) to the next Pod watch request; a + /// `DELETED` event also removes the Pod. Later watches never answer. + pub watch: Vec, + pub calls: Vec, +} + +impl Default for State { + fn default() -> Self { + Self { + now: DateTime::from_timestamp(1_800_000_000, 0).unwrap(), + no_date: false, + reject_writes: false, + sandboxes: BTreeMap::new(), + pods: BTreeMap::new(), + rv: 100, + race: None, + sticky_race: false, + watch: Vec::new(), + calls: Vec::new(), + } + } +} + +pub type Shared = Arc>; + +fn status(code: u16, reason: &str) -> Response { + let body = json!({"kind":"Status","apiVersion":"v1","status":"Failure", + "message":reason,"reason":reason,"code":code}); + Response::builder() + .status(code) + .body(Body::from(serde_json::to_vec(&body).unwrap())) + .unwrap() +} + +/// JSON merge patch (RFC 7386). +pub fn merge(target: &mut Value, patch: &Value) { + match (target, patch) { + (Value::Object(target), Value::Object(patch)) => { + for (key, value) in patch { + if value.is_null() { + target.remove(key); + } else { + merge(target.entry(key.clone()).or_insert(Value::Null), value); + } + } + } + (target, patch) => *target = patch.clone(), + } +} + +fn matches_selector(object: &Value, query: &str) -> bool { + let Some(selector) = query + .split('&') + .find_map(|pair| pair.strip_prefix("labelSelector=")) + else { + return true; + }; + let selector = selector + .replace("%3D", "=") + .replace("%2C", ",") + .replace("%2F", "/"); + selector.split(',').all(|term| { + let (key, value) = term.split_once('=').unwrap_or((term, "")); + object["metadata"]["labels"][key] == value + }) +} + +impl State { + fn ok(&self, value: &Value) -> Response { + let mut response = Response::builder().status(200); + if !self.no_date { + response = response.header( + http::header::DATE, + self.now.format("%a, %d %b %Y %H:%M:%S GMT").to_string(), + ); + } + response + .body(Body::from(serde_json::to_vec(value).unwrap())) + .unwrap() + } + + fn list(&self, items: &BTreeMap, query: &str) -> Response { + let items: Vec<&Value> = items + .values() + .filter(|o| matches_selector(o, query)) + .collect(); + self.ok(&json!({"apiVersion":"v1","kind":"List", + "metadata":{"resourceVersion": self.rv.to_string()},"items":items})) + } + + fn apply_race(&mut self) { + let race = if self.sticky_race { + self.race.clone() + } else { + self.race.take() + }; + if let Some((name, mutation)) = race { + self.rv += 1; + let rv = self.rv.to_string(); + if let Some(current) = self.sandboxes.get_mut(&name) { + merge(current, &mutation); + current["metadata"]["resourceVersion"] = rv.into(); + } + } + } + + /// Serve the queued watch events, or `None` (never answer) once drained. + fn watch_pods(&mut self) -> Option> { + if self.watch.is_empty() { + return None; + } + let mut lines = Vec::new(); + for event in std::mem::take(&mut self.watch) { + if event["type"] == "DELETED" { + if let Some(name) = event["object"]["metadata"]["name"].as_str() { + self.pods.remove(name); + } + } + lines.extend(serde_json::to_vec(&event).unwrap()); + lines.push(b'\n'); + } + Some( + Response::builder() + .status(200) + .body(Body::from(lines)) + .unwrap(), + ) + } + + /// `None` holds the request open forever (an idle watch). + fn handle(&mut self, method: &Method, uri: &str, body: &[u8]) -> Option> { + let (path, query) = uri.split_once('?').unwrap_or((uri, "")); + if *method == Method::GET && path.ends_with("/pods") && query.contains("watch=true") { + return self.watch_pods(); + } + Some(self.respond(method, uri, body)) + } + + fn respond(&mut self, method: &Method, uri: &str, body: &[u8]) -> Response { + let (path, query) = uri.split_once('?').unwrap_or((uri, "")); + let body: Value = serde_json::from_slice(body).unwrap_or(Value::Null); + self.calls.push(Call { + method: method.clone(), + path: path.to_string(), + body: body.clone(), + }); + if self.reject_writes && *method != Method::GET { + return status(403, "Forbidden"); + } + let segments: Vec<&str> = path.trim_start_matches('/').split('/').collect(); + match (method.clone(), segments.as_slice()) { + (Method::GET, ["apis", "agents.x-k8s.io", "v1beta1", "namespaces", _, "sandboxes"]) => { + self.list(&self.sandboxes, query) + } + ( + Method::GET, + ["apis", "agents.x-k8s.io", "v1beta1", "namespaces", _, "sandboxes", name], + ) => match self.sandboxes.get(*name) { + Some(sandbox) => self.ok(sandbox), + None => status(404, "NotFound"), + }, + ( + Method::PATCH, + ["apis", "agents.x-k8s.io", "v1beta1", "namespaces", _, "sandboxes", name], + ) => { + self.apply_race(); + self.rv += 1; + let rv = self.rv.to_string(); + let Some(current) = self.sandboxes.get_mut(*name) else { + return status(404, "NotFound"); + }; + let expected = &body["metadata"]["resourceVersion"]; + if !expected.is_null() && *expected != current["metadata"]["resourceVersion"] { + return status(409, "Conflict"); + } + merge(current, &body); + current["metadata"]["resourceVersion"] = rv.into(); + let updated = current.clone(); + self.ok(&updated) + } + ( + Method::DELETE, + ["apis", "agents.x-k8s.io", "v1beta1", "namespaces", _, "sandboxes", name], + ) => { + self.apply_race(); + let Some(current) = self.sandboxes.get(*name) else { + return status(404, "NotFound"); + }; + let pre = &body["preconditions"]; + for field in ["uid", "resourceVersion"] { + if pre[field].is_null() || pre[field] != current["metadata"][field] { + return status(409, "Conflict"); + } + } + let removed = self.sandboxes.remove(*name).unwrap(); + self.ok(&removed) + } + (Method::GET, ["api", "v1", "namespaces", _, "pods"]) => self.list(&self.pods, query), + (Method::GET, ["api", "v1", "namespaces", _, "pods", name]) => { + match self.pods.get(*name) { + Some(pod) => self.ok(pod), + None => status(404, "NotFound"), + } + } + (Method::POST, ["api", "v1", "namespaces", _, "events"]) => self.ok(&body), + _ => status(405, "MethodNotAllowed"), + } + } +} + +/// A client backed by `state`. +pub fn client(state: &Shared) -> Client { + let state = Arc::clone(state); + let service = tower::service_fn(move |request: Request| { + let state = Arc::clone(&state); + async move { + let (parts, body) = request.into_parts(); + let bytes = http_body_util::BodyExt::collect(body) + .await + .map(|b| b.to_bytes().to_vec()) + .unwrap_or_default(); + let response = + state + .lock() + .unwrap() + .handle(&parts.method, &parts.uri.to_string(), &bytes); + match response { + Some(response) => Ok::<_, std::convert::Infallible>(response), + None => std::future::pending().await, + } + } + }); + Client::new(service, "agents") +} diff --git a/crates/buzz-sandbox-reaper/src/reconcile.rs b/crates/buzz-sandbox-reaper/src/reconcile.rs index c8cda7d5a67..20071e3181d 100644 --- a/crates/buzz-sandbox-reaper/src/reconcile.rs +++ b/crates/buzz-sandbox-reaper/src/reconcile.rs @@ -784,3 +784,7 @@ mod deadline_tests { assert!(!health.ready()); } } + +#[cfg(test)] +#[path = "reconcile_tests.rs"] +mod tests; diff --git a/crates/buzz-sandbox-reaper/src/reconcile_tests.rs b/crates/buzz-sandbox-reaper/src/reconcile_tests.rs new file mode 100644 index 00000000000..bf16db566bf --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/reconcile_tests.rs @@ -0,0 +1,1122 @@ +//! Reconcile passes against an in-memory apiserver over the real +//! `kube::Client` path. Every test asserts on the requests actually sent. + +use super::*; +use crate::evidence::tests::FakeStore; +use buzz_backend_kubernetes::lifecycle::{BIND_TIMEOUT_SECS, MISSING_CHECKPOINT_HOLD_SECS, OWNER}; +use proptest::prelude::*; +use serde_json::Value; + +#[path = "fake_apiserver.rs"] +mod fake; + +const NS: &str = "buzz-agent-staging-dev"; +const GENERATION_ID: &str = "0123456789abcdef0123456789abcdef"; + +fn owner() -> String { + "a".repeat(64) +} + +fn config() -> Namespace { + Namespace { + developer: "dev".into(), + owner_pubkey: owner(), + checkpoint_prefix: "dev/".into(), + } +} + +fn key() -> String { + format!("dev/{GENERATION_ID}/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz") +} + +fn receipt(key: &str) -> String { + json!({"version": 1, "checkpoint": key}).to_string() +} + +fn ago(state: &fake::State, secs: i64) -> String { + (state.now - chrono::Duration::seconds(secs)).to_rfc3339_opts(SecondsFormat::Secs, true) +} + +/// A managed, owner-matching Sandbox `age` seconds old. +fn sandbox(state: &fake::State, name: &str, age: i64, mode: &str, annotations: Value) -> Value { + let mut all = json!({ + (OWNER): owner(), + (GENERATION): GENERATION_ID, + (CHECKPOINTING): "enabled", + }); + fake::merge(&mut all, &annotations); + json!({ + "apiVersion": "agents.x-k8s.io/v1beta1", "kind": "Sandbox", + "metadata": { + "name": name, "namespace": NS, "uid": format!("sb-{name}"), + "resourceVersion": "1", "creationTimestamp": ago(state, age), + "labels": {(LABEL_MANAGED_BY): MANAGED_BY, (LABEL_BINDING_VERSION): "2"}, + "annotations": all, + }, + "spec": {"operatingMode": mode}, + }) +} + +/// The Sandbox-owned Pod named `name`, `age` seconds old, with this agent +/// container state. +fn pod(state: &fake::State, name: &str, uid: &str, age: i64, container: Value) -> Value { + json!({ + "apiVersion": "v1", "kind": "Pod", + "metadata": { + "name": name, "namespace": NS, "uid": uid, + "creationTimestamp": ago(state, age), + "labels": {(LABEL_MANAGED_BY): MANAGED_BY, (LABEL_BINDING_VERSION): "2"}, + "ownerReferences": [{"apiVersion": "agents.x-k8s.io/v1beta1", "kind": "Sandbox", + "name": name, "uid": format!("sb-{name}"), "controller": true}], + }, + "status": {"containerStatuses": [{"name": "agent", "image": "i", "imageID": "", + "ready": false, "restartCount": 0, "state": container}]}, + }) +} + +fn terminated(message: Option<&str>) -> Value { + let mut state = json!({"terminated": {"exitCode": 0, "finishedAt": "2027-01-15T08:00:00Z"}}); + if let Some(message) = message { + state["terminated"]["message"] = message.into(); + } + state +} + +fn reconciler(state: &fake::Shared, store: FakeStore) -> Reconciler { + Reconciler { + client: fake::client(state), + store, + dry_run: false, + } +} + +fn store_with(keys: &[String]) -> FakeStore { + FakeStore { + keys: keys.to_vec(), + ..Default::default() + } +} + +/// Every non-GET request except Event creation. +fn writes(state: &fake::Shared) -> Vec { + state + .lock() + .unwrap() + .calls + .iter() + .filter(|c| c.method != http::Method::GET && !c.path.ends_with("/events")) + .cloned() + .collect() +} + +fn events(state: &fake::Shared) -> Vec { + state + .lock() + .unwrap() + .calls + .iter() + .filter(|c| c.method == http::Method::POST && c.path.ends_with("/events")) + .map(|c| c.body.clone()) + .collect() +} + +fn stored(state: &fake::Shared, name: &str) -> Value { + state.lock().unwrap().sandboxes[name].clone() +} + +fn observe(memory: &mut Memory, pod: Value) { + memory.observe_pod(&serde_json::from_value(pod).unwrap()); +} + +async fn pass(reconciler: &Reconciler, memory: &mut Memory) -> Result { + reconciler.pass(NS, &config(), memory).await +} + +#[tokio::test] +async fn a_completed_session_with_a_verified_receipt_is_tombstoned_once() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + let p = pod(&s, "s", "pod-1", 600, terminated(Some(&receipt(&key())))); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let r = reconciler(&state, store_with(&[key()])); + let mut memory = Memory::default(); + let summary = pass(&r, &mut memory).await.unwrap(); + assert_eq!(summary.tombstoned, 1); + let sent = writes(&state); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].method, http::Method::PATCH); + // One CAS merge patch carrying the observed resourceVersion. + assert_eq!(sent[0].body["metadata"]["resourceVersion"], "1"); + let ended = stored(&state, "s"); + let now = state.lock().unwrap().now; + assert_eq!(ended["spec"]["operatingMode"], "Suspended"); + let annotations = &ended["metadata"]["annotations"]; + assert_eq!(annotations[LIFECYCLE], LIFECYCLE_ENDED); + assert_eq!(annotations[ENDED_REASON], "completed"); + assert_eq!(annotations[CHECKPOINT], key()); + // ended-at is the apiserver's clock, not ours. + assert_eq!( + annotations[ENDED_AT], + now.to_rfc3339_opts(SecondsFormat::Secs, true) + ); + assert_eq!(events(&state).len(), 1); + // The next pass sees an unexpired tombstone and does nothing. + let summary = pass(&r, &mut memory).await.unwrap(); + assert_eq!(summary.tombstoned, 0); + assert_eq!(writes(&state).len(), 1); +} + +/// The hold runs from the reaper's first apiserver-time observation. A +/// node-written `finishedAt` two days in the past does not shorten it. +#[tokio::test] +async fn a_completed_session_without_a_checkpoint_is_held_for_24_hours() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + let mut p = pod(&s, "s", "pod-1", 600, terminated(None)); + p["status"]["containerStatuses"][0]["state"]["terminated"]["finishedAt"] = + ago(&s, 48 * 3600).into(); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let r = reconciler(&state, FakeStore::default()); + let mut memory = Memory::default(); + let summary = pass(&r, &mut memory).await.unwrap(); + assert_eq!(summary.held, 1); + assert!(writes(&state).is_empty()); + let warned = events(&state); + assert_eq!(warned.len(), 1); + assert_eq!(warned[0]["reason"], "CheckpointMissing"); + // Still held one second before the hold ends; a repeated hold does not + // repeat its Event. + state.lock().unwrap().now += chrono::Duration::seconds(24 * 3600 - 1); + assert_eq!(pass(&r, &mut memory).await.unwrap().held, 1); + assert!(writes(&state).is_empty()); + assert_eq!(events(&state).len(), 1); + state.lock().unwrap().now += chrono::Duration::seconds(1); + assert_eq!(pass(&r, &mut memory).await.unwrap().tombstoned, 1); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][CHECKPOINT], + "missing" + ); +} + +/// A recovered session whose restore failed exits without a receipt. Its +/// tombstone carries the checkpoint it was recovered from, so the developer +/// can retry recovery instead of losing the original work. +#[tokio::test] +async fn a_failed_recovery_carries_its_restore_checkpoint_forward() { + const PREVIOUS: &str = "fedcba9876543210fedcba9876543210"; + let original = format!("dev/{PREVIOUS}/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"); + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox( + &s, + "s", + 600, + "Running", + json!({ + (INITIAL_POD): "pod-1", + (RESTORE_CHECKPOINT): original, + (RECOVERED_FROM): PREVIOUS, + }), + ); + let mut p = pod(&s, "s", "pod-1", 600, terminated(None)); + p["status"]["containerStatuses"][0]["state"]["terminated"]["finishedAt"] = + ago(&s, 60).into(); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let r = reconciler(&state, store_with(std::slice::from_ref(&original))); + let mut memory = Memory::default(); + // Still held and reported: this session saved nothing of its own. + assert_eq!(pass(&r, &mut memory).await.unwrap().held, 1); + assert_eq!(events(&state)[0]["reason"], "CheckpointMissing"); + state.lock().unwrap().now += chrono::Duration::seconds(24 * 3600); + assert_eq!(pass(&r, &mut memory).await.unwrap().tombstoned, 1); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][CHECKPOINT], + original + ); +} + +/// `stop` suspends the Sandbox; the controller deletes the Pod after the +/// harness wrote its receipt. The watch captured it, so the tombstone +/// records the checkpoint even though the Pod is gone. +#[tokio::test] +async fn a_stopped_session_records_the_receipt_its_deleted_pod_left_behind() { + let state = fake::Shared::default(); + let gone_pod; + { + let mut s = state.lock().unwrap(); + let sb = sandbox( + &s, + "s", + 600, + "Suspended", + json!({(INITIAL_POD): "pod-1", (LIFECYCLE): "ending", (ENDED_REASON): "stopped"}), + ); + gone_pod = pod(&s, "s", "pod-1", 600, terminated(Some(&receipt(&key())))); + s.sandboxes.insert("s".into(), sb); + } + let r = reconciler(&state, store_with(&[key()])); + let mut memory = Memory::default(); + observe(&mut memory, gone_pod); + assert_eq!(pass(&r, &mut memory).await.unwrap().tombstoned, 1); + let annotations = stored(&state, "s")["metadata"]["annotations"].clone(); + assert_eq!(annotations[ENDED_REASON], "stopped"); + assert_eq!(annotations[CHECKPOINT], key()); +} + +#[tokio::test] +async fn a_receipt_s3_does_not_confirm_is_recorded_as_missing() { + let state = fake::Shared::default(); + let planted; + { + let mut s = state.lock().unwrap(); + let sb = sandbox( + &s, + "s", + 600, + "Suspended", + json!({(INITIAL_POD): "pod-1", (LIFECYCLE): "ending", (ENDED_REASON): "stopped"}), + ); + planted = pod(&s, "s", "pod-1", 600, terminated(Some(&receipt(&key())))); + s.sandboxes.insert("s".into(), sb); + } + let r = reconciler(&state, FakeStore::default()); + let mut memory = Memory::default(); + observe(&mut memory, planted); + pass(&r, &mut memory).await.unwrap(); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][CHECKPOINT], + "missing" + ); +} + +#[tokio::test] +async fn an_s3_outage_defers_the_tombstone() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + let p = pod(&s, "s", "pod-1", 600, terminated(Some(&receipt(&key())))); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let store = FakeStore { + keys: vec![key()], + fail: true, + ..Default::default() + }; + let r = reconciler(&state, store); + assert!(pass(&r, &mut Memory::default()).await.is_err()); + assert!(writes(&state).is_empty()); +} + +#[tokio::test] +async fn a_replaced_session_is_tombstoned_with_the_old_pods_receipt() { + let state = fake::Shared::default(); + let old_pod; + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + old_pod = pod(&s, "s", "pod-1", 600, terminated(Some(&receipt(&key())))); + let new_pod = pod(&s, "s", "pod-2", 5, json!({"running": {}})); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), new_pod); + } + let r = reconciler(&state, store_with(&[key()])); + let mut memory = Memory::default(); + observe(&mut memory, old_pod); + pass(&r, &mut memory).await.unwrap(); + let annotations = stored(&state, "s")["metadata"]["annotations"].clone(); + assert_eq!(annotations[ENDED_REASON], "replaced"); + assert_eq!(annotations[CHECKPOINT], key()); +} + +#[tokio::test] +async fn a_lost_pod_is_tombstoned_only_after_the_grace() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + s.sandboxes.insert("s".into(), sb); + } + let r = reconciler(&state, FakeStore::default()); + let mut memory = Memory::default(); + pass(&r, &mut memory).await.unwrap(); + state.lock().unwrap().now += chrono::Duration::seconds(59); + pass(&r, &mut memory).await.unwrap(); + assert!(writes(&state).is_empty()); + state.lock().unwrap().now += chrono::Duration::seconds(1); + pass(&r, &mut memory).await.unwrap(); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][ENDED_REASON], + "lost" + ); +} + +#[tokio::test] +async fn abandoned_bindings_are_tombstoned_without_probing_s3() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let unbound = sandbox(&s, "unbound", 1200, "Running", json!({})); + let young = sandbox(&s, "young", 1199, "Running", json!({})); + let stuck = sandbox( + &s, + "stuck", + 1300, + "Running", + json!({(INITIAL_POD): "pod-s"}), + ); + let stuck_pod = pod(&s, "stuck", "pod-s", 1200, json!({"waiting": {}})); + s.sandboxes.insert("unbound".into(), unbound); + s.sandboxes.insert("young".into(), young); + s.sandboxes.insert("stuck".into(), stuck); + s.pods.insert("stuck".into(), stuck_pod); + } + let r = reconciler(&state, FakeStore::default()); + pass(&r, &mut Memory::default()).await.unwrap(); + assert_eq!( + stored(&state, "unbound")["metadata"]["annotations"][ENDED_REASON], + "bind-abandoned" + ); + assert_eq!( + stored(&state, "stuck")["metadata"]["annotations"][ENDED_REASON], + "binding-failed" + ); + assert!(stored(&state, "young")["metadata"]["annotations"] + .get(LIFECYCLE) + .is_none()); + assert!(r.store.probes.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn sessions_without_checkpointing_are_tombstoned_as_unconfigured() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox( + &s, + "s", + 600, + "Running", + json!({(INITIAL_POD): "pod-1", (CHECKPOINTING): "disabled"}), + ); + let p = pod(&s, "s", "pod-1", 600, terminated(None)); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let r = reconciler(&state, FakeStore::default()); + pass(&r, &mut Memory::default()).await.unwrap(); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][CHECKPOINT], + "unconfigured" + ); +} + +/// The CAS loses to a concurrent writer that already tombstoned the +/// session: the reaper re-reads, re-classifies (Ended), and writes nothing +/// further instead of replaying its stale tombstone. +#[tokio::test] +async fn a_cas_conflict_re_reads_and_re_classifies() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox( + &s, + "s", + 600, + "Running", + json!({(INITIAL_POD): "pod-1", (CHECKPOINTING): "disabled"}), + ); + let p = pod(&s, "s", "pod-1", 600, terminated(None)); + let now = s.now.to_rfc3339_opts(SecondsFormat::Secs, true); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + s.race = Some(( + "s".into(), + json!({"spec": {"operatingMode": "Suspended"}, + "metadata": {"annotations": {(LIFECYCLE): "ended", (ENDED_REASON): "stopped", + (ENDED_AT): now, (CHECKPOINT): "missing"}}}), + )); + } + let r = reconciler(&state, FakeStore::default()); + let summary = pass(&r, &mut Memory::default()).await.unwrap(); + assert_eq!(summary.conflicts, 1); + assert_eq!(summary.tombstoned, 0); + assert_eq!(summary.states.get("ended"), Some(&1)); + assert_eq!(writes(&state).len(), 1, "only the losing CAS attempt"); + // The peer's tombstone stands. + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][ENDED_REASON], + "stopped" + ); +} + +#[tokio::test] +async fn expired_tombstones_are_deleted_with_uid_and_version_preconditions() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let tombstone = |s: &fake::State, name: &str, ended_ago: i64| { + sandbox( + s, + name, + ended_ago + 600, + "Suspended", + json!({(LIFECYCLE): "ended", (ENDED_AT): ago(s, ended_ago), + (ENDED_REASON): "stopped", (CHECKPOINT): "missing"}), + ) + }; + let old = tombstone(&s, "old", 30 * 24 * 3600); + let recent = tombstone(&s, "recent", 30 * 24 * 3600 - 1); + s.sandboxes.insert("old".into(), old); + s.sandboxes.insert("recent".into(), recent); + } + let r = reconciler(&state, FakeStore::default()); + assert_eq!(pass(&r, &mut Memory::default()).await.unwrap().deleted, 1); + let sent = writes(&state); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].method, http::Method::DELETE); + assert!(sent[0].path.ends_with("/sandboxes/old")); + assert_eq!( + sent[0].body["preconditions"], + json!({"uid": "sb-old", "resourceVersion": "1"}) + ); + assert!(state.lock().unwrap().sandboxes.contains_key("recent")); +} + +#[tokio::test] +async fn foreign_owners_are_left_alone_and_reported_once() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let foreign = sandbox(&s, "s", 5000, "Running", json!({(OWNER): "b".repeat(64)})); + s.sandboxes.insert("s".into(), foreign); + // Unmarked objects are not even listed. + let mut unmarked = sandbox(&s, "u", 5000, "Running", json!({})); + unmarked["metadata"]["labels"] = json!({}); + s.sandboxes.insert("u".into(), unmarked); + } + let r = reconciler(&state, FakeStore::default()); + let mut memory = Memory::default(); + pass(&r, &mut memory).await.unwrap(); + pass(&r, &mut memory).await.unwrap(); + assert!(writes(&state).is_empty()); + let reported = events(&state); + assert_eq!(reported.len(), 1); + assert_eq!(reported[0]["reason"], "SandboxIgnored"); + assert_eq!(reported[0]["involvedObject"]["name"], "s"); +} + +#[tokio::test] +async fn without_an_apiserver_date_the_pass_writes_nothing() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + s.no_date = true; + let sb = sandbox(&s, "s", 5000, "Running", json!({})); + s.sandboxes.insert("s".into(), sb); + } + let r = reconciler(&state, FakeStore::default()); + assert!(pass(&r, &mut Memory::default()).await.is_err()); + assert!(writes(&state).is_empty()); +} + +#[tokio::test] +async fn dry_run_decides_but_writes_nothing() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 5000, "Running", json!({})); + s.sandboxes.insert("s".into(), sb); + } + let r = Reconciler { + dry_run: true, + ..reconciler(&state, FakeStore::default()) + }; + pass(&r, &mut Memory::default()).await.unwrap(); + assert!(writes(&state).is_empty()); + assert!(events(&state).is_empty()); +} + +#[test] +fn receipt_memory_is_bounded_and_pruned_to_bound_pods() { + let s = fake::State::default(); + let mut memory = Memory::default(); + for i in 0..MAX_RECEIPTS + 10 { + observe( + &mut memory, + pod(&s, "s", &format!("pod-{i}"), 0, terminated(Some("{}"))), + ); + } + assert_eq!(memory.receipts.len(), MAX_RECEIPTS); + // Oversized and blank messages are not receipts. + let mut fresh = Memory::default(); + for message in ["x".repeat(MAX_RECEIPT_BYTES + 1), " ".into()] { + observe( + &mut fresh, + pod(&s, "s", "pod-x", 0, terminated(Some(&message))), + ); + } + assert!(fresh.receipts.is_empty()); + memory.retain(&HashSet::new(), &HashSet::from(["pod-3".to_string()])); + assert_eq!( + memory.receipts.keys().collect::>(), + [&"pod-3".to_string()] + ); +} + +// ── Never resurrect ──────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +struct Session { + owner_matches: bool, + mode: &'static str, + lifecycle: Option<&'static str>, + age: i64, + ended_ago: Option, + bound: bool, + /// (uid matches the binding, Pod age, container: waiting/running/terminated) + pod: Option<(bool, i64, u8)>, + checkpointing: bool, + receipt: bool, +} + +fn any_session() -> impl Strategy { + // Weighted toward owner-matching, Running, live-lifecycle sessions near + // the bind timeout, so live sessions are common in every case. + ( + ( + proptest::bool::weighted(0.8), + prop_oneof![3 => Just("Running"), 1 => Just("Suspended"), 1 => Just("Paused")], + prop_oneof![ + 3 => Just(None), + 1 => Just(Some("ending")), + 1 => Just(Some("ended")), + 1 => Just(Some("bogus")) + ], + prop_oneof![0i64..2400, 0i64..3_000_000], + proptest::option::of(0i64..3_000_000), + ), + ( + any::(), + proptest::option::weighted(0.7, (any::(), 0i64..3000, 0u8..3)), + any::(), + any::(), + ), + ) + .prop_map( + |( + (owner_matches, mode, lifecycle, age, ended_ago), + (bound, pod, checkpointing, receipt), + )| { + Session { + owner_matches, + mode, + lifecycle, + age, + ended_ago, + bound, + pod, + checkpointing, + receipt, + } + }, + ) +} + +fn install(s: &mut fake::State, i: usize, session: &Session) { + let name = format!("s{i}"); + let mut annotations = json!({ + (CHECKPOINTING): if session.checkpointing { "enabled" } else { "disabled" }, + }); + if !session.owner_matches { + annotations[OWNER] = "c".repeat(64).into(); + } + if let Some(lifecycle) = session.lifecycle { + annotations[LIFECYCLE] = lifecycle.into(); + } + if let Some(ended) = session.ended_ago { + annotations[ENDED_AT] = ago(s, ended).into(); + } + if session.bound { + annotations[INITIAL_POD] = format!("pod-{i}-a").into(); + } + let sb = sandbox(s, &name, session.age, session.mode, annotations); + s.sandboxes.insert(name.clone(), sb); + if let Some((same_uid, age, container)) = session.pod { + let uid = if same_uid { + format!("pod-{i}-a") + } else { + format!("pod-{i}-b") + }; + let container = match container { + 0 => json!({"waiting": {}}), + 1 => json!({"running": {}}), + _ => terminated(session.receipt.then(|| receipt(&key())).as_deref()), + }; + let p = pod(s, &name, &uid, age, container); + s.pods.insert(name, p); + } +} + +impl Session { + /// Must no pass tombstone this session, `elapsed` seconds after install? + /// Active (bound, same-uid Pod running), still binding (unbound younger + /// than the bind timeout, or a never-started bound Pod younger than it), + /// or draining (a stop with its Pod still present). + fn live(&self, elapsed: i64) -> bool { + let young = |age: i64| age + elapsed < BIND_TIMEOUT_SECS; + match (self.lifecycle, self.bound, self.pod) { + (Some("ending"), _, Some(_)) => true, + (None, true, Some((true, _, 1))) => true, + (None, true, Some((true, pod_age, 0))) => young(pod_age), + (None, false, _) => young(self.age), + _ => false, + } + } +} + +/// Every write in `calls` is a tombstone of a session that is owner-matching, +/// not already ended, and not `live`, or the deletion of an expired one. +fn check_writes( + calls: &[fake::Call], + before: &BTreeMap, + live: &HashSet, + now: DateTime, +) -> Result<(), TestCaseError> { + for call in calls + .iter() + .filter(|c| c.method != http::Method::GET && !c.path.ends_with("/events")) + { + let name = call.path.rsplit('/').next().unwrap().to_string(); + let original = &before[&name]; + match call.method { + http::Method::PATCH => { + prop_assert!(!live.contains(&name), "tombstoned live session {}", name); + let body = &call.body; + prop_assert_eq!(&body["spec"], &json!({"operatingMode": "Suspended"})); + prop_assert!(body["metadata"]["resourceVersion"].is_string()); + let annotations = body["metadata"]["annotations"].as_object().unwrap(); + prop_assert_eq!(&annotations[LIFECYCLE], LIFECYCLE_ENDED); + let keys: HashSet<&str> = annotations.keys().map(String::as_str).collect(); + prop_assert_eq!( + keys, + HashSet::from([LIFECYCLE, ENDED_REASON, ENDED_AT, CHECKPOINT]) + ); + prop_assert_eq!(&original["metadata"]["annotations"][OWNER], &json!(owner())); + prop_assert_ne!(&original["metadata"]["annotations"][LIFECYCLE], "ended"); + } + http::Method::DELETE => { + let annotations = &original["metadata"]["annotations"]; + prop_assert_eq!(&original["spec"]["operatingMode"], "Suspended"); + prop_assert_eq!(&annotations[LIFECYCLE], "ended"); + let ended = DateTime::parse_from_rfc3339(annotations[ENDED_AT].as_str().unwrap()) + .unwrap() + .with_timezone(&Utc); + prop_assert!(now - ended >= chrono::Duration::days(30)); + } + ref other => prop_assert!(false, "unexpected write {} {}", other, call.path), + } + } + Ok(()) +} + +proptest! { + #![proptest_config(ProptestConfig::with_cases(96))] + + /// Over arbitrary namespaces and two passes with the clock advanced in + /// between (reaching the lost grace, the 24h hold, and tombstone + /// expiry), no pass creates a Sandbox or writes `Running`, tombstones a + /// live session, writes anything but the tombstone shape to an + /// owner-matching session that is not already ended, or deletes anything + /// but a Suspended `ended` tombstone past retention. + #[test] + fn no_pass_resurrects_or_discards_a_live_session( + sessions in proptest::collection::vec(any_session(), 1..6), + verified in any::(), + advance in prop_oneof![Just(60i64), Just(MISSING_CHECKPOINT_HOLD_SECS), 0i64..3_000_000], + ) { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + for (i, session) in sessions.iter().enumerate() { + install(&mut s, i, session); + } + } + let live = |elapsed: i64| -> HashSet { + sessions + .iter() + .enumerate() + .filter(|(_, session)| session.live(elapsed)) + .map(|(i, _)| format!("s{i}")) + .collect() + }; + let store = if verified { store_with(&[key()]) } else { FakeStore::default() }; + let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + let _runtime = runtime.enter(); + let r = reconciler(&state, store); + let mut memory = Memory::default(); + for elapsed in [0, advance] { + let (before, now, seen) = { + let mut s = state.lock().unwrap(); + s.now += chrono::Duration::seconds(elapsed); + (s.sandboxes.clone(), s.now, s.calls.len()) + }; + runtime.block_on(pass(&r, &mut memory)).unwrap(); + let calls = state.lock().unwrap().calls[seen..].to_vec(); + check_writes(&calls, &before, &live(elapsed), now)?; + } + } +} + +#[tokio::test] +async fn repeated_failed_recovery_preserves_original_checkpoint() { + const ORIGINAL: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + const PREVIOUS: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + let original = format!("dev/{ORIGINAL}/saved.tar.gz"); + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + // Generation B inherited A's checkpoint. Recovery created C with + // recovered-from=B and restore-checkpoint=A's key, as recover() writes. + let sb = sandbox( + &s, + "s", + 600, + "Suspended", + json!({ + (INITIAL_POD): "pod-c", + (LIFECYCLE): "ending", + (RESTORE_CHECKPOINT): original, + (RECOVERED_FROM): PREVIOUS, + }), + ); + s.sandboxes.insert("s".into(), sb); + } + let r = reconciler(&state, store_with(std::slice::from_ref(&original))); + assert_eq!( + pass(&r, &mut Memory::default()).await.unwrap().tombstoned, + 1 + ); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][CHECKPOINT], + original, + "a second failed recovery must not orphan the original checkpoint" + ); +} + +#[tokio::test] +async fn failed_final_save_does_not_verify_an_earlier_turn_receipt() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + // A background save populated /dev/termination-log. The final save + // failed and the harness exited 1, leaving that same receipt in place. + let mut p = pod(&s, "s", "pod-1", 600, terminated(Some(&receipt(&key())))); + p["status"]["containerStatuses"][0]["state"]["terminated"]["exitCode"] = 1.into(); + p["status"]["containerStatuses"][0]["state"]["terminated"]["finishedAt"] = + ago(&s, 10).into(); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let r = reconciler(&state, store_with(&[key()])); + let result = pass(&r, &mut Memory::default()).await.unwrap(); + assert_eq!( + result.held, 1, + "retain the unsaved workspace for the 24h hold" + ); + assert_eq!(result.tombstoned, 0); +} + +#[tokio::test] +async fn denied_tombstone_is_a_failed_pass() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox( + &s, + "s", + 600, + "Suspended", + json!({ + (LIFECYCLE): "ending", (CHECKPOINTING): "disabled" + }), + ); + s.sandboxes.insert("s".into(), sb); + s.reject_writes = true; + } + let r = reconciler(&state, FakeStore::default()); + assert!( + pass(&r, &mut Memory::default()).await.is_err(), + "RBAC-denied writes must fail --once and not mark readiness successful" + ); +} + +/// An Ending session whose Pod was deleted while the reaper was down: memory +/// holds no receipt, so the checkpoint comes from listing the session's own +/// prefix. Before the listing fallback this recorded `missing` and made the +/// saved work unrecoverable. +#[tokio::test] +async fn a_restarted_reaper_lists_the_checkpoint_of_a_deleted_pod() { + let ending = |s: &fake::State| { + sandbox( + s, + "s", + 600, + "Suspended", + json!({(INITIAL_POD): "pod-1", (LIFECYCLE): "ending", (ENDED_REASON): "stopped"}), + ) + }; + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = ending(&s); + s.sandboxes.insert("s".into(), sb); + } + let foreign = "dev/ffffffffffffffffffffffffffffffff/later.tar.gz".to_string(); + let r = reconciler(&state, store_with(&[key(), foreign])); + assert_eq!( + pass(&r, &mut Memory::default()).await.unwrap().tombstoned, + 1 + ); + let annotations = &stored(&state, "s")["metadata"]["annotations"]; + assert_eq!(annotations[ENDED_REASON], "stopped"); + assert_eq!(annotations[CHECKPOINT], key()); + assert_eq!( + r.store.probes.lock().unwrap().as_slice(), + [format!("dev/{GENERATION_ID}/")] + ); + + // S3 down: no tombstone this pass, rather than a false `missing`. + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = ending(&s); + s.sandboxes.insert("s".into(), sb); + } + let store = FakeStore { + keys: vec![key()], + fail: true, + ..Default::default() + }; + let r = reconciler(&state, store); + assert!(pass(&r, &mut Memory::default()).await.is_err()); + assert!(writes(&state).is_empty()); +} + +/// A generation the provider could not have written never names a prefix +/// to list or verify: the session ends with `missing`, without asking S3. +#[tokio::test] +async fn an_invalid_generation_is_recorded_as_missing_without_probing() { + const BAD: &str = "0123456789ABCDEF0123456789ABCDEF"; + let planted = format!("dev/{BAD}/x.tar.gz"); + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox( + &s, + "s", + 600, + "Suspended", + json!({(INITIAL_POD): "pod-1", (LIFECYCLE): "ending", (GENERATION): BAD}), + ); + s.sandboxes.insert("s".into(), sb); + } + let r = reconciler(&state, store_with(std::slice::from_ref(&planted))); + let mut memory = Memory::default(); + let gone = pod( + &state.lock().unwrap(), + "s", + "pod-1", + 600, + terminated(Some(&receipt(&planted))), + ); + observe(&mut memory, gone); + pass(&r, &mut memory).await.unwrap(); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][CHECKPOINT], + "missing" + ); + assert!(r.store.probes.lock().unwrap().is_empty()); +} + +/// The exact receipt bytes buzz-acp writes (#13's wire format), not a +/// re-serialization of the reaper's own type. +#[tokio::test] +async fn the_harness_receipt_wire_format_is_accepted() { + const KEY: &str = + "dev/0123456789abcdef0123456789abcdef/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"; + const RECEIPT: &str = r#"{"version":1,"checkpoint":"dev/0123456789abcdef0123456789abcdef/6f9619ff-8b86-d011-b42d-00cf4fc964ff.tar.gz"}"#; + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 600, "Running", json!({(INITIAL_POD): "pod-1"})); + let p = pod(&s, "s", "pod-1", 600, terminated(Some(RECEIPT))); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), p); + } + let r = reconciler(&state, store_with(&[KEY.to_string()])); + assert_eq!( + pass(&r, &mut Memory::default()).await.unwrap().tombstoned, + 1 + ); + assert_eq!( + stored(&state, "s")["metadata"]["annotations"][CHECKPOINT], + KEY + ); +} + +/// Recovery replaces the expired tombstone (new uid, Running) between the +/// reaper's read and its DELETE. The uid precondition fails, and the reaper +/// re-reads and leaves the new session alone. +#[tokio::test] +async fn a_recovery_racing_the_expiry_delete_is_left_alone() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let old = sandbox( + &s, + "s", + 31 * 24 * 3600, + "Suspended", + json!({(LIFECYCLE): "ended", (ENDED_AT): ago(&s, 30 * 24 * 3600), + (ENDED_REASON): "stopped", (CHECKPOINT): "missing"}), + ); + let now = ago(&s, 0); + s.sandboxes.insert("s".into(), old); + s.race = Some(( + "s".into(), + json!({"metadata": {"uid": "sb-s-recovered", "creationTimestamp": now, + "annotations": {(LIFECYCLE): null, (ENDED_AT): null, (ENDED_REASON): null, + (CHECKPOINT): null, (RECOVERED_FROM): GENERATION_ID, + (GENERATION): "fedcba9876543210fedcba9876543210"}}, + "spec": {"operatingMode": "Running"}}), + )); + } + let r = reconciler(&state, FakeStore::default()); + let summary = pass(&r, &mut Memory::default()).await.unwrap(); + assert_eq!(summary.conflicts, 1); + assert_eq!(summary.deleted, 0); + assert_eq!(summary.states.get("binding"), Some(&1)); + let sent = writes(&state); + assert_eq!(sent.len(), 1, "only the refused DELETE"); + assert_eq!(sent[0].method, http::Method::DELETE); + let live = stored(&state, "s"); + assert_eq!(live["metadata"]["uid"], "sb-s-recovered"); + assert_eq!(live["spec"]["operatingMode"], "Running"); +} + +/// A deploy binds the first Pod while the reaper's bind-abandoned tombstone +/// is in flight. The CAS fails, and the re-read session is Active. +#[tokio::test] +async fn a_bind_racing_the_abandoned_tombstone_is_left_alone() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let unbound = sandbox(&s, "s", 1300, "Running", json!({})); + let running = pod(&s, "s", "pod-1", 30, json!({"running": {}})); + s.sandboxes.insert("s".into(), unbound); + s.pods.insert("s".into(), running); + s.race = Some(( + "s".into(), + json!({"metadata": {"annotations": {(INITIAL_POD): "pod-1"}}}), + )); + } + let r = reconciler(&state, FakeStore::default()); + let summary = pass(&r, &mut Memory::default()).await.unwrap(); + assert_eq!(summary.conflicts, 1); + assert_eq!(summary.tombstoned, 0); + assert_eq!(summary.states.get("active"), Some(&1)); + let sent = writes(&state); + assert_eq!(sent.len(), 1, "only the refused PATCH"); + assert_eq!( + sent[0].body["metadata"]["annotations"][ENDED_REASON], + "bind-abandoned" + ); + let live = stored(&state, "s"); + assert_eq!(live["spec"]["operatingMode"], "Running"); + assert!(live["metadata"]["annotations"].get(LIFECYCLE).is_none()); +} + +/// A Sandbox that changes before every write is retried MAX_ATTEMPTS times, +/// then deferred to the next pass as a failure. +#[tokio::test] +async fn a_sandbox_that_keeps_changing_is_deferred_after_max_attempts() { + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + let sb = sandbox(&s, "s", 5000, "Running", json!({})); + s.sandboxes.insert("s".into(), sb); + s.race = Some(( + "s".into(), + json!({"metadata": {"annotations": {"example.com/touched": "yes"}}}), + )); + s.sticky_race = true; + } + let r = reconciler(&state, FakeStore::default()); + assert!(pass(&r, &mut Memory::default()).await.is_err()); + let sent = writes(&state); + assert_eq!(sent.len(), MAX_ATTEMPTS); + assert!(sent.iter().all(|c| c.method == http::Method::PATCH)); + assert!(stored(&state, "s")["metadata"]["annotations"] + .get(LIFECYCLE) + .is_none()); +} + +/// The namespace loop: the Pod watch's Delete event carries the receipt into +/// memory, the next pass records that receipt's key (not the newer key a +/// listing would pick), and each pass is recorded in health. +#[tokio::test(start_paused = true)] +async fn the_namespace_loop_keeps_receipts_from_watched_deletions() { + let interval = Duration::from_secs(30); + let newer = format!("dev/{GENERATION_ID}/ffffffff-0000-0000-0000-000000000000.tar.gz"); + let state = fake::Shared::default(); + { + let mut s = state.lock().unwrap(); + // A stop is draining: the Pod is still there and still running. + let sb = sandbox( + &s, + "s", + 600, + "Suspended", + json!({(INITIAL_POD): "pod-1", (LIFECYCLE): "ending", (ENDED_REASON): "stopped"}), + ); + let running = pod(&s, "s", "pod-1", 600, json!({"running": {}})); + let mut exited = pod(&s, "s", "pod-1", 600, terminated(Some(&receipt(&key())))); + exited["metadata"]["resourceVersion"] = "2".into(); + s.sandboxes.insert("s".into(), sb); + s.pods.insert("s".into(), running); + s.watch = vec![json!({"type": "DELETED", "object": exited})]; + } + let reconciler = Arc::new(reconciler(&state, store_with(&[key(), newer]))); + let health = Arc::new(Health::new([NS.to_string()], interval)); + let cancel = CancellationToken::new(); + let task = tokio::spawn(run_namespace( + Arc::clone(&reconciler), + NS.to_string(), + config(), + interval, + Arc::clone(&health), + cancel.clone(), + )); + tokio::time::sleep(interval * 3 / 2).await; + cancel.cancel(); + task.await.unwrap(); + let annotations = &stored(&state, "s")["metadata"]["annotations"]; + assert_eq!(annotations[LIFECYCLE], LIFECYCLE_ENDED); + assert_eq!(annotations[CHECKPOINT], key()); + assert!(health.live()); + assert!(health.ready()); +}