From 5e5ce73fd0e4c738552be83963a6e43071398815 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 12 Aug 2026 00:46:39 +0900 Subject: [PATCH] runtime: harden retention reconciliation and retry --- crates/worker-runtime/src/retention.rs | 450 ++++++++++++++++++++--- crates/worker-runtime/src/runtime.rs | 29 +- crates/workspace-server/src/retention.rs | 217 +++++++++-- 3 files changed, 619 insertions(+), 77 deletions(-) diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index e77a8372..699f660d 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -46,6 +46,69 @@ pub struct WorkerRetentionInventory { pub diagnostics_bytes: u64, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct RuntimeWorkerAggregateDiagnostic { + worker_id: String, + category: String, + detail: String, +} + +impl RuntimeWorkerAggregateDiagnostic { + pub fn worker_id(&self) -> &str { + &self.worker_id + } + + pub fn category(&self) -> &str { + &self.category + } + + pub fn detail(&self) -> &str { + &self.detail + } +} + +/// Opaque host-derived inventory snapshot. Callers can inspect but cannot +/// construct or alter the Runtime/Workspace scope used by reconciliation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct WorkerRetentionInventorySnapshot { + workspace_id: String, + runtime_id: String, + workers: Vec, + diagnostics: Vec, +} + +impl WorkerRetentionInventorySnapshot { + pub(crate) fn new( + workspace_id: String, + runtime_id: String, + workers: Vec, + diagnostics: Vec, + ) -> Self { + Self { + workspace_id, + runtime_id, + workers, + diagnostics, + } + } + + pub fn workspace_id(&self) -> &str { + &self.workspace_id + } + + pub fn runtime_id(&self) -> &str { + &self.runtime_id + } + + pub fn workers(&self) -> &[WorkerRetentionInventory] { + &self.workers + } + + pub fn diagnostics(&self) -> &[RuntimeWorkerAggregateDiagnostic] { + &self.diagnostics + } +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerRetentionExecutionRequest { pub operation_id: String, @@ -113,12 +176,6 @@ pub(crate) trait WorkerRetentionProvider: Send + Sync { &self, request: &WorkerRetentionExecutionRequest, ) -> Result; - - fn completed( - &self, - operation_id: &str, - input_fingerprint: &str, - ) -> Result, RuntimeError>; } /// Filesystem provider for the canonical Runtime Worker aggregate. @@ -134,6 +191,137 @@ impl FsWorkerRetentionProvider { } } + pub(crate) fn completed_for( + &self, + request: &WorkerRetentionExecutionRequest, + ) -> Result, RuntimeError> { + let path = self.operation_path(&request.operation_id)?; + let Some(receipt) = read_operation_receipt(&path)? else { + return Ok(None); + }; + validate_receipt_request(&receipt, request)?; + Ok(receipt.result.source_removed.then_some(receipt.result)) + } + + pub(crate) fn snapshot( + &self, + workspace_id: &str, + runtime_id: &str, + ) -> Result { + fs::create_dir_all(&self.runtime_root).map_err(|source| RuntimeError::StoreIo { + operation: "prepare Worker retention inventory", + path: self.runtime_root.clone(), + source, + })?; + let lock_path = self.runtime_root.join(RETENTION_LOCK); + let lock = OpenOptions::new() + .create(true) + .read(true) + .write(true) + .open(&lock_path) + .map_err(|source| RuntimeError::StoreIo { + operation: "lock Worker retention inventory", + path: lock_path.clone(), + source, + })?; + lock.lock().map_err(|source| RuntimeError::StoreIo { + operation: "lock Worker retention inventory", + path: lock_path, + source, + })?; + let workers_root = self.runtime_root.join("workers"); + if !workers_root.is_dir() { + return Ok(WorkerRetentionInventorySnapshot::new( + workspace_id.to_string(), + runtime_id.to_string(), + Vec::new(), + Vec::new(), + )); + } + let mut entries = fs::read_dir(&workers_root) + .map_err(|source| RuntimeError::StoreIo { + operation: "scan Worker retention inventory", + path: workers_root.clone(), + source, + })? + .collect::, _>>() + .map_err(|source| RuntimeError::StoreIo { + operation: "scan Worker retention inventory", + path: workers_root.clone(), + source, + })?; + entries.sort_by_key(|entry| entry.file_name()); + let mut workers = Vec::new(); + let mut diagnostics = Vec::new(); + for entry in entries { + let raw_id = entry.file_name().to_string_lossy().to_string(); + let bounded_id = bounded_diagnostic_id(&raw_id); + let file_type = match entry.file_type() { + Ok(file_type) => file_type, + Err(_) => { + diagnostics.push(runtime_aggregate_diagnostic( + &bounded_id, + "aggregate_metadata_unreadable", + )); + continue; + } + }; + if !file_type.is_dir() || file_type.is_symlink() { + diagnostics.push(runtime_aggregate_diagnostic( + &bounded_id, + "aggregate_entry_unsupported", + )); + continue; + } + let Ok(worker_number) = raw_id.parse::() else { + diagnostics.push(runtime_aggregate_diagnostic( + &bounded_id, + "aggregate_worker_id_invalid", + )); + continue; + }; + let worker_id = WorkerId::new(worker_number); + let worker_dir = self.worker_dir(worker_id); + let snapshot: WorkerGenerationSnapshot = match read_json( + &worker_dir.join("worker.json"), + "scan Worker retention inventory", + ) { + Ok(snapshot) => snapshot, + Err(_) => { + diagnostics.push(runtime_aggregate_diagnostic( + &bounded_id, + "aggregate_worker_record_corrupt", + )); + continue; + } + }; + if snapshot.workspace_id.as_deref() != Some(workspace_id) { + diagnostics.push(runtime_aggregate_diagnostic( + &bounded_id, + "aggregate_workspace_mismatch", + )); + continue; + } + match self.inventory(workspace_id, runtime_id, worker_id, snapshot.run_generation) { + Ok(item) => workers.push(item), + Err(_) => diagnostics.push(runtime_aggregate_diagnostic( + &bounded_id, + "aggregate_session_inventory_failed", + )), + } + } + workers.sort_by_key(|item| item.worker_id); + diagnostics.sort_by(|left, right| { + (&left.worker_id, &left.category).cmp(&(&right.worker_id, &right.category)) + }); + Ok(WorkerRetentionInventorySnapshot::new( + workspace_id.to_string(), + runtime_id.to_string(), + workers, + diagnostics, + )) + } + pub(crate) fn recover_after_source_removal( &self, request: &WorkerRetentionExecutionRequest, @@ -192,6 +380,19 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { if !worker_dir.is_dir() { return Err(RuntimeError::WorkerNotFound { worker_id }); } + let worker: WorkerGenerationSnapshot = read_json( + &worker_dir.join("worker.json"), + "inventory Worker retention", + )?; + if worker.workspace_id.as_deref() != Some(workspace_id) { + return Err(RuntimeError::WorkerNotFound { worker_id }); + } + if worker.run_generation != run_generation { + return Err(RuntimeError::InvalidRequest(format!( + "Worker retention inventory expected generation {run_generation}, current generation is {}", + worker.run_generation + ))); + } let session_dir = worker_dir.join("session"); let (session_id, segment_ids, session_bytes) = if session_dir.is_dir() { let manifest: CanonicalSessionManifest = read_json( @@ -264,11 +465,9 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { source, })?; - let pending = read_operation_receipt( - &self.operation_path(&request.operation_id)?, - &request.input_fingerprint, - )?; + let pending = read_operation_receipt(&self.operation_path(&request.operation_id)?)?; if let Some(receipt) = &pending { + validate_receipt_request(receipt, request)?; if receipt.result.source_removed { return Ok(receipt.result.clone()); } @@ -293,6 +492,11 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { } let snapshot: WorkerGenerationSnapshot = read_json(&worker_dir.join("worker.json"), "execute Worker retention")?; + if snapshot.workspace_id.as_deref() != Some(request.workspace_id.as_str()) { + return Err(RuntimeError::WorkerNotFound { + worker_id: request.worker_id, + }); + } if snapshot.run_generation != request.expected_run_generation { return Err(RuntimeError::InvalidRequest(format!( "Worker retention plan expected generation {}, current generation is {}", @@ -326,6 +530,7 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { }; let mut receipt = RetentionOperationReceipt { schema_version: OPERATION_SCHEMA_VERSION, + request: request.clone(), result: result.clone(), }; // Pending receipt makes the delete/final-receipt crash window @@ -355,22 +560,12 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { )?; Ok(result) } - - fn completed( - &self, - operation_id: &str, - input_fingerprint: &str, - ) -> Result, RuntimeError> { - let path = self.operation_path(operation_id)?; - let Some(receipt) = read_operation_receipt(&path, input_fingerprint)? else { - return Ok(None); - }; - Ok(receipt.result.source_removed.then_some(receipt.result)) - } } #[derive(Deserialize)] struct WorkerGenerationSnapshot { + #[serde(default)] + workspace_id: Option, #[serde(default)] run_generation: u64, } @@ -383,13 +578,11 @@ struct CanonicalSessionManifest { #[derive(Serialize, Deserialize)] struct RetentionOperationReceipt { schema_version: u32, + request: WorkerRetentionExecutionRequest, result: WorkerRetentionExecutionResult, } -fn read_operation_receipt( - path: &Path, - input_fingerprint: &str, -) -> Result, RuntimeError> { +fn read_operation_receipt(path: &Path) -> Result, RuntimeError> { if !path.is_file() { return Ok(None); } @@ -404,16 +597,26 @@ fn read_operation_receipt( ), }); } - if receipt.result.input_fingerprint != input_fingerprint { - return Err(RuntimeError::InvalidRequest(format!( - "retention operation {} was already used with different input", - receipt.result.operation_id - ))); - } Ok(Some(receipt)) } -#[derive(Serialize, Deserialize)] +fn validate_receipt_request( + receipt: &RetentionOperationReceipt, + request: &WorkerRetentionExecutionRequest, +) -> Result<(), RuntimeError> { + if &receipt.request != request + || receipt.result.operation_id != request.operation_id + || receipt.result.input_fingerprint != request.input_fingerprint + { + return Err(RuntimeError::InvalidRequest(format!( + "retention operation {} was already used with different input", + request.operation_id + ))); + } + Ok(()) +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] struct DiagnosticsArchiveManifest { schema_version: u32, operation_id: String, @@ -426,6 +629,36 @@ struct DiagnosticsArchiveManifest { content_file_count: u64, } +fn runtime_aggregate_diagnostic( + worker_id: &str, + category: &str, +) -> RuntimeWorkerAggregateDiagnostic { + RuntimeWorkerAggregateDiagnostic { + worker_id: worker_id.to_string(), + category: category.to_string(), + detail: "Canonical Runtime Worker aggregate requires diagnostic reconciliation".to_string(), + } +} + +fn bounded_diagnostic_id(value: &str) -> String { + let sanitized = value + .chars() + .take(64) + .map(|character| { + if character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.') { + character + } else { + '_' + } + }) + .collect::(); + if sanitized.is_empty() { + "unknown".to_string() + } else { + sanitized + } +} + fn validate_request(request: &WorkerRetentionExecutionRequest) -> Result<(), RuntimeError> { validate_id("operation_id", &request.operation_id)?; validate_id("workspace_id", &request.workspace_id)?; @@ -611,16 +844,7 @@ fn commit_diagnostics_archive( let files = diagnostics_files(worker_dir, "archive Worker diagnostics")?; let target = provider.diagnostics_dir(&request.operation_id)?; if target.exists() { - let manifest: DiagnosticsArchiveManifest = read_json( - &target.join("manifest.json"), - "verify Worker diagnostics archive", - )?; - if manifest.input_fingerprint != request.input_fingerprint { - return Err(RuntimeError::InvalidRequest(format!( - "diagnostics archive operation {} was reused with different input", - request.operation_id - ))); - } + validate_existing_diagnostics_archive(&target, request)?; return Ok(()); } let parent = target.parent().ok_or_else(|| RuntimeError::StoreCorrupt { @@ -677,7 +901,8 @@ fn commit_diagnostics_archive( path: target.clone(), source, })?; - sync_directory(parent, "archive Worker diagnostics") + sync_directory(parent, "archive Worker diagnostics")?; + validate_existing_diagnostics_archive(&target, request) })(); if result.is_err() { let _ = fs::remove_dir_all(&staging); @@ -685,6 +910,42 @@ fn commit_diagnostics_archive( result } +fn validate_existing_diagnostics_archive( + target: &Path, + request: &WorkerRetentionExecutionRequest, +) -> Result<(), RuntimeError> { + let manifest: DiagnosticsArchiveManifest = read_json( + &target.join("manifest.json"), + "verify Worker diagnostics archive", + )?; + if manifest.schema_version != ARCHIVE_SCHEMA_VERSION + || manifest.operation_id != request.operation_id + || manifest.workspace_id != request.workspace_id + || manifest.source_runtime_id != request.source_runtime_id + || manifest.source_worker_id != request.worker_id + || manifest.input_fingerprint != request.input_fingerprint + { + return Err(RuntimeError::InvalidRequest(format!( + "diagnostics archive operation {} does not match the retention request", + request.operation_id + ))); + } + let mut files = collect_files(target, "verify Worker diagnostics archive")?; + files.retain(|(relative, _)| relative != Path::new("manifest.json")); + let (checksum, bytes, count) = checksum_files(&files, "verify Worker diagnostics archive")?; + if checksum != manifest.content_checksum_sha256 + || bytes != manifest.content_bytes + || count != manifest.content_file_count + { + return Err(RuntimeError::StoreCorrupt { + operation: "verify Worker diagnostics archive", + path: target.to_path_buf(), + message: "diagnostics archive checksum or content summary mismatch".to_string(), + }); + } + Ok(()) +} + fn diagnostics_files( worker_dir: &Path, operation: &'static str, @@ -1001,7 +1262,7 @@ mod tests { let worker = root.join("workers").join(worker_id.to_string()); write_json( &worker.join("worker.json"), - &serde_json::json!({"run_generation": generation}), + &serde_json::json!({"workspace_id": "workspace-a", "run_generation": generation}), ); write_json( &worker.join("session/session.json"), @@ -1102,6 +1363,36 @@ mod tests { ); } + #[test] + fn target_inventory_and_execute_reject_cross_workspace_aggregate() { + let temp = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::new(16); + source(temp.path(), worker_id, 3); + let provider = FsWorkerRetentionProvider::new(temp.path()); + assert!(matches!( + provider.inventory("other-workspace", "runtime-a", worker_id, 3), + Err(RuntimeError::WorkerNotFound { .. }) + )); + let mut request = request(worker_id, 3, SessionDisposition::Purge); + request.workspace_id = "other-workspace".to_string(); + assert!(matches!( + provider.execute(&request), + Err(RuntimeError::WorkerNotFound { .. }) + )); + assert!(temp.path().join("workers/16/session").is_dir()); + assert!( + !temp + .path() + .join("retention/operations/operation-a.json") + .exists() + ); + + request.workspace_id = "workspace-a".to_string(); + provider.execute(&request).unwrap(); + request.workspace_id = "other-workspace".to_string(); + assert!(provider.execute(&request).is_err()); + } + #[test] fn purge_removes_aggregate_and_rejects_stale_generation() { let temp = tempfile::tempdir().unwrap(); @@ -1141,12 +1432,71 @@ mod tests { let recovered = provider.execute(&request).unwrap(); assert_eq!(recovered, completed); - assert!( - provider - .completed("operation-a", "fingerprint-a") - .unwrap() - .is_some() + assert!(provider.completed_for(&request).unwrap().is_some()); + } + + #[test] + fn provider_snapshot_scans_aggregate_storage_independent_of_runtime_catalog() { + let temp = tempfile::tempdir().unwrap(); + source(temp.path(), WorkerId::new(13), 2); + source(temp.path(), WorkerId::new(14), 1); + write_json( + &temp.path().join("workers/14/worker.json"), + &serde_json::json!({"workspace_id": "other-workspace", "run_generation": 1}), ); + fs::create_dir_all(temp.path().join("workers/not-a-worker")).unwrap(); + fs::write( + temp.path().join("workers/not-a-worker/worker.json"), + b"not-json", + ) + .unwrap(); + fs::create_dir_all(temp.path().join("workers/15")).unwrap(); + fs::write(temp.path().join("workers/15/worker.json"), b"not-json").unwrap(); + let provider = FsWorkerRetentionProvider::new(temp.path()); + + let snapshot = provider.snapshot("workspace-a", "runtime-a").unwrap(); + assert_eq!(snapshot.workers().len(), 1); + assert_eq!(snapshot.workers()[0].worker_id, WorkerId::new(13)); + assert!(snapshot.diagnostics().iter().any(|diagnostic| { + diagnostic.worker_id() == "14" + && diagnostic.category() == "aggregate_workspace_mismatch" + })); + assert!(snapshot.diagnostics().iter().any(|diagnostic| { + diagnostic.worker_id() == "not-a-worker" + && diagnostic.category() == "aggregate_worker_id_invalid" + })); + assert!(snapshot.diagnostics().iter().any(|diagnostic| { + diagnostic.worker_id() == "15" + && diagnostic.category() == "aggregate_worker_record_corrupt" + })); + } + + #[test] + fn diagnostics_retry_rejects_corrupt_existing_archive_before_source_delete() { + let temp = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::new(12); + source(temp.path(), worker_id, 1); + let provider = FsWorkerRetentionProvider::new(temp.path()); + let mut request = request(worker_id, 1, SessionDisposition::Archive); + request.diagnostics_disposition = DiagnosticsDisposition::Retain; + provider.execute(&request).unwrap(); + + let receipt_path = temp.path().join("retention/operations/operation-a.json"); + let mut receipt: RetentionOperationReceipt = + serde_json::from_slice(&fs::read(&receipt_path).unwrap()).unwrap(); + receipt.result.source_removed = false; + fs::write(&receipt_path, serde_json::to_vec_pretty(&receipt).unwrap()).unwrap(); + source(temp.path(), worker_id, 1); + fs::write( + temp.path() + .join("archives/diagnostics/operation-a/runs/1/worker.out.log"), + b"corrupt\n", + ) + .unwrap(); + + assert!(provider.execute(&request).is_err()); + assert!(temp.path().join("workers/12/session").is_dir()); + assert!(provider.completed_for(&request).unwrap().is_none()); } #[test] diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index e93a421d..2c9a4961 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -29,7 +29,7 @@ use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; #[cfg(feature = "fs-store")] use crate::retention::{ FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, - WorkerRetentionInventory, WorkerRetentionProvider, + WorkerRetentionInventory, WorkerRetentionInventorySnapshot, WorkerRetentionProvider, }; use protocol::subscription::{ EventSubscriptionSelector, SubscriptionEventPayload, SubscriptionSnapshot, @@ -1706,6 +1706,29 @@ impl Runtime { ) } + /// Enumerate host-authoritative Runtime inventory for Backend orphan + /// reconciliation. Runtime identity and Workspace scope are derived here, + /// not accepted in a diagnostic payload. + #[cfg(feature = "fs-store")] + pub fn list_worker_retention_inventory( + &self, + workspace_id: &str, + ) -> Result { + let state = self.lock()?; + let runtime_id = state.runtime_identity.as_deref().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Runtime identity is not bound for Worker retention".to_string(), + ) + })?; + let store = state.fs_store().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Worker retention inventory requires an fs-backed Runtime".to_string(), + ) + })?; + let provider = FsWorkerRetentionProvider::new(store.runtime_dir()); + provider.snapshot(workspace_id, runtime_id) + } + /// Execute a Backend-resolved retention plan. Only stopped Workers are /// eligible. Provider receipt lookup happens before live lookup so exact /// retries converge after aggregate removal. @@ -1731,9 +1754,7 @@ impl Runtime { ) })?; let provider = FsWorkerRetentionProvider::new(store.runtime_dir()); - if let Some(completed) = - provider.completed(&request.operation_id, &request.input_fingerprint)? - { + if let Some(completed) = provider.completed_for(request)? { state.workers.remove(&request.worker_id); state.persist_runtime_snapshot()?; return Ok(completed); diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index b3b21ff6..561d6e48 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -7,10 +7,12 @@ use chrono::Utc; use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use std::collections::{BTreeMap, BTreeSet}; use worker_runtime::identity::RuntimeWorkerRef; use worker_runtime::retention::{ - DiagnosticsDisposition, SessionDisposition, WorkerRetentionExecutionRequest, - WorkerRetentionExecutionResult, WorkerRetentionInventory, + DiagnosticsDisposition, RuntimeWorkerAggregateDiagnostic, SessionDisposition, + WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, WorkerRetentionInventory, + WorkerRetentionInventorySnapshot, }; pub const CONSERVATIVE_POLICY_ID: &str = "workspace-default-conservative"; @@ -366,7 +368,7 @@ impl SqliteWorkspaceStore { "Runtime Worker id is not a canonical unsigned integer".to_string(), ) })?; - let removed_at = Utc::now().to_rfc3339(); + let removed_at = plan.created_at.clone(); Ok(PreparedWorkerRemoval { runtime_request: WorkerRetentionExecutionRequest { operation_id: plan.operation_id.clone(), @@ -462,14 +464,109 @@ impl SqliteWorkspaceStore { }).map_err(map_error) } - pub fn record_worker_orphan_diagnostic( + /// Compare trusted Runtime inventory with the Backend worker registry and + /// persist bounded diagnostics for both orphan directions. This operation + /// is diagnostic-only: a Runtime aggregate without Backend authority is + /// never assigned an implicit purge disposition. + pub fn reconcile_worker_retention_inventory( &self, - d: &WorkerOrphanDiagnostic, - ) -> Result<(), WorkerRetentionError> { - bounded("orphan category", &d.category, 160)?; - bounded("orphan detail", &d.detail, 2000)?; - self.with_conn(|conn|{conn.execute("INSERT OR IGNORE INTO worker_orphan_diagnostics(diagnostic_id,workspace_id,runtime_id,worker_id,category,detail,observed_at) VALUES(?1,?2,?3,?4,?5,?6,?7)",params![d.diagnostic_id,d.workspace_id,d.runtime_id,d.worker_id,d.category,d.detail,d.observed_at])?;Ok(())})?; - Ok(()) + snapshot: &WorkerRetentionInventorySnapshot, + ) -> Result, WorkerRetentionError> { + self.reconcile_worker_retention_inventory_parts( + snapshot.workspace_id(), + snapshot.runtime_id(), + snapshot.workers(), + snapshot.diagnostics(), + ) + } + + fn reconcile_worker_retention_inventory_parts( + &self, + workspace_id: &str, + runtime_id: &str, + inventory: &[WorkerRetentionInventory], + runtime_diagnostics: &[RuntimeWorkerAggregateDiagnostic], + ) -> Result, WorkerRetentionError> { + bounded("Workspace id", workspace_id, 160)?; + bounded("Runtime id", runtime_id, 160)?; + let mut runtime_workers = BTreeMap::new(); + for item in inventory { + if item.workspace_id != workspace_id || item.runtime_id != runtime_id { + return Err(WorkerRetentionError::CrossWorkspace); + } + let worker_id = item.worker_id.to_string(); + if runtime_workers.insert(worker_id.clone(), item).is_some() { + return Err(WorkerRetentionError::Invalid(format!( + "duplicate Runtime inventory for Worker {worker_id}" + ))); + } + } + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let policy_configured = load_policy(&tx, workspace_id)?.is_some(); + let mut statement = tx.prepare( + "SELECT CAST(runtime_worker_id AS TEXT), retention_state + FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2", + )?; + let registry = statement + .query_map(params![workspace_id, runtime_id], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })? + .collect::, _>>()?; + drop(statement); + let runtime_ids = runtime_workers.keys().cloned().collect::>(); + let registry_ids = registry.keys().cloned().collect::>(); + let observed_at = Utc::now().to_rfc3339(); + let mut diagnostics = runtime_diagnostics + .iter() + .map(|diagnostic| { + orphan_diagnostic( + workspace_id, + runtime_id, + diagnostic.worker_id(), + diagnostic.category(), + diagnostic.detail(), + &observed_at, + ) + }) + .collect::>(); + for worker_id in runtime_ids.difference(®istry_ids) { + let category = if policy_configured { + "runtime_aggregate_without_backend_registry" + } else { + "runtime_aggregate_policy_missing_fail_closed" + }; + diagnostics.push(orphan_diagnostic( + workspace_id, + runtime_id, + worker_id, + category, + "Runtime canonical aggregate is absent from Backend worker_registry; removal is blocked pending reconciliation", + &observed_at, + )); + } + for worker_id in registry_ids.difference(&runtime_ids) { + let category = if registry.get(worker_id).map(String::as_str) == Some("pinned") { + "backend_registry_without_runtime_aggregate_pinned" + } else { + "backend_registry_without_runtime_aggregate" + }; + diagnostics.push(orphan_diagnostic( + workspace_id, + runtime_id, + worker_id, + category, + "Backend worker_registry record has no Runtime canonical aggregate", + &observed_at, + )); + } + for diagnostic in &diagnostics { + insert_orphan_diagnostic(&tx, diagnostic)?; + } + tx.commit()?; + Ok(diagnostics) + }) + .map_err(WorkerRetentionError::Store) } pub fn worker_tombstone( @@ -481,6 +578,49 @@ impl SqliteWorkspaceStore { } } +fn orphan_diagnostic( + workspace_id: &str, + runtime_id: &str, + worker_id: &str, + category: &str, + detail: &str, + observed_at: &str, +) -> WorkerOrphanDiagnostic { + let identity = format!("{workspace_id}\0{runtime_id}\0{worker_id}\0{category}"); + WorkerOrphanDiagnostic { + diagnostic_id: stable("wod", &identity), + workspace_id: workspace_id.to_string(), + runtime_id: runtime_id.to_string(), + worker_id: worker_id.to_string(), + category: category.to_string(), + detail: detail.to_string(), + observed_at: observed_at.to_string(), + } +} + +fn insert_orphan_diagnostic( + conn: &Connection, + diagnostic: &WorkerOrphanDiagnostic, +) -> crate::Result<()> { + conn.execute( + "INSERT INTO worker_orphan_diagnostics + (diagnostic_id,workspace_id,runtime_id,worker_id,category,detail,observed_at) + VALUES(?1,?2,?3,?4,?5,?6,?7) + ON CONFLICT(diagnostic_id) DO UPDATE SET + detail=excluded.detail, observed_at=excluded.observed_at", + params![ + diagnostic.diagnostic_id, + diagnostic.workspace_id, + diagnostic.runtime_id, + diagnostic.worker_id, + diagnostic.category, + diagnostic.detail, + diagnostic.observed_at + ], + )?; + Ok(()) +} + #[derive(Clone)] struct WorkerRow { display_name: String, @@ -948,6 +1088,10 @@ mod tests { ); assert_eq!(prepared.runtime_request.policy_revision, 1); assert_eq!(prepared.runtime_request.worker_id, WorkerId::new(1)); + let retry = s + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + assert_eq!(retry.runtime_request, prepared.runtime_request); } #[test] @@ -1018,27 +1162,54 @@ mod tests { assert!( matches!(&p.blockers[..],[WorkerRemovalBlocker::CurrentAssignment{assignment_id,ticket_id}] if assignment_id=="assignment"&&ticket_id=="ticket") ); - let d = WorkerOrphanDiagnostic { - diagnostic_id: "orphan".into(), + let runtime_only = WorkerRetentionInventory { workspace_id: "w".into(), runtime_id: "r".into(), - worker_id: "missing".into(), - category: "runtime_without_catalog".into(), - detail: "bounded diagnostic".into(), - observed_at: "t".into(), + worker_id: WorkerId::new(2), + run_generation: 1, + session_id: Some("orphan-session".into()), + segment_ids: vec![], + session_bytes: 10, + diagnostics_bytes: 0, }; - s.record_worker_orphan_diagnostic(&d).unwrap(); - let n: i64 = s - .with_conn(|c| { - c.query_row( - "SELECT COUNT(*) FROM worker_orphan_diagnostics WHERE diagnostic_id='orphan'", + let diagnostics = s + .reconcile_worker_retention_inventory_parts("w", "r", &[runtime_only], &[]) + .unwrap(); + assert_eq!(diagnostics.len(), 2); + assert!(diagnostics.iter().any(|item| { + item.worker_id == "2" && item.category == "runtime_aggregate_without_backend_registry" + })); + assert!(diagnostics.iter().any(|item| { + item.worker_id == "1" && item.category == "backend_registry_without_runtime_aggregate" + })); + let count: i64 = s + .with_conn(|conn| { + conn.query_row( + "SELECT COUNT(*) FROM worker_orphan_diagnostics WHERE workspace_id='w' AND runtime_id='r'", [], - |r| r.get(0), + |row| row.get(0), ) .map_err(StoreError::from) }) .unwrap(); - assert_eq!(n, 1); + assert_eq!(count, 2); + let mut wrong_scope = inv(); + wrong_scope.workspace_id = "other".into(); + assert!(matches!( + s.reconcile_worker_retention_inventory_parts("w", "r", &[wrong_scope], &[]), + Err(WorkerRetentionError::CrossWorkspace) + )); + let unchanged: i64 = s + .with_conn(|conn| { + conn.query_row( + "SELECT COUNT(*) FROM worker_orphan_diagnostics WHERE workspace_id='w' AND runtime_id='r'", + [], + |row| row.get(0), + ) + .map_err(StoreError::from) + }) + .unwrap(); + assert_eq!(unchanged, 2); } #[test] fn concurrent_plan_converges_and_purge_omits_tombstone() {