diff --git a/app/src-tauri/src/aws/environments.rs b/app/src-tauri/src/aws/environments.rs index 9521132..260e73a 100644 --- a/app/src-tauri/src/aws/environments.rs +++ b/app/src-tauri/src/aws/environments.rs @@ -41,7 +41,7 @@ const MAX_UNTRUSTED_BYTES: usize = 4096; const MAX_LOCK_STRING_CHARS: usize = 128; /// A lock younger than this is a deploy in progress somewhere else; an older /// one is an apply that died holding it. -const LIVE_LOCK_SECS: u64 = 2 * 3600; +pub const LIVE_LOCK_SECS: u64 = 2 * 3600; const RECORD_SCHEMA: u64 = 1; /// Bounds on what one scan lists, so a bucket or parameter path full of /// junk can't make it unbounded. diff --git a/app/src-tauri/src/commands/environment_commands.rs b/app/src-tauri/src/commands/environment_commands.rs index fd8b9eb..5833413 100644 --- a/app/src-tauri/src/commands/environment_commands.rs +++ b/app/src-tauri/src/commands/environment_commands.rs @@ -8,9 +8,12 @@ //! offered. //! //! Applies are detached (D10). Each started or followed run gets one task -//! here that follows its log and emits [`PROGRESS_EVENT`] batches, then a -//! final `done` with the result. A deployment is followed by at most one task -//! per process ([`FOLLOWERS`]). +//! here that follows its log from byte 0 and emits [`PROGRESS_EVENT`] +//! batches, then a final `done` with the result. The task emits nothing +//! until the command is about to return its op id ([`Gate`]), so a listener +//! registered before the invoke sees everything. Per work folder this +//! process keeps one [`Slot`]: a launch in progress (which refuses a second +//! launch or a follow) or the one follower (a re-follow replaces it). //! //! Credentials: a keys connection's keys are read from the keychain here //! and go only into [`tofu::child_env`]; they are never logged, written or @@ -31,8 +34,11 @@ use serde::Serialize; use serde_json::Value; use tauri::{AppHandle, Emitter, State}; +use aws_sdk_s3::error::ProvideErrorMetadata; +use tokio::sync::oneshot; + use crate::aws::environments::{ - ensure_state_bucket, scan, DetectedEnvironment, EnvironmentAction, EnvironmentInfo, + ensure_state_bucket, scan, LIVE_LOCK_SECS, DetectedEnvironment, EnvironmentAction, EnvironmentInfo, EnvironmentScan, EnvironmentState, Knowable, OnboardingState, SdkDetectApi, }; use crate::aws::tofu::{ @@ -55,7 +61,7 @@ const FOLLOW_POLL: Duration = Duration::from_millis(500); /// whole log at once). const MAX_EVENTS_PER_BATCH: usize = 200; /// A state object read back for its outputs larger than this is refused. -const MAX_STATE_BYTES: usize = 32 * 1024 * 1024; +const MAX_STATE_BYTES: usize = 1024 * 1024; fn other(message: impl Into) -> AwsError { AwsError::Other { @@ -275,6 +281,81 @@ pub(crate) fn start_error(e: StartError, label: &str) -> AwsError { } } +/// A held state lock on the environment, if detection saw one. +fn held_lock(env: &DetectedEnvironment) -> bool { + match &env.state { + EnvironmentState::Deploying { .. } => true, + EnvironmentState::Incomplete { lock } => lock.is_some(), + _ => false, + } +} + +/// Update and Repair: the action is offered and no lock is held. A re-apply +/// under a held lock would only fail inside the detached run. +pub(crate) fn reapply_refusal<'a>( + scan: &'a EnvironmentScan, + name: &str, + action: EnvironmentAction, +) -> Result<&'a DetectedEnvironment, String> { + let env = require_action(scan, name, action)?; + if held_lock(env) { + return Err(format!( + "Environment \"{name}\"'s state is locked by an interrupted run; Unlock it first" + )); + } + Ok(env) +} + +/// Destroy from the app (D6, R12): offered by detection, no valid record +/// (a working environment is an admin's `tofu destroy`), and no held lock. +pub(crate) fn destroy_refusal<'a>( + scan: &'a EnvironmentScan, + name: &str, +) -> Result<&'a DetectedEnvironment, String> { + let env = require_action(scan, name, EnvironmentAction::Destroy)?; + if env.info.is_some() { + return Err(format!( + "Environment \"{name}\" has a valid record, so it may be working; it can't be \ + destroyed from the app (an admin destroys it from a bundle)" + )); + } + if held_lock(env) { + return Err(format!( + "Environment \"{name}\"'s state is locked by an interrupted run; Unlock it first" + )); + } + Ok(env) +} + +/// The lock object about to be deleted must still be stale, by its own +/// `LastModified` read just now. +pub(crate) fn stale_lock_refusal(age_secs: u64) -> Result<(), String> { + if age_secs < LIVE_LOCK_SECS { + Err("The state lock is recent, so a deploy may be running; detect again".into()) + } else { + Ok(()) + } +} + +/// S3's answer to a conditional delete whose ETag no longer matches. +pub(crate) fn is_precondition_failed(code: Option<&str>, status: Option) -> bool { + code == Some("PreconditionFailed") || status == Some(412) +} + +/// The boundary policy is account-wide; its state is per region. A boundary +/// that exists without state in this region's bucket was deployed from +/// another region or by hand, and an apply here would fail on the name. +/// `has_state` is `None` when the caller can't tell. +pub(crate) fn boundary_refusal(boundary: &Knowable, has_state: Option) -> Result<(), String> { + if *boundary == Knowable::Yes && has_state == Some(false) { + Err("The permissions boundary already exists in this account (deployed from another \ + region or by hand), so it can't be deployed from here" + .into()) + } else { + Ok(()) + } +} + fn module_kind(d: Deployment) -> ModuleKind { match d { Deployment::Environment(_) => ModuleKind::Environment, @@ -437,19 +518,113 @@ pub struct DeployProgress { pub result: Option, } +/// Holds a follower's first emit back until the command that started it is +/// about to return its op id, so no event (a finished run's whole replay +/// and its `done` included) can arrive before the caller knows the op id. +/// Opening it, or dropping it, releases the follower. +pub(crate) struct Gate(oneshot::Sender<()>); + +impl Gate { + pub(crate) fn new() -> (Gate, oneshot::Receiver<()>) { + let (tx, rx) = oneshot::channel(); + (Gate(tx), rx) + } + + /// Call immediately before the command's final `Ok(op_id)`. + pub(crate) fn open(self) { + let _ = self.0.send(()); + } +} + +/// A follower: wait for the gate, follow the log from byte 0 emitting +/// batches of at most [`MAX_EVENTS_PER_BATCH`], then emit `done` with the +/// result (a follow error is a failed result). +pub(crate) async fn follow_and_report( + gate: oneshot::Receiver<()>, + wd: Workdir, + base: DeployProgress, + poll: Duration, + mut emit: impl FnMut(&DeployProgress), +) { + // A dropped gate means the command returned without opening it. + let _ = gate.await; + let outcome = tofu::follow(&wd, poll, |events| { + for chunk in events.chunks(MAX_EVENTS_PER_BATCH) { + emit(&DeployProgress { + events: chunk.to_vec(), + ..base.clone() + }); + } + }) + .await + .unwrap_or_else(|e| TofuOutcome { + ok: false, + exit_code: -1, + summary: Some(e.to_string()), + }); + emit(&DeployProgress { + done: true, + cancellable: false, + result: Some(outcome), + ..base + }); +} + // --------------------------------------------------------------------------- // Plumbing -/// Work folders being followed by this process, with their op id. -static FOLLOWERS: OnceLock>> = OnceLock::new(); +/// What this process is doing with a work folder. +enum Slot { + /// A launch is between its checks and its start; nothing else may + /// materialize, start or follow the folder. + Launching, + /// One task follows the folder's run under `op_id`. + Following { + op_id: String, + task: tauri::async_runtime::JoinHandle<()>, + }, +} -fn followers() -> std::sync::MutexGuard<'static, HashMap> { - FOLLOWERS +static SLOTS: OnceLock>> = OnceLock::new(); + +fn slots() -> std::sync::MutexGuard<'static, HashMap> { + SLOTS .get_or_init(Default::default) .lock() .unwrap_or_else(|p| p.into_inner()) } +/// The folder's `Launching` slot, released on every exit path unless a +/// follower has replaced it. +struct LaunchGuard(PathBuf); + +impl LaunchGuard { + fn take(dir: &Path, label: &str) -> Result { + let mut map = slots(); + match map.get(dir) { + Some(Slot::Launching) => Err(other(format!( + "A deployment is starting for {label}; wait for it" + ))), + Some(Slot::Following { .. }) => Err(other(format!( + "The run for {label} is still being followed; wait for it to finish" + ))), + None => { + map.insert(dir.to_path_buf(), Slot::Launching); + Ok(LaunchGuard(dir.to_path_buf())) + } + } + } +} + +impl Drop for LaunchGuard { + fn drop(&mut self) { + let mut map = slots(); + if matches!(map.get(&self.0), Some(Slot::Launching)) { + map.remove(&self.0); + } + } +} + /// `/triple-c/aws/`: OpenTofu, its plugin cache and the work /// folders. fn work_root() -> Result { @@ -523,21 +698,49 @@ async fn recorded_vpc_id(ctx: &Ctx, env: &str) -> Result, AwsErro .send() .await .map_err(|e| crate::aws::classify_sdk_error("s3:GetObject", &e))?; + let too_large = || other("The environment's state is implausibly large"); if obj .content_length() .is_some_and(|n| n < 0 || n as usize > MAX_STATE_BYTES) { - return Err(other("The environment's state is implausibly large")); + return Err(too_large()); } - let bytes = obj - .body - .collect() + // Capped while reading too: the length header may be absent. + let mut body = obj.body; + let mut bytes = Vec::new(); + while let Some(chunk) = body + .try_next() .await .map_err(|e| other(format!("Could not read the environment's state: {e}")))? - .into_bytes(); + { + if bytes.len() + chunk.len() > MAX_STATE_BYTES { + return Err(too_large()); + } + bytes.extend_from_slice(&chunk); + } Ok(vpc_id_from_state(&bytes)) } +/// Whether this region's bucket holds the boundary's state; `None` when +/// the caller may not look. +async fn boundary_state_exists(ctx: &Ctx) -> Result, AwsError> { + let res = aws_sdk_s3::Client::new(&ctx.cfg) + .head_object() + .bucket(ctx.bucket()) + .key(state_key(Deployment::Boundary)) + .expected_bucket_owner(&ctx.who.account) + .send() + .await; + match res { + Ok(_) => Ok(Some(true)), + Err(e) => match e.raw_response().map(|r| r.status().as_u16()) { + Some(404) => Ok(Some(false)), + Some(403) => Ok(None), + _ => Err(crate::aws::classify_sdk_error("s3:GetObject", &e)), + }, + } +} + /// The network to re-apply an environment with: the recorded one. async fn recorded_network(ctx: &Ctx, info: &EnvironmentInfo) -> Result { let vpc = if info.network_mode == "existing" { @@ -558,79 +761,65 @@ fn emit_progress(app: &AppHandle, p: &DeployProgress) { } } -/// Follow `wd` in a task of its own, unless this process already does; the -/// op id of whichever task follows it. -fn spawn_follower(app: AppHandle, wd: Workdir, key: String, can_cancel: bool) -> String { - let op_id = { - let mut map = followers(); - if let Some(existing) = map.get(&wd.dir) { - return existing.clone(); - } - let op_id = uuid::Uuid::new_v4().to_string(); - map.insert(wd.dir.clone(), op_id.clone()); - op_id +/// Start the folder's follower under a new op id, replacing (aborting) any +/// follower this process already has for it, so a re-follow (a reloaded +/// window) replays the whole log. The follower emits nothing until the +/// returned [`Gate`] is opened. The caller has checked the folder isn't +/// launching, or holds its [`LaunchGuard`]. +fn install_follower(app: AppHandle, wd: Workdir, key: String, can_cancel: bool) -> (String, Gate) { + let op_id = uuid::Uuid::new_v4().to_string(); + let (gate, rx) = Gate::new(); + let base = DeployProgress { + op_id: op_id.clone(), + state_key: key, + events: Vec::new(), + done: false, + cancellable: can_cancel, + result: None, }; + let dir = wd.dir.clone(); + let mut map = slots(); + // Spawned under the lock, so the task's own clean-up (which takes it) + // can't run before its slot is in place. let id = op_id.clone(); - tauri::async_runtime::spawn(async move { - let base = DeployProgress { - op_id: id, - state_key: key, - events: Vec::new(), - done: false, - cancellable: can_cancel, - result: None, - }; - let outcome = tofu::follow(&wd, FOLLOW_POLL, |events| { - for chunk in events.chunks(MAX_EVENTS_PER_BATCH) { - emit_progress( - &app, - &DeployProgress { - events: chunk.to_vec(), - ..base.clone() - }, - ); - } - }) - .await - .unwrap_or_else(|e| TofuOutcome { - ok: false, - exit_code: -1, - summary: Some(e.to_string()), - }); - followers().remove(&wd.dir); - emit_progress( - &app, - &DeployProgress { - done: true, - cancellable: false, - result: Some(outcome), - ..base - }, - ); + let task = tauri::async_runtime::spawn(async move { + let dir = wd.dir.clone(); + follow_and_report(rx, wd, base, FOLLOW_POLL, |p| emit_progress(&app, p)).await; + let mut map = slots(); + if matches!(map.get(&dir), Some(Slot::Following { op_id, .. }) if *op_id == id) { + map.remove(&dir); + } }); - op_id + if let Some(Slot::Following { task: old, .. }) = map.insert( + dir, + Slot::Following { + op_id: op_id.clone(), + task, + }, + ) { + old.abort(); + } + (op_id, gate) } /// Ensure OpenTofu and the state bucket, write the work folder and start a -/// detached `init` + `op`; returns the op id its progress is emitted under. +/// detached `init` + `op`. Returns the op id its progress is emitted under +/// and the gate the command opens just before returning it. async fn launch( app: AppHandle, ctx: &Ctx, key: &DeploymentKey, vars: &Value, op: TofuOp, -) -> Result { +) -> Result<(String, Gate), AwsError> { let d = key.deployment(); let label = key.label(); let wd = ctx.workdir(d)?; if tofu::is_running(&wd) { return Err(start_error(StartError::AlreadyRunning, &label)); } - if followers().contains_key(&wd.dir) { - return Err(other(format!( - "The previous run for {label} is still finishing; try again in a moment" - ))); - } + // Held until a follower replaces it, or released on any error below. + let _launching = LaunchGuard::take(&wd.dir, &label)?; let tofu_bin = tofu::ensure_tofu(&ctx.root).await?; // StateBucketTaken is passed on as it is. let bucket = ensure_state_bucket(&ctx.cfg, &ctx.who.account, &ctx.region).await?; @@ -646,7 +835,6 @@ async fn launch( AwsAuth::Profile { .. } => None, }; let env = tofu::child_env(&ctx.conn, keys.as_ref(), &ctx.root, std::env::vars())?; - drop(keys); let start_wd = wd.clone(); let started = tokio::task::spawn_blocking(move || { @@ -656,7 +844,15 @@ async fn launch( .map_err(|e| other(format!("The deploy could not be started: {e}")))? .map_err(|e| start_error(e, &label))?; let can_cancel = cfg!(unix) && started.pid.is_some(); - Ok(spawn_follower(app, wd, state_key(d), can_cancel)) + Ok(install_follower(app, wd, state_key(d), can_cancel)) +} + +/// Open a launch's gate and hand back its op id: the last thing a command +/// does, so its events follow the op id to the caller. +fn opened(launched: Result<(String, Gate), AwsError>) -> Result { + let (op_id, gate) = launched?; + gate.open(); + Ok(op_id) } // --------------------------------------------------------------------------- @@ -690,7 +886,7 @@ pub async fn deploy_aws_environment( deploy_refusal(&found, &name).map_err(other)?; let vars = environment_vars(&name, &ctx.region, &network).map_err(other)?; let key = DeploymentKey::Environment(name); - launch(app, &ctx, &key, &vars, TofuOp::Apply).await + opened(launch(app, &ctx, &key, &vars, TofuOp::Apply).await) } /// Re-apply an environment at the app's module version, with the network @@ -705,7 +901,7 @@ pub async fn update_aws_environment( ) -> Result { let ctx = context(&state, &connection_id, ®ion).await?; let found = detect(&ctx).await?; - let env = require_action(&found, &name, EnvironmentAction::Update).map_err(other)?; + let env = reapply_refusal(&found, &name, EnvironmentAction::Update).map_err(other)?; let info = env .info .as_ref() @@ -713,7 +909,7 @@ pub async fn update_aws_environment( let network = recorded_network(&ctx, info).await?; let vars = environment_vars(&env.name, &ctx.region, &network).map_err(other)?; let key = DeploymentKey::Environment(env.name.clone()); - launch(app, &ctx, &key, &vars, TofuOp::Apply).await + opened(launch(app, &ctx, &key, &vars, TofuOp::Apply).await) } /// Re-apply an Incomplete environment. With a valid record its network is @@ -730,7 +926,7 @@ pub async fn repair_aws_environment( ) -> Result { let ctx = context(&state, &connection_id, ®ion).await?; let found = detect(&ctx).await?; - let env = require_action(&found, &name, EnvironmentAction::Repair).map_err(other)?; + let env = reapply_refusal(&found, &name, EnvironmentAction::Repair).map_err(other)?; let network = match (&env.info, network) { (Some(info), _) => recorded_network(&ctx, info).await?, (None, Some(n)) => n, @@ -742,12 +938,12 @@ pub async fn repair_aws_environment( }; let vars = environment_vars(&env.name, &ctx.region, &network).map_err(other)?; let key = DeploymentKey::Environment(env.name.clone()); - launch(app, &ctx, &key, &vars, TofuOp::Apply).await + opened(launch(app, &ctx, &key, &vars, TofuOp::Apply).await) } /// Destroy an incomplete deployment: only when detection offers Destroy, -/// i.e. there is no valid record (D6). A working environment is destroyed -/// by an admin from a bundle. +/// there is no valid record (D6) and no lock is held ([`destroy_refusal`]). +/// A working environment is destroyed by an admin from a bundle. #[tauri::command] pub async fn destroy_incomplete_aws_environment( connection_id: String, @@ -758,20 +954,19 @@ pub async fn destroy_incomplete_aws_environment( ) -> Result { let ctx = context(&state, &connection_id, ®ion).await?; let found = detect(&ctx).await?; - let env = require_action(&found, &name, EnvironmentAction::Destroy).map_err(other)?; - if env.info.is_some() { - return Err(other("A working environment can't be destroyed from the app")); - } + let env = destroy_refusal(&found, &name).map_err(other)?; // A destroy plans the removal of everything in state whatever the // network variables say; they only have to be valid. let vars = environment_vars(&env.name, &ctx.region, &NetworkChoice::Public).map_err(other)?; let key = DeploymentKey::Environment(env.name.clone()); - launch(app, &ctx, &key, &vars, TofuOp::Destroy).await + opened(launch(app, &ctx, &key, &vars, TofuOp::Destroy).await) } /// Delete an environment's stale `.tflock`. Only an environment's lock, -/// only when detection offers Unlock, the lock is stale and no runner on -/// this machine holds the deployment. +/// only when detection offers Unlock and no runner on this machine holds +/// the deployment. The delete is conditional on the ETag of the lock object +/// as just re-read (and re-checked stale), so a lock taken since is never +/// removed; there is no unconditional fallback. #[tauri::command] pub async fn unlock_aws_state( connection_id: String, @@ -790,21 +985,60 @@ pub async fn unlock_aws_state( return Err(other("A deploy on this machine holds the lock")); } let lock_key = format!("{state_key}.tflock"); - aws_sdk_s3::Client::new(&ctx.cfg) - .delete_object() + let s3 = aws_sdk_s3::Client::new(&ctx.cfg); + let head = match s3 + .head_object() .bucket(ctx.bucket()) .key(&lock_key) .expected_bucket_owner(&ctx.who.account) .send() .await - .map_err(|e| crate::aws::classify_sdk_error("s3:DeleteObject", &e))?; + { + Ok(h) => h, + Err(e) if e.raw_response().map(|r| r.status().as_u16()) == Some(404) => { + return Err(other("The state lock is already gone; detect again")); + } + Err(e) => return Err(crate::aws::classify_sdk_error("s3:GetObject", &e)), + }; + let etag = head + .e_tag() + .filter(|t| !t.is_empty()) + .ok_or_else(|| other("The state lock has no ETag, so it can't be removed safely"))? + .to_string(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + let age = head + .last_modified() + .map(|t| now.saturating_sub(t.secs()).max(0) as u64) + .ok_or_else(|| other("The state lock's age is unknown, so it can't be removed safely"))?; + stale_lock_refusal(age).map_err(other)?; + if let Err(e) = s3 + .delete_object() + .bucket(ctx.bucket()) + .key(&lock_key) + .if_match(&etag) + .expected_bucket_owner(&ctx.who.account) + .send() + .await + { + let status = e.raw_response().map(|r| r.status().as_u16()); + if is_precondition_failed(e.code(), status) { + return Err(other( + "The state lock changed since it was checked; detect again", + )); + } + return Err(crate::aws::classify_sdk_error("s3:DeleteObject", &e)); + } log::info!("Removed the stale state lock {lock_key}"); Ok(()) } /// Reattach to a run on this machine (in progress, or finished with its -/// log still there): replays its log as progress. Returns the op id; a run -/// already followed keeps its follower and op id. +/// log still there): replays its whole log as progress under a new op id. +/// A follower this process already had for it is replaced, so a reloaded +/// window gets the full history. Refused while a launch is starting there. #[tauri::command] pub async fn follow_aws_deployment( connection_id: String, @@ -822,8 +1056,19 @@ pub async fn follow_aws_deployment( key.label() ))); } + if matches!(slots().get(&wd.dir), Some(Slot::Launching)) { + return Err(other(format!( + "A deployment is starting for {}; wait for it", + key.label() + ))); + } let can_cancel = tofu::is_running(&wd) && cancellable(&wd); - Ok(spawn_follower(app, wd, state_key, can_cancel)) + // A launch can't start here between the check and the install: it + // needs a runner-free folder or an empty slot, and either way the + // follower then follows that run from byte 0. + let (op_id, gate) = install_follower(app, wd, state_key, can_cancel); + gate.open(); + Ok(op_id) } /// Cancel: interrupt the runner, which stops OpenTofu gracefully. @@ -859,10 +1104,11 @@ pub async fn onboard_aws_user( owner_key: owner_key(&owner), env, }; - launch(app, &ctx, &key, &vars, TofuOp::Apply).await + opened(launch(app, &ctx, &key, &vars, TofuOp::Apply).await) } -/// Deploy (or re-apply) the account's permissions boundary. +/// Deploy (or re-apply) the account's permissions boundary. Refused when +/// the boundary exists but this region's bucket has no state for it. #[tauri::command] pub async fn deploy_aws_permissions_boundary( connection_id: String, @@ -871,8 +1117,15 @@ pub async fn deploy_aws_permissions_boundary( state: State<'_, AppState>, ) -> Result { let ctx = context(&state, &connection_id, ®ion).await?; + let found = detect(&ctx).await?; + let has_state = match found.state_bucket { + Knowable::Yes => boundary_state_exists(&ctx).await?, + Knowable::No => Some(false), + Knowable::Unknown => None, + }; + boundary_refusal(&found.boundary, has_state).map_err(other)?; let vars = boundary_vars(&ctx.region); - launch(app, &ctx, &DeploymentKey::Boundary, &vars, TofuOp::Apply).await + opened(launch(app, &ctx, &DeploymentKey::Boundary, &vars, TofuOp::Apply).await) } /// Ask for a folder from Rust (see `file_commands::pick_save_path`). diff --git a/app/src-tauri/src/commands/environment_commands/tests.rs b/app/src-tauri/src/commands/environment_commands/tests.rs index 83513f9..031e464 100644 --- a/app/src-tauri/src/commands/environment_commands/tests.rs +++ b/app/src-tauri/src/commands/environment_commands/tests.rs @@ -1,8 +1,10 @@ use std::fs; +use std::time::Duration; use serde_json::json; use super::*; +use crate::aws::tofu::Workdir; use crate::aws::environments::{ EnvironmentState, Knowable, LockInfo, OnboardingState, }; @@ -357,3 +359,142 @@ fn an_already_running_start_names_the_deployment_and_suggests_follow() { let inner = AwsError::KeysMissing; assert_eq!(start_error(StartError::Failed(inner.clone()), "x"), inner); } + +#[tokio::test] +async fn a_follower_emits_nothing_until_its_gate_opens() { + use std::sync::{Arc, Mutex}; + // A finished run: the whole replay and `done` are available at once, + // the case where events could otherwise beat the op id to the caller. + let dir = tempfile::tempdir().unwrap(); + let lines: String = (0..450).map(|i| format!("line {i}\n")).collect(); + fs::write(dir.path().join("apply.log"), lines).unwrap(); + fs::write(dir.path().join("apply.exit"), "0\n").unwrap(); + let wd = Workdir { dir: dir.path().to_path_buf() }; + let base = DeployProgress { + op_id: "op-1".into(), + state_key: "boundary/terraform.tfstate".into(), + events: vec![], + done: false, + cancellable: true, + result: None, + }; + let seen: Arc>> = Arc::default(); + let sink = seen.clone(); + let (gate, rx) = Gate::new(); + let task = tokio::spawn(follow_and_report( + rx, + wd, + base, + Duration::from_millis(10), + move |p| sink.lock().unwrap().push(p.clone()), + )); + tokio::time::sleep(Duration::from_millis(150)).await; + assert!(seen.lock().unwrap().is_empty(), "emitted before the gate opened"); + + gate.open(); + task.await.unwrap(); + let seen = seen.lock().unwrap(); + let batches: Vec = seen.iter().filter(|p| !p.done).map(|p| p.events.len()).collect(); + assert_eq!(batches, vec![200, 200, 50], "the whole log, in capped batches"); + assert!(seen.iter().all(|p| p.op_id == "op-1" && p.state_key == "boundary/terraform.tfstate")); + let last = seen.last().unwrap(); + assert!(last.done && !last.cancellable); + assert_eq!(last.result, Some(TofuOutcome { ok: true, exit_code: 0, summary: None })); + assert_eq!(seen.iter().filter(|p| p.done).count(), 1); +} + +#[tokio::test] +async fn a_dropped_gate_still_releases_the_follower() { + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("apply.log"), "x\n").unwrap(); + fs::write(dir.path().join("apply.exit"), "1\n").unwrap(); + let (gate, rx) = Gate::new(); + drop(gate); + let mut n = 0; + follow_and_report( + rx, + Workdir { dir: dir.path().to_path_buf() }, + DeployProgress { + op_id: "o".into(), + state_key: "k".into(), + events: vec![], + done: false, + cancellable: false, + result: None, + }, + Duration::from_millis(10), + |_| n += 1, + ) + .await; + assert_eq!(n, 2); +} + +#[test] +fn destroy_needs_the_action_no_record_and_no_lock() { + use EnvironmentAction::*; + let mut stale_with_record = env( + "rec", + EnvironmentState::Incomplete { lock: Some(lock(9000, false)) }, + vec![Repair, Destroy, Unlock], + ); + // Stale lock + record + no state: detection offers Destroy; the guard + // must still refuse (the record says it may be working). + stale_with_record.info = Some(info("rec", "public", true)); + let mut record_unlocked = env("rec2", EnvironmentState::Incomplete { lock: None }, vec![Destroy]); + record_unlocked.info = Some(info("rec2", "public", true)); + let s = scan( + Knowable::Yes, + vec![ + stale_with_record, + record_unlocked, + env("locked", EnvironmentState::Incomplete { lock: Some(lock(9000, false)) }, vec![Repair, Destroy, Unlock]), + env("half", EnvironmentState::Incomplete { lock: None }, vec![Repair, Destroy]), + env("ready", EnvironmentState::Ready, vec![Join]), + ], + ); + let e = destroy_refusal(&s, "rec").unwrap_err(); + assert!(e.contains("valid record"), "{e}"); + assert!(destroy_refusal(&s, "rec2").unwrap_err().contains("valid record")); + assert!(destroy_refusal(&s, "locked").unwrap_err().contains("Unlock it first")); + assert!(destroy_refusal(&s, "ready").is_err()); + assert!(destroy_refusal(&s, "missing").is_err()); + assert_eq!(destroy_refusal(&s, "half").unwrap().name, "half"); +} + +#[test] +fn repair_and_update_are_refused_under_a_held_lock() { + use EnvironmentAction::*; + let s = scan( + Knowable::Yes, + vec![ + env("locked", EnvironmentState::Incomplete { lock: Some(lock(9000, false)) }, vec![Repair, Unlock]), + env("free", EnvironmentState::Incomplete { lock: None }, vec![Repair]), + env("old", EnvironmentState::UpdateAvailable, vec![Join, Update]), + ], + ); + let e = reapply_refusal(&s, "locked", Repair).unwrap_err(); + assert!(e.contains("Unlock it first"), "{e}"); + assert!(reapply_refusal(&s, "free", Repair).is_ok()); + assert!(reapply_refusal(&s, "old", Update).is_ok()); + assert!(reapply_refusal(&s, "free", Update).is_err()); +} + +#[test] +fn unlock_deletes_only_a_still_stale_lock_and_reads_412_as_changed() { + assert!(stale_lock_refusal(0).is_err()); + assert!(stale_lock_refusal(LIVE_LOCK_SECS - 1).is_err()); + assert!(stale_lock_refusal(LIVE_LOCK_SECS).is_ok()); + assert!(is_precondition_failed(Some("PreconditionFailed"), Some(412))); + assert!(is_precondition_failed(None, Some(412))); + assert!(!is_precondition_failed(Some("AccessDenied"), Some(403))); + assert!(!is_precondition_failed(None, Some(404))); +} + +#[test] +fn an_existing_boundary_without_state_here_is_refused() { + assert!(boundary_refusal(&Knowable::Yes, Some(false)).unwrap_err().contains("already exists")); + assert!(boundary_refusal(&Knowable::Yes, Some(true)).is_ok(), "re-apply our own"); + assert!(boundary_refusal(&Knowable::Yes, None).is_ok(), "can't tell"); + assert!(boundary_refusal(&Knowable::No, Some(false)).is_ok()); + assert!(boundary_refusal(&Knowable::Unknown, Some(false)).is_ok()); +} diff --git a/app/src/lib/tauri-commands.ts b/app/src/lib/tauri-commands.ts index 1918d22..3d7dbf7 100644 --- a/app/src/lib/tauri-commands.ts +++ b/app/src/lib/tauri-commands.ts @@ -80,8 +80,12 @@ export const checkAwsPermissions = (id: string, region?: string) => export const awsPolicyDocument = (tier: Tier) => invoke("aws_policy_document", { tier }); -// AWS environments (OpenTofu). Reject with an `AwsError`. Starting commands resolve to an op id; -// progress arrives on the `AWS_DEPLOY_PROGRESS_EVENT` event (`DeployProgress`, in ./types) under that op id. +// AWS environments (OpenTofu). Reject with an `AwsError`. Starting commands (and follow) resolve +// to an op id; progress arrives on the `AWS_DEPLOY_PROGRESS_EVENT` event (`DeployProgress`, in +// ./types). Contract: subscribe to aws-deploy-progress BEFORE invoking; events carry `op_id` and +// `state_key`. The backend holds a run's first event until the command is about to return, so a +// listener registered before the invoke sees the whole run, `done` included. A re-follow (e.g. +// after a reload) replays the whole log under a new op id. export const detectAwsEnvironments = (connectionId: string, region: string) => invoke("detect_aws_environments", { connectionId, region }); export const deployAwsEnvironment = (connectionId: string, region: string, name: string, network: NetworkChoice) => diff --git a/app/src/lib/types.ts b/app/src/lib/types.ts index d6c661f..6b6c220 100644 --- a/app/src/lib/types.ts +++ b/app/src/lib/types.ts @@ -1284,7 +1284,7 @@ export interface TofuOutcome { /** The Tauri event a deploy's progress is emitted on (`environment_commands::PROGRESS_EVENT`). */ export const AWS_DEPLOY_PROGRESS_EVENT = "aws-deploy-progress"; -/** Payload of the `aws-deploy-progress` event. `result` is set only when `done`. */ +/** Payload of the `aws-deploy-progress` event. Subscribe before invoking the command; match on `op_id` (or `state_key`). `result` is set only when `done`. */ export interface DeployProgress { op_id: string; /** `environments//terraform.tfstate`, `users///terraform.tfstate` or `boundary/terraform.tfstate`. */