From babceda518ec16c348e61d9d6b02e4992116a647 Mon Sep 17 00:00:00 2001 From: Matt Jane Date: Thu, 15 Oct 2026 13:59:41 -0700 Subject: [PATCH 1/4] feat(provider-sync): restart the agents attached to a rotated provider A pushed value reached new requests, but a connection an agent opened with the old value kept working and a bad value failed quietly inside a long-running process. After a successful update, provider-sync now restarts each sandbox the gateway reports attached to that provider, with StopSandbox then StartSandbox, so the agent starts fresh on the new values. Sandboxes of other providers are not touched, a sandbox restarts once when several of its providers rotate together, and a sandbox that is stopped, stopping, completed or being deleted is left alone. The start is retried for up to 10 minutes so a restart never leaves an agent down. Only a value that replaces one already applied counts as a rotation. With --state-secret the salted fingerprints, never values, are kept in a Secret, so restarting provider-sync neither re-sends unchanged values nor restarts every agent. --no-restart turns restarts off. Outcomes are Kubernetes Events on the Secret: AgentRestarted, AgentRestartSkipped and AgentRestartFailed. Signed-off-by: Matt Jane --- Cargo.lock | 3 + crates/openshell-provider-sync/Cargo.toml | 3 + crates/openshell-provider-sync/README.md | 48 +++- .../openshell-provider-sync/src/controller.rs | 248 +++++++++++++++++- crates/openshell-provider-sync/src/gateway.rs | 221 +++++++++++++++- crates/openshell-provider-sync/src/lib.rs | 8 +- crates/openshell-provider-sync/src/main.rs | 51 +++- crates/openshell-provider-sync/src/state.rs | 156 ++++++++++- .../src/state_store.rs | 83 ++++++ skills/debug-openshell-cluster/SKILL.md | 12 +- 10 files changed, 795 insertions(+), 38 deletions(-) create mode 100644 crates/openshell-provider-sync/src/state_store.rs diff --git a/Cargo.lock b/Cargo.lock index 4633cdf62e..bbfcc3e409 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4899,12 +4899,15 @@ dependencies = [ "openshell-core", "openshell-sdk", "rand 0.9.4", + "serde", + "serde_json", "sha2 0.10.9", "thiserror 2.0.20", "tokio", "tonic", "tracing", "tracing-subscriber", + "uuid", ] [[package]] diff --git a/crates/openshell-provider-sync/Cargo.toml b/crates/openshell-provider-sync/Cargo.toml index 6cb67f75d7..2261b4333c 100644 --- a/crates/openshell-provider-sync/Cargo.toml +++ b/crates/openshell-provider-sync/Cargo.toml @@ -24,12 +24,15 @@ k8s-openapi = { workspace = true } kube = { workspace = true } miette = { workspace = true } rand = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } sha2 = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tonic = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +uuid = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/openshell-provider-sync/README.md b/crates/openshell-provider-sync/README.md index 68443d688d..ae5e9fa635 100644 --- a/crates/openshell-provider-sync/README.md +++ b/crates/openshell-provider-sync/README.md @@ -43,9 +43,16 @@ accident, and one Secret key may be listed only once. changed keys of that provider as a value-only `UpdateProvider`. The gateway merges them into the provider, and attached sandboxes refetch on their next provider poll, every 10s by default. -- **Running processes.** A process that is already running gets the new value - only when the credential's profile sets `stable_placeholder: true`. Without it, - only processes started after the update see the new value. +- **Restarting affected agents.** After a successful update the controller + restarts every sandbox the gateway reports attached to that provider, with + `StopSandbox` then `StartSandbox`, so the agent starts fresh and no open + connection keeps using the old value. Sandboxes of other providers are not + touched, and a sandbox restarts once even when several of its providers + rotate together. A sandbox that is stopped, stopping, completed or being + deleted is left alone. Once stopped, the start is retried for up to 10 + minutes, so a restart does not leave an agent down. Agents must therefore + tolerate a restart at any time: keep state outside the sandbox or resume + from it. `--no-restart` turns this off. - **What it leaves alone.** It never creates or deletes providers, never clears a credential when its Secret is deleted or unlabelled, and never touches credential keys that no Secret lists. Providers are still created by whatever @@ -65,10 +72,17 @@ accident, and one Secret key may be listed only once. - **One Secret per credential.** A second live Secret that maps the same workspace, provider and credential key is refused with a `ProviderSyncConflict` event. +- **Rotations, not baselines.** Only a value that replaces one this controller + applied before restarts agents. The first values it sees for a provider are a + baseline: they are sent, and nothing restarts. - **Startup and restarts.** The watch starts with a full list, so every labelled - Secret is synced at startup. State is in memory: a salted fingerprint of each - value sent, never the value. After a restart every Secret is sent once more. - That is a value-only update and does not disturb running sandboxes. + Secret is synced at startup. The controller keeps a salted fingerprint of each + value sent, never the value. With `--state-secret`, the salt and fingerprints + are kept in that Secret, so after a restart unchanged values are recognized: + nothing is re-sent and no agent restarts. Without it, every Secret is sent + once more after a restart as a baseline that restarts nothing; a rotation + that happened while the controller was down then reaches the providers but + restarts no agents. - **Retries.** - A missing provider is retried every 30s. - Transient gateway failures are retried every 10s. @@ -77,7 +91,8 @@ accident, and one Secret key may be listed only once. - An absent provider is not retried; the next change to the Secret is seen. - **Audit.** Each outcome is published as a Kubernetes Event on the Secret (`ProviderSynced`, `ProviderSyncSkipped`, `ProviderNotFound`, - `ProviderSyncConflict`, `InvalidProviderSync`, `ProviderSyncFailed`). The + `ProviderSyncConflict`, `InvalidProviderSync`, `ProviderSyncFailed`, + `AgentRestarted`, `AgentRestartSkipped`, `AgentRestartFailed`). The gateway separately audits each `UpdateProvider` with the controller's principal. Neither carries values. @@ -102,15 +117,22 @@ openshell-provider-sync \ | `--default-workspace` | `OPENSHELL_PROVIDER_SYNC_DEFAULT_WORKSPACE` | Workspace when a Secret has no workspace annotation | | `--reporter`, `--reporter-instance` | `OPENSHELL_PROVIDER_SYNC_REPORTER`, `POD_NAME` | Event reporter | | `--no-events` | | Disable Kubernetes Events | +| `--no-restart` | `OPENSHELL_PROVIDER_SYNC_NO_RESTART` | Do not restart agents after a rotation | +| `--state-secret` | `OPENSHELL_PROVIDER_SYNC_STATE_SECRET` | Secret that keeps applied-value fingerprints across restarts | +| `--state-namespace` | `POD_NAMESPACE` | Namespace of the state Secret, usually the controller's own | | `--log-json` | `OPENSHELL_LOG_JSON` | Log as JSON lines | On a gateway that requires authentication, the principal needs `provider:write` -with the workspace `admin` role in every workspace it updates. The token file +with the workspace `admin` role in every workspace it updates, plus +`sandbox:read` and `sandbox:write` there to restart agents. The token file suits a projected ServiceAccount token or a Secret-mounted token that something else rotates on disk. The ServiceAccount needs `get`, `list` and `watch` on Secrets in the watched -namespace, and `create` and `patch` on `events.k8s.io` Events. +namespace, and `create` and `patch` on `events.k8s.io` Events. With +`--state-secret` it also needs `get`, `create` and `patch` on Secrets in its own +namespace, which a `resourceNames` rule can limit to the state Secret for `get` +and `patch`. The gateway client is the Rust SDK, which speaks bearer-token auth only. A gateway that requires mTLS client certificates is not supported yet. @@ -120,5 +142,9 @@ gateway that requires mTLS client certificates is not supported yet. - **One replica.** There is no leader election. Two replicas would send the same idempotent updates twice; that is wasteful but not wrong. - **No activation wait.** The controller does not wait for supervisors to report - that they activated the update. The gateway's mutation id is logged for - correlation. + that they activated the update before restarting. A restarted sandbox fetches + the provider's current values when it starts. The gateway's mutation id is + logged for correlation. +- **Attachment at update time.** The restart set is the gateway's target set for + the update. A sandbox attached to the provider afterwards already starts with + the new values. diff --git a/crates/openshell-provider-sync/src/controller.rs b/crates/openshell-provider-sync/src/controller.rs index b0c2fb026e..cfb7ea7360 100644 --- a/crates/openshell-provider-sync/src/controller.rs +++ b/crates/openshell-provider-sync/src/controller.rs @@ -16,9 +16,10 @@ use kube::{Api, Client, Resource, ResourceExt}; use tokio::sync::Mutex; use tracing::{debug, info, warn}; -use crate::gateway::{ProviderUpdater, UpdateError}; +use crate::gateway::{ProviderUpdater, Restart, SandboxRestarter, UpdateError}; use crate::mapping::{self, Extracted, MappingError}; use crate::state::{Conflict, SecretRef, SyncState}; +use crate::state_store::StateStore; /// How long to wait before retrying a provider that does not exist yet. const PROVIDER_MISSING_RETRY: Duration = Duration::from_secs(30); @@ -35,6 +36,10 @@ pub enum Outcome { Updated { keys: Vec, mutation_id: String, + /// Sandboxes attached to the provider whose agents must restart to + /// pick up the new values. Empty for a baseline: values this + /// controller had not applied before, such as at first sight. + restart: Vec, }, /// Nothing changed since the last push. Unchanged, @@ -92,10 +97,31 @@ impl SyncReport { } } +impl SyncReport { + /// Every sandbox to restart, once each even when several of its + /// providers rotated. + pub fn sandboxes_to_restart(&self) -> Vec { + let mut sandboxes: Vec = self + .providers + .iter() + .filter_map(|(_, result)| match result { + Ok(Outcome::Updated { restart, .. }) => Some(restart.iter().cloned()), + _ => None, + }) + .flatten() + .collect(); + sandboxes.sort(); + sandboxes.dedup(); + sandboxes + } +} + /// Shared by every reconcile. pub struct Context { updater: Arc, + restarter: Option>, state: Mutex, + store: Option>, default_workspace: String, } @@ -103,11 +129,29 @@ impl Context { pub fn new(updater: Arc, default_workspace: impl Into) -> Self { Self { updater, + restarter: None, state: Mutex::new(SyncState::new()), + store: None, default_workspace: default_workspace.into(), } } + /// Restart the agents attached to a provider whose values rotated. + #[must_use] + pub fn with_restarter(mut self, restarter: Arc) -> Self { + self.restarter = Some(restarter); + self + } + + /// Resume from persisted state and keep it current, so a restart of this + /// controller does not mistake values it already applied for rotations. + #[must_use] + pub fn with_state(mut self, state: SyncState, store: Arc) -> Self { + self.state = Mutex::new(state); + self.store = Some(store); + self + } + /// Sync one Secret: parse its annotations, then for each provider read /// the mapped keys and push the ones that changed. /// @@ -153,19 +197,35 @@ impl Context { if changed.is_empty() { return Ok(Outcome::Unchanged); } - let mutation_id = self + let rotation = state.has_applied(secret_ref, &target.workspace, provider); + let updated = self .updater .update_credentials(&target.workspace, provider, &changed) .await?; state.commit(secret_ref, &target.workspace, provider, &values); Ok(Outcome::Updated { keys: changed.keys().cloned().collect(), - mutation_id, + mutation_id: updated.mutation_id, + restart: if rotation { + updated.sandboxes + } else { + Vec::new() + }, }) } .await; providers.push((provider.clone(), result)); } + if let Some(store) = &self.store + && providers + .iter() + .any(|(_, result)| matches!(result, Ok(Outcome::Updated { .. }))) + && let Err(err) = store.save(&state.to_persisted()).await + { + // In-memory state is still right. The cost of a lost save is one + // extra restart of these agents after this controller restarts. + warn!(error = %err, "failed to persist provider sync state"); + } Ok(SyncReport { workspace: target.workspace, providers, @@ -225,13 +285,18 @@ async fn reconcile(secret: Arc, runtime: Arc) -> Result { + Ok(Outcome::Updated { + keys, + mutation_id, + restart, + }) => { info!( secret = %secret_ref, workspace, provider, keys = ?keys, mutation_id, + restart = ?restart, "provider credentials updated from Secret" ); let note = format!( @@ -277,11 +342,66 @@ async fn reconcile(secret: Arc, runtime: Arc) -> Result, + runtime: Arc, + secret: Arc, + workspace: String, + sandbox: String, +) { + let secret_ref = secret_ref(&secret); + match restarter.restart(&workspace, &sandbox).await { + Ok(Restart::Restarted) => { + info!(secret = %secret_ref, workspace, sandbox, "restarted agent after credential rotation"); + let note = + format!("Restarted sandbox {workspace}/{sandbox} after its credentials rotated"); + publish(&runtime, &secret, EventType::Normal, "AgentRestarted", note).await; + } + Ok(Restart::Skipped { phase }) => { + info!(secret = %secret_ref, workspace, sandbox, phase, "agent not running; left alone"); + let note = format!("Did not restart sandbox {workspace}/{sandbox}: it is {phase}"); + publish( + &runtime, + &secret, + EventType::Normal, + "AgentRestartSkipped", + note, + ) + .await; + } + Err(err) => { + warn!(secret = %secret_ref, workspace, sandbox, error = %err, "agent restart failed"); + let note = format!("Restarting sandbox {workspace}/{sandbox} failed: {err}"); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentRestartFailed", + note, + ) + .await; + } + } +} + fn error_policy(_secret: Arc, err: &SyncError, _runtime: Arc) -> Action { Action::requeue(err.retry_after()) } @@ -363,6 +483,23 @@ mod tests { struct FakeUpdater { calls: StdMutex>, fail_with: StdMutex>, + /// Sandboxes every provider reports attached; `-agent` + /// when unset. + attached: StdMutex>>, + } + + #[derive(Default)] + struct MemoryStore(StdMutex>); + + #[async_trait::async_trait] + impl StateStore for MemoryStore { + async fn load(&self) -> Result, String> { + Ok(self.0.lock().unwrap().clone()) + } + async fn save(&self, state: &crate::state::PersistedState) -> Result<(), String> { + *self.0.lock().unwrap() = Some(state.clone()); + Ok(()) + } } #[async_trait::async_trait] @@ -372,7 +509,7 @@ mod tests { workspace: &str, provider: &str, values: &CredentialValues, - ) -> Result { + ) -> Result { let fail_with = *self.fail_with.lock().unwrap(); if let Some((only, fail)) = fail_with && only.is_none_or(|only| only == provider) @@ -384,7 +521,16 @@ mod tests { provider.to_string(), values.0.clone(), )); - Ok("mutation-1".to_string()) + let sandboxes = self + .attached + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| vec![format!("{provider}-agent")]); + Ok(crate::gateway::Updated { + mutation_id: "mutation-1".to_string(), + sandboxes, + }) } } @@ -549,6 +695,96 @@ mod tests { ); } + const PM_CREDENTIALS: [(&str, &str); 4] = [ + ("SLACK_BOT_TOKEN", "bot"), + ("SLACK_APP_TOKEN", "app"), + ("LINEAR_API_KEY", "lin"), + ("ANTHROPIC_API_KEY", "ant"), + ]; + + fn with( + base: [(&'static str, &'static str); 4], + key: &str, + value: &'static str, + ) -> [(&'static str, &'static str); 4] { + base.map(|(name, current)| { + if name == key { + (name, value) + } else { + (name, current) + } + }) + } + + #[tokio::test] + async fn a_rotation_restarts_only_the_agents_attached_to_its_provider() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + let (r, s) = agent(&PM_CREDENTIALS); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!( + report.sandboxes_to_restart().is_empty(), + "first sight is a baseline, not a rotation" + ); + + let (r, s) = agent(&with(PM_CREDENTIALS, "SLACK_BOT_TOKEN", "bot-2")); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["slack-pm-agent"]); + assert!(matches!( + outcome(&report, "slack-pm"), + Ok(Outcome::Updated { restart, .. }) if restart == &["slack-pm-agent"] + )); + assert!(matches!(outcome(&report, "linear"), Ok(Outcome::Unchanged))); + + // Unchanged values restart nothing. + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(report.sandboxes_to_restart().is_empty()); + } + + #[tokio::test] + async fn an_agent_restarts_once_when_several_of_its_providers_rotate() { + let updater = Arc::new(FakeUpdater::default()); + *updater.attached.lock().unwrap() = Some(vec!["pm".to_string()]); + let ctx = Context::new(updater.clone(), "default"); + let (r, s) = agent(&PM_CREDENTIALS); + ctx.sync(&r, &s, |_| true).await.unwrap(); + + let rotated = with( + with(PM_CREDENTIALS, "SLACK_BOT_TOKEN", "bot-2"), + "LINEAR_API_KEY", + "lin-2", + ); + let (r, s) = agent(&rotated); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["pm"]); + } + + #[tokio::test] + async fn persisted_state_keeps_a_controller_restart_from_restarting_agents() { + let store = Arc::new(MemoryStore::default()); + let updater = Arc::new(FakeUpdater::default()); + let first = + Context::new(updater.clone(), "default").with_state(SyncState::new(), store.clone()); + let (r, s) = agent(&PM_CREDENTIALS); + first.sync(&r, &s, |_| true).await.unwrap(); + updater.calls.lock().unwrap().clear(); + + // A new controller process resumes from the saved fingerprints. + let persisted = store.load().await.unwrap().expect("state saved"); + let second = Context::new(updater.clone(), "default") + .with_state(SyncState::from_persisted(persisted), store.clone()); + let report = second.sync(&r, &s, |_| true).await.unwrap(); + assert!(report.sandboxes_to_restart().is_empty()); + assert!( + updater.calls.lock().unwrap().is_empty(), + "nothing re-pushed" + ); + + let (r, s) = agent(&with(PM_CREDENTIALS, "LINEAR_API_KEY", "lin-2")); + let report = second.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["linear-agent"]); + } + #[tokio::test] async fn a_failing_provider_does_not_hold_back_the_others() { let updater = Arc::new(FakeUpdater::default()); diff --git a/crates/openshell-provider-sync/src/gateway.rs b/crates/openshell-provider-sync/src/gateway.rs index 106bbaf243..54a4ff5659 100644 --- a/crates/openshell-provider-sync/src/gateway.rs +++ b/crates/openshell-provider-sync/src/gateway.rs @@ -7,7 +7,10 @@ use std::collections::HashMap; use std::path::PathBuf; use openshell_core::proto::datamodel::v1::{ObjectMeta, Provider}; -use openshell_core::proto::{UpdateProviderRequest, workspace_selector}; +use openshell_core::proto::{ + GetSandboxRequest, ProviderResponse, SandboxPhase, StartSandboxRequest, StopSandboxRequest, + UpdateProviderRequest, workspace_selector, +}; use openshell_sdk::{OpenShellClient, Refresh, RefreshError, RefreshedToken}; use tonic::Code; @@ -43,17 +46,61 @@ impl UpdateError { } } +/// What one provider update reached. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Updated { + pub mutation_id: String, + /// Sandboxes the gateway found attached to the provider when it applied + /// the update, by name. + pub sandboxes: Vec, +} + +impl Updated { + fn from_response(response: ProviderResponse) -> Self { + let mut sandboxes: Vec = response + .target_receipts + .iter() + .filter_map(|receipt| receipt.desired.as_ref()) + .map(|desired| desired.sandbox.clone()) + .filter(|sandbox| !sandbox.is_empty()) + .collect(); + sandboxes.sort(); + sandboxes.dedup(); + Self { + mutation_id: response.mutation_id, + sandboxes, + } + } +} + /// Writes credential values to a provider. #[async_trait::async_trait] pub trait ProviderUpdater: Send + Sync + 'static { /// Merge `values` into the provider's credentials. Keys not in `values` - /// are left unchanged. Returns the gateway's mutation receipt id. + /// are left unchanged. async fn update_credentials( &self, workspace: &str, provider: &str, values: &CredentialValues, - ) -> Result; + ) -> Result; +} + +/// What restarting one sandbox did. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Restart { + Restarted, + /// Stopped, stopping, completed or being deleted: someone or something + /// else decided it should not run, so it is left alone. + Skipped { + phase: String, + }, +} + +/// Restarts a sandbox so its agent starts fresh with current credentials. +#[async_trait::async_trait] +pub trait SandboxRestarter: Send + Sync + 'static { + async fn restart(&self, workspace: &str, sandbox: &str) -> Result; } /// [`ProviderUpdater`] over the gateway's `UpdateProvider` RPC. @@ -101,7 +148,7 @@ impl ProviderUpdater for GatewayUpdater { workspace: &str, provider: &str, values: &CredentialValues, - ) -> Result { + ) -> Result { let mut retried_auth = false; loop { let mut grpc = self @@ -113,7 +160,7 @@ impl ProviderUpdater for GatewayUpdater { .update_provider(update_request(workspace, provider, values)) .await { - Ok(response) => return Ok(response.into_inner().mutation_id), + Ok(response) => return Ok(Updated::from_response(response.into_inner())), Err(status) if status.code() == Code::Unauthenticated && !retried_auth => { retried_auth = true; let refreshed = self @@ -131,6 +178,124 @@ impl ProviderUpdater for GatewayUpdater { } } +/// How long a restart keeps trying to start a sandbox it stopped. +const START_DEADLINE: std::time::Duration = std::time::Duration::from_mins(10); +/// Pause between start attempts while the sandbox is still stopping. +const START_RETRY: std::time::Duration = std::time::Duration::from_secs(5); + +/// Call one gateway RPC, refreshing the bearer token once if it is rejected. +macro_rules! call_refreshing { + ($client:expr, $method:ident, $request:expr) => {{ + let client = $client; + let first = client + .raw_grpc_fresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))? + .$method($request.clone()) + .await; + match first { + Err(status) if status.code() == Code::Unauthenticated => { + client + .force_refresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))?; + client + .raw_grpc_fresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))? + .$method($request) + .await + } + other => other, + } + }}; +} + +/// [`SandboxRestarter`] over the gateway's `StopSandbox` and `StartSandbox`. +pub struct GatewayRestarter { + client: OpenShellClient, +} + +impl GatewayRestarter { + pub fn new(client: OpenShellClient) -> Self { + Self { client } + } + + async fn phase(&self, workspace: &str, sandbox: &str) -> Result { + let request = GetSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + }; + let response = call_refreshing!(&self.client, get_sandbox, request) + .map_err(|status| UpdateError::from_status(&status, workspace, sandbox))?; + Ok(response + .into_inner() + .sandbox + .and_then(|sandbox| sandbox.status) + .map_or(SandboxPhase::Unspecified, |status| status.phase())) + } +} + +/// A sandbox in one of these phases is meant to stay down. +fn left_alone(phase: SandboxPhase) -> bool { + matches!( + phase, + SandboxPhase::Stopping + | SandboxPhase::Stopped + | SandboxPhase::Completed + | SandboxPhase::Deleting + ) +} + +#[async_trait::async_trait] +impl SandboxRestarter for GatewayRestarter { + async fn restart(&self, workspace: &str, sandbox: &str) -> Result { + let phase = self.phase(workspace, sandbox).await?; + if left_alone(phase) { + return Ok(Restart::Skipped { + phase: phase.as_str_name().to_string(), + }); + } + // Request ids make a retried stop or start a replay, not a repeat. + let stop = StopSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + request_id: uuid::Uuid::new_v4().to_string(), + }; + call_refreshing!(&self.client, stop_sandbox, stop) + .map_err(|status| UpdateError::from_status(&status, workspace, sandbox))?; + + // A stopped agent must not stay down because this restart gave up: + // keep starting until the gateway accepts or the deadline passes. + let start = StartSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + request_id: uuid::Uuid::new_v4().to_string(), + }; + let deadline = tokio::time::Instant::now() + START_DEADLINE; + loop { + match call_refreshing!(&self.client, start_sandbox, start.clone()) { + Ok(_) => return Ok(Restart::Restarted), + Err(status) + if matches!( + status.code(), + Code::FailedPrecondition | Code::Unavailable | Code::Aborted + ) && tokio::time::Instant::now() < deadline => + { + tokio::time::sleep(START_RETRY).await; + } + Err(status) => { + return Err(UpdateError::Transient(format!( + "sandbox {workspace}/{sandbox} was stopped but did not start: {:?}: {}", + status.code(), + status.message() + ))); + } + } + } + } +} + /// Bearer token read from a file, re-read whenever the client refreshes. /// /// Suits a projected `ServiceAccount` token or a Secret-mounted token that @@ -189,6 +354,52 @@ mod tests { assert_eq!(request.workspace_scope, Some(workspace_selector("oncall"))); } + #[test] + fn update_reports_the_attached_sandboxes_once_each() { + use openshell_core::proto::{ProviderDesiredIdentity, ProviderMutationReceipt}; + let receipt = |sandbox: &str| ProviderMutationReceipt { + desired: Some(ProviderDesiredIdentity { + sandbox: sandbox.to_string(), + ..Default::default() + }), + ..Default::default() + }; + let updated = Updated::from_response(ProviderResponse { + mutation_id: "m".to_string(), + target_receipts: vec![ + receipt("pm"), + receipt("gtm-bot"), + receipt("pm"), + receipt(""), + ], + ..Default::default() + }); + assert_eq!(updated.mutation_id, "m"); + assert_eq!(updated.sandboxes, ["gtm-bot", "pm"]); + } + + #[test] + fn stopped_sandboxes_are_left_alone() { + for phase in [ + SandboxPhase::Stopping, + SandboxPhase::Stopped, + SandboxPhase::Completed, + SandboxPhase::Deleting, + ] { + assert!(left_alone(phase), "{phase:?}"); + } + for phase in [ + SandboxPhase::Ready, + SandboxPhase::Error, + SandboxPhase::Provisioning, + SandboxPhase::Starting, + SandboxPhase::Unknown, + SandboxPhase::Unspecified, + ] { + assert!(!left_alone(phase), "{phase:?}"); + } + } + #[test] fn classifies_gateway_errors() { let not_found = UpdateError::from_status(&tonic::Status::not_found("x"), "w", "p"); diff --git a/crates/openshell-provider-sync/src/lib.rs b/crates/openshell-provider-sync/src/lib.rs index 5eeeac09f0..347c7fbcb7 100644 --- a/crates/openshell-provider-sync/src/lib.rs +++ b/crates/openshell-provider-sync/src/lib.rs @@ -6,10 +6,9 @@ //! Secrets managed outside `OpenShell` (for example by External Secrets //! Operator) opt in with a label and name the providers and credential keys //! they back. When a Secret's mapped values change, the controller sends the -//! changed keys to the gateway as a value-only `UpdateProvider`. Running -//! sandboxes pick the new values up on their next provider poll, and -//! credentials with `stable_placeholder: true` reach already-running -//! processes too. +//! changed keys to the gateway as a value-only `UpdateProvider`, then +//! restarts the sandboxes attached to that provider so their agents start +//! fresh with the new values. Agents of other providers are not touched. //! //! The controller never creates or deletes providers, and never clears a //! credential when its Secret goes away. @@ -18,3 +17,4 @@ pub mod controller; pub mod gateway; pub mod mapping; pub mod state; +pub mod state_store; diff --git a/crates/openshell-provider-sync/src/main.rs b/crates/openshell-provider-sync/src/main.rs index ddb8a0eea3..c088c45809 100644 --- a/crates/openshell-provider-sync/src/main.rs +++ b/crates/openshell-provider-sync/src/main.rs @@ -10,7 +10,9 @@ use clap::Parser; use kube::runtime::events::Reporter; use miette::{IntoDiagnostic, Result, WrapErr}; use openshell_provider_sync::controller::{self, Context, Options}; -use openshell_provider_sync::gateway::{GatewayUpdater, TokenFile}; +use openshell_provider_sync::gateway::{GatewayRestarter, GatewayUpdater, TokenFile}; +use openshell_provider_sync::state::SyncState; +use openshell_provider_sync::state_store::{SecretStateStore, StateStore}; use openshell_sdk::{AuthConfig, ClientConfig, OpenShellClient}; use tracing_subscriber::EnvFilter; @@ -63,6 +65,21 @@ struct Cli { #[arg(long)] no_events: bool, + /// Do not restart the agents attached to a provider whose values rotated. + /// They then keep running with connections opened on the old values. + #[arg(long, env = "OPENSHELL_PROVIDER_SYNC_NO_RESTART")] + no_restart: bool, + + /// Secret, in this controller's own namespace, that keeps fingerprints of + /// applied values across restarts. Without it, the first sync after a + /// restart is a baseline that restarts no agents. + #[arg(long, env = "OPENSHELL_PROVIDER_SYNC_STATE_SECRET")] + state_secret: Option, + + /// Namespace of the state Secret; usually this pod's namespace. + #[arg(long, env = "POD_NAMESPACE", requires = "state_secret")] + state_namespace: Option, + /// Log as JSON lines. #[arg(long, env = "OPENSHELL_LOG_JSON")] log_json: bool, @@ -99,6 +116,31 @@ async fn main() -> Result<()> { .into_diagnostic() .wrap_err("build Kubernetes client")?; + let mut context = Context::new( + Arc::new(GatewayUpdater::new(gateway.clone())), + cli.default_workspace.clone(), + ); + if !cli.no_restart { + context = context.with_restarter(Arc::new(GatewayRestarter::new(gateway))); + } + if let Some(name) = &cli.state_secret { + let namespace = cli.state_namespace.clone().ok_or_else(|| { + miette::miette!("--state-secret needs --state-namespace or POD_NAMESPACE") + })?; + let store = SecretStateStore::new(kube_client.clone(), &namespace, name.clone()); + let state = match store.load().await { + Ok(Some(persisted)) => SyncState::from_persisted(persisted), + Ok(None) => SyncState::new(), + Err(err) => { + // Starting fresh only costs a baseline sync that restarts + // nothing; refusing to start would stop all syncing. + tracing::warn!(error = %err, "could not load provider sync state; starting fresh"); + SyncState::new() + } + }; + context = context.with_state(state, Arc::new(store)); + } + let reporter = (!cli.no_events).then_some(Reporter { controller: cli.reporter, instance: cli.reporter_instance, @@ -107,14 +149,13 @@ async fn main() -> Result<()> { gateway = %cli.gateway, namespace = cli.namespace.as_deref().unwrap_or(""), default_workspace = %cli.default_workspace, + restart_agents = !cli.no_restart, + state_secret = cli.state_secret.as_deref().unwrap_or(""), "starting provider sync" ); controller::run( kube_client, - Context::new( - Arc::new(GatewayUpdater::new(gateway)), - cli.default_workspace, - ), + context, Options { namespace: cli.namespace, reporter, diff --git a/crates/openshell-provider-sync/src/state.rs b/crates/openshell-provider-sync/src/state.rs index fece7bbf6a..b707dde7a1 100644 --- a/crates/openshell-provider-sync/src/state.rs +++ b/crates/openshell-provider-sync/src/state.rs @@ -3,10 +3,12 @@ //! What has been pushed to the gateway, so only changed keys are sent. //! -//! State is in memory only. Values are never stored: each applied value is -//! kept as a salted SHA-256 fingerprint, and the salt is random per process so -//! a fingerprint is useless outside it. After a restart every Secret is pushed -//! once more, which is a value-only update and harmless to running sandboxes. +//! Values are never stored: each applied value is kept as a salted SHA-256 +//! fingerprint. The salt and fingerprints can be persisted (see +//! [`PersistedState`]) so a restart of this controller recognizes values it +//! already applied, and does not treat them as rotations that restart agents. +//! Without persisted state every Secret is pushed once more after a restart, +//! as a baseline that restarts nothing. use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt; @@ -69,6 +71,46 @@ pub struct SyncState { owners: HashMap, } +/// Fingerprints of applied values, kept across controller restarts. +/// +/// Holds no values. It is stored in a Secret because the salt and the +/// fingerprints together would let anyone who reads them test guesses. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct PersistedState { + salt: String, + applied: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +struct PersistedApplied { + namespace: String, + secret: String, + provider: String, + workspace: String, + fingerprints: BTreeMap, +} + +fn to_hex(bytes: &[u8; 32]) -> String { + use std::fmt::Write as _; + bytes + .iter() + .fold(String::with_capacity(64), |mut hex, byte| { + let _ = write!(hex, "{byte:02x}"); + hex + }) +} + +fn from_hex(text: &str) -> Option<[u8; 32]> { + if text.len() != 64 { + return None; + } + let mut bytes = [0u8; 32]; + for (index, byte) in bytes.iter_mut().enumerate() { + *byte = u8::from_str_radix(text.get(index * 2..index * 2 + 2)?, 16).ok()?; + } + Some(bytes) +} + impl Default for SyncState { fn default() -> Self { Self::new() @@ -94,6 +136,81 @@ impl SyncState { } } + /// Resume from persisted fingerprints. Entries that do not parse are + /// dropped, which only makes their next value a baseline. + pub fn from_persisted(persisted: PersistedState) -> Self { + let Some(salt) = from_hex(&persisted.salt) else { + return Self::new(); + }; + let applied = persisted + .applied + .into_iter() + .filter_map(|entry| { + let fingerprints = entry + .fingerprints + .into_iter() + .map(|(key, fingerprint)| Some((key, from_hex(&fingerprint)?))) + .collect::>>()?; + Some(( + ( + SecretRef { + namespace: entry.namespace, + name: entry.secret, + }, + entry.provider, + ), + Applied { + workspace: entry.workspace, + fingerprints, + }, + )) + }) + .collect(); + Self { + salt, + applied, + owners: HashMap::new(), + } + } + + /// The fingerprints to persist. Claims are rebuilt from the watch. + pub fn to_persisted(&self) -> PersistedState { + let mut applied: Vec = self + .applied + .iter() + .map(|((secret, provider), applied)| PersistedApplied { + namespace: secret.namespace.clone(), + secret: secret.name.clone(), + provider: provider.clone(), + workspace: applied.workspace.clone(), + fingerprints: applied + .fingerprints + .iter() + .map(|(key, fingerprint)| (key.clone(), to_hex(fingerprint))) + .collect(), + }) + .collect(); + applied.sort_by(|left, right| { + (&left.namespace, &left.secret, &left.provider).cmp(&( + &right.namespace, + &right.secret, + &right.provider, + )) + }); + PersistedState { + salt: to_hex(&self.salt), + applied, + } + } + + /// Whether `secret` has applied values to `provider` before, so a change + /// now is a rotation rather than a baseline. + pub fn has_applied(&self, secret: &SecretRef, workspace: &str, provider: &str) -> bool { + self.applied + .get(&(secret.clone(), provider.to_string())) + .is_some_and(|applied| applied.workspace == workspace) + } + fn fingerprint(&self, value: &str) -> [u8; 32] { let mut hasher = Sha256::new(); hasher.update(self.salt); @@ -408,6 +525,37 @@ mod tests { ); } + #[test] + fn persisted_state_recognizes_applied_values_after_a_restart() { + let mut state = SyncState::new(); + let secret = secret("a"); + let applied = values(&[("TOKEN", "v1")]); + state.commit(&secret, "w", "p", &applied); + assert!(state.has_applied(&secret, "w", "p")); + + let persisted = state.to_persisted(); + let serialized = serde_json::to_string(&persisted).unwrap(); + assert!(!serialized.contains("v1"), "values are never persisted"); + let restarted = SyncState::from_persisted(serde_json::from_str(&serialized).unwrap()); + + assert!(restarted.has_applied(&secret, "w", "p")); + assert!(changed_keys(&restarted.changed(&secret, "w", "p", &applied)).is_empty()); + assert_eq!( + changed_keys(&restarted.changed(&secret, "w", "p", &values(&[("TOKEN", "v2")]))), + ["TOKEN"] + ); + assert!(!restarted.has_applied(&secret, "other", "p")); + } + + #[test] + fn unreadable_persisted_state_starts_fresh() { + let restarted = SyncState::from_persisted(PersistedState { + salt: "not hex".to_string(), + applied: Vec::new(), + }); + assert!(!restarted.has_applied(&secret("a"), "w", "p")); + } + #[test] fn forget_releases_claims() { let mut state = SyncState::new(); diff --git a/crates/openshell-provider-sync/src/state_store.rs b/crates/openshell-provider-sync/src/state_store.rs new file mode 100644 index 0000000000..fe9733643d --- /dev/null +++ b/crates/openshell-provider-sync/src/state_store.rs @@ -0,0 +1,83 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Keeping applied-value fingerprints across controller restarts. +//! +//! Without them every Secret looks new after a restart, and a rotation could +//! not be told from a baseline. Restarting every agent whenever this +//! controller restarts would be wrong, so the fingerprints live in a Secret +//! this controller owns. + +use k8s_openapi::ByteString; +use k8s_openapi::api::core::v1::Secret; +use kube::Api; +use kube::api::{ObjectMeta, Patch, PatchParams}; + +use crate::state::PersistedState; + +const STATE_KEY: &str = "state.json"; +const FIELD_MANAGER: &str = "openshell-provider-sync"; + +/// Where persisted state is kept. +#[async_trait::async_trait] +pub trait StateStore: Send + Sync + 'static { + /// The saved state, or `None` when nothing has been saved yet. + async fn load(&self) -> Result, String>; + async fn save(&self, state: &PersistedState) -> Result<(), String>; +} + +/// [`StateStore`] in one Kubernetes Secret. +pub struct SecretStateStore { + api: Api, + name: String, +} + +impl SecretStateStore { + pub fn new(client: kube::Client, namespace: &str, name: impl Into) -> Self { + Self { + api: Api::namespaced(client, namespace), + name: name.into(), + } + } +} + +#[async_trait::async_trait] +impl StateStore for SecretStateStore { + async fn load(&self) -> Result, String> { + let Some(secret) = self + .api + .get_opt(&self.name) + .await + .map_err(|err| format!("read Secret {}: {err}", self.name))? + else { + return Ok(None); + }; + let Some(raw) = secret.data.as_ref().and_then(|data| data.get(STATE_KEY)) else { + return Ok(None); + }; + serde_json::from_slice(&raw.0) + .map(Some) + .map_err(|err| format!("parse Secret {}: {err}", self.name)) + } + + async fn save(&self, state: &PersistedState) -> Result<(), String> { + let raw = serde_json::to_vec(state).map_err(|err| err.to_string())?; + let secret = Secret { + metadata: ObjectMeta { + name: Some(self.name.clone()), + ..Default::default() + }, + data: Some([(STATE_KEY.to_string(), ByteString(raw))].into()), + ..Default::default() + }; + self.api + .patch( + &self.name, + &PatchParams::apply(FIELD_MANAGER).force(), + &Patch::Apply(&secret), + ) + .await + .map(|_| ()) + .map_err(|err| format!("write Secret {}: {err}", self.name)) + } +} diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 2d8289347e..2c62675451 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -966,13 +966,19 @@ kubectl -n logs deployment/ --tail=200 | `ProviderSyncConflict` | Another live Secret already feeds the same workspace, provider and key | | `InvalidProviderSync` | Bad annotations, or only some of a provider's keys are set, or a value is not UTF-8 | | `ProviderSyncFailed` | The gateway rejected the update or was unreachable | +| `AgentRestarted` | A rotation restarted a sandbox attached to that provider | +| `AgentRestartSkipped` | The attached sandbox was stopped, stopping, completed or deleting, so it was left alone | +| `AgentRestartFailed` | Stopping or starting the sandbox failed; a sandbox stopped but not started stays down until started by hand | No Events at all means the controller is not watching the Secret: check the label, its `OPENSHELL_PROVIDER_SYNC_NAMESPACE`, and its RBAC (`get`, `list` and `watch` on Secrets, `create` and `patch` on `events.k8s.io` Events). A -`ProviderSynced` event without the change reaching a running process usually -means the credential's profile lacks `stable_placeholder: true`; only processes -started after the update see the new value. Avoid rotating while gateways are +`ProviderSynced` event with no `AgentRestarted` after it means the values were a +baseline (the first values the controller saw, or the first sync after it +restarted without `--state-secret`), restarts are off (`--no-restart`), or no +sandbox was attached to the provider. Every agent restarting after a +provider-sync restart means its state Secret is missing or unreadable. Avoid +rotating while gateways are rolling: a supervisor whose provider refetch fails revokes its static credentials. From 0f5da5e5dab37e20de673b719cf6506cb8777850 Mon Sep 17 00:00:00 2001 From: Matt Jane Date: Thu, 15 Oct 2026 14:16:15 -0700 Subject: [PATCH 2/4] fix(provider-sync): never leave an agent stopped by a restart A restart gave up starting the sandbox after 10 minutes, and the gateway deliberately never revives an explicitly stopped sandbox, so a gateway outage during a rotation could leave an agent down for good. The start is now retried until the sandbox runs again, backing off from 10 seconds to 5 minutes, with an AgentStartPending Warning at each retry. Each attempt reads the sandbox's phase first, so a start that succeeded but whose reply was lost is not repeated. The pending start is saved in the state Secret before the stop and resumed when provider-sync restarts. Only deleting the sandbox ends it. A second rotation while a restart is pending leaves it to the first. Signed-off-by: Matt Jane --- crates/openshell-provider-sync/Cargo.toml | 2 +- crates/openshell-provider-sync/README.md | 11 +- .../openshell-provider-sync/src/controller.rs | 474 ++++++++++++++++-- crates/openshell-provider-sync/src/gateway.rs | 129 ++--- crates/openshell-provider-sync/src/state.rs | 78 ++- skills/debug-openshell-cluster/SKILL.md | 3 +- 6 files changed, 595 insertions(+), 102 deletions(-) diff --git a/crates/openshell-provider-sync/Cargo.toml b/crates/openshell-provider-sync/Cargo.toml index 2261b4333c..70a82656ea 100644 --- a/crates/openshell-provider-sync/Cargo.toml +++ b/crates/openshell-provider-sync/Cargo.toml @@ -35,7 +35,7 @@ tracing-subscriber = { workspace = true } uuid = { workspace = true } [dev-dependencies] -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] } [lints] workspace = true diff --git a/crates/openshell-provider-sync/README.md b/crates/openshell-provider-sync/README.md index ae5e9fa635..a1bb9bfcc0 100644 --- a/crates/openshell-provider-sync/README.md +++ b/crates/openshell-provider-sync/README.md @@ -49,8 +49,11 @@ accident, and one Secret key may be listed only once. connection keeps using the old value. Sandboxes of other providers are not touched, and a sandbox restarts once even when several of its providers rotate together. A sandbox that is stopped, stopping, completed or being - deleted is left alone. Once stopped, the start is retried for up to 10 - minutes, so a restart does not leave an agent down. Agents must therefore + deleted is left alone. Once the controller has stopped a sandbox it keeps + starting it until it runs again, however long the gateway is unavailable, + backing off from 10 seconds to 5 minutes between attempts. The pending start + is saved in the state Secret before the stop, so a controller restart resumes + it. Only deleting the sandbox ends it. Agents must therefore tolerate a restart at any time: keep state outside the sandbox or resume from it. `--no-restart` turns this off. - **What it leaves alone.** It never creates or deletes providers, never clears @@ -92,7 +95,9 @@ accident, and one Secret key may be listed only once. - **Audit.** Each outcome is published as a Kubernetes Event on the Secret (`ProviderSynced`, `ProviderSyncSkipped`, `ProviderNotFound`, `ProviderSyncConflict`, `InvalidProviderSync`, `ProviderSyncFailed`, - `AgentRestarted`, `AgentRestartSkipped`, `AgentRestartFailed`). The + `AgentRestarted`, `AgentRestartSkipped`, `AgentStartPending`, + `AgentRestartFailed`). `AgentStartPending` is a Warning at each retry of a + start that has not succeeded yet. The gateway separately audits each `UpdateProvider` with the controller's principal. Neither carries values. diff --git a/crates/openshell-provider-sync/src/controller.rs b/crates/openshell-provider-sync/src/controller.rs index cfb7ea7360..b2d8943001 100644 --- a/crates/openshell-provider-sync/src/controller.rs +++ b/crates/openshell-provider-sync/src/controller.rs @@ -16,10 +16,11 @@ use kube::{Api, Client, Resource, ResourceExt}; use tokio::sync::Mutex; use tracing::{debug, info, warn}; -use crate::gateway::{ProviderUpdater, Restart, SandboxRestarter, UpdateError}; +use crate::gateway::{ProviderUpdater, SandboxRestarter, Stop, UpdateError}; use crate::mapping::{self, Extracted, MappingError}; use crate::state::{Conflict, SecretRef, SyncState}; use crate::state_store::StateStore; +use openshell_core::proto::SandboxPhase; /// How long to wait before retrying a provider that does not exist yet. const PROVIDER_MISSING_RETRY: Duration = Duration::from_secs(30); @@ -28,6 +29,14 @@ const TRANSIENT_RETRY: Duration = Duration::from_secs(10); /// How long to wait before retrying a Secret that can only succeed once it, /// the provider or the gateway's configuration changes. const INVALID_RETRY: Duration = Duration::from_mins(5); +/// First and longest wait between attempts to start a sandbox this +/// controller stopped. Starting is retried until it succeeds. +const START_BACKOFF_FIRST: Duration = Duration::from_secs(10); +const START_BACKOFF_MAX: Duration = Duration::from_mins(5); +/// How often to look again while a stopped sandbox finishes stopping, and +/// how long that may take before each look is reported. +const STOPPING_POLL: Duration = Duration::from_secs(5); +const STOPPING_GRACE: Duration = Duration::from_mins(1); /// The result of syncing one provider. #[derive(Debug, PartialEq, Eq)] @@ -216,15 +225,11 @@ impl Context { .await; providers.push((provider.clone(), result)); } - if let Some(store) = &self.store - && providers - .iter() - .any(|(_, result)| matches!(result, Ok(Outcome::Updated { .. }))) - && let Err(err) = store.save(&state.to_persisted()).await + if providers + .iter() + .any(|(_, result)| matches!(result, Ok(Outcome::Updated { .. }))) { - // In-memory state is still right. The cost of a lost save is one - // extra restart of these agents after this controller restarts. - warn!(error = %err, "failed to persist provider sync state"); + self.persist(&state).await; } Ok(SyncReport { workspace: target.workspace, @@ -235,6 +240,39 @@ impl Context { async fn forget(&self, secret_ref: &SecretRef) { self.state.lock().await.forget(secret_ref); } + + async fn persist(&self, state: &SyncState) { + if let Some(store) = &self.store + && let Err(err) = store.save(&state.to_persisted()).await + { + // In-memory state is still right. A lost save costs one extra + // restart of these agents, or a forgotten pending start, only if + // this controller restarts before the next save. + warn!(error = %err, "failed to persist provider sync state"); + } + } + + /// Claim `sandbox` for a restart and persist the pending start before it + /// is stopped. False when a restart of it is already in progress. + pub async fn begin_restart(&self, secret: &SecretRef, workspace: &str, sandbox: &str) -> bool { + let mut state = self.state.lock().await; + if !state.begin_restart(secret, workspace, sandbox) { + return false; + } + self.persist(&state).await; + true + } + + pub async fn end_restart(&self, workspace: &str, sandbox: &str) { + let mut state = self.state.lock().await; + state.end_restart(workspace, sandbox); + self.persist(&state).await; + } + + /// Sandboxes stopped by an earlier run that must still be started. + pub async fn pending_starts(&self) -> Vec { + self.state.lock().await.pending_starts() + } } struct Runtime { @@ -273,7 +311,7 @@ async fn reconcile(secret: Arc, runtime: Arc) -> Result, runtime: Arc) -> Result { debug!(secret = %secret_ref, workspace, provider, "provider credentials unchanged"); @@ -322,7 +367,7 @@ async fn reconcile(secret: Arc, runtime: Arc) -> Result, runtime: Arc) -> Result, runtime: Arc) -> Result, runtime: Arc) -> Result, runtime: Arc, - secret: Arc, + secret: SecretRef, workspace: String, sandbox: String, ) { - let secret_ref = secret_ref(&secret); - match restarter.restart(&workspace, &sandbox).await { - Ok(Restart::Restarted) => { - info!(secret = %secret_ref, workspace, sandbox, "restarted agent after credential rotation"); - let note = - format!("Restarted sandbox {workspace}/{sandbox} after its credentials rotated"); - publish(&runtime, &secret, EventType::Normal, "AgentRestarted", note).await; + if !runtime + .ctx + .begin_restart(&secret, &workspace, &sandbox) + .await + { + // A restart of this sandbox is already under way and will start it + // with current values. + debug!(secret = %secret, workspace, sandbox, "agent restart already pending"); + return; + } + match restarter.stop(&workspace, &sandbox).await { + Ok(Stop::Stopped) => { + start_until_running(restarter, runtime, secret, workspace, sandbox).await; } - Ok(Restart::Skipped { phase }) => { - info!(secret = %secret_ref, workspace, sandbox, phase, "agent not running; left alone"); + Ok(Stop::Skipped { phase }) => { + runtime.ctx.end_restart(&workspace, &sandbox).await; + info!(secret = %secret, workspace, sandbox, phase, "agent not running; left alone"); let note = format!("Did not restart sandbox {workspace}/{sandbox}: it is {phase}"); publish( &runtime, @@ -388,8 +447,12 @@ async fn restart_sandbox( .await; } Err(err) => { - warn!(secret = %secret_ref, workspace, sandbox, error = %err, "agent restart failed"); - let note = format!("Restarting sandbox {workspace}/{sandbox} failed: {err}"); + // Nothing was stopped, so the agent keeps running. + runtime.ctx.end_restart(&workspace, &sandbox).await; + warn!(secret = %secret, workspace, sandbox, error = %err, "agent restart failed"); + let note = format!( + "Restarting sandbox {workspace}/{sandbox} failed before it stopped; it keeps running: {err}" + ); publish( &runtime, &secret, @@ -402,11 +465,131 @@ async fn restart_sandbox( } } +/// What one look at a stopped sandbox found. +enum StartStep { + /// It runs again, or reached a state no start applies to. + Done, + /// It no longer exists. + Gone, + /// It is still stopping. + Stopping, + Failed(UpdateError), +} + +/// Start a sandbox this controller stopped, retrying with capped backoff +/// until it runs again or is deleted. Never gives up on a stopped agent. +async fn start_until_running( + restarter: Arc, + runtime: Arc, + secret: SecretRef, + workspace: String, + sandbox: String, +) { + let mut backoff = START_BACKOFF_FIRST; + let mut attempts: u32 = 0; + let mut stopping_since: Option = None; + loop { + // Look before starting: a start that succeeded but whose reply was + // lost must not be retried against a running sandbox forever. + let step = match restarter.phase(&workspace, &sandbox).await { + Ok(None | Some(SandboxPhase::Deleting)) => StartStep::Gone, + Ok(Some(SandboxPhase::Stopping)) => StartStep::Stopping, + Ok(Some(SandboxPhase::Stopped | SandboxPhase::Unknown | SandboxPhase::Unspecified)) => { + match restarter.start(&workspace, &sandbox).await { + Ok(()) => StartStep::Done, + Err(err) => StartStep::Failed(err), + } + } + Ok(Some(_)) => StartStep::Done, + Err(err) => StartStep::Failed(err), + }; + let delay = match step { + StartStep::Done => { + runtime.ctx.end_restart(&workspace, &sandbox).await; + info!(secret = %secret, workspace, sandbox, "restarted agent after credential rotation"); + let note = format!( + "Restarted sandbox {workspace}/{sandbox} after its credentials rotated" + ); + publish(&runtime, &secret, EventType::Normal, "AgentRestarted", note).await; + return; + } + StartStep::Gone => { + runtime.ctx.end_restart(&workspace, &sandbox).await; + warn!(secret = %secret, workspace, sandbox, "sandbox deleted before it started again"); + let note = + format!("Did not start sandbox {workspace}/{sandbox} again: it was deleted"); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentRestartFailed", + note, + ) + .await; + return; + } + StartStep::Stopping => { + let since = *stopping_since.get_or_insert_with(tokio::time::Instant::now); + if since.elapsed() >= STOPPING_GRACE { + let note = format!( + "Sandbox {workspace}/{sandbox} is still stopping after {}s; it will be started once stopped", + since.elapsed().as_secs() + ); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentStartPending", + note, + ) + .await; + stopping_since = Some(tokio::time::Instant::now()); + } + STOPPING_POLL + } + StartStep::Failed(err) => { + attempts += 1; + warn!( + secret = %secret, + workspace, + sandbox, + attempts, + retry_in_secs = backoff.as_secs(), + error = %err, + "starting stopped agent failed; retrying" + ); + let note = format!( + "Starting sandbox {workspace}/{sandbox} failed (attempt {attempts}); retrying in {}s: {err}", + backoff.as_secs() + ); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentStartPending", + note, + ) + .await; + let delay = backoff; + backoff = (backoff * 2).min(START_BACKOFF_MAX); + delay + } + }; + tokio::time::sleep(delay).await; + } +} + fn error_policy(_secret: Arc, err: &SyncError, _runtime: Arc) -> Action { Action::requeue(err.retry_after()) } -async fn publish(runtime: &Runtime, secret: &Secret, type_: EventType, reason: &str, note: String) { +async fn publish( + runtime: &Runtime, + secret: &SecretRef, + type_: EventType, + reason: &str, + note: String, +) { let Some(recorder) = &runtime.recorder else { return; }; @@ -417,7 +600,14 @@ async fn publish(runtime: &Runtime, secret: &Secret, type_: EventType, reason: & action: "UpdateProviderCredentials".to_string(), secondary: None, }; - if let Err(err) = recorder.publish(&event, &secret.object_ref(&())).await { + let reference = k8s_openapi::api::core::v1::ObjectReference { + api_version: Some("v1".to_string()), + kind: Some("Secret".to_string()), + namespace: Some(secret.namespace.clone()), + name: Some(secret.name.clone()), + ..Default::default() + }; + if let Err(err) = recorder.publish(&event, &reference).await { debug!(error = %err, "failed to publish Kubernetes event"); } } @@ -450,6 +640,23 @@ pub async fn run(client: Client, ctx: Context, options: Options) { .reporter .map(|reporter| Recorder::new(client, reporter)), }); + // An earlier run may have stopped agents it had not started yet. + if let Some(restarter) = &runtime.ctx.restarter { + for pending in runtime.ctx.pending_starts().await { + info!( + workspace = pending.workspace, + sandbox = pending.sandbox, + "resuming the start of an agent stopped for a restart" + ); + tokio::spawn(start_until_running( + Arc::clone(restarter), + Arc::clone(&runtime), + pending.secret, + pending.workspace, + pending.sandbox, + )); + } + } controller .shutdown_on_signal() .run(reconcile, error_policy, runtime) @@ -488,16 +695,17 @@ mod tests { attached: StdMutex>>, } + /// Keeps every save, newest last. #[derive(Default)] - struct MemoryStore(StdMutex>); + struct MemoryStore(StdMutex>); #[async_trait::async_trait] impl StateStore for MemoryStore { async fn load(&self) -> Result, String> { - Ok(self.0.lock().unwrap().clone()) + Ok(self.0.lock().unwrap().last().cloned()) } async fn save(&self, state: &crate::state::PersistedState) -> Result<(), String> { - *self.0.lock().unwrap() = Some(state.clone()); + self.0.lock().unwrap().push(state.clone()); Ok(()) } } @@ -785,6 +993,210 @@ mod tests { assert_eq!(report.sandboxes_to_restart(), ["linear-agent"]); } + /// Scripted sandbox lifecycle: phases returned in order (the last + /// repeats), and start results in order (then success). + #[derive(Default)] + struct FakeRestarter { + stop_skipped: StdMutex>, + phases: StdMutex>>, + start_failures: StdMutex, + stops: StdMutex, + starts: StdMutex, + } + + #[async_trait::async_trait] + impl SandboxRestarter for FakeRestarter { + async fn stop(&self, _: &str, _: &str) -> Result { + *self.stops.lock().unwrap() += 1; + Ok(self + .stop_skipped + .lock() + .unwrap() + .clone() + .map_or(Stop::Stopped, |phase| Stop::Skipped { phase })) + } + async fn phase(&self, _: &str, _: &str) -> Result, UpdateError> { + let mut phases = self.phases.lock().unwrap(); + Ok(if phases.len() > 1 { + phases.remove(0) + } else { + phases.first().copied().flatten() + }) + } + async fn start(&self, _: &str, _: &str) -> Result<(), UpdateError> { + *self.starts.lock().unwrap() += 1; + let mut failures = self.start_failures.lock().unwrap(); + if *failures > 0 { + *failures -= 1; + return Err(UpdateError::Transient( + "Unavailable: gateway down".to_string(), + )); + } + Ok(()) + } + } + + fn runtime(ctx: Context) -> Arc { + let (store, _writer) = kube::runtime::reflector::store(); + Arc::new(Runtime { + ctx, + store, + recorder: None, + }) + } + + #[tokio::test(start_paused = true)] + async fn a_stopped_agent_is_started_however_long_the_gateway_is_down() { + let store = Arc::new(MemoryStore::default()); + let runtime = runtime( + Context::new(Arc::new(FakeUpdater::default()), "default") + .with_state(SyncState::new(), store.clone()), + ); + let restarter = Arc::new(FakeRestarter::default()); + *restarter.phases.lock().unwrap() = vec![Some(SandboxPhase::Stopped)]; + // Far longer than any fixed deadline: 30 failures at up to 5 minutes. + *restarter.start_failures.lock().unwrap() = 30; + + let (secret, _) = github("v1"); + restart_sandbox( + restarter.clone(), + runtime.clone(), + secret.clone(), + "oncall".to_string(), + "triage".to_string(), + ) + .await; + // The pending start was persisted before anything was stopped. + let first_save = store.0.lock().unwrap().first().cloned().unwrap(); + assert_eq!( + SyncState::from_persisted(first_save).pending_starts().len(), + 1, + "pending start saved before the stop" + ); + assert_eq!(*restarter.stops.lock().unwrap(), 1); + assert_eq!(*restarter.starts.lock().unwrap(), 31); + assert!(runtime.ctx.pending_starts().await.is_empty()); + let saved = store.load().await.unwrap().unwrap(); + assert!(SyncState::from_persisted(saved).pending_starts().is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_controller_restart_resumes_a_pending_start() { + let store = Arc::new(MemoryStore::default()); + let mut before = SyncState::new(); + let (secret, _) = github("v1"); + assert!(before.begin_restart(&secret, "oncall", "triage")); + store.save(&before.to_persisted()).await.unwrap(); + + let persisted = store.load().await.unwrap().unwrap(); + let runtime = runtime( + Context::new(Arc::new(FakeUpdater::default()), "default") + .with_state(SyncState::from_persisted(persisted), store.clone()), + ); + let pending = runtime.ctx.pending_starts().await; + assert_eq!(pending.len(), 1); + + let restarter = Arc::new(FakeRestarter::default()); + // Still stopping at first, then stopped. + *restarter.phases.lock().unwrap() = vec![ + Some(SandboxPhase::Stopping), + Some(SandboxPhase::Stopping), + Some(SandboxPhase::Stopped), + ]; + let pending = pending.into_iter().next().unwrap(); + start_until_running( + restarter.clone(), + runtime.clone(), + pending.secret, + pending.workspace, + pending.sandbox, + ) + .await; + assert_eq!( + *restarter.stops.lock().unwrap(), + 0, + "resuming never stops again" + ); + assert_eq!(*restarter.starts.lock().unwrap(), 1); + assert!(runtime.ctx.pending_starts().await.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_start_that_already_happened_is_not_repeated() { + let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); + let restarter = Arc::new(FakeRestarter::default()); + *restarter.phases.lock().unwrap() = vec![Some(SandboxPhase::Ready)]; + let (secret, _) = github("v1"); + start_until_running( + restarter.clone(), + runtime, + secret, + "oncall".to_string(), + "triage".to_string(), + ) + .await; + assert_eq!(*restarter.starts.lock().unwrap(), 0); + } + + #[tokio::test(start_paused = true)] + async fn a_deleted_sandbox_ends_the_pending_start() { + let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); + let (secret, _) = github("v1"); + assert!(runtime.ctx.begin_restart(&secret, "oncall", "triage").await); + let restarter = Arc::new(FakeRestarter::default()); + *restarter.phases.lock().unwrap() = vec![None]; + start_until_running( + restarter.clone(), + runtime.clone(), + secret, + "oncall".to_string(), + "triage".to_string(), + ) + .await; + assert_eq!(*restarter.starts.lock().unwrap(), 0); + assert!(runtime.ctx.pending_starts().await.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_sandbox_someone_else_stopped_is_left_alone() { + let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); + let restarter = Arc::new(FakeRestarter::default()); + *restarter.stop_skipped.lock().unwrap() = Some("SANDBOX_PHASE_STOPPED".to_string()); + let (secret, _) = github("v1"); + restart_sandbox( + restarter.clone(), + runtime.clone(), + secret, + "oncall".to_string(), + "triage".to_string(), + ) + .await; + assert_eq!(*restarter.starts.lock().unwrap(), 0); + assert!(runtime.ctx.pending_starts().await.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_second_rotation_does_not_restart_a_sandbox_already_restarting() { + let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); + let (secret, _) = github("v1"); + assert!(runtime.ctx.begin_restart(&secret, "oncall", "triage").await); + let restarter = Arc::new(FakeRestarter::default()); + restart_sandbox( + restarter.clone(), + runtime.clone(), + secret, + "oncall".to_string(), + "triage".to_string(), + ) + .await; + assert_eq!(*restarter.stops.lock().unwrap(), 0); + assert_eq!( + runtime.ctx.pending_starts().await.len(), + 1, + "the first restart still owns it" + ); + } + #[tokio::test] async fn a_failing_provider_does_not_hold_back_the_others() { let updater = Arc::new(FakeUpdater::default()); diff --git a/crates/openshell-provider-sync/src/gateway.rs b/crates/openshell-provider-sync/src/gateway.rs index 54a4ff5659..f97c2b6386 100644 --- a/crates/openshell-provider-sync/src/gateway.rs +++ b/crates/openshell-provider-sync/src/gateway.rs @@ -86,23 +86,6 @@ pub trait ProviderUpdater: Send + Sync + 'static { ) -> Result; } -/// What restarting one sandbox did. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum Restart { - Restarted, - /// Stopped, stopping, completed or being deleted: someone or something - /// else decided it should not run, so it is left alone. - Skipped { - phase: String, - }, -} - -/// Restarts a sandbox so its agent starts fresh with current credentials. -#[async_trait::async_trait] -pub trait SandboxRestarter: Send + Sync + 'static { - async fn restart(&self, workspace: &str, sandbox: &str) -> Result; -} - /// [`ProviderUpdater`] over the gateway's `UpdateProvider` RPC. pub struct GatewayUpdater { client: OpenShellClient, @@ -178,10 +161,32 @@ impl ProviderUpdater for GatewayUpdater { } } -/// How long a restart keeps trying to start a sandbox it stopped. -const START_DEADLINE: std::time::Duration = std::time::Duration::from_mins(10); -/// Pause between start attempts while the sandbox is still stopping. -const START_RETRY: std::time::Duration = std::time::Duration::from_secs(5); +/// What stopping a sandbox for a restart did. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Stop { + Stopped, + /// Stopped, stopping, completed, being deleted or gone: someone or + /// something else decided it should not run, so it is left alone. + Skipped { + phase: String, + }, +} + +/// Stops and starts sandboxes so their agents start fresh with current +/// credentials. Retrying is the caller's job. +#[async_trait::async_trait] +pub trait SandboxRestarter: Send + Sync + 'static { + /// Stop the sandbox unless it is already meant to stay down. + async fn stop(&self, workspace: &str, sandbox: &str) -> Result; + /// The sandbox's phase, or `None` when it no longer exists. + async fn phase( + &self, + workspace: &str, + sandbox: &str, + ) -> Result, UpdateError>; + /// One start attempt. + async fn start(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError>; +} /// Call one gateway RPC, refreshing the bearer token once if it is rejected. macro_rules! call_refreshing { @@ -211,7 +216,8 @@ macro_rules! call_refreshing { }}; } -/// [`SandboxRestarter`] over the gateway's `StopSandbox` and `StartSandbox`. +/// [`SandboxRestarter`] over the gateway's `GetSandbox`, `StopSandbox` and +/// `StartSandbox`. pub struct GatewayRestarter { client: OpenShellClient, } @@ -220,24 +226,10 @@ impl GatewayRestarter { pub fn new(client: OpenShellClient) -> Self { Self { client } } - - async fn phase(&self, workspace: &str, sandbox: &str) -> Result { - let request = GetSandboxRequest { - workspace_scope: Some(workspace_selector(workspace)), - name: sandbox.to_string(), - }; - let response = call_refreshing!(&self.client, get_sandbox, request) - .map_err(|status| UpdateError::from_status(&status, workspace, sandbox))?; - Ok(response - .into_inner() - .sandbox - .and_then(|sandbox| sandbox.status) - .map_or(SandboxPhase::Unspecified, |status| status.phase())) - } } /// A sandbox in one of these phases is meant to stay down. -fn left_alone(phase: SandboxPhase) -> bool { +pub(crate) fn left_alone(phase: SandboxPhase) -> bool { matches!( phase, SandboxPhase::Stopping @@ -249,14 +241,18 @@ fn left_alone(phase: SandboxPhase) -> bool { #[async_trait::async_trait] impl SandboxRestarter for GatewayRestarter { - async fn restart(&self, workspace: &str, sandbox: &str) -> Result { - let phase = self.phase(workspace, sandbox).await?; + async fn stop(&self, workspace: &str, sandbox: &str) -> Result { + let Some(phase) = self.phase(workspace, sandbox).await? else { + return Ok(Stop::Skipped { + phase: "deleted".to_string(), + }); + }; if left_alone(phase) { - return Ok(Restart::Skipped { + return Ok(Stop::Skipped { phase: phase.as_str_name().to_string(), }); } - // Request ids make a retried stop or start a replay, not a repeat. + // A request id makes a retried stop a replay, not a repeat. let stop = StopSandboxRequest { workspace_scope: Some(workspace_selector(workspace)), name: sandbox.to_string(), @@ -264,35 +260,40 @@ impl SandboxRestarter for GatewayRestarter { }; call_refreshing!(&self.client, stop_sandbox, stop) .map_err(|status| UpdateError::from_status(&status, workspace, sandbox))?; + Ok(Stop::Stopped) + } + + async fn phase( + &self, + workspace: &str, + sandbox: &str, + ) -> Result, UpdateError> { + let request = GetSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + }; + match call_refreshing!(&self.client, get_sandbox, request) { + Ok(response) => Ok(Some( + response + .into_inner() + .sandbox + .and_then(|sandbox| sandbox.status) + .map_or(SandboxPhase::Unspecified, |status| status.phase()), + )), + Err(status) if status.code() == Code::NotFound => Ok(None), + Err(status) => Err(UpdateError::from_status(&status, workspace, sandbox)), + } + } - // A stopped agent must not stay down because this restart gave up: - // keep starting until the gateway accepts or the deadline passes. + async fn start(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError> { let start = StartSandboxRequest { workspace_scope: Some(workspace_selector(workspace)), name: sandbox.to_string(), request_id: uuid::Uuid::new_v4().to_string(), }; - let deadline = tokio::time::Instant::now() + START_DEADLINE; - loop { - match call_refreshing!(&self.client, start_sandbox, start.clone()) { - Ok(_) => return Ok(Restart::Restarted), - Err(status) - if matches!( - status.code(), - Code::FailedPrecondition | Code::Unavailable | Code::Aborted - ) && tokio::time::Instant::now() < deadline => - { - tokio::time::sleep(START_RETRY).await; - } - Err(status) => { - return Err(UpdateError::Transient(format!( - "sandbox {workspace}/{sandbox} was stopped but did not start: {:?}: {}", - status.code(), - status.message() - ))); - } - } - } + call_refreshing!(&self.client, start_sandbox, start) + .map(|_| ()) + .map_err(|status| UpdateError::from_status(&status, workspace, sandbox)) } } diff --git a/crates/openshell-provider-sync/src/state.rs b/crates/openshell-provider-sync/src/state.rs index b707dde7a1..7a5bb76b44 100644 --- a/crates/openshell-provider-sync/src/state.rs +++ b/crates/openshell-provider-sync/src/state.rs @@ -18,7 +18,9 @@ use sha2::{Digest, Sha256}; use crate::mapping::{CredentialValues, KeyMap, SyncTarget}; /// `namespace/name` of a Secret. -#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] +#[derive( + Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, serde::Serialize, serde::Deserialize, +)] pub struct SecretRef { pub namespace: String, pub name: String, @@ -69,6 +71,17 @@ pub struct SyncState { salt: [u8; 32], applied: HashMap, owners: HashMap, + pending_starts: BTreeMap<(String, String), SecretRef>, +} + +/// A sandbox this controller stopped for a restart and has not yet seen +/// start again. Kept across restarts so a stopped agent is never forgotten. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct PendingStart { + pub workspace: String, + pub sandbox: String, + /// The Secret whose rotation caused the restart, for its Events. + pub secret: SecretRef, } /// Fingerprints of applied values, kept across controller restarts. @@ -79,6 +92,8 @@ pub struct SyncState { pub struct PersistedState { salt: String, applied: Vec, + #[serde(default)] + pending_starts: Vec, } #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -133,6 +148,7 @@ impl SyncState { salt: rand::random(), applied: HashMap::new(), owners: HashMap::new(), + pending_starts: BTreeMap::new(), } } @@ -166,10 +182,16 @@ impl SyncState { )) }) .collect(); + let pending_starts = persisted + .pending_starts + .into_iter() + .map(|pending| ((pending.workspace, pending.sandbox), pending.secret)) + .collect(); Self { salt, applied, owners: HashMap::new(), + pending_starts, } } @@ -200,7 +222,38 @@ impl SyncState { PersistedState { salt: to_hex(&self.salt), applied, + pending_starts: self.pending_starts(), + } + } + + /// Record that this controller is about to stop `sandbox` and must start + /// it again. Returns false when a start is already pending, so one restart + /// at a time owns the sandbox. + pub fn begin_restart(&mut self, secret: &SecretRef, workspace: &str, sandbox: &str) -> bool { + let key = (workspace.to_string(), sandbox.to_string()); + if self.pending_starts.contains_key(&key) { + return false; } + self.pending_starts.insert(key, secret.clone()); + true + } + + /// The sandbox runs again, or will never need starting. + pub fn end_restart(&mut self, workspace: &str, sandbox: &str) { + self.pending_starts + .remove(&(workspace.to_string(), sandbox.to_string())); + } + + /// Sandboxes stopped for a restart and not yet seen running again. + pub fn pending_starts(&self) -> Vec { + self.pending_starts + .iter() + .map(|((workspace, sandbox), secret)| PendingStart { + workspace: workspace.clone(), + sandbox: sandbox.clone(), + secret: secret.clone(), + }) + .collect() } /// Whether `secret` has applied values to `provider` before, so a change @@ -547,11 +600,32 @@ mod tests { assert!(!restarted.has_applied(&secret, "other", "p")); } + #[test] + fn a_pending_start_survives_a_restart_and_blocks_a_second_restart() { + let mut state = SyncState::new(); + assert!(state.begin_restart(&secret("pm"), "pm", "pm")); + assert!(!state.begin_restart(&secret("pm"), "pm", "pm")); + + let restarted = SyncState::from_persisted(state.to_persisted()); + assert_eq!( + restarted.pending_starts(), + [PendingStart { + workspace: "pm".to_string(), + sandbox: "pm".to_string(), + secret: secret("pm"), + }] + ); + + state.end_restart("pm", "pm"); + assert!(state.pending_starts().is_empty()); + assert!(state.begin_restart(&secret("pm"), "pm", "pm")); + } + #[test] fn unreadable_persisted_state_starts_fresh() { let restarted = SyncState::from_persisted(PersistedState { salt: "not hex".to_string(), - applied: Vec::new(), + ..Default::default() }); assert!(!restarted.has_applied(&secret("a"), "w", "p")); } diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 2c62675451..55baa7f899 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -968,7 +968,8 @@ kubectl -n logs deployment/ --tail=200 | `ProviderSyncFailed` | The gateway rejected the update or was unreachable | | `AgentRestarted` | A rotation restarted a sandbox attached to that provider | | `AgentRestartSkipped` | The attached sandbox was stopped, stopping, completed or deleting, so it was left alone | -| `AgentRestartFailed` | Stopping or starting the sandbox failed; a sandbox stopped but not started stays down until started by hand | +| `AgentStartPending` | provider-sync stopped the sandbox and its start keeps failing or it is still stopping; retried until it runs, backing off to every 5 minutes, and resumed after a provider-sync restart | +| `AgentRestartFailed` | The stop failed, so the agent kept running on its old connections; or the sandbox was deleted before it could start again | No Events at all means the controller is not watching the Secret: check the label, its `OPENSHELL_PROVIDER_SYNC_NAMESPACE`, and its RBAC (`get`, `list` and From f961f671e7267af46f1813e2f025749db480049a Mon Sep 17 00:00:00 2001 From: Matt Jane Date: Thu, 15 Oct 2026 15:00:04 -0700 Subject: [PATCH 3/4] fix(provider-sync): make agent restarts durable through crashes and outages Review of the restart flow found ways to lose a restart or report one that never happened: - The rotation's fingerprints were saved before its restarts were recorded, so a crash in between applied the rotation and forgot the restart. Restarts are now requested in the same state, and saved in the same write. - A crash between recording a restart and stopping the sandbox resumed as "already started". Each restart is now a saved state machine (requested, stopping, awaiting running), driven by the sandbox's phase on every step, and the controller saves that it owns the stop before it stops anything. A resumed request with a ready sandbox is stopped and started. - A failed stop ended the restart with the agent still on its old connections. Every gateway failure and every failed save is now retried with capped backoff. - Error, Starting and Provisioning counted as restarted. Only Ready does; a sandbox that ends in Error is reported and left to the gateway's recovery. - A rotation during a restart was dropped. It now restarts the agent once more afterwards. - Unreadable state was replaced by fresh state on the next save, losing restarts in progress. The controller now refuses to start on it, keeps restarts even when the salt is unreadable, and warns when restarts run without a state Secret. Also: sandbox errors name the sandbox rather than a provider, retry events no longer carry counters so repeats aggregate, fingerprints of deleted Secrets are pruned, and the README and debug skill now match the behaviour, including the remaining stop race the gateway cannot close. Signed-off-by: Matt Jane --- crates/openshell-provider-sync/README.md | 38 +- .../openshell-provider-sync/src/controller.rs | 798 ++++++++++-------- crates/openshell-provider-sync/src/gateway.rs | 116 +-- crates/openshell-provider-sync/src/lib.rs | 1 + crates/openshell-provider-sync/src/main.rs | 25 +- crates/openshell-provider-sync/src/restart.rs | 284 +++++++ crates/openshell-provider-sync/src/state.rs | 203 ++++- skills/debug-openshell-cluster/SKILL.md | 20 +- 8 files changed, 977 insertions(+), 508 deletions(-) create mode 100644 crates/openshell-provider-sync/src/restart.rs diff --git a/crates/openshell-provider-sync/README.md b/crates/openshell-provider-sync/README.md index a1bb9bfcc0..7c7d26d182 100644 --- a/crates/openshell-provider-sync/README.md +++ b/crates/openshell-provider-sync/README.md @@ -43,19 +43,32 @@ accident, and one Secret key may be listed only once. changed keys of that provider as a value-only `UpdateProvider`. The gateway merges them into the provider, and attached sandboxes refetch on their next provider poll, every 10s by default. -- **Restarting affected agents.** After a successful update the controller +- **Restarting affected agents.** After a successful rotation the controller restarts every sandbox the gateway reports attached to that provider, with `StopSandbox` then `StartSandbox`, so the agent starts fresh and no open connection keeps using the old value. Sandboxes of other providers are not touched, and a sandbox restarts once even when several of its providers - rotate together. A sandbox that is stopped, stopping, completed or being - deleted is left alone. Once the controller has stopped a sandbox it keeps - starting it until it runs again, however long the gateway is unavailable, - backing off from 10 seconds to 5 minutes between attempts. The pending start - is saved in the state Secret before the stop, so a controller restart resumes - it. Only deleting the sandbox ends it. Agents must therefore - tolerate a restart at any time: keep state outside the sandbox or resume - from it. `--no-restart` turns this off. + rotate together. A rotation that lands while a sandbox is already restarting + restarts it once more afterwards. Agents must therefore tolerate a restart at + any time: keep state outside the sandbox or resume from it. `--no-restart` + turns this off. +- **Never leaving an agent stopped.** Each restart is a saved state machine: + requested, then stopping, then awaiting running. The request is saved in the + same write as the rotation's fingerprints, and the controller saves that it + owns the stop before it stops anything, so after a controller restart it + resumes every restart where it was. Every gateway failure and every failed + save is retried, backing off from 10 seconds to 5 minutes. A restart is done + only when the sandbox is ready again. This holds with `--state-secret`; + without it a restart in progress is lost when the controller restarts, which + can leave an agent stopped, and the controller warns about it at startup. +- **What a restart leaves to others.** A sandbox that is stopped, stopping, + completed, in Error or being deleted when the restart begins is left alone, + and so is one someone else stops after the restart started it. A sandbox that + ends in Error is left to the gateway's own recovery, with an + `AgentRestartFailed` event. One race remains: if someone stops a sandbox in + the moment between the controller reading its phase and stopping it, the + controller starts it again afterwards. The gateway has no conditional stop + to close that window. - **What it leaves alone.** It never creates or deletes providers, never clears a credential when its Secret is deleted or unlabelled, and never touches credential keys that no Secret lists. Providers are still created by whatever @@ -95,9 +108,10 @@ accident, and one Secret key may be listed only once. - **Audit.** Each outcome is published as a Kubernetes Event on the Secret (`ProviderSynced`, `ProviderSyncSkipped`, `ProviderNotFound`, `ProviderSyncConflict`, `InvalidProviderSync`, `ProviderSyncFailed`, - `AgentRestarted`, `AgentRestartSkipped`, `AgentStartPending`, - `AgentRestartFailed`). `AgentStartPending` is a Warning at each retry of a - start that has not succeeded yet. The + `AgentRestarted`, `AgentRestartSkipped`, `AgentRestartPending`, + `AgentRestartFailed`). `AgentRestartPending` is a Warning at each retry after + a failure, and when a sandbox takes over 2 minutes to stop or start; repeats + of the same failure aggregate into one Event with a count. The gateway separately audits each `UpdateProvider` with the controller's principal. Neither carries values. diff --git a/crates/openshell-provider-sync/src/controller.rs b/crates/openshell-provider-sync/src/controller.rs index b2d8943001..9e6bdbe6e4 100644 --- a/crates/openshell-provider-sync/src/controller.rs +++ b/crates/openshell-provider-sync/src/controller.rs @@ -16,11 +16,11 @@ use kube::{Api, Client, Resource, ResourceExt}; use tokio::sync::Mutex; use tracing::{debug, info, warn}; -use crate::gateway::{ProviderUpdater, SandboxRestarter, Stop, UpdateError}; +use crate::gateway::{ProviderUpdater, SandboxRestarter, UpdateError}; use crate::mapping::{self, Extracted, MappingError}; use crate::state::{Conflict, SecretRef, SyncState}; +use crate::state::{PendingRestart, RestartStage}; use crate::state_store::StateStore; -use openshell_core::proto::SandboxPhase; /// How long to wait before retrying a provider that does not exist yet. const PROVIDER_MISSING_RETRY: Duration = Duration::from_secs(30); @@ -29,15 +29,8 @@ const TRANSIENT_RETRY: Duration = Duration::from_secs(10); /// How long to wait before retrying a Secret that can only succeed once it, /// the provider or the gateway's configuration changes. const INVALID_RETRY: Duration = Duration::from_mins(5); -/// First and longest wait between attempts to start a sandbox this -/// controller stopped. Starting is retried until it succeeds. -const START_BACKOFF_FIRST: Duration = Duration::from_secs(10); -const START_BACKOFF_MAX: Duration = Duration::from_mins(5); -/// How often to look again while a stopped sandbox finishes stopping, and -/// how long that may take before each look is reported. -const STOPPING_POLL: Duration = Duration::from_secs(5); -const STOPPING_GRACE: Duration = Duration::from_mins(1); - +/// How often applied fingerprints of deleted Secrets are dropped. +const PRUNE_INTERVAL: Duration = Duration::from_mins(10); /// The result of syncing one provider. #[derive(Debug, PartialEq, Eq)] pub enum Outcome { @@ -45,9 +38,10 @@ pub enum Outcome { Updated { keys: Vec, mutation_id: String, - /// Sandboxes attached to the provider whose agents must restart to - /// pick up the new values. Empty for a baseline: values this - /// controller had not applied before, such as at first sight. + /// Sandboxes attached to the provider whose restart this rotation + /// requested. Empty for a baseline: values this controller had not + /// applied before, such as at first sight. A sandbox already + /// restarting is marked to restart once more instead. restart: Vec, }, /// Nothing changed since the last push. @@ -212,14 +206,24 @@ impl Context { .update_credentials(&target.workspace, provider, &changed) .await?; state.commit(secret_ref, &target.workspace, provider, &values); + // Requested in the same state as the commit, so one save + // records both: a controller restart can never see the + // rotation as applied and its restarts as never asked for. + let restart = if rotation && self.restarter.is_some() { + updated + .sandboxes + .into_iter() + .filter(|sandbox| { + state.request_restart(secret_ref, &target.workspace, sandbox) + }) + .collect() + } else { + Vec::new() + }; Ok(Outcome::Updated { keys: changed.keys().cloned().collect(), mutation_id: updated.mutation_id, - restart: if rotation { - updated.sandboxes - } else { - Vec::new() - }, + restart, }) } .await; @@ -228,8 +232,12 @@ impl Context { if providers .iter() .any(|(_, result)| matches!(result, Ok(Outcome::Updated { .. }))) + && let Err(err) = self.persist(&state).await { - self.persist(&state).await; + // In-memory state is still right. If this controller restarts + // before a later save, the rotation is seen again and requested + // again; a restart never stops anything before it is saved. + warn!(error = %err, "failed to persist provider sync state"); } Ok(SyncReport { workspace: target.workspace, @@ -241,44 +249,83 @@ impl Context { self.state.lock().await.forget(secret_ref); } - async fn persist(&self, state: &SyncState) { - if let Some(store) = &self.store - && let Err(err) = store.save(&state.to_persisted()).await - { - // In-memory state is still right. A lost save costs one extra - // restart of these agents, or a forgotten pending start, only if - // this controller restarts before the next save. - warn!(error = %err, "failed to persist provider sync state"); + async fn persist(&self, state: &SyncState) -> Result<(), String> { + match &self.store { + Some(store) => store.save(&state.to_persisted()).await, + None => Ok(()), } } - /// Claim `sandbox` for a restart and persist the pending start before it - /// is stopped. False when a restart of it is already in progress. - pub async fn begin_restart(&self, secret: &SecretRef, workspace: &str, sandbox: &str) -> bool { + pub(crate) async fn restart_stage( + &self, + workspace: &str, + sandbox: &str, + ) -> Option { + self.state.lock().await.restart_stage(workspace, sandbox) + } + + /// Move a restart to `stage` and save it. Fails when the save fails, so + /// a caller never acts on a stage a controller restart would not see. + pub(crate) async fn set_restart_stage( + &self, + workspace: &str, + sandbox: &str, + next: RestartStage, + ) -> Result<(), String> { let mut state = self.state.lock().await; - if !state.begin_restart(secret, workspace, sandbox) { - return false; + let previous = state.restart_stage(workspace, sandbox); + state.set_restart_stage(workspace, sandbox, next); + if let Err(err) = self.persist(&state).await { + if let Some(previous) = previous { + state.set_restart_stage(workspace, sandbox, previous); + } + return Err(err); } - self.persist(&state).await; - true + Ok(()) } - pub async fn end_restart(&self, workspace: &str, sandbox: &str) { + /// The sandbox runs again; true when it must restart once more. + pub(crate) async fn finish_restart(&self, workspace: &str, sandbox: &str) -> bool { let mut state = self.state.lock().await; - state.end_restart(workspace, sandbox); - self.persist(&state).await; + let again = state.finish_restart(workspace, sandbox); + if let Err(err) = self.persist(&state).await { + // A lost save leaves a finished restart recorded: after a + // controller restart it sees the sandbox ready and finishes again. + warn!(error = %err, "failed to persist provider sync state"); + } + again } - /// Sandboxes stopped by an earlier run that must still be started. - pub async fn pending_starts(&self) -> Vec { - self.state.lock().await.pending_starts() + pub(crate) async fn abandon_restart(&self, workspace: &str, sandbox: &str) { + let mut state = self.state.lock().await; + state.abandon_restart(workspace, sandbox); + if let Err(err) = self.persist(&state).await { + warn!(error = %err, "failed to persist provider sync state"); + } + } + + /// Restarts a previous run left in progress. + pub async fn pending_restarts(&self) -> Vec { + self.state.lock().await.pending_restarts() + } + + async fn prune(&self, exists: impl Fn(&SecretRef) -> bool) { + let mut state = self.state.lock().await; + if state.prune_applied(exists) + && let Err(err) = self.persist(&state).await + { + warn!(error = %err, "failed to persist provider sync state"); + } } } -struct Runtime { - ctx: Context, +pub(crate) struct Runtime { + pub(crate) ctx: Context, store: Store, recorder: Option, + /// Event reasons published, kept only by tests. + #[cfg(test)] + events: std::sync::Mutex>, } fn secret_ref(secret: &Secret) -> SecretRef { @@ -396,7 +443,7 @@ async fn reconcile(secret: Arc, runtime: Arc) -> Result, runtime: Arc) -> Result, - runtime: Arc, - secret: SecretRef, - workspace: String, - sandbox: String, -) { - if !runtime - .ctx - .begin_restart(&secret, &workspace, &sandbox) - .await - { - // A restart of this sandbox is already under way and will start it - // with current values. - debug!(secret = %secret, workspace, sandbox, "agent restart already pending"); - return; - } - match restarter.stop(&workspace, &sandbox).await { - Ok(Stop::Stopped) => { - start_until_running(restarter, runtime, secret, workspace, sandbox).await; - } - Ok(Stop::Skipped { phase }) => { - runtime.ctx.end_restart(&workspace, &sandbox).await; - info!(secret = %secret, workspace, sandbox, phase, "agent not running; left alone"); - let note = format!("Did not restart sandbox {workspace}/{sandbox}: it is {phase}"); - publish( - &runtime, - &secret, - EventType::Normal, - "AgentRestartSkipped", - note, - ) - .await; - } - Err(err) => { - // Nothing was stopped, so the agent keeps running. - runtime.ctx.end_restart(&workspace, &sandbox).await; - warn!(secret = %secret, workspace, sandbox, error = %err, "agent restart failed"); - let note = format!( - "Restarting sandbox {workspace}/{sandbox} failed before it stopped; it keeps running: {err}" - ); - publish( - &runtime, - &secret, - EventType::Warning, - "AgentRestartFailed", - note, - ) - .await; - } - } -} - -/// What one look at a stopped sandbox found. -enum StartStep { - /// It runs again, or reached a state no start applies to. - Done, - /// It no longer exists. - Gone, - /// It is still stopping. - Stopping, - Failed(UpdateError), -} - -/// Start a sandbox this controller stopped, retrying with capped backoff -/// until it runs again or is deleted. Never gives up on a stopped agent. -async fn start_until_running( - restarter: Arc, - runtime: Arc, - secret: SecretRef, - workspace: String, - sandbox: String, -) { - let mut backoff = START_BACKOFF_FIRST; - let mut attempts: u32 = 0; - let mut stopping_since: Option = None; - loop { - // Look before starting: a start that succeeded but whose reply was - // lost must not be retried against a running sandbox forever. - let step = match restarter.phase(&workspace, &sandbox).await { - Ok(None | Some(SandboxPhase::Deleting)) => StartStep::Gone, - Ok(Some(SandboxPhase::Stopping)) => StartStep::Stopping, - Ok(Some(SandboxPhase::Stopped | SandboxPhase::Unknown | SandboxPhase::Unspecified)) => { - match restarter.start(&workspace, &sandbox).await { - Ok(()) => StartStep::Done, - Err(err) => StartStep::Failed(err), - } - } - Ok(Some(_)) => StartStep::Done, - Err(err) => StartStep::Failed(err), - }; - let delay = match step { - StartStep::Done => { - runtime.ctx.end_restart(&workspace, &sandbox).await; - info!(secret = %secret, workspace, sandbox, "restarted agent after credential rotation"); - let note = format!( - "Restarted sandbox {workspace}/{sandbox} after its credentials rotated" - ); - publish(&runtime, &secret, EventType::Normal, "AgentRestarted", note).await; - return; - } - StartStep::Gone => { - runtime.ctx.end_restart(&workspace, &sandbox).await; - warn!(secret = %secret, workspace, sandbox, "sandbox deleted before it started again"); - let note = - format!("Did not start sandbox {workspace}/{sandbox} again: it was deleted"); - publish( - &runtime, - &secret, - EventType::Warning, - "AgentRestartFailed", - note, - ) - .await; - return; - } - StartStep::Stopping => { - let since = *stopping_since.get_or_insert_with(tokio::time::Instant::now); - if since.elapsed() >= STOPPING_GRACE { - let note = format!( - "Sandbox {workspace}/{sandbox} is still stopping after {}s; it will be started once stopped", - since.elapsed().as_secs() - ); - publish( - &runtime, - &secret, - EventType::Warning, - "AgentStartPending", - note, - ) - .await; - stopping_since = Some(tokio::time::Instant::now()); - } - STOPPING_POLL - } - StartStep::Failed(err) => { - attempts += 1; - warn!( - secret = %secret, - workspace, - sandbox, - attempts, - retry_in_secs = backoff.as_secs(), - error = %err, - "starting stopped agent failed; retrying" - ); - let note = format!( - "Starting sandbox {workspace}/{sandbox} failed (attempt {attempts}); retrying in {}s: {err}", - backoff.as_secs() - ); - publish( - &runtime, - &secret, - EventType::Warning, - "AgentStartPending", - note, - ) - .await; - let delay = backoff; - backoff = (backoff * 2).min(START_BACKOFF_MAX); - delay - } - }; - tokio::time::sleep(delay).await; - } -} - fn error_policy(_secret: Arc, err: &SyncError, _runtime: Arc) -> Action { Action::requeue(err.retry_after()) } -async fn publish( +pub(crate) async fn publish( runtime: &Runtime, secret: &SecretRef, type_: EventType, reason: &str, note: String, ) { + #[cfg(test)] + runtime.events.lock().unwrap().push(reason.to_string()); let Some(recorder) = &runtime.recorder else { return; }; @@ -639,16 +519,20 @@ pub async fn run(client: Client, ctx: Context, options: Options) { recorder: options .reporter .map(|reporter| Recorder::new(client, reporter)), + #[cfg(test)] + events: std::sync::Mutex::default(), }); - // An earlier run may have stopped agents it had not started yet. + // An earlier run may have left restarts in progress, including agents + // it stopped and had not started yet. if let Some(restarter) = &runtime.ctx.restarter { - for pending in runtime.ctx.pending_starts().await { + for pending in runtime.ctx.pending_restarts().await { info!( workspace = pending.workspace, sandbox = pending.sandbox, - "resuming the start of an agent stopped for a restart" + stage = ?pending.stage, + "resuming an agent restart" ); - tokio::spawn(start_until_running( + tokio::spawn(crate::restart::drive( Arc::clone(restarter), Arc::clone(&runtime), pending.secret, @@ -657,6 +541,21 @@ pub async fn run(client: Client, ctx: Context, options: Options) { )); } } + // Fingerprints of deleted Secrets would otherwise be kept forever: the + // watch does not see a Secret that is gone or no longer labelled. + let pruner = Arc::clone(&runtime); + tokio::spawn(async move { + if pruner.store.wait_until_ready().await.is_err() { + return; + } + loop { + pruner + .ctx + .prune(|secret| is_live(&pruner.store, secret)) + .await; + tokio::time::sleep(PRUNE_INTERVAL).await; + } + }); controller .shutdown_on_signal() .run(reconcile, error_policy, runtime) @@ -676,6 +575,7 @@ mod tests { }; use k8s_openapi::ByteString; use kube::api::ObjectMeta; + use openshell_core::proto::SandboxPhase; use std::collections::BTreeMap; use std::sync::Mutex as StdMutex; @@ -927,7 +827,8 @@ mod tests { #[tokio::test] async fn a_rotation_restarts_only_the_agents_attached_to_its_provider() { let updater = Arc::new(FakeUpdater::default()); - let ctx = Context::new(updater.clone(), "default"); + let ctx = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)); let (r, s) = agent(&PM_CREDENTIALS); let report = ctx.sync(&r, &s, |_| true).await.unwrap(); assert!( @@ -953,7 +854,8 @@ mod tests { async fn an_agent_restarts_once_when_several_of_its_providers_rotate() { let updater = Arc::new(FakeUpdater::default()); *updater.attached.lock().unwrap() = Some(vec!["pm".to_string()]); - let ctx = Context::new(updater.clone(), "default"); + let ctx = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)); let (r, s) = agent(&PM_CREDENTIALS); ctx.sync(&r, &s, |_| true).await.unwrap(); @@ -971,8 +873,9 @@ mod tests { async fn persisted_state_keeps_a_controller_restart_from_restarting_agents() { let store = Arc::new(MemoryStore::default()); let updater = Arc::new(FakeUpdater::default()); - let first = - Context::new(updater.clone(), "default").with_state(SyncState::new(), store.clone()); + let first = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)) + .with_state(SyncState::new(), store.clone()); let (r, s) = agent(&PM_CREDENTIALS); first.sync(&r, &s, |_| true).await.unwrap(); updater.calls.lock().unwrap().clear(); @@ -980,6 +883,7 @@ mod tests { // A new controller process resumes from the saved fingerprints. let persisted = store.load().await.unwrap().expect("state saved"); let second = Context::new(updater.clone(), "default") + .with_restarter(FakeSandbox::new(SandboxPhase::Ready)) .with_state(SyncState::from_persisted(persisted), store.clone()); let report = second.sync(&r, &s, |_| true).await.unwrap(); assert!(report.sandboxes_to_restart().is_empty()); @@ -993,208 +897,366 @@ mod tests { assert_eq!(report.sandboxes_to_restart(), ["linear-agent"]); } - /// Scripted sandbox lifecycle: phases returned in order (the last - /// repeats), and start results in order (then success). - #[derive(Default)] - struct FakeRestarter { - stop_skipped: StdMutex>, - phases: StdMutex>>, + /// A sandbox whose phase moves like the gateway's: a stop goes through + /// Stopping to Stopped, a start through Starting to `after_start`. + struct FakeSandbox { + phase: StdMutex>, + after_start: StdMutex, + phase_failures: StdMutex, + stop_failures: StdMutex, start_failures: StdMutex, stops: StdMutex, starts: StdMutex, } - #[async_trait::async_trait] - impl SandboxRestarter for FakeRestarter { - async fn stop(&self, _: &str, _: &str) -> Result { - *self.stops.lock().unwrap() += 1; - Ok(self - .stop_skipped - .lock() - .unwrap() - .clone() - .map_or(Stop::Stopped, |phase| Stop::Skipped { phase })) + impl FakeSandbox { + fn new(phase: SandboxPhase) -> Arc { + Arc::new(Self { + phase: StdMutex::new(Some(phase)), + after_start: StdMutex::new(SandboxPhase::Ready), + phase_failures: StdMutex::new(0), + stop_failures: StdMutex::new(0), + start_failures: StdMutex::new(0), + stops: StdMutex::new(0), + starts: StdMutex::new(0), + }) + } + fn counts(&self) -> (u32, u32) { + (*self.stops.lock().unwrap(), *self.starts.lock().unwrap()) + } + } + + fn fail_once(counter: &StdMutex) -> bool { + let mut remaining = counter.lock().unwrap(); + if *remaining > 0 { + *remaining -= 1; + return true; } + false + } + + #[async_trait::async_trait] + impl SandboxRestarter for FakeSandbox { async fn phase(&self, _: &str, _: &str) -> Result, UpdateError> { - let mut phases = self.phases.lock().unwrap(); - Ok(if phases.len() > 1 { - phases.remove(0) - } else { - phases.first().copied().flatten() - }) + if fail_once(&self.phase_failures) { + return Err(UpdateError::Transient( + "Unavailable: gateway down".to_string(), + )); + } + let mut phase = self.phase.lock().unwrap(); + let current = *phase; + *phase = match current { + Some(SandboxPhase::Stopping) => Some(SandboxPhase::Stopped), + Some(SandboxPhase::Starting) => Some(*self.after_start.lock().unwrap()), + other => other, + }; + Ok(current) + } + async fn stop(&self, _: &str, _: &str) -> Result<(), UpdateError> { + if fail_once(&self.stop_failures) { + return Err(UpdateError::Transient( + "Unavailable: gateway down".to_string(), + )); + } + *self.stops.lock().unwrap() += 1; + *self.phase.lock().unwrap() = Some(SandboxPhase::Stopping); + Ok(()) } async fn start(&self, _: &str, _: &str) -> Result<(), UpdateError> { - *self.starts.lock().unwrap() += 1; - let mut failures = self.start_failures.lock().unwrap(); - if *failures > 0 { - *failures -= 1; + if fail_once(&self.start_failures) { return Err(UpdateError::Transient( "Unavailable: gateway down".to_string(), )); } + *self.starts.lock().unwrap() += 1; + *self.phase.lock().unwrap() = Some(SandboxPhase::Starting); Ok(()) } } + /// A store whose saves can be made to fail. + #[derive(Default)] + struct FlakyStore { + inner: MemoryStore, + failing: StdMutex, + } + + #[async_trait::async_trait] + impl StateStore for FlakyStore { + async fn load(&self) -> Result, String> { + self.inner.load().await + } + async fn save(&self, state: &crate::state::PersistedState) -> Result<(), String> { + if *self.failing.lock().unwrap() { + return Err("apiserver unavailable".to_string()); + } + self.inner.save(state).await + } + } + fn runtime(ctx: Context) -> Arc { let (store, _writer) = kube::runtime::reflector::store(); Arc::new(Runtime { ctx, store, recorder: None, + events: StdMutex::default(), }) } - #[tokio::test(start_paused = true)] - async fn a_stopped_agent_is_started_however_long_the_gateway_is_down() { - let store = Arc::new(MemoryStore::default()); - let runtime = runtime( + fn restarting_runtime( + sandbox: &Arc, + store: Arc, + state: SyncState, + ) -> Arc { + runtime( Context::new(Arc::new(FakeUpdater::default()), "default") - .with_state(SyncState::new(), store.clone()), - ); - let restarter = Arc::new(FakeRestarter::default()); - *restarter.phases.lock().unwrap() = vec![Some(SandboxPhase::Stopped)]; - // Far longer than any fixed deadline: 30 failures at up to 5 minutes. - *restarter.start_failures.lock().unwrap() = 30; + .with_restarter(sandbox.clone()) + .with_state(state, store), + ) + } + + fn events(runtime: &Runtime) -> Vec { + runtime.events.lock().unwrap().clone() + } + async fn drive(sandbox: &Arc, runtime: &Arc) { let (secret, _) = github("v1"); - restart_sandbox( - restarter.clone(), - runtime.clone(), - secret.clone(), + crate::restart::drive( + sandbox.clone(), + Arc::clone(runtime), + secret, "oncall".to_string(), "triage".to_string(), ) .await; - // The pending start was persisted before anything was stopped. - let first_save = store.0.lock().unwrap().first().cloned().unwrap(); + } + + fn requested(at: RestartStage) -> SyncState { + let mut state = SyncState::new(); + let (secret, _) = github("v1"); + assert!(state.request_restart(&secret, "oncall", "triage")); + state.set_restart_stage("oncall", "triage", at); + state + } + + #[tokio::test(start_paused = true)] + async fn a_rotation_records_its_restart_in_the_same_save_and_restarts_the_agent() { + let store = Arc::new(MemoryStore::default()); + let updater = Arc::new(FakeUpdater::default()); + *updater.attached.lock().unwrap() = Some(vec!["triage".to_string()]); + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let ctx = Context::new(updater, "default") + .with_restarter(sandbox.clone()) + .with_state(SyncState::new(), store.clone()); + let (r, s) = github("v1"); + ctx.sync(&r, &s, |_| true).await.unwrap(); + let (r, s) = github("v2"); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.sandboxes_to_restart(), ["triage"]); + + // The save that records the new fingerprint also records the + // restart, so no crash can apply the rotation and forget the restart. + let saved = SyncState::from_persisted(store.load().await.unwrap().unwrap()); assert_eq!( - SyncState::from_persisted(first_save).pending_starts().len(), - 1, - "pending start saved before the stop" + saved.restart_stage("oncall", "triage"), + Some(RestartStage::Requested) ); - assert_eq!(*restarter.stops.lock().unwrap(), 1); - assert_eq!(*restarter.starts.lock().unwrap(), 31); - assert!(runtime.ctx.pending_starts().await.is_empty()); - let saved = store.load().await.unwrap().unwrap(); - assert!(SyncState::from_persisted(saved).pending_starts().is_empty()); + assert!(saved.has_applied(&r, "oncall", "github")); + assert!( + saved + .changed( + &r, + "oncall", + "github", + &CredentialValues(BTreeMap::from([( + "GITHUB_TOKEN".to_string(), + "v2".to_string() + )])) + ) + .0 + .is_empty() + ); + + let runtime = runtime(ctx); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(*sandbox.phase.lock().unwrap(), Some(SandboxPhase::Ready)); + assert!(runtime.ctx.pending_restarts().await.is_empty()); + assert_eq!(events(&runtime), ["AgentRestarted"]); } #[tokio::test(start_paused = true)] - async fn a_controller_restart_resumes_a_pending_start() { - let store = Arc::new(MemoryStore::default()); - let mut before = SyncState::new(); - let (secret, _) = github("v1"); - assert!(before.begin_restart(&secret, "oncall", "triage")); - store.save(&before.to_persisted()).await.unwrap(); + async fn a_long_gateway_outage_delays_the_restart_but_never_leaves_the_agent_stopped() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + *sandbox.stop_failures.lock().unwrap() = 3; + *sandbox.start_failures.lock().unwrap() = 30; + *sandbox.phase_failures.lock().unwrap() = 2; + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(*sandbox.phase.lock().unwrap(), Some(SandboxPhase::Ready)); + assert!(runtime.ctx.pending_restarts().await.is_empty()); + let events = events(&runtime); + assert_eq!(events.last().map(String::as_str), Some("AgentRestarted")); + assert!( + events + .iter() + .filter(|reason| *reason == "AgentRestartPending") + .count() + >= 35 + ); + } - let persisted = store.load().await.unwrap().unwrap(); - let runtime = runtime( - Context::new(Arc::new(FakeUpdater::default()), "default") - .with_state(SyncState::from_persisted(persisted), store.clone()), + #[tokio::test(start_paused = true)] + async fn a_requested_restart_resumed_after_a_crash_still_stops_and_starts() { + // Crash after the request was saved but before the stop reached the + // gateway: the agent still runs on old connections. + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), ); - let pending = runtime.ctx.pending_starts().await; - assert_eq!(pending.len(), 1); - - let restarter = Arc::new(FakeRestarter::default()); - // Still stopping at first, then stopped. - *restarter.phases.lock().unwrap() = vec![ - Some(SandboxPhase::Stopping), - Some(SandboxPhase::Stopping), - Some(SandboxPhase::Stopped), - ]; - let pending = pending.into_iter().next().unwrap(); - start_until_running( - restarter.clone(), - runtime.clone(), - pending.secret, - pending.workspace, - pending.sandbox, - ) - .await; + drive(&sandbox, &runtime).await; assert_eq!( - *restarter.stops.lock().unwrap(), - 0, - "resuming never stops again" + sandbox.counts(), + (1, 1), + "a ready sandbox is not mistaken for restarted" ); - assert_eq!(*restarter.starts.lock().unwrap(), 1); - assert!(runtime.ctx.pending_starts().await.is_empty()); } #[tokio::test(start_paused = true)] - async fn a_start_that_already_happened_is_not_repeated() { - let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); - let restarter = Arc::new(FakeRestarter::default()); - *restarter.phases.lock().unwrap() = vec![Some(SandboxPhase::Ready)]; - let (secret, _) = github("v1"); - start_until_running( - restarter.clone(), - runtime, - secret, - "oncall".to_string(), - "triage".to_string(), - ) - .await; - assert_eq!(*restarter.starts.lock().unwrap(), 0); + async fn a_resumed_stop_starts_the_sandbox_without_stopping_it_again() { + let sandbox = FakeSandbox::new(SandboxPhase::Stopped); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Stopping), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 1)); + assert_eq!(events(&runtime), ["AgentRestarted"]); } #[tokio::test(start_paused = true)] - async fn a_deleted_sandbox_ends_the_pending_start() { - let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); - let (secret, _) = github("v1"); - assert!(runtime.ctx.begin_restart(&secret, "oncall", "triage").await); - let restarter = Arc::new(FakeRestarter::default()); - *restarter.phases.lock().unwrap() = vec![None]; - start_until_running( - restarter.clone(), - runtime.clone(), - secret, - "oncall".to_string(), - "triage".to_string(), - ) - .await; - assert_eq!(*restarter.starts.lock().unwrap(), 0); - assert!(runtime.ctx.pending_starts().await.is_empty()); + async fn nothing_is_stopped_until_the_restart_is_saved() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let store = Arc::new(FlakyStore::default()); + *store.failing.lock().unwrap() = true; + let runtime = + restarting_runtime(&sandbox, store.clone(), requested(RestartStage::Requested)); + let task = tokio::spawn({ + let sandbox = sandbox.clone(); + let runtime = runtime.clone(); + async move { drive(&sandbox, &runtime).await } + }); + tokio::time::sleep(Duration::from_mins(30)).await; + assert_eq!(sandbox.counts(), (0, 0), "no stop while the save fails"); + assert_eq!( + runtime.ctx.restart_stage("oncall", "triage").await, + Some(RestartStage::Requested) + ); + + *store.failing.lock().unwrap() = false; + task.await.unwrap(); + assert_eq!(sandbox.counts(), (1, 1)); } #[tokio::test(start_paused = true)] - async fn a_sandbox_someone_else_stopped_is_left_alone() { - let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); - let restarter = Arc::new(FakeRestarter::default()); - *restarter.stop_skipped.lock().unwrap() = Some("SANDBOX_PHASE_STOPPED".to_string()); - let (secret, _) = github("v1"); - restart_sandbox( - restarter.clone(), - runtime.clone(), - secret, - "oncall".to_string(), - "triage".to_string(), - ) - .await; - assert_eq!(*restarter.starts.lock().unwrap(), 0); - assert!(runtime.ctx.pending_starts().await.is_empty()); + async fn an_agent_in_error_after_its_start_is_not_reported_restarted() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + *sandbox.after_start.lock().unwrap() = SandboxPhase::Error; + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(events(&runtime), ["AgentRestartFailed"]); + assert!(runtime.ctx.pending_restarts().await.is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_starting_agent_is_restarted_once_it_runs() { + let sandbox = FakeSandbox::new(SandboxPhase::Starting); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (1, 1)); + assert_eq!(events(&runtime), ["AgentRestarted"]); + } + + #[tokio::test(start_paused = true)] + async fn agents_someone_else_stopped_or_failed_are_left_alone() { + for phase in [ + SandboxPhase::Stopped, + SandboxPhase::Stopping, + SandboxPhase::Completed, + SandboxPhase::Error, + SandboxPhase::Deleting, + ] { + let sandbox = FakeSandbox::new(phase); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::Requested), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 0), "{phase:?}"); + assert_eq!(events(&runtime), ["AgentRestartSkipped"], "{phase:?}"); + } + // Stopped by someone else after this controller started it. + let sandbox = FakeSandbox::new(SandboxPhase::Stopped); + let runtime = restarting_runtime( + &sandbox, + Arc::new(MemoryStore::default()), + requested(RestartStage::AwaitingRunning), + ); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 0)); + assert_eq!(events(&runtime), ["AgentRestartSkipped"]); } #[tokio::test(start_paused = true)] - async fn a_second_rotation_does_not_restart_a_sandbox_already_restarting() { - let runtime = runtime(Context::new(Arc::new(FakeUpdater::default()), "default")); + async fn a_rotation_during_a_restart_restarts_the_agent_once_more() { + let sandbox = FakeSandbox::new(SandboxPhase::Ready); + let mut state = requested(RestartStage::Requested); let (secret, _) = github("v1"); - assert!(runtime.ctx.begin_restart(&secret, "oncall", "triage").await); - let restarter = Arc::new(FakeRestarter::default()); - restart_sandbox( - restarter.clone(), - runtime.clone(), - secret, - "oncall".to_string(), - "triage".to_string(), - ) - .await; - assert_eq!(*restarter.stops.lock().unwrap(), 0); - assert_eq!( - runtime.ctx.pending_starts().await.len(), - 1, - "the first restart still owns it" + assert!(!state.request_restart(&secret, "oncall", "triage")); + let runtime = restarting_runtime(&sandbox, Arc::new(MemoryStore::default()), state); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (2, 2)); + assert_eq!(events(&runtime), ["AgentRestarted", "AgentRestarted"]); + } + + #[tokio::test(start_paused = true)] + async fn a_controller_restart_resumes_restarts_from_saved_state() { + let store = Arc::new(MemoryStore::default()); + store + .save(&requested(RestartStage::Stopping).to_persisted()) + .await + .unwrap(); + let persisted = store.load().await.unwrap().unwrap(); + let sandbox = FakeSandbox::new(SandboxPhase::Stopped); + let runtime = restarting_runtime( + &sandbox, + store.clone(), + SyncState::from_persisted(persisted), ); + assert_eq!(runtime.ctx.pending_restarts().await.len(), 1); + drive(&sandbox, &runtime).await; + assert_eq!(sandbox.counts(), (0, 1)); + let saved = SyncState::from_persisted(store.load().await.unwrap().unwrap()); + assert!(saved.pending_restarts().is_empty()); } #[tokio::test] diff --git a/crates/openshell-provider-sync/src/gateway.rs b/crates/openshell-provider-sync/src/gateway.rs index f97c2b6386..009e9a1ed0 100644 --- a/crates/openshell-provider-sync/src/gateway.rs +++ b/crates/openshell-provider-sync/src/gateway.rs @@ -161,30 +161,21 @@ impl ProviderUpdater for GatewayUpdater { } } -/// What stopping a sandbox for a restart did. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum Stop { - Stopped, - /// Stopped, stopping, completed, being deleted or gone: someone or - /// something else decided it should not run, so it is left alone. - Skipped { - phase: String, - }, -} - -/// Stops and starts sandboxes so their agents start fresh with current -/// credentials. Retrying is the caller's job. +/// Gateway calls a restart needs. Each is one attempt; deciding what to do +/// next and retrying is the caller's job (see [`crate::restart`]). #[async_trait::async_trait] pub trait SandboxRestarter: Send + Sync + 'static { - /// Stop the sandbox unless it is already meant to stay down. - async fn stop(&self, workspace: &str, sandbox: &str) -> Result; /// The sandbox's phase, or `None` when it no longer exists. async fn phase( &self, workspace: &str, sandbox: &str, ) -> Result, UpdateError>; - /// One start attempt. + /// Ask the gateway to stop the sandbox. A sandbox that no longer exists + /// is not an error; the next phase read reports it. + async fn stop(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError>; + /// Ask the gateway to start the sandbox, with the same treatment of a + /// sandbox that no longer exists. async fn start(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError>; } @@ -228,41 +219,22 @@ impl GatewayRestarter { } } -/// A sandbox in one of these phases is meant to stay down. -pub(crate) fn left_alone(phase: SandboxPhase) -> bool { - matches!( - phase, - SandboxPhase::Stopping - | SandboxPhase::Stopped - | SandboxPhase::Completed - | SandboxPhase::Deleting - ) +fn sandbox_error(status: &tonic::Status, workspace: &str, sandbox: &str) -> UpdateError { + let message = format!( + "sandbox {workspace}/{sandbox}: {:?}: {}", + status.code(), + status.message() + ); + match status.code() { + Code::InvalidArgument | Code::FailedPrecondition | Code::PermissionDenied => { + UpdateError::Rejected(message) + } + _ => UpdateError::Transient(message), + } } #[async_trait::async_trait] impl SandboxRestarter for GatewayRestarter { - async fn stop(&self, workspace: &str, sandbox: &str) -> Result { - let Some(phase) = self.phase(workspace, sandbox).await? else { - return Ok(Stop::Skipped { - phase: "deleted".to_string(), - }); - }; - if left_alone(phase) { - return Ok(Stop::Skipped { - phase: phase.as_str_name().to_string(), - }); - } - // A request id makes a retried stop a replay, not a repeat. - let stop = StopSandboxRequest { - workspace_scope: Some(workspace_selector(workspace)), - name: sandbox.to_string(), - request_id: uuid::Uuid::new_v4().to_string(), - }; - call_refreshing!(&self.client, stop_sandbox, stop) - .map_err(|status| UpdateError::from_status(&status, workspace, sandbox))?; - Ok(Stop::Stopped) - } - async fn phase( &self, workspace: &str, @@ -281,7 +253,21 @@ impl SandboxRestarter for GatewayRestarter { .map_or(SandboxPhase::Unspecified, |status| status.phase()), )), Err(status) if status.code() == Code::NotFound => Ok(None), - Err(status) => Err(UpdateError::from_status(&status, workspace, sandbox)), + Err(status) => Err(sandbox_error(&status, workspace, sandbox)), + } + } + + async fn stop(&self, workspace: &str, sandbox: &str) -> Result<(), UpdateError> { + // A request id makes a retried call a replay, not a repeat. + let stop = StopSandboxRequest { + workspace_scope: Some(workspace_selector(workspace)), + name: sandbox.to_string(), + request_id: uuid::Uuid::new_v4().to_string(), + }; + match call_refreshing!(&self.client, stop_sandbox, stop) { + Ok(_) => Ok(()), + Err(status) if status.code() == Code::NotFound => Ok(()), + Err(status) => Err(sandbox_error(&status, workspace, sandbox)), } } @@ -291,9 +277,11 @@ impl SandboxRestarter for GatewayRestarter { name: sandbox.to_string(), request_id: uuid::Uuid::new_v4().to_string(), }; - call_refreshing!(&self.client, start_sandbox, start) - .map(|_| ()) - .map_err(|status| UpdateError::from_status(&status, workspace, sandbox)) + match call_refreshing!(&self.client, start_sandbox, start) { + Ok(_) => Ok(()), + Err(status) if status.code() == Code::NotFound => Ok(()), + Err(status) => Err(sandbox_error(&status, workspace, sandbox)), + } } } @@ -380,25 +368,13 @@ mod tests { } #[test] - fn stopped_sandboxes_are_left_alone() { - for phase in [ - SandboxPhase::Stopping, - SandboxPhase::Stopped, - SandboxPhase::Completed, - SandboxPhase::Deleting, - ] { - assert!(left_alone(phase), "{phase:?}"); - } - for phase in [ - SandboxPhase::Ready, - SandboxPhase::Error, - SandboxPhase::Provisioning, - SandboxPhase::Starting, - SandboxPhase::Unknown, - SandboxPhase::Unspecified, - ] { - assert!(!left_alone(phase), "{phase:?}"); - } + fn sandbox_errors_name_the_sandbox() { + let error = sandbox_error(&tonic::Status::failed_precondition("starting"), "w", "pm"); + assert!( + matches!(&error, UpdateError::Rejected(message) if message.contains("sandbox w/pm")) + ); + let error = sandbox_error(&tonic::Status::unavailable("down"), "w", "pm"); + assert!(matches!(error, UpdateError::Transient(_))); } #[test] diff --git a/crates/openshell-provider-sync/src/lib.rs b/crates/openshell-provider-sync/src/lib.rs index 347c7fbcb7..dab72e615b 100644 --- a/crates/openshell-provider-sync/src/lib.rs +++ b/crates/openshell-provider-sync/src/lib.rs @@ -16,5 +16,6 @@ pub mod controller; pub mod gateway; pub mod mapping; +pub mod restart; pub mod state; pub mod state_store; diff --git a/crates/openshell-provider-sync/src/main.rs b/crates/openshell-provider-sync/src/main.rs index c088c45809..8ff3eddd50 100644 --- a/crates/openshell-provider-sync/src/main.rs +++ b/crates/openshell-provider-sync/src/main.rs @@ -122,22 +122,27 @@ async fn main() -> Result<()> { ); if !cli.no_restart { context = context.with_restarter(Arc::new(GatewayRestarter::new(gateway))); + if cli.state_secret.is_none() { + tracing::warn!( + "agent restarts are on without --state-secret: a restart in progress is \ + forgotten if this controller restarts, which can leave an agent stopped" + ); + } } if let Some(name) = &cli.state_secret { let namespace = cli.state_namespace.clone().ok_or_else(|| { miette::miette!("--state-secret needs --state-namespace or POD_NAMESPACE") })?; let store = SecretStateStore::new(kube_client.clone(), &namespace, name.clone()); - let state = match store.load().await { - Ok(Some(persisted)) => SyncState::from_persisted(persisted), - Ok(None) => SyncState::new(), - Err(err) => { - // Starting fresh only costs a baseline sync that restarts - // nothing; refusing to start would stop all syncing. - tracing::warn!(error = %err, "could not load provider sync state; starting fresh"); - SyncState::new() - } - }; + // Refuse to start on state that exists but cannot be read: starting + // fresh would overwrite it with the next save and forget restarts in + // progress, which could leave an agent stopped. Crash-looping until it + // is readable is the safe failure. + let state = store + .load() + .await + .map_err(|err| miette::miette!("load provider sync state: {err}"))? + .map_or_else(SyncState::new, SyncState::from_persisted); context = context.with_state(state, Arc::new(store)); } diff --git a/crates/openshell-provider-sync/src/restart.rs b/crates/openshell-provider-sync/src/restart.rs new file mode 100644 index 0000000000..79b5772380 --- /dev/null +++ b/crates/openshell-provider-sync/src/restart.rs @@ -0,0 +1,284 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Restarting an agent after its credentials rotate, without ever leaving it +//! stopped. +//! +//! A restart is a persisted state machine, one per sandbox: +//! +//! - **Requested**: a rotation needs the agent restarted. Nothing is stopped +//! until the next stage has been saved. +//! - **Stopping**: this controller stops the sandbox, then starts it. +//! - **`AwaitingRunning`**: the start was accepted; it is done when the sandbox +//! is ready. +//! +//! Every gateway failure is retried with capped backoff. The stage is read +//! from the sandbox's phase on every step, so a reply lost in a crash or a +//! network failure never makes a step repeat wrongly, and a controller +//! restart resumes each restart where it was. A sandbox that someone else +//! stops or deletes, or that fails on its own, is left to them. + +use std::sync::Arc; +use std::time::Duration; + +use kube::runtime::events::EventType; +use openshell_core::proto::SandboxPhase; +use tracing::{info, warn}; + +use crate::controller::{Runtime, publish}; +use crate::gateway::SandboxRestarter; +use crate::state::{RestartStage, SecretRef}; + +/// First and longest wait between attempts after a failure. Retried until +/// it succeeds. +const BACKOFF_FIRST: Duration = Duration::from_secs(10); +const BACKOFF_MAX: Duration = Duration::from_mins(5); +/// How often to look again while the sandbox changes phase, and how long a +/// phase may take before each look is reported. +const POLL: Duration = Duration::from_secs(5); +const SLOW: Duration = Duration::from_mins(2); + +/// What one look at the sandbox decided. +enum Step { + /// Moved to the next stage; look again now. + Advanced, + /// Waiting for the sandbox to change phase. + Waiting(&'static str), + /// A gateway call or a save failed; retry after backoff. + Failed(String), + /// The sandbox runs again. + Done, + /// The restart ends without this controller starting the sandbox. + Abandoned { + type_: EventType, + reason: &'static str, + note: String, + }, +} + +/// Drive the restart of one sandbox to its end. Started for each new request +/// and, at startup, for each restart a previous run left in progress. +pub(crate) async fn drive( + restarter: Arc, + runtime: Arc, + secret: SecretRef, + workspace: String, + sandbox: String, +) { + let ctx = &runtime.ctx; + let mut backoff = BACKOFF_FIRST; + let mut waiting_since: Option<(&'static str, tokio::time::Instant)> = None; + loop { + let Some(stage) = ctx.restart_stage(&workspace, &sandbox).await else { + return; + }; + let step = match restarter.phase(&workspace, &sandbox).await { + Err(err) => Step::Failed(err.to_string()), + Ok(phase) => { + step( + restarter.as_ref(), + &runtime, + stage, + phase, + &workspace, + &sandbox, + ) + .await + } + }; + let delay = match step { + Step::Advanced => { + backoff = BACKOFF_FIRST; + waiting_since = None; + continue; + } + Step::Waiting(what) => { + let since = match waiting_since { + Some((previous, since)) if previous == what => since, + _ => { + let now = tokio::time::Instant::now(); + waiting_since = Some((what, now)); + now + } + }; + if since.elapsed() >= SLOW { + let note = format!( + "Restart of sandbox {workspace}/{sandbox} is waiting: it is still {what}" + ); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentRestartPending", + note, + ) + .await; + waiting_since = Some((what, tokio::time::Instant::now())); + } + POLL + } + Step::Failed(error) => { + warn!( + secret = %secret, + workspace, + sandbox, + ?stage, + retry_in_secs = backoff.as_secs(), + error, + "agent restart step failed; retrying" + ); + // No counter or delay in the note, so repeats aggregate into + // one Event with a count. + let note = format!( + "Restart of sandbox {workspace}/{sandbox} is retrying after a failure: {error}" + ); + publish( + &runtime, + &secret, + EventType::Warning, + "AgentRestartPending", + note, + ) + .await; + let delay = backoff; + backoff = (backoff * 2).min(BACKOFF_MAX); + delay + } + Step::Done => { + let again = ctx.finish_restart(&workspace, &sandbox).await; + info!(secret = %secret, workspace, sandbox, again, "restarted agent after credential rotation"); + let note = if again { + format!( + "Restarted sandbox {workspace}/{sandbox}; restarting once more for a later rotation" + ) + } else { + format!("Restarted sandbox {workspace}/{sandbox} after its credentials rotated") + }; + publish(&runtime, &secret, EventType::Normal, "AgentRestarted", note).await; + if !again { + return; + } + backoff = BACKOFF_FIRST; + waiting_since = None; + continue; + } + Step::Abandoned { + type_, + reason, + note, + } => { + ctx.abandon_restart(&workspace, &sandbox).await; + info!(secret = %secret, workspace, sandbox, reason, note, "agent restart ended"); + publish(&runtime, &secret, type_, reason, note).await; + return; + } + }; + tokio::time::sleep(delay).await; + } +} + +fn skipped(workspace: &str, sandbox: &str, why: &str) -> Step { + Step::Abandoned { + type_: EventType::Normal, + reason: "AgentRestartSkipped", + note: format!("Did not restart sandbox {workspace}/{sandbox}: {why}"), + } +} + +fn failed(workspace: &str, sandbox: &str, why: &str) -> Step { + Step::Abandoned { + type_: EventType::Warning, + reason: "AgentRestartFailed", + note: format!("Restart of sandbox {workspace}/{sandbox} ended: {why}"), + } +} + +async fn step( + restarter: &dyn SandboxRestarter, + runtime: &Runtime, + stage: RestartStage, + phase: Option, + workspace: &str, + sandbox: &str, +) -> Step { + use SandboxPhase as P; + let ctx = &runtime.ctx; + match stage { + RestartStage::Requested => match phase { + None | Some(P::Deleting) => skipped(workspace, sandbox, "it was deleted"), + Some(P::Stopped | P::Stopping | P::Completed) => skipped( + workspace, + sandbox, + "it is not running, and was stopped by someone else", + ), + Some(P::Error) => skipped( + workspace, + sandbox, + "it is in Error; it fetches the new values when it next starts", + ), + // It may have fetched values from before the rotation; restart it + // once it runs. + Some(P::Provisioning | P::Starting) => Step::Waiting("starting"), + Some(P::Ready | P::Unknown | P::Unspecified) => { + // Nothing is stopped until this controller has saved that it + // owns the stop, so it can never forget to start it. + match ctx + .set_restart_stage(workspace, sandbox, RestartStage::Stopping) + .await + { + Ok(()) => Step::Advanced, + Err(err) => Step::Failed(format!("could not record the restart: {err}")), + } + } + }, + RestartStage::Stopping => match phase { + None | Some(P::Deleting) => failed(workspace, sandbox, "it was deleted while stopped"), + Some(P::Stopping) => Step::Waiting("stopping"), + Some(P::Stopped | P::Completed) => match restarter.start(workspace, sandbox).await { + Ok(()) => { + // A lost save only means a resumed restart sees the + // sandbox starting or ready in this stage and stops it + // once more: an extra restart, never a stopped agent. + if let Err(err) = ctx + .set_restart_stage(workspace, sandbox, RestartStage::AwaitingRunning) + .await + { + warn!(workspace, sandbox, error = %err, "could not record the start"); + } + Step::Advanced + } + Err(err) => Step::Failed(err.to_string()), + }, + // The gateway refuses to stop a sandbox that is still starting. + Some(P::Provisioning | P::Starting) => Step::Waiting("starting"), + Some(P::Error) => failed( + workspace, + sandbox, + "it is in Error after being stopped; the gateway's own recovery owns it now", + ), + Some(P::Ready | P::Unknown | P::Unspecified) => { + match restarter.stop(workspace, sandbox).await { + Ok(()) => Step::Waiting("stopping"), + Err(err) => Step::Failed(err.to_string()), + } + } + }, + RestartStage::AwaitingRunning => match phase { + Some(P::Ready) => Step::Done, + Some(P::Provisioning | P::Starting | P::Unknown | P::Unspecified) => { + Step::Waiting("starting") + } + None | Some(P::Deleting) => skipped(workspace, sandbox, "it was deleted"), + Some(P::Stopped | P::Stopping | P::Completed) => skipped( + workspace, + sandbox, + "someone else stopped it after it started", + ), + Some(P::Error) => failed( + workspace, + sandbox, + "it is in Error after starting; the gateway's own recovery owns it now", + ), + }, + } +} diff --git a/crates/openshell-provider-sync/src/state.rs b/crates/openshell-provider-sync/src/state.rs index 7a5bb76b44..47862cf2e9 100644 --- a/crates/openshell-provider-sync/src/state.rs +++ b/crates/openshell-provider-sync/src/state.rs @@ -71,17 +71,41 @@ pub struct SyncState { salt: [u8; 32], applied: HashMap, owners: HashMap, - pending_starts: BTreeMap<(String, String), SecretRef>, + restarts: BTreeMap<(String, String), Restart>, } -/// A sandbox this controller stopped for a restart and has not yet seen -/// start again. Kept across restarts so a stopped agent is never forgotten. +/// How far a restart has got. Persisted before each step, so a controller +/// restart resumes it and never forgets a sandbox it stopped. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RestartStage { + /// A rotation needs this agent restarted; nothing has been stopped yet. + Requested, + /// This controller is stopping the sandbox, or has stopped it. + Stopping, + /// The start was accepted; waiting for the sandbox to run. + AwaitingRunning, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Restart { + secret: SecretRef, + stage: RestartStage, + /// Another rotation landed while this restart was under way, so the + /// agent may have started on the earlier value: restart once more. + rerun: bool, +} + +/// One sandbox restart in progress. #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct PendingStart { +pub struct PendingRestart { pub workspace: String, pub sandbox: String, /// The Secret whose rotation caused the restart, for its Events. pub secret: SecretRef, + pub stage: RestartStage, + #[serde(default)] + pub rerun: bool, } /// Fingerprints of applied values, kept across controller restarts. @@ -93,7 +117,7 @@ pub struct PersistedState { salt: String, applied: Vec, #[serde(default)] - pending_starts: Vec, + restarts: Vec, } #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -148,15 +172,34 @@ impl SyncState { salt: rand::random(), applied: HashMap::new(), owners: HashMap::new(), - pending_starts: BTreeMap::new(), + restarts: BTreeMap::new(), } } /// Resume from persisted fingerprints. Entries that do not parse are /// dropped, which only makes their next value a baseline. pub fn from_persisted(persisted: PersistedState) -> Self { + // Restarts in progress are kept whatever else is unreadable: losing + // one could leave an agent stopped. + let restarts = persisted + .restarts + .into_iter() + .map(|pending| { + ( + (pending.workspace, pending.sandbox), + Restart { + secret: pending.secret, + stage: pending.stage, + rerun: pending.rerun, + }, + ) + }) + .collect(); let Some(salt) = from_hex(&persisted.salt) else { - return Self::new(); + return Self { + restarts, + ..Self::new() + }; }; let applied = persisted .applied @@ -182,16 +225,11 @@ impl SyncState { )) }) .collect(); - let pending_starts = persisted - .pending_starts - .into_iter() - .map(|pending| ((pending.workspace, pending.sandbox), pending.secret)) - .collect(); Self { salt, applied, owners: HashMap::new(), - pending_starts, + restarts, } } @@ -222,40 +260,93 @@ impl SyncState { PersistedState { salt: to_hex(&self.salt), applied, - pending_starts: self.pending_starts(), + restarts: self.pending_restarts(), } } - /// Record that this controller is about to stop `sandbox` and must start - /// it again. Returns false when a start is already pending, so one restart - /// at a time owns the sandbox. - pub fn begin_restart(&mut self, secret: &SecretRef, workspace: &str, sandbox: &str) -> bool { + /// Ask for `sandbox` to be restarted. Returns true when this starts a + /// new restart; when one is already under way it is marked to run once + /// more, so the agent cannot keep a value from before this rotation. + pub fn request_restart(&mut self, secret: &SecretRef, workspace: &str, sandbox: &str) -> bool { let key = (workspace.to_string(), sandbox.to_string()); - if self.pending_starts.contains_key(&key) { + if let Some(restart) = self.restarts.get_mut(&key) { + restart.rerun = true; return false; } - self.pending_starts.insert(key, secret.clone()); + self.restarts.insert( + key, + Restart { + secret: secret.clone(), + stage: RestartStage::Requested, + rerun: false, + }, + ); true } - /// The sandbox runs again, or will never need starting. - pub fn end_restart(&mut self, workspace: &str, sandbox: &str) { - self.pending_starts + pub fn restart_stage(&self, workspace: &str, sandbox: &str) -> Option { + self.restarts + .get(&(workspace.to_string(), sandbox.to_string())) + .map(|restart| restart.stage) + } + + pub fn set_restart_stage(&mut self, workspace: &str, sandbox: &str, stage: RestartStage) { + if let Some(restart) = self + .restarts + .get_mut(&(workspace.to_string(), sandbox.to_string())) + { + restart.stage = stage; + } + } + + /// The sandbox runs again. Returns true when another rotation asked for + /// one more restart, which is then requested afresh. + pub fn finish_restart(&mut self, workspace: &str, sandbox: &str) -> bool { + let key = (workspace.to_string(), sandbox.to_string()); + match self.restarts.get_mut(&key) { + Some(restart) if restart.rerun => { + restart.rerun = false; + restart.stage = RestartStage::Requested; + true + } + Some(_) => { + self.restarts.remove(&key); + false + } + None => false, + } + } + + /// The restart ends without the sandbox running, as when someone else + /// stopped or deleted it. + pub fn abandon_restart(&mut self, workspace: &str, sandbox: &str) { + self.restarts .remove(&(workspace.to_string(), sandbox.to_string())); } - /// Sandboxes stopped for a restart and not yet seen running again. - pub fn pending_starts(&self) -> Vec { - self.pending_starts + /// Restarts in progress. + pub fn pending_restarts(&self) -> Vec { + self.restarts .iter() - .map(|((workspace, sandbox), secret)| PendingStart { + .map(|((workspace, sandbox), restart)| PendingRestart { workspace: workspace.clone(), sandbox: sandbox.clone(), - secret: secret.clone(), + secret: restart.secret.clone(), + stage: restart.stage, + rerun: restart.rerun, }) .collect() } + /// Drop applied fingerprints of Secrets that no longer exist, so + /// persisted state does not grow without bound. Returns whether any were + /// dropped. Claims are left to the watch. + pub fn prune_applied(&mut self, exists: impl Fn(&SecretRef) -> bool) -> bool { + let before = self.applied.len(); + self.applied.retain(|(secret, _), _| exists(secret)); + self.applied.len() != before + } + /// Whether `secret` has applied values to `provider` before, so a change /// now is a rotation rather than a baseline. pub fn has_applied(&self, secret: &SecretRef, workspace: &str, provider: &str) -> bool { @@ -601,24 +692,54 @@ mod tests { } #[test] - fn a_pending_start_survives_a_restart_and_blocks_a_second_restart() { + fn a_restart_survives_a_controller_restart_at_every_stage() { let mut state = SyncState::new(); - assert!(state.begin_restart(&secret("pm"), "pm", "pm")); - assert!(!state.begin_restart(&secret("pm"), "pm", "pm")); + assert!(state.request_restart(&secret("pm"), "pm", "pm")); + for stage in [ + RestartStage::Requested, + RestartStage::Stopping, + RestartStage::AwaitingRunning, + ] { + state.set_restart_stage("pm", "pm", stage); + let restarted = SyncState::from_persisted(state.to_persisted()); + assert_eq!(restarted.restart_stage("pm", "pm"), Some(stage)); + } + // Even with an unreadable salt, restarts in progress are kept. + let mut persisted = state.to_persisted(); + persisted.salt = "not hex".to_string(); + assert_eq!( + SyncState::from_persisted(persisted).restart_stage("pm", "pm"), + Some(RestartStage::AwaitingRunning) + ); + } - let restarted = SyncState::from_persisted(state.to_persisted()); + #[test] + fn a_rotation_during_a_restart_restarts_once_more() { + let mut state = SyncState::new(); + assert!(state.request_restart(&secret("pm"), "pm", "pm")); + state.set_restart_stage("pm", "pm", RestartStage::AwaitingRunning); + assert!( + !state.request_restart(&secret("pm"), "pm", "pm"), + "one restart at a time" + ); + assert!(state.finish_restart("pm", "pm"), "runs once more"); assert_eq!( - restarted.pending_starts(), - [PendingStart { - workspace: "pm".to_string(), - sandbox: "pm".to_string(), - secret: secret("pm"), - }] + state.restart_stage("pm", "pm"), + Some(RestartStage::Requested) ); + assert!(!state.finish_restart("pm", "pm")); + assert!(state.pending_restarts().is_empty()); + } - state.end_restart("pm", "pm"); - assert!(state.pending_starts().is_empty()); - assert!(state.begin_restart(&secret("pm"), "pm", "pm")); + #[test] + fn prune_drops_fingerprints_of_deleted_secrets() { + let mut state = SyncState::new(); + state.commit(&secret("kept"), "w", "p", &values(&[("A", "1")])); + state.commit(&secret("gone"), "w", "p", &values(&[("A", "1")])); + assert!(state.prune_applied(|secret| secret.name == "kept")); + assert!(state.has_applied(&secret("kept"), "w", "p")); + assert!(!state.has_applied(&secret("gone"), "w", "p")); + assert!(!state.prune_applied(|_| true)); } #[test] diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 55baa7f899..7d27687073 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -966,10 +966,10 @@ kubectl -n logs deployment/ --tail=200 | `ProviderSyncConflict` | Another live Secret already feeds the same workspace, provider and key | | `InvalidProviderSync` | Bad annotations, or only some of a provider's keys are set, or a value is not UTF-8 | | `ProviderSyncFailed` | The gateway rejected the update or was unreachable | -| `AgentRestarted` | A rotation restarted a sandbox attached to that provider | -| `AgentRestartSkipped` | The attached sandbox was stopped, stopping, completed or deleting, so it was left alone | -| `AgentStartPending` | provider-sync stopped the sandbox and its start keeps failing or it is still stopping; retried until it runs, backing off to every 5 minutes, and resumed after a provider-sync restart | -| `AgentRestartFailed` | The stop failed, so the agent kept running on its old connections; or the sandbox was deleted before it could start again | +| `AgentRestarted` | A rotation restarted a sandbox attached to that provider, and it is ready again | +| `AgentRestartSkipped` | The sandbox was stopped, stopping, completed, in Error or deleted, or someone else stopped it after the restart started it; left alone | +| `AgentRestartPending` | A stop, start, phase read or state save failed and is being retried (backing off to every 5 minutes), or the sandbox is taking over 2 minutes to stop or start. Retried until it succeeds, and resumed after a provider-sync restart | +| `AgentRestartFailed` | The sandbox was deleted while stopped for the restart, or ended in Error; the gateway's own recovery owns it | No Events at all means the controller is not watching the Secret: check the label, its `OPENSHELL_PROVIDER_SYNC_NAMESPACE`, and its RBAC (`get`, `list` and @@ -977,9 +977,15 @@ label, its `OPENSHELL_PROVIDER_SYNC_NAMESPACE`, and its RBAC (`get`, `list` and `ProviderSynced` event with no `AgentRestarted` after it means the values were a baseline (the first values the controller saw, or the first sync after it restarted without `--state-secret`), restarts are off (`--no-restart`), or no -sandbox was attached to the provider. Every agent restarting after a -provider-sync restart means its state Secret is missing or unreadable. Avoid -rotating while gateways are +sandbox was attached to the provider. Agents restarting right after a +provider-sync restart mean values changed while it was down, or restarts it had +in progress being resumed; a missing state Secret restarts nothing. provider-sync +refuses to start on a state Secret it cannot parse rather than overwrite it. +Restarts in progress are in that Secret (`state.json`, `restarts`). One ends +when its sandbox is ready again or is deleted. Starting the sandbox by hand does +not cancel it: provider-sync stops and starts it once more, so the agent still +ends up running. +Avoid rotating while gateways are rolling: a supervisor whose provider refetch fails revokes its static credentials. From c43dc88b4d6aea836aa3ad61b64f74cd4da7c116 Mon Sep 17 00:00:00 2001 From: Matt Jane Date: Thu, 15 Oct 2026 15:05:47 -0700 Subject: [PATCH 4/4] fix(provider-sync): recover from unreadable or corrupt state without a person Refusing to start on a state Secret that did not parse stopped all syncing in the namespace until someone deleted it. A state Secret that cannot be read is now retried with backoff and never replaced. One that does not parse is copied to -corrupt-, reported with a StateSecretCorrupt Warning, and replaced by fresh state that keeps every restart entry that still parses. Fingerprints start over as a baseline. An agent whose restart entry was lost stays stopped, since a stop by provider-sync cannot be told from one by a person; the Event says so. Signed-off-by: Matt Jane --- crates/openshell-provider-sync/README.md | 6 ++ crates/openshell-provider-sync/src/main.rs | 85 +++++++++++++++---- crates/openshell-provider-sync/src/state.rs | 58 +++++++++++++ .../src/state_store.rs | 61 +++++++++++-- skills/debug-openshell-cluster/SKILL.md | 8 +- 5 files changed, 194 insertions(+), 24 deletions(-) diff --git a/crates/openshell-provider-sync/README.md b/crates/openshell-provider-sync/README.md index 7c7d26d182..fa328f12e2 100644 --- a/crates/openshell-provider-sync/README.md +++ b/crates/openshell-provider-sync/README.md @@ -61,6 +61,12 @@ accident, and one Secret key may be listed only once. only when the sandbox is ready again. This holds with `--state-secret`; without it a restart in progress is lost when the controller restarts, which can leave an agent stopped, and the controller warns about it at startup. + At startup a state Secret that cannot be read is retried with backoff, never + replaced. One that does not parse is copied to + `-corrupt-` and reported with a `StateSecretCorrupt` + Warning; every restart entry that still parses is kept and the rest starts + over as a baseline. An agent whose restart entry was lost stays stopped, + because a stop by this controller cannot be told from one by a person. - **What a restart leaves to others.** A sandbox that is stopped, stopping, completed, in Error or being deleted when the restart begins is left alone, and so is one someone else stops after the restart started it. A sandbox that diff --git a/crates/openshell-provider-sync/src/main.rs b/crates/openshell-provider-sync/src/main.rs index 8ff3eddd50..29b07ff263 100644 --- a/crates/openshell-provider-sync/src/main.rs +++ b/crates/openshell-provider-sync/src/main.rs @@ -7,12 +7,12 @@ use std::path::PathBuf; use std::sync::Arc; use clap::Parser; -use kube::runtime::events::Reporter; +use kube::runtime::events::{Event, EventType, Recorder, Reporter}; use miette::{IntoDiagnostic, Result, WrapErr}; use openshell_provider_sync::controller::{self, Context, Options}; use openshell_provider_sync::gateway::{GatewayRestarter, GatewayUpdater, TokenFile}; -use openshell_provider_sync::state::SyncState; -use openshell_provider_sync::state_store::{SecretStateStore, StateStore}; +use openshell_provider_sync::state::{PersistedState, SyncState}; +use openshell_provider_sync::state_store::{LoadError, SecretStateStore}; use openshell_sdk::{AuthConfig, ClientConfig, OpenShellClient}; use tracing_subscriber::EnvFilter; @@ -116,6 +116,10 @@ async fn main() -> Result<()> { .into_diagnostic() .wrap_err("build Kubernetes client")?; + let reporter = (!cli.no_events).then_some(Reporter { + controller: cli.reporter.clone(), + instance: cli.reporter_instance.clone(), + }); let mut context = Context::new( Arc::new(GatewayUpdater::new(gateway.clone())), cli.default_workspace.clone(), @@ -134,22 +138,10 @@ async fn main() -> Result<()> { miette::miette!("--state-secret needs --state-namespace or POD_NAMESPACE") })?; let store = SecretStateStore::new(kube_client.clone(), &namespace, name.clone()); - // Refuse to start on state that exists but cannot be read: starting - // fresh would overwrite it with the next save and forget restarts in - // progress, which could leave an agent stopped. Crash-looping until it - // is readable is the safe failure. - let state = store - .load() - .await - .map_err(|err| miette::miette!("load provider sync state: {err}"))? - .map_or_else(SyncState::new, SyncState::from_persisted); + let state = load_state(&store, &kube_client, &namespace, reporter.as_ref()).await; context = context.with_state(state, Arc::new(store)); } - let reporter = (!cli.no_events).then_some(Reporter { - controller: cli.reporter, - instance: cli.reporter_instance, - }); tracing::info!( gateway = %cli.gateway, namespace = cli.namespace.as_deref().unwrap_or(""), @@ -170,6 +162,67 @@ async fn main() -> Result<()> { Ok(()) } +/// Load saved state, never starting over state that is only unreadable for +/// now, and never stopping all syncing over state that is corrupt. +async fn load_state( + store: &SecretStateStore, + client: &kube::Client, + namespace: &str, + reporter: Option<&Reporter>, +) -> SyncState { + let mut backoff = std::time::Duration::from_secs(5); + loop { + match store.load_checked().await { + Ok(Some(persisted)) => return SyncState::from_persisted(persisted), + Ok(None) => return SyncState::new(), + Err(LoadError::Unavailable(err)) => { + // Starting fresh now would overwrite the saved state, and + // with it any restart in progress. + tracing::warn!(error = %err, retry_in_secs = backoff.as_secs(), "provider sync state unavailable; retrying"); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(std::time::Duration::from_mins(5)); + } + Err(LoadError::Corrupt { raw, error }) => { + let (salvaged, recovered) = PersistedState::salvage(&raw); + let copy = store.quarantine(&raw).await; + let kept = match © { + Ok(name) => format!("kept in Secret {name}"), + Err(err) => format!("not kept ({err})"), + }; + let note = format!( + "State in Secret {} does not parse ({error}); {kept}. Starting over: the next values are a \ + baseline that restarts no agents. Recovered {recovered} restart(s) in progress; any \ + other agent this controller had stopped stays stopped, because a stop by this \ + controller cannot be told from one by a person", + store.name() + ); + tracing::warn!(note, "provider sync state is corrupt"); + if let Some(reporter) = reporter { + let recorder = Recorder::new(client.clone(), reporter.clone()); + let reference = k8s_openapi::api::core::v1::ObjectReference { + api_version: Some("v1".to_string()), + kind: Some("Secret".to_string()), + namespace: Some(namespace.to_string()), + name: Some(store.name().to_string()), + ..Default::default() + }; + let event = Event { + type_: EventType::Warning, + reason: "StateSecretCorrupt".to_string(), + note: Some(note), + action: "LoadState".to_string(), + secondary: None, + }; + if let Err(err) = recorder.publish(&event, &reference).await { + tracing::debug!(error = %err, "failed to publish Kubernetes event"); + } + } + return SyncState::from_persisted(salvaged); + } + } + } +} + fn init_tracing(json: bool) { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let builder = tracing_subscriber::fmt().with_env_filter(filter); diff --git a/crates/openshell-provider-sync/src/state.rs b/crates/openshell-provider-sync/src/state.rs index 47862cf2e9..2ee508769f 100644 --- a/crates/openshell-provider-sync/src/state.rs +++ b/crates/openshell-provider-sync/src/state.rs @@ -129,6 +129,32 @@ struct PersistedApplied { fingerprints: BTreeMap, } +impl PersistedState { + /// Recover what can be recovered from state that does not parse: every + /// restart entry that still parses on its own. Fingerprints are dropped, + /// which only makes the next values a baseline; restarts are kept because + /// losing one could leave an agent stopped. Returns the state and how + /// many restarts were recovered. + pub fn salvage(raw: &[u8]) -> (Self, usize) { + let restarts: Vec = serde_json::from_slice::(raw) + .ok() + .and_then(|value| value.get("restarts").cloned()) + .and_then(|restarts| restarts.as_array().cloned()) + .unwrap_or_default() + .into_iter() + .filter_map(|entry| serde_json::from_value(entry).ok()) + .collect(); + let recovered = restarts.len(); + ( + Self { + restarts, + ..Self::default() + }, + recovered, + ) + } +} + fn to_hex(bytes: &[u8; 32]) -> String { use std::fmt::Write as _; bytes @@ -742,6 +768,38 @@ mod tests { assert!(!state.prune_applied(|_| true)); } + #[test] + fn corrupt_state_keeps_every_restart_that_still_parses() { + let mut state = SyncState::new(); + state.commit(&secret("a"), "w", "p", &values(&[("A", "1")])); + assert!(state.request_restart(&secret("a"), "w", "pm")); + state.set_restart_stage("w", "pm", RestartStage::Stopping); + let mut raw = serde_json::to_value(state.to_persisted()).unwrap(); + // Corrupt everything but the one restart. + raw["salt"] = serde_json::json!(42); + raw["applied"] = serde_json::json!("garbage"); + raw["restarts"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({"workspace": "w"})); + let raw = serde_json::to_vec(&raw).unwrap(); + assert!(serde_json::from_slice::(&raw).is_err()); + + let (salvaged, recovered) = PersistedState::salvage(&raw); + assert_eq!(recovered, 1); + let restored = SyncState::from_persisted(salvaged); + assert_eq!( + restored.restart_stage("w", "pm"), + Some(RestartStage::Stopping) + ); + assert!( + !restored.has_applied(&secret("a"), "w", "p"), + "fingerprints become a baseline" + ); + + assert_eq!(PersistedState::salvage(b"not json").1, 0); + } + #[test] fn unreadable_persisted_state_starts_fresh() { let restarted = SyncState::from_persisted(PersistedState { diff --git a/crates/openshell-provider-sync/src/state_store.rs b/crates/openshell-provider-sync/src/state_store.rs index fe9733643d..8c50eb867a 100644 --- a/crates/openshell-provider-sync/src/state_store.rs +++ b/crates/openshell-provider-sync/src/state_store.rs @@ -18,6 +18,18 @@ use crate::state::PersistedState; const STATE_KEY: &str = "state.json"; const FIELD_MANAGER: &str = "openshell-provider-sync"; +/// Why saved state could not be loaded. +#[derive(Debug, thiserror::Error)] +pub enum LoadError { + /// It could not be read, as when the API server is unavailable. Retry: + /// starting without it would overwrite it. + #[error("read state: {0}")] + Unavailable(String), + /// It was read but does not parse. Retrying cannot help. + #[error("state does not parse: {error}")] + Corrupt { raw: Vec, error: String }, +} + /// Where persisted state is kept. #[async_trait::async_trait] pub trait StateStore: Send + Sync + 'static { @@ -39,16 +51,18 @@ impl SecretStateStore { name: name.into(), } } -} -#[async_trait::async_trait] -impl StateStore for SecretStateStore { - async fn load(&self) -> Result, String> { + pub fn name(&self) -> &str { + &self.name + } + + /// Load, telling a read failure from state that does not parse. + pub async fn load_checked(&self) -> Result, LoadError> { let Some(secret) = self .api .get_opt(&self.name) .await - .map_err(|err| format!("read Secret {}: {err}", self.name))? + .map_err(|err| LoadError::Unavailable(format!("Secret {}: {err}", self.name)))? else { return Ok(None); }; @@ -57,7 +71,42 @@ impl StateStore for SecretStateStore { }; serde_json::from_slice(&raw.0) .map(Some) - .map_err(|err| format!("parse Secret {}: {err}", self.name)) + .map_err(|err| LoadError::Corrupt { + raw: raw.0.clone(), + error: err.to_string(), + }) + } + + /// Keep unparseable state for diagnosis in a new Secret beside this one, + /// before it is overwritten. Returns the copy's name. + pub async fn quarantine(&self, raw: &[u8]) -> Result { + let stamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |elapsed| elapsed.as_secs()); + let name = format!("{}-corrupt-{stamp}", self.name); + let copy = Secret { + metadata: ObjectMeta { + name: Some(name.clone()), + ..Default::default() + }, + data: Some([(STATE_KEY.to_string(), ByteString(raw.to_vec()))].into()), + ..Default::default() + }; + match self + .api + .create(&kube::api::PostParams::default(), ©) + .await + { + Ok(_) => Ok(name), + Err(err) => Err(format!("create Secret {name}: {err}")), + } + } +} + +#[async_trait::async_trait] +impl StateStore for SecretStateStore { + async fn load(&self) -> Result, String> { + self.load_checked().await.map_err(|err| err.to_string()) } async fn save(&self, state: &PersistedState) -> Result<(), String> { diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 7d27687073..e61e6b0b3f 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -979,8 +979,12 @@ baseline (the first values the controller saw, or the first sync after it restarted without `--state-secret`), restarts are off (`--no-restart`), or no sandbox was attached to the provider. Agents restarting right after a provider-sync restart mean values changed while it was down, or restarts it had -in progress being resumed; a missing state Secret restarts nothing. provider-sync -refuses to start on a state Secret it cannot parse rather than overwrite it. +in progress being resumed; a missing state Secret restarts nothing. If the state Secret +cannot be read, provider-sync waits and retries rather than starting over. If it +does not parse, provider-sync copies it to `-corrupt-`, +records a `StateSecretCorrupt` Warning on it, keeps every restart in progress it +can still parse, and starts over; an agent it had stopped whose restart entry +was lost stays stopped and must be started by hand, as the Event says. Restarts in progress are in that Secret (`state.json`, `restarts`). One ends when its sandbox is ready again or is deleted. Starting the sandbox by hand does not cancel it: provider-sync stops and starts it once more, so the agent still