diff --git a/docs/oabctl.md b/docs/oabctl.md index 1a03f6022..81a5a9ff6 100644 --- a/docs/oabctl.md +++ b/docs/oabctl.md @@ -365,18 +365,49 @@ must still be registered manually in the LINE Developers console. > reminded, not verified). > > **Teardown:** `oabctl delete oabservice ` (or `oabctl delete -f `, -> using the same file `apply -f` deployed it from) permanently removes the bot's -> per-bot ingress resources ā€” its exact Cloud Map service (resolved by the ECS -> service's own registry ARN, not a name search, so same-named bots in different -> VPCs/environments can't collide) and its HTTP API (`oab-webhook--`, -> including the API resource itself this time, since the bot is gone for good) ā€” -> on a best-effort basis (it never blocks service deletion). If you instead edit -> a manifest to remove `spec.ingress` while keeping the bot, `apply` runs the same -> Cloud Map + routes/integration/stage cleanup automatically, but **keeps the HTTP -> API** in case ingress is re-added later. The **shared** VPC Link and the -> security-group inbound rule are always left in place for other bots. If the -> Cloud Map service still has registered instances, teardown retries for ~25s -> before falling back to a warning with the manual cleanup command. +> using the same file `apply -f` deployed it from) first captures the exact ECS +> cluster/service ARNs, ECS registry ARN, matching HTTP API ID, configured +> control-plane bucket, caller partition/account/region, and ECS service +> incarnation (`createdAt`) in `delete-checkpoints//.json`. The +> checkpoint is written before ECS mutation and removed only after all +> dependent and S3 cleanup succeeds, so rerunning the same command safely +> resumes a partial delete. ECS HTTP-200 failures, ambiguous drain responses, +> empty service responses, and a missing service fail closed; a matching checkpoint +> authorizes retry only when ECS explicitly reports exactly one `MISSING` +> failure with zero services, or an `INACTIVE` response from the original +> service incarnation. +> +> API cleanup never selects the first same-named API: duplicate names fail +> closed, and the sole candidate is checkpointed only when its integration URI +> equals the ECS registry ARN. Cloud Map is deleted only by the service ID +> parsed from a structurally and boundary-validated Cloud Map service ARN. +> Exact-ID NotFound is idempotent; other cleanup errors retain the checkpoint. +> There is no name-only API, Cloud Map, or S3 orphan cleanup path. Programmatic +> delete namespaces must not contain `-` (names may contain it), which makes the +> existing `oab-{namespace}-{name}` identity injective for every accepted delete +> target. If a per-target +> `ingress-teardown-checkpoints//.json` record exists, delete +> fails before destructive mutation so it cannot discard unfinished exact +> cleanup identity. Re-run the ingress-free apply to completion, then retry +> delete. +> +> Programmatic apply and delete do not provide an internal same-target lock. +> Callers must serialize mutations for the same AWS account, Region, +> control-plane bucket, ECS cluster, namespace, and name. After an accidental +> overlap, stop concurrent writers, inspect retained checkpoints, then +> explicitly re-apply the desired state or retry delete. +> +> If you edit a manifest to remove `spec.ingress` while keeping the bot, `apply` first +> stores every exact ECS registry ARN only when the previously stored OAB +> manifest owned ingress (or resumes an already-valid checkpoint); an arbitrary +> attached registry is never enough. +> `ingress-teardown-checkpoints//.json`, clears only API wiring +> bound to those ARNs, detaches all registries from ECS, waits until the detach +> is observable, and only then deletes each exact Cloud Map service. It keeps +> the HTTP API resource so its URL can survive re-enabling ingress and removes +> the checkpoint only after the full apply succeeds, so retries cannot lose +> cleanup identity. The **shared** VPC Link and security-group inbound rule are +> always left in place for other bots. > > **Changing `paths`:** `apply` prunes routes on the bot's API that are no longer > in the manifest's `ingress.paths`, so renaming or removing a webhook path never diff --git a/operator/README.md b/operator/README.md index 2ad4de1f1..41aab4daf 100644 --- a/operator/README.md +++ b/operator/README.md @@ -75,16 +75,19 @@ commands reference. ## Library API -The crate exposes a deliberately narrow manifest + apply facade for control -planes that should not shell out to the CLI: +The crate exposes a deliberately narrow manifest + apply/delete facade for +control planes that should not shell out to the CLI: ```rust,no_run -use oabctl::{apply_manifests, ApplyOptions, OABServiceManifest}; +use oabctl::{ + apply_manifests, delete_services, ApplyOptions, DeleteOptions, DeleteTarget, + OABServiceManifest, +}; -async fn deploy( +async fn reconcile( aws: &aws_config::SdkConfig, manifest: OABServiceManifest, -) -> Result<(), oabctl::ApplyError> { +) -> Result<(), Box> { let report = apply_manifests( aws, &[manifest], @@ -94,20 +97,56 @@ async fn deploy( for service in report.services { println!("{}: {:?}", service.ecs_service_name, service.action); } + + delete_services( + aws, + &[DeleteTarget::new("prod", "bot")], + &DeleteOptions::new("production-cluster") + .with_control_plane_bucket("my-control-plane-bucket"), + ) + .await?; Ok(()) } ``` -Programmatic apply emits no progress to process-global stdout/stderr. Success -returns per-service actions, webhook URLs, and warnings; reconciliation errors -identify the failed service and include the report completed before the failure. -Both CLI and programmatic apply verify that the target cluster exists and is -`ACTIVE` before mutation, so the caller identity requires -`ecs:DescribeClusters`. This ECS action does not support resource-level +Programmatic apply/delete emit no progress to process-global stdout/stderr. +Apply success returns per-service actions, webhook URLs, and warnings; +reconciliation errors identify the failed service and include the report +completed before the failure. Both CLI and programmatic apply verify that the +target cluster exists and is `ACTIVE` before mutation, so the caller identity +requires `ecs:DescribeClusters`. This ECS action does not support resource-level permissions; its IAM statement must use `Resource: "*"`. -The library never reads `~/.oabctl/config.toml`: set -`with_control_plane_bucket(...)` explicitly when needed, otherwise bucket -resolution uses `OAB_CONTROL_PLANE_BUCKET` and then the caller's AWS account. + +The library never reads `~/.oabctl/config.toml`. Apply and delete resolve the +control-plane bucket through the shared chain: an explicit +`with_control_plane_bucket(...)` override, then `OAB_CONTROL_PLANE_BUCKET`, then +`oab-control-plane-{account}` from the caller identity. Before deleting ECS, +`delete_services` stores the resolved bucket, caller partition/account/region, +canonical ECS cluster/service ARNs, the service `createdAt` incarnation, and +exact ingress IDs in `delete-checkpoints//.json`. The checkpoint +is written before ECS mutation and removed last. Initial DescribeServices +responses must identify exactly one expected live service; missing, empty-success, +ambiguous, or mixed-failure responses fail closed. Retries use only checkpointed +IDs after ECS explicitly reports exactly one `MISSING` failure with zero services +or a matching original `INACTIVE` incarnation. Duplicate APIs fail closed, +and a sole same-name API is checkpointed only when an integration URI exactly +matches the ECS registry ARN. +There is no name-only API, Cloud Map, or S3 orphan cleanup fallback. If a +per-target apply-side `ingress-teardown-checkpoints//.json` +record exists, delete fails before destructive mutation so it cannot discard +unfinished exact cleanup identity. Re-run the ingress-free apply to completion, +then retry delete. + +Delete namespaces must not contain `-`; names may contain it. This makes the +existing `oab-{namespace}-{name}` physical identity injective for every target +accepted by the programmatic delete API. + +Programmatic apply and delete do not provide an internal same-target lock. +Callers must serialize mutations for the same AWS account, Region, control-plane +bucket, ECS cluster, namespace, and name. If overlapping calls occur, stop +concurrent writers, inspect the retained checkpoint, then explicitly re-apply +the desired state or retry delete. + For `aws-sm://#`, a non-ARN `` requires the caller to have `secretsmanager:DescribeSecret`; full-ARN shorthand does not need that lookup. diff --git a/operator/src/apply.rs b/operator/src/apply.rs index 25e180604..a8f8560d6 100644 --- a/operator/src/apply.rs +++ b/operator/src/apply.rs @@ -5,6 +5,7 @@ use aws_sdk_ecs::types::{ AssignPublicIp, AwsVpcConfiguration, CapacityProviderStrategyItem, ContainerDefinition, KeyValuePair, NetworkConfiguration, RuntimePlatform, Secret, }; +use aws_sdk_s3::error::ProvideErrorMetadata; use aws_sdk_s3::primitives::ByteStream; use std::fmt; use std::path::Path; @@ -21,6 +22,15 @@ pub(crate) fn progress_enabled() -> bool { .unwrap_or(true) } +/// Run `future` with human-readable progress output suppressed (library +/// callers). Shared by the programmatic apply and delete entry points. +pub(crate) async fn with_progress_suppressed(future: F) -> F::Output +where + F: std::future::Future, +{ + PROGRESS_ENABLED.scope(false, future).await +} + macro_rules! println { ($($arg:tt)*) => {{ if progress_enabled() { std::println!($($arg)*); } }}; } @@ -389,6 +399,14 @@ pub(crate) async fn run( /// Validate and reconcile in-memory manifests without writing progress to /// process-global stdout or stderr. +/// +/// # Concurrency +/// +/// This function is not serialized with [`crate::delete::delete_services`]. +/// Callers must serialize mutations for the same AWS account, Region, +/// control-plane bucket, ECS cluster, namespace, and name. If that precondition +/// is violated, stop concurrent writers and re-apply the intended desired state +/// before resuming other mutations. pub async fn apply_manifests( aws_config: &aws_config::SdkConfig, manifests: &[OABServiceManifest], @@ -497,6 +515,298 @@ fn parse_manifest_file(path: &Path) -> Result> { } } + +const INGRESS_TEARDOWN_CHECKPOINT_VERSION: u8 = 1; + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +struct IngressTeardownCheckpoint { + version: u8, + bucket: String, + cluster: String, + service_name: String, + service_arn: String, + service_created_at: i128, + registry_arns: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ApplyServicePresence { + Absent, + Present, +} + +fn classify_apply_service_response( + service_count: usize, + failure_reasons: &[&str], +) -> Result { + match (service_count, failure_reasons) { + (1, []) => Ok(ApplyServicePresence::Present), + (0, [reason]) if reason.eq_ignore_ascii_case("MISSING") => { + Ok(ApplyServicePresence::Absent) + } + (0, []) => { + anyhow::bail!("ECS returned no service and no MISSING failure for apply") + } + _ => anyhow::bail!( + "ECS returned an ambiguous apply response: {service_count} service(s), {} failure(s)", + failure_reasons.len() + ), + } +} + +fn ingress_teardown_is_owned(previously_had_ingress: bool, checkpoint_present: bool) -> bool { + previously_had_ingress || checkpoint_present +} + +fn validate_apply_service_identity( + service_arn: &str, + cluster_arn: &str, + expected_service_name: &str, +) -> Result<()> { + let service_resource = service_arn + .split_once(":service/") + .and_then(|(_, resource)| resource.rsplit_once('/')) + .filter(|(_, name)| *name == expected_service_name) + .context("ECS apply response has a conflicting service ARN")?; + let service_parts: Vec<&str> = service_arn.splitn(6, ':').collect(); + let cluster_parts: Vec<&str> = cluster_arn.splitn(6, ':').collect(); + if service_parts.len() != 6 + || cluster_parts.len() != 6 + || service_parts[0] != "arn" + || cluster_parts[0] != "arn" + || service_parts[2] != "ecs" + || cluster_parts[2] != "ecs" + || service_parts[1..5] != cluster_parts[1..5] + { + anyhow::bail!("ECS apply response has an invalid cluster/service ARN boundary"); + } + let cluster_resource = cluster_arn + .split_once(":cluster/") + .map(|(_, resource)| resource) + .filter(|resource| !resource.is_empty() && !resource.contains('/')) + .context("ECS apply response has an invalid cluster ARN")?; + if service_resource.0 != cluster_resource { + anyhow::bail!("ECS apply response service ARN targets another cluster"); + } + Ok(()) +} + +fn ingress_teardown_checkpoint_key(namespace: &str, name: &str) -> String { + format!("ingress-teardown-checkpoints/{namespace}/{name}.json") +} + +pub(crate) fn require_no_pending_ingress_teardown(checkpoint_present: bool) -> Result<()> { + if checkpoint_present { + anyhow::bail!( + "a previous apply ingress teardown is still pending; re-run the ingress-free apply before delete" + ); + } + Ok(()) +} + +async fn load_ingress_teardown_checkpoint( + s3: &aws_sdk_s3::Client, + bucket: &str, + namespace: &str, + name: &str, +) -> Result> { + use aws_sdk_s3::error::ProvideErrorMetadata; + + let key = ingress_teardown_checkpoint_key(namespace, name); + match s3.get_object().bucket(bucket).key(&key).send().await { + Ok(response) => { + let bytes = response + .body + .collect() + .await + .context("failed to read ingress teardown checkpoint")? + .into_bytes(); + Ok(Some(serde_json::from_slice(&bytes).with_context(|| { + format!("invalid ingress teardown checkpoint s3://{bucket}/{key}") + })?)) + } + Err(error) if matches!(error.code(), Some("NoSuchKey" | "NotFound")) => Ok(None), + Err(error) => Err(error).with_context(|| { + format!("failed to read ingress teardown checkpoint s3://{bucket}/{key}") + }), + } +} + +pub(crate) async fn ensure_no_pending_ingress_teardown( + s3: &aws_sdk_s3::Client, + bucket: &str, + namespace: &str, + name: &str, +) -> Result<()> { + let checkpoint = load_ingress_teardown_checkpoint(s3, bucket, namespace, name).await?; + require_no_pending_ingress_teardown(checkpoint.is_some()) +} + +async fn save_ingress_teardown_checkpoint( + s3: &aws_sdk_s3::Client, + bucket: &str, + namespace: &str, + name: &str, + checkpoint: &IngressTeardownCheckpoint, +) -> Result<()> { + let key = ingress_teardown_checkpoint_key(namespace, name); + let body = serde_json::to_vec_pretty(checkpoint)?; + s3.put_object() + .bucket(bucket) + .key(&key) + .body(ByteStream::from(body)) + .content_type("application/json") + .send() + .await + .with_context(|| format!("failed to persist s3://{bucket}/{key}"))?; + Ok(()) +} + +async fn remove_ingress_teardown_checkpoint( + s3: &aws_sdk_s3::Client, + bucket: &str, + namespace: &str, + name: &str, +) -> Result<()> { + let key = ingress_teardown_checkpoint_key(namespace, name); + s3.delete_object() + .bucket(bucket) + .key(&key) + .send() + .await + .with_context(|| format!("failed to remove s3://{bucket}/{key}"))?; + Ok(()) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ApplyServiceIdentity { + service_arn: String, + cluster_arn: String, + service_name: String, + created_at: i128, +} + +fn resolve_apply_service_identity( + service: &aws_sdk_ecs::types::Service, + expected_service_name: &str, +) -> Result { + let service_arn = service + .service_arn() + .filter(|value| !value.trim().is_empty()) + .context("ECS apply response missing service ARN")? + .to_string(); + let cluster_arn = service + .cluster_arn() + .filter(|value| !value.trim().is_empty()) + .context("ECS apply response missing cluster ARN")? + .to_string(); + validate_apply_service_identity(&service_arn, &cluster_arn, expected_service_name)?; + let service_name = service + .service_name() + .filter(|value| !value.trim().is_empty()) + .context("ECS apply response missing service name")?; + if service_name != expected_service_name { + anyhow::bail!("ECS apply response has a conflicting service name"); + } + let created_at = service + .created_at() + .map(|value| value.as_nanos()) + .context("ECS apply response missing createdAt incarnation")?; + Ok(ApplyServiceIdentity { + service_arn, + cluster_arn, + service_name: expected_service_name.to_string(), + created_at, + }) +} + +fn validate_expected_apply_service_identity( + expected: &ApplyServiceIdentity, + service: &aws_sdk_ecs::types::Service, +) -> Result<()> { + let actual = resolve_apply_service_identity(service, &expected.service_name)?; + if actual != *expected { + anyhow::bail!( + "ECS apply response belongs to a recreated same-name service or different cluster" + ); + } + Ok(()) +} + +async fn revalidate_apply_service_identity( + ecs: &aws_sdk_ecs::Client, + expected: &ApplyServiceIdentity, +) -> Result<()> { + let response = ecs + .describe_services() + .cluster(&expected.cluster_arn) + .services(&expected.service_arn) + .send() + .await + .context("failed to revalidate ECS service identity before update")?; + if !response.failures().is_empty() { + anyhow::bail!( + "ECS returned failure(s) while revalidating service identity before update: {:?}", + response.failures() + ); + } + let service = match response.services() { + [service] => service, + [] => anyhow::bail!( + "ECS service disappeared while revalidating identity before update" + ), + services => anyhow::bail!( + "ECS returned {} services while revalidating one update target", + services.len() + ), + }; + validate_expected_apply_service_identity(expected, service) +} + +async fn wait_for_registry_detach( + ecs: &aws_sdk_ecs::Client, + expected: &ApplyServiceIdentity, +) -> Result<()> { + const ATTEMPTS: u32 = 12; + const INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); + + for attempt in 0..ATTEMPTS { + let response = ecs + .describe_services() + .cluster(&expected.cluster_arn) + .services(&expected.service_arn) + .send() + .await + .context("failed to verify ECS service registry detach")?; + if !response.failures().is_empty() { + anyhow::bail!( + "ECS returned failure(s) while verifying registry detach: {:?}", + response.failures() + ); + } + let service = match response.services() { + [service] => service, + [] => anyhow::bail!("ECS service disappeared while verifying registry detach"), + services => anyhow::bail!( + "ECS returned {} services for one registry-detach target", + services.len() + ), + }; + validate_expected_apply_service_identity(expected, service)?; + if service.service_registries().is_empty() { + return Ok(()); + } + if attempt + 1 < ATTEMPTS { + tokio::time::sleep(INTERVAL).await; + } + } + + anyhow::bail!( + "ECS service {} still reports service registries after waiting; Cloud Map cleanup was not started", + expected.service_name + ) +} + async fn apply_ecs( ecs: &aws_sdk_ecs::Client, s3: &aws_sdk_s3::Client, @@ -520,12 +830,9 @@ async fn apply_ecs( } let bootstrap_state = bootstrap.state.as_ref(); - // Read current generation from S3 manifest (if exists), increment. - // Also capture whether the *previous* apply had ingress configured, so we - // can detect "ingress was removed from the manifest" and tear it down - // below ā€” apply only ever provisioned ingress resources before this, so a - // manifest edit that drops `spec.ingress` used to orphan the per-bot HTTP - // API and Cloud Map service. + // Read the current generation from the stored desired-state manifest. + // Ingress teardown retry state is kept separately in an exact-identity + // checkpoint, so writing a newer manifest cannot lose pending cleanup. let manifest_key = format!( "manifests/{}/{}.yaml", m.metadata.namespace, m.metadata.name @@ -545,15 +852,16 @@ async fn apply_ecs( existing.spec.ingress.is_some(), ) } - Err(_) => (0, false), + Err(error) if matches!(error.code(), Some("NoSuchKey" | "NotFound")) => (0, false), + Err(error) => { + return Err(error).with_context(|| { + format!("failed to read existing manifest s3://{bucket}/{manifest_key}") + }); + } }; let generation = current_gen + 1; - // Look up the ECS service's current registry ARN(s) up front so both the - // ingress-removal teardown below and the update/create logic further down - // can use the *exact* registry rather than falling back to a name-only - // Cloud Map scan (which can collide across VPCs/environments that share - // an account and reuse the same namespace/name). + // Resolve the current ECS registry identities without first-match behavior. let describe_resp = ecs .describe_services() .cluster(cluster) @@ -561,44 +869,159 @@ async fn apply_ecs( .send() .await .context("failed to describe ECS service")?; - let existing_registry_arns: Vec = describe_resp - .services() - .first() - .map(|service| { - service - .service_registries() - .iter() - .filter_map(|registry| registry.registry_arn()) - .map(str::to_owned) - .collect() - }) - .unwrap_or_default(); + let failure_reasons: Vec<&str> = describe_resp + .failures() + .iter() + .map(|failure| failure.reason().unwrap_or("UNKNOWN")) + .collect(); + let existing_service = match classify_apply_service_response( + describe_resp.services().len(), + &failure_reasons, + )? { + ApplyServicePresence::Absent => None, + ApplyServicePresence::Present => { + let service = &describe_resp.services()[0]; + validate_apply_service_identity( + service + .service_arn() + .filter(|value| !value.trim().is_empty()) + .context("ECS apply response missing service ARN")?, + service + .cluster_arn() + .filter(|value| !value.trim().is_empty()) + .context("ECS apply response missing cluster ARN")?, + &service_name, + )?; + if !matches!(service.status(), Some("ACTIVE" | "DRAINING")) { + anyhow::bail!("ECS apply response has unexpected service status"); + } + Some(service) + } + }; + let mut existing_registry_arns = Vec::new(); + if let Some(service) = existing_service { + for registry in service.service_registries() { + existing_registry_arns.push( + registry + .registry_arn() + .filter(|value| !value.trim().is_empty()) + .context("ECS returned a service registry without an ARN")? + .to_string(), + ); + } + } + existing_registry_arns.sort(); + existing_registry_arns.dedup(); let has_registries = !existing_registry_arns.is_empty(); + let service_active = existing_service.is_some(); + let existing_identity = existing_service + .map(|service| resolve_apply_service_identity(service, &service_name)) + .transpose()?; + + let stored_teardown = load_ingress_teardown_checkpoint( + s3, + bucket, + &m.metadata.namespace, + &m.metadata.name, + ) + .await?; + if m.spec.ingress.is_some() && stored_teardown.is_some() { + anyhow::bail!( + "a previous ingress teardown is still pending; apply the ingress-free manifest again before re-enabling ingress" + ); + } + let teardown_checkpoint = if m.spec.ingress.is_none() + && ingress_teardown_is_owned(previously_had_ingress, stored_teardown.is_some()) + { + match stored_teardown { + Some(checkpoint) => { + if checkpoint.version != INGRESS_TEARDOWN_CHECKPOINT_VERSION + || checkpoint.bucket != bucket + || checkpoint.cluster != cluster + || checkpoint.service_name != service_name + || checkpoint.service_arn.trim().is_empty() + || checkpoint.service_created_at <= 0 + || checkpoint.registry_arns.is_empty() + { + anyhow::bail!("ingress teardown checkpoint does not match this apply target"); + } + if let Some(identity) = &existing_identity { + if identity.service_arn != checkpoint.service_arn + || identity.created_at != checkpoint.service_created_at + { + anyhow::bail!( + "ingress teardown checkpoint belongs to a different ECS service incarnation" + ); + } + } + Some(checkpoint) + } + None if previously_had_ingress && has_registries => { + let identity = existing_identity + .clone() + .context("cannot checkpoint ingress teardown without an identified ECS service")?; + let checkpoint = IngressTeardownCheckpoint { + version: INGRESS_TEARDOWN_CHECKPOINT_VERSION, + bucket: bucket.to_string(), + cluster: cluster.to_string(), + service_name: service_name.clone(), + service_arn: identity.service_arn, + service_created_at: identity.created_at, + registry_arns: existing_registry_arns.clone(), + }; + save_ingress_teardown_checkpoint( + s3, + bucket, + &m.metadata.namespace, + &m.metadata.name, + &checkpoint, + ) + .await?; + Some(checkpoint) + } + None => None, + } + } else { + None + }; - // If ingress was configured before but is absent now, tear down the - // orphaned per-bot ingress resources (best-effort, mirrors `oabctl delete`) - // and detach the stale registry from the ECS service itself ā€” omitting - // `serviceRegistries` on `UpdateService` leaves the existing configuration - // untouched (AWS only clears it when explicitly passed an empty list), so - // without this the service would keep pointing at a Cloud Map service that - // teardown() is about to delete. - if previously_had_ingress && m.spec.ingress.is_none() { - eprintln!(" šŸŒ ingress removed from manifest ā€” tearing down orphaned resources..."); - match crate::ingress::teardown( - config, - &m.metadata.namespace, - &m.metadata.name, - existing_registry_arns.first().map(String::as_str), - ) - .await - { - Ok(teardown_warnings) => warnings.extend(teardown_warnings), - Err(error) => { - let warning = format!("ingress teardown skipped: {error}"); - eprintln!(" āš  {warning}"); - warnings.push(warning); + let expected_identity = if service_active { + let identity = existing_identity + .clone() + .context("cannot update ECS service without an exact identity")?; + if let Some(checkpoint) = &teardown_checkpoint { + if identity.service_arn != checkpoint.service_arn + || identity.created_at != checkpoint.service_created_at + { + anyhow::bail!( + "ingress teardown checkpoint belongs to a different ECS service incarnation" + ); } } + Some(identity) + } else { + if teardown_checkpoint.is_some() { + anyhow::bail!( + "cannot detach ingress from an absent ECS service; exact service identity is unavailable" + ); + } + None + }; + + if let Some(checkpoint) = &teardown_checkpoint { + eprintln!(" šŸŒ ingress removed from manifest ā€” tearing down exact API wiring..."); + for registry_arn in &checkpoint.registry_arns { + warnings.extend( + crate::ingress::teardown( + config, + &m.metadata.namespace, + &m.metadata.name, + Some(registry_arn), + ) + .await + .context("failed to tear down exact ingress API wiring")?, + ); + } } // 1. Upload manifest to S3 (record of desired state) @@ -849,10 +1272,6 @@ async fn apply_ecs( // Check if service exists. Reuses `describe_resp` captured above (before // the ingress-removal teardown) ā€” `ensure_cloud_map` above doesn't touch // the ECS service, so its ACTIVE status can't have changed since then. - let service_active = describe_resp - .services() - .first() - .is_some_and(|service| service.status() == Some("ACTIVE")); let action; if service_active { @@ -874,12 +1293,20 @@ async fn apply_ecs( // "clear"; only an explicit empty list detaches it. Without this the // service keeps pointing at the Cloud Map service that the // ingress-removal teardown (above) just deleted. - let needs_detach = cloud_map.is_none() && has_registries; + let needs_detach = teardown_checkpoint.is_some() && cloud_map.is_none() && has_registries; + let update_cluster = expected_identity + .as_ref() + .map(|identity| identity.cluster_arn.as_str()) + .unwrap_or(cluster); + let update_service = expected_identity + .as_ref() + .map(|identity| identity.service_arn.as_str()) + .unwrap_or(service_name.as_str()); let mut update_req = ecs .update_service() - .cluster(cluster) - .service(&service_name) + .cluster(update_cluster) + .service(update_service) .task_definition(&task_def_arn) .enable_execute_command(true) .network_configuration(network_config); @@ -899,6 +1326,11 @@ async fn apply_ecs( update_req = update_req.set_service_registries(Some(Vec::new())); } + if let Some(expected) = &expected_identity { + // Revalidate immediately before the mutation. In particular, a + // recreated same-name service must never receive this update. + revalidate_apply_service_identity(ecs, expected).await?; + } update_req .send() .await @@ -1012,6 +1444,20 @@ async fn apply_ecs( ); } + if let Some(checkpoint) = &teardown_checkpoint { + let expected = expected_identity + .as_ref() + .context("missing exact identity for ingress detach wait")?; + wait_for_registry_detach(ecs, expected).await?; + for registry_arn in &checkpoint.registry_arns { + crate::ingress::delete_cloud_map_exact(config, registry_arn, &service_name) + .await + .with_context(|| { + format!("failed exact Cloud Map cleanup for {registry_arn}") + })?; + } + } + // Ingress step 2: VPC Link + API Gateway + routes + SG rule. let mut webhook_urls = Vec::new(); if let (Some(ingress), Some(cm)) = (&m.spec.ingress, &cloud_map) { @@ -1060,6 +1506,16 @@ async fn apply_ecs( eprintln!(" āœ“ {} is stable", m.metadata.name); } + if teardown_checkpoint.is_some() { + remove_ingress_teardown_checkpoint( + s3, + bucket, + &m.metadata.namespace, + &m.metadata.name, + ) + .await?; + } + Ok(AppliedService { namespace: m.metadata.namespace.clone(), name: m.metadata.name.clone(), @@ -1089,10 +1545,25 @@ async fn wait_for_stable(ecs: &aws_sdk_ecs::Client, cluster: &str, service: &str .await?; let elapsed = (i + 1) * 5; - let Some(svc) = resp.services().first() else { - eprintln!(" [{elapsed}s] service not found in describe-services response yet"); - continue; + if !resp.failures().is_empty() { + anyhow::bail!("ECS returned failure(s) while waiting for service stability: {:?}", resp.failures()); + } + let svc = match resp.services() { + [service] => service, + [] => { + eprintln!(" [{elapsed}s] service not found in describe-services response yet"); + continue; + } + services => anyhow::bail!( + "ECS returned {} services while waiting for one apply target", + services.len() + ), }; + validate_apply_service_identity( + svc.service_arn().context("ECS stability response missing service ARN")?, + svc.cluster_arn().context("ECS stability response missing cluster ARN")?, + service, + )?; let running = svc.running_count() as usize; let desired = svc.desired_count() as usize; @@ -1382,4 +1853,54 @@ spec: let result = resolve_task_role_arn(&manifest, bootstrap); assert_eq!(result, None); } + + #[test] + fn apply_accepts_only_single_missing_failure_as_first_absence() { + assert_eq!( + classify_apply_service_response(0, &["MISSING"]).unwrap(), + ApplyServicePresence::Absent + ); + assert!(classify_apply_service_response(0, &[]).is_err()); + assert!(classify_apply_service_response(0, &["MISSING", "ACCESS_DENIED"]).is_err()); + assert!(classify_apply_service_response(1, &["MISSING"]).is_err()); + } + + #[test] + fn apply_requires_one_service_for_present_identity() { + assert_eq!( + classify_apply_service_response(1, &[]).unwrap(), + ApplyServicePresence::Present + ); + assert!(classify_apply_service_response(2, &[]).is_err()); + assert!(validate_apply_service_identity( + "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-bot", + "arn:aws:ecs:us-east-1:123456789012:cluster/cluster", + "oab-prod-bot", + ) + .is_ok()); + assert!(validate_apply_service_identity( + "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-other", + "arn:aws:ecs:us-east-1:123456789012:cluster/cluster", + "oab-prod-bot", + ) + .is_err()); + } + + #[test] + fn apply_teardown_requires_manifest_ownership_or_checkpoint() { + assert!(ingress_teardown_is_owned(true, false)); + assert!(ingress_teardown_is_owned(false, true)); + assert!(!ingress_teardown_is_owned(false, false)); + } + + #[tokio::test] + async fn progress_gate_suppresses_nested_progress() { + assert!(progress_enabled()); + PROGRESS_ENABLED + .scope(false, async { + assert!(!progress_enabled()); + }) + .await; + assert!(progress_enabled()); + } } diff --git a/operator/src/delete.rs b/operator/src/delete.rs index 953e863dd..1c09dc309 100644 --- a/operator/src/delete.rs +++ b/operator/src/delete.rs @@ -1,7 +1,515 @@ use anyhow::{Context, Result}; use aws_sdk_ecs::error::ProvideErrorMetadata; +use aws_sdk_s3::primitives::ByteStream; +use serde::{Deserialize, Serialize}; +use std::fmt; use std::path::Path; +// Route all human-readable progress through the same task-local gate as +// apply: CLI callers keep today's output, programmatic callers are silent. +macro_rules! println { + ($($arg:tt)*) => {{ if crate::apply::progress_enabled() { std::println!($($arg)*); } }}; +} +macro_rules! eprintln { + ($($arg:tt)*) => {{ if crate::apply::progress_enabled() { std::eprintln!($($arg)*); } }}; +} +macro_rules! eprint { + ($($arg:tt)*) => {{ if crate::apply::progress_enabled() { std::eprint!($($arg)*); } }}; +} + +/// Identity of a service targeted for deletion. +/// +/// Deletion does not require the original manifest: the control-plane copy +/// under `manifests/{namespace}/{name}.yaml` (and every other resource) is +/// addressed by `namespace` + `name` alone. Because physical resource names +/// use `-` as the component delimiter, [`delete_services`] accepts namespaces +/// without `-`; names may still contain it. This keeps every accepted logical +/// target's physical delete identity injective without renaming existing +/// deployments. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DeleteTarget { + pub namespace: String, + pub name: String, +} + +impl DeleteTarget { + pub fn new(namespace: impl Into, name: impl Into) -> Self { + Self { + namespace: namespace.into(), + name: name.into(), + } + } + + /// ECS service name derived from the target (`oab-{namespace}-{name}`). + pub fn ecs_service_name(&self) -> String { + format!("oab-{}-{}", self.namespace, self.name) + } +} + +/// Target options for [`delete_services`]. Mirrors +/// [`ApplyOptions`](crate::apply::ApplyOptions): the cluster is required and +/// the library never reads `~/.oabctl/config.toml`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DeleteOptions { + /// ECS cluster name or ARN the services were deployed into. + pub cluster: String, + /// Optional control-plane bucket override. When absent, resolution uses + /// `OAB_CONTROL_PLANE_BUCKET`, then `oab-control-plane-{account}` from + /// the caller's AWS identity, matching apply's shared resolver. + pub control_plane_bucket: Option, +} + +impl DeleteOptions { + pub fn new(cluster: impl Into) -> Self { + Self { + cluster: cluster.into(), + control_plane_bucket: None, + } + } + + pub fn with_control_plane_bucket(mut self, bucket: impl Into) -> Self { + self.control_plane_bucket = Some(bucket.into()); + self + } +} + +/// Teardown outcome for one service. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DeletedService { + pub namespace: String, + pub name: String, + pub ecs_service_name: String, + /// Non-fatal diagnostics retained for API compatibility. Exact-identity + /// dependent and S3 cleanup failures are fatal so the durable checkpoint + /// remains available for retry. + pub warnings: Vec, +} + +/// Structured result of a successful (or partially completed) delete. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct DeleteReport { + pub services: Vec, +} + +/// High-level phase in which delete failed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DeleteErrorKind { + /// Contract violation caught before any AWS call. + Validation, + /// The deployment target (bucket resolution) could not be established. + Target, + /// Teardown of a specific service failed. Cleanup is resumable: calling + /// [`delete_services`] again with the same target continues from the + /// remaining resources. + Teardown, +} + +/// Structured delete failure. Teardown failures identify the failed service +/// and retain the report for all services completed before it. +#[derive(Debug)] +pub struct DeleteError { + pub kind: DeleteErrorKind, + pub failed_service: Option, + pub completed: DeleteReport, + source: anyhow::Error, +} + +impl DeleteError { + fn validation(source: impl Into) -> Self { + Self { + kind: DeleteErrorKind::Validation, + failed_service: None, + completed: DeleteReport::default(), + source: source.into(), + } + } + + fn target(source: impl Into) -> Self { + Self { + kind: DeleteErrorKind::Target, + failed_service: None, + completed: DeleteReport::default(), + source: source.into(), + } + } + + fn teardown( + failed_service: DeleteTarget, + completed: DeleteReport, + source: impl Into, + ) -> Self { + Self { + kind: DeleteErrorKind::Teardown, + failed_service: Some(failed_service), + completed, + source: source.into(), + } + } + + pub fn source_error(&self) -> &anyhow::Error { + &self.source + } +} + +impl fmt::Display for DeleteError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match &self.failed_service { + Some(service) => write!( + f, + "delete {:?} error for {}/{}: {}", + self.kind, service.namespace, service.name, self.source + ), + None => write!(f, "delete {:?} error: {}", self.kind, self.source), + } + } +} + +impl std::error::Error for DeleteError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.source.as_ref()) + } +} + +const DELETE_CHECKPOINT_VERSION: u8 = 1; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct DeleteCheckpoint { + version: u8, + namespace: String, + name: String, + bucket: String, + partition: String, + account: String, + region: String, + requested_cluster: String, + cluster_arn: String, + service_arn: String, + /// ECS `createdAt` as epoch nanoseconds; distinguishes a recreated same-name service. + service_created_at: i128, + registry_arn: Option, + api_id: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ServiceIdentityState { + Live, + RetryGone, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ArnParts<'a> { + partition: &'a str, + service: &'a str, + region: &'a str, + account: &'a str, + resource: &'a str, +} + +fn parse_arn(value: &str) -> Option> { + let mut parts = value.splitn(6, ':'); + if parts.next()? != "arn" { + return None; + } + let arn = ArnParts { + partition: parts.next()?, + service: parts.next()?, + region: parts.next()?, + account: parts.next()?, + resource: parts.next()?, + }; + if arn.partition.is_empty() + || arn.service.is_empty() + || arn.region.is_empty() + || arn.account.is_empty() + || arn.resource.is_empty() + { + return None; + } + Some(arn) +} + +fn cluster_reference_matches(cluster_arn: &str, reference: &str) -> bool { + let Some(cluster) = parse_arn(cluster_arn) else { + return false; + }; + if reference.starts_with("arn:") { + return reference == cluster_arn; + } + cluster.resource.strip_prefix("cluster/") == Some(reference) +} + +fn validate_checkpoint_arns( + checkpoint: &DeleteCheckpoint, + expected_service_name: &str, + partition: &str, + account: &str, + region: &str, +) -> Result<()> { + let cluster = parse_arn(&checkpoint.cluster_arn) + .context("delete checkpoint contains an invalid ECS cluster ARN")?; + if cluster.partition != partition + || cluster.service != "ecs" + || cluster.account != account + || cluster.region != region + { + anyhow::bail!("delete checkpoint ECS cluster ARN is outside the caller boundary"); + } + let cluster_name = cluster + .resource + .strip_prefix("cluster/") + .filter(|value| !value.is_empty() && !value.contains('/')) + .context("delete checkpoint contains an invalid ECS cluster resource")?; + + let service = parse_arn(&checkpoint.service_arn) + .context("delete checkpoint contains an invalid ECS service ARN")?; + if service.partition != cluster.partition + || service.service != "ecs" + || service.account != account + || service.region != region + { + anyhow::bail!("delete checkpoint ECS service ARN is outside the cluster boundary"); + } + let (service_cluster, service_name) = service + .resource + .strip_prefix("service/") + .and_then(|resource| resource.split_once('/')) + .context("delete checkpoint ECS service ARN lacks its cluster identity")?; + if service_cluster != cluster_name || service_name != expected_service_name { + anyhow::bail!("delete checkpoint ECS service ARN does not match the target cluster/service"); + } + + if let Some(registry_arn) = checkpoint.registry_arn.as_deref() { + let registry = parse_arn(registry_arn) + .context("delete checkpoint contains an invalid Cloud Map service ARN")?; + if registry.partition != cluster.partition + || registry.service != "servicediscovery" + || registry.account != account + || registry.region != region + || registry + .resource + .strip_prefix("service/") + .filter(|value| !value.is_empty() && !value.contains('/')) + .is_none() + { + anyhow::bail!( + "delete checkpoint Cloud Map ARN is outside the ECS account/region boundary" + ); + } + } + Ok(()) +} + +fn classify_service_identity( + has_checkpoint: bool, + service_present: bool, + status: Option<&str>, + failure_reasons: &[&str], +) -> std::result::Result { + if !failure_reasons.is_empty() { + if has_checkpoint + && !service_present + && failure_reasons.len() == 1 + && failure_reasons[0].eq_ignore_ascii_case("MISSING") + { + return Ok(ServiceIdentityState::RetryGone); + } + return Err(format!( + "DescribeServices returned failure(s): {}", + failure_reasons.join(", ") + )); + } + + match (service_present, status) { + (true, Some("ACTIVE")) | (true, Some("DRAINING")) => Ok(ServiceIdentityState::Live), + // INACTIVE is retry-gone only after the caller has validated the exact + // service ARN, cluster ARN, and incarnation discriminator. + (true, Some("INACTIVE")) if has_checkpoint => Ok(ServiceIdentityState::RetryGone), + // An empty successful response is ambiguous, and a checkpoint only + // authorizes exactly zero services plus one MISSING failure. + (false, None) => Err( + "ECS returned no service without an explicit MISSING failure; refusing cleanup" + .to_string(), + ), + (true, Some("INACTIVE")) => Err( + "ECS returned INACTIVE without a matching delete checkpoint".to_string(), + ), + (true, None) => Err("ECS returned a service without status".to_string()), + (true, Some(other)) => Err(format!( + "unexpected ECS service status during delete: {other}" + )), + (false, Some(_)) => Err("ECS returned status without a service identity".to_string()), + } +} + +fn checkpoint_key(namespace: &str, name: &str) -> String { + format!("delete-checkpoints/{namespace}/{name}.json") +} + +#[allow(clippy::too_many_arguments)] +fn validate_checkpoint( + checkpoint: &DeleteCheckpoint, + namespace: &str, + name: &str, + cluster: &str, + bucket: &str, + partition: &str, + account: &str, + region: &str, +) -> Result<()> { + if checkpoint.version != DELETE_CHECKPOINT_VERSION { + anyhow::bail!( + "unsupported delete checkpoint version {}", + checkpoint.version + ); + } + if checkpoint.namespace != namespace + || checkpoint.name != name + || checkpoint.bucket != bucket + || checkpoint.partition != partition + || checkpoint.account != account + || checkpoint.region != region + { + anyhow::bail!( + "delete checkpoint identity does not match namespace/name, bucket, or caller boundary" + ); + } + if !cluster_reference_matches(&checkpoint.cluster_arn, &checkpoint.requested_cluster) + || !cluster_reference_matches(&checkpoint.cluster_arn, cluster) + { + anyhow::bail!("delete checkpoint canonical cluster does not match the requested cluster"); + } + if checkpoint.cluster_arn.trim().is_empty() + || checkpoint.service_arn.trim().is_empty() + || checkpoint.service_created_at <= 0 + { + anyhow::bail!("delete checkpoint is missing exact ECS identity or service incarnation"); + } + validate_checkpoint_arns( + checkpoint, + &format!("oab-{namespace}-{name}"), + partition, + account, + region, + )?; + if let Some(api_id) = checkpoint.api_id.as_deref() { + if api_id.trim().is_empty() || checkpoint.registry_arn.is_none() { + anyhow::bail!( + "delete checkpoint API identity requires a non-empty API ID and registry ARN" + ); + } + } + Ok(()) +} + +fn validate_delete_request( + targets: &[DeleteTarget], + opts: &DeleteOptions, +) -> std::result::Result<(), DeleteError> { + if targets.is_empty() { + return Err(DeleteError::validation(anyhow::anyhow!( + "no targets to delete (empty target set)" + ))); + } + if opts.cluster.trim().is_empty() { + return Err(DeleteError::validation(anyhow::anyhow!( + "DeleteOptions.cluster must not be empty" + ))); + } + for target in targets { + if target.namespace.trim().is_empty() || target.name.trim().is_empty() { + return Err(DeleteError::validation(anyhow::anyhow!( + "delete target namespace and name must not be empty (got '{}'/'{}')", + target.namespace, + target.name + ))); + } + if target.namespace.contains('-') { + return Err(DeleteError::validation(anyhow::anyhow!( + "delete target namespace must not contain '-' because it is the physical identity delimiter (got '{}')", + target.namespace + ))); + } + } + Ok(()) +} + +fn record_delete_result( + mut report: DeleteReport, + target: &DeleteTarget, + outcome: Result>, +) -> std::result::Result { + match outcome { + Ok(warnings) => { + report.services.push(DeletedService { + namespace: target.namespace.clone(), + name: target.name.clone(), + ecs_service_name: target.ecs_service_name(), + warnings, + }); + Ok(report) + } + Err(error) => Err(DeleteError::teardown(target.clone(), report, error)), + } +} + +/// Tear down OAB services programmatically, without reading CLI home +/// configuration or writing progress to process-global stdout/stderr. +/// +/// Contract (enforced before any AWS call): +/// - the target set must be non-empty +/// - [`DeleteOptions::cluster`] must be non-empty; its optional bucket override +/// follows the shared control-plane resolver +/// - every target's `namespace`/`name` must be non-empty +/// - target namespaces must not contain `-`; that delimiter may appear in names, +/// so this restriction makes `oab-{namespace}-{name}` injective for every +/// accepted target +/// +/// # Concurrency +/// +/// This function is not serialized with [`crate::apply::apply_manifests`]. +/// Callers must serialize mutations for the same AWS account, Region, +/// control-plane bucket, ECS cluster, namespace, and name. If that precondition +/// is violated, stop concurrent writers, inspect the retained checkpoint, then +/// either re-apply the intended desired state or retry delete. +/// +/// Before ECS mutation, delete persists an exact-identity checkpoint containing +/// the caller account/region, bucket, canonical cluster and service ARNs, and +/// any ECS registry/API IDs. An absent ECS service or ambiguous +/// `DescribeServices` response is rejected unless that matching durable +/// checkpoint already exists. Cleanup uses only IDs from the checkpoint and +/// removes the checkpoint last, making partial failures safe to retry. +pub async fn delete_services( + aws_config: &aws_config::SdkConfig, + targets: &[DeleteTarget], + opts: &DeleteOptions, +) -> std::result::Result { + crate::apply::with_progress_suppressed(async { + validate_delete_request(targets, opts)?; + let bucket = crate::control_plane::resolve_bucket( + aws_config, + opts.control_plane_bucket.as_deref(), + ) + .await + .map_err(DeleteError::target)?; + + let mut report = DeleteReport::default(); + for target in targets { + let outcome = run_with_bucket( + aws_config, + "oabservice", + &target.name, + &opts.cluster, + &target.namespace, + &bucket, + ) + .await; + report = record_delete_result(report, target, outcome)?; + } + Ok(report) + }) + .await +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum EcsDeletePhase { Delete, @@ -18,6 +526,25 @@ fn ecs_delete_phase(status: Option<&str>) -> Result { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum DrainPollAction { + Complete, + Retry, + TimedOut, +} + +fn drain_poll_action(is_gone: bool, attempt: u32, max_attempts: u32) -> DrainPollAction { + debug_assert!(max_attempts > 0); + debug_assert!(attempt < max_attempts); + if is_gone { + DrainPollAction::Complete + } else if attempt + 1 == max_attempts { + DrainPollAction::TimedOut + } else { + DrainPollAction::Retry + } +} + /// Delete every OABService defined in a manifest file or directory. pub(crate) async fn run_from_file( aws_config: &aws_config::SdkConfig, @@ -81,215 +608,869 @@ pub(crate) async fn run( let bucket = crate::control_plane::resolve_bucket(aws_config, oab_cfg.bootstrap.bucket.as_deref()) .await?; - run_with_bucket(aws_config, resource, name, cluster, namespace, &bucket).await + run_with_bucket( + aws_config, + resource, + name, + cluster, + namespace, + &bucket, + ) + .await + .map(|_warnings| ()) } -async fn run_with_bucket( +async fn caller_context(aws_config: &aws_config::SdkConfig) -> Result<(String, String, String)> { + let identity = aws_sdk_sts::Client::new(aws_config) + .get_caller_identity() + .send() + .await + .context("failed to identify delete caller")?; + let account = identity + .account() + .context("STS response missing caller account")? + .to_string(); + let caller_arn = identity.arn().context("STS response missing caller ARN")?; + let mut caller_arn_parts = caller_arn.splitn(3, ':'); + if caller_arn_parts.next() != Some("arn") { + anyhow::bail!("STS returned an invalid caller ARN"); + } + let partition = caller_arn_parts + .next() + .filter(|value| !value.is_empty()) + .context("STS caller ARN is missing its partition")? + .to_string(); + let region = aws_config + .region() + .context("AWS region must be resolved before delete")? + .to_string(); + Ok((partition, account, region)) +} + +async fn load_checkpoint( + s3: &aws_sdk_s3::Client, bucket: &str, namespace: &str, name: &str, +) -> Result> { + let key = checkpoint_key(namespace, name); + match s3.get_object().bucket(bucket).key(&key).send().await { + Ok(response) => { + let bytes = response.body.collect().await + .context("failed to read delete checkpoint body")?.into_bytes(); + Ok(Some(serde_json::from_slice(&bytes) + .with_context(|| format!("invalid delete checkpoint s3://{bucket}/{key}"))?)) + } + Err(error) if matches!(error.code(), Some("NoSuchKey" | "NotFound")) => Ok(None), + Err(error) => Err(error) + .with_context(|| format!("failed to read delete checkpoint s3://{bucket}/{key}")), + } +} + +async fn save_checkpoint(s3: &aws_sdk_s3::Client, checkpoint: &DeleteCheckpoint) -> Result<()> { + let key = checkpoint_key(&checkpoint.namespace, &checkpoint.name); + let body = serde_json::to_vec_pretty(checkpoint)?; + s3.put_object().bucket(&checkpoint.bucket).key(&key) + .body(ByteStream::from(body)).content_type("application/json").send().await + .with_context(|| format!( + "failed to persist exact delete identity to s3://{}/{key}", checkpoint.bucket + ))?; + Ok(()) +} + +fn failure_reasons( + response: &aws_sdk_ecs::operation::describe_services::DescribeServicesOutput, +) -> Vec<&str> { + response.failures().iter() + .map(|failure| failure.reason().unwrap_or("UNKNOWN")) + .collect() +} + + +fn single_described_service( + response: &aws_sdk_ecs::operation::describe_services::DescribeServicesOutput, +) -> Result> { + match response.services() { + [] => Ok(None), + [service] => Ok(Some(service)), + services => anyhow::bail!( + "ECS returned {} services for one delete target", + services.len() + ), + } +} +enum PreparedIdentity { + Exact(Box<(DeleteCheckpoint, ServiceIdentityState)>), +} + +#[allow(clippy::too_many_arguments)] +async fn prepare_identity( aws_config: &aws_config::SdkConfig, - resource: &str, + ecs: &aws_sdk_ecs::Client, + s3: &aws_sdk_s3::Client, + namespace: &str, name: &str, cluster: &str, - namespace: &str, bucket: &str, -) -> Result<()> { - if resource != "oabservice" { - anyhow::bail!("unknown resource type: {resource}. Use 'oabservice'"); +) -> Result { + let (partition, account, region) = caller_context(aws_config).await?; + if let Some(checkpoint) = load_checkpoint(s3, bucket, namespace, name).await? { + validate_checkpoint( + &checkpoint, + namespace, + name, + cluster, + bucket, + &partition, + &account, + ®ion, + )?; + let response = ecs.describe_services() + .cluster(&checkpoint.cluster_arn).services(&checkpoint.service_arn) + .send().await.context("failed to describe checkpointed ECS service")?; + let service = single_described_service(&response)?; + let reasons = failure_reasons(&response); + if let Some(service) = service { + validate_returned_service(&checkpoint, service)?; + } + let state = classify_service_identity( + true, + service.is_some(), + service.and_then(|service| service.status()), + &reasons, + ) + .map_err(anyhow::Error::msg)?; + return Ok(PreparedIdentity::Exact(Box::new((checkpoint, state)))); } let service_name = format!("oab-{namespace}-{name}"); - let ecs = aws_sdk_ecs::Client::new(aws_config); - let s3 = aws_sdk_s3::Client::new(aws_config); + let response = ecs.describe_services().cluster(cluster).services(&service_name) + .send().await.context("failed to describe ECS service before delete")?; + let service = single_described_service(&response)?; + let reasons = failure_reasons(&response); + let state = classify_service_identity( + false, + service.is_some(), + service.and_then(|service| service.status()), + &reasons, + ) + .map_err(anyhow::Error::msg)?; + let service = service.context("ECS service identity unexpectedly absent")?; + let service_arn = service.service_arn().filter(|value| !value.trim().is_empty()) + .context("ECS service response missing service ARN")?.to_string(); + let cluster_arn = service.cluster_arn().filter(|value| !value.trim().is_empty()) + .context("ECS service response missing canonical cluster ARN")?.to_string(); + let service_created_at = service + .created_at() + .map(|created_at| created_at.as_nanos()) + .context("ECS service response missing createdAt incarnation")?; + let registry_arn = match service.service_registries() { + [] => None, + [registry] => Some( + registry + .registry_arn() + .filter(|value| !value.trim().is_empty()) + .context("ECS service registry is missing its exact ARN")? + .to_string(), + ), + _ => anyhow::bail!( + "ECS service has multiple registry identities; refusing ambiguous dependent cleanup" + ), + }; + let api_id = match registry_arn.as_deref() { + Some(registry_arn) => crate::ingress::resolve_api_id_for_registry( + aws_config, namespace, name, registry_arn, + ).await?, + None => None, + }; + let checkpoint = DeleteCheckpoint { + version: DELETE_CHECKPOINT_VERSION, + namespace: namespace.to_string(), + name: name.to_string(), + bucket: bucket.to_string(), + partition, + account, + region, + requested_cluster: cluster.to_string(), + cluster_arn, + service_arn, + service_created_at, + registry_arn, + api_id, + }; + validate_checkpoint( + &checkpoint, + namespace, + name, + cluster, + bucket, + &checkpoint.partition, + &checkpoint.account, + &checkpoint.region, + )?; + save_checkpoint(s3, &checkpoint).await?; + Ok(PreparedIdentity::Exact(Box::new((checkpoint, state)))) +} - println!("Deleting {name}..."); +fn validate_service_incarnation(expected: i128, actual: Option) -> Result<()> { + if actual != Some(expected) { + anyhow::bail!( + "ECS response belongs to a recreated same-name service (createdAt mismatch)" + ); + } + Ok(()) +} + +fn validate_returned_service( + checkpoint: &DeleteCheckpoint, + service: &aws_sdk_ecs::types::Service, +) -> Result<()> { + if service.service_arn() != Some(checkpoint.service_arn.as_str()) + || service.cluster_arn() != Some(checkpoint.cluster_arn.as_str()) + { + anyhow::bail!("ECS response returned a conflicting service identity"); + } + let created_at = service + .created_at() + .map(|created_at| created_at.as_nanos()); + validate_service_incarnation(checkpoint.service_created_at, created_at)?; + let registry_arn = match service.service_registries() { + [] => None, + [registry] => Some( + registry + .registry_arn() + .filter(|value| !value.trim().is_empty()) + .context("ECS response returned a registry without an ARN")?, + ), + _ => anyhow::bail!("ECS response returned multiple registry identities"), + }; + if registry_arn != checkpoint.registry_arn.as_deref() { + anyhow::bail!("ECS response returned a conflicting registry identity"); + } + Ok(()) +} - let describe_response = ecs +async fn refresh_checkpointed_ecs( + ecs: &aws_sdk_ecs::Client, + checkpoint: &DeleteCheckpoint, + context: &str, +) -> Result { + let response = ecs .describe_services() - .cluster(cluster) - .services(&service_name) + .cluster(&checkpoint.cluster_arn) + .services(&checkpoint.service_arn) .send() .await - .context("failed to describe ECS service before delete")?; - let service = describe_response.services().first(); - let registry_arn: Option = service.and_then(|service| { - service - .service_registries() - .first() - .and_then(|registry| registry.registry_arn()) - .map(str::to_owned) - }); - let service_status = service.and_then(|service| service.status()); - let delete_phase = ecs_delete_phase(service_status)?; - let service_needs_delete = delete_phase == EcsDeletePhase::Delete; - let service_is_draining = delete_phase == EcsDeletePhase::Drain; - - if service_needs_delete { - let _ = ecs - .update_service() - .cluster(cluster) - .service(&service_name) - .desired_count(0) - .send() - .await; - println!(" āœ“ Scaled to 0"); + .with_context(|| context.to_string())?; + let reasons = failure_reasons(&response); + let service = single_described_service(&response)?; + if let Some(service) = service { + validate_returned_service(checkpoint, service)?; + } + classify_service_identity( + true, + service.is_some(), + service.and_then(|service| service.status()), + &reasons, + ) + .map_err(anyhow::Error::msg) +} + +async fn delete_checkpointed_ecs( + ecs: &aws_sdk_ecs::Client, + checkpoint: &DeleteCheckpoint, + state: ServiceIdentityState, +) -> Result<()> { + if state == ServiceIdentityState::RetryGone { + println!(" āœ“ ECS service already absent; resuming exact dependent cleanup"); + return Ok(()); + } + + let response = ecs + .describe_services() + .cluster(&checkpoint.cluster_arn) + .services(&checkpoint.service_arn) + .send() + .await + .context("failed to refresh ECS service before mutation")?; + let reasons = failure_reasons(&response); + let service = single_described_service(&response)?; + if let Some(service) = service { + validate_returned_service(checkpoint, service)?; + } + let state = classify_service_identity( + true, + service.is_some(), + service.and_then(|service| service.status()), + &reasons, + ) + .map_err(anyhow::Error::msg)?; + if state == ServiceIdentityState::RetryGone { + return Ok(()); + } + let service = service.context("checkpointed ECS service disappeared ambiguously")?; + let delete_phase = ecs_delete_phase(service.status())?; + if delete_phase == EcsDeletePhase::Delete { match ecs - .delete_service() - .cluster(cluster) - .service(&service_name) - .force(true) + .update_service() + .cluster(&checkpoint.cluster_arn) + .service(&checkpoint.service_arn) + .desired_count(0) .send() .await { - Ok(_) => println!(" āœ“ ECS service deleted"), + Ok(_) => { + println!(" āœ“ Scaled to 0"); + // The service may disappear or be recreated between scale and + // delete. Re-describe the exact checkpointed ARN/incarnation + // immediately before issuing delete_service. + match refresh_checkpointed_ecs( + ecs, + checkpoint, + "failed to revalidate ECS service before delete", + ) + .await? + { + ServiceIdentityState::RetryGone => { + println!(" āœ“ ECS service disappeared after scaling; skipping delete") + } + ServiceIdentityState::Live => { + match ecs + .delete_service() + .cluster(&checkpoint.cluster_arn) + .service(&checkpoint.service_arn) + .force(true) + .send() + .await + { + Ok(_) => println!(" āœ“ ECS service deleted"), + Err(error) if error.code() == Some("ServiceNotFoundException") => { + println!(" āœ“ ECS service disappeared while deleting") + } + Err(error) => { + return Err(error).context("failed to delete ECS service"); + } + } + } + } + } Err(error) if error.code() == Some("ServiceNotFoundException") => { - println!(" āœ“ ECS service already absent") + // A successful scale is not a prerequisite for exact polling; + // never follow this response with delete_service. + println!(" āœ“ ECS service disappeared while scaling; skipping delete") } - Err(error) => return Err(error).context("failed to delete ECS service"), + Err(error) => return Err(error).context("failed to scale ECS service to zero"), } - } else if service_is_draining { - println!(" āœ“ ECS service is already draining; resuming delete cleanup"); } else { - println!(" āœ“ ECS service already absent; resuming dependent cleanup"); - } - - if service_needs_delete || service_is_draining { - const DRAIN_POLL_ATTEMPTS: u32 = 12; - const DRAIN_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); - eprint!(" ā³ Waiting for drain to complete..."); - for attempt in 0..DRAIN_POLL_ATTEMPTS { - let response = ecs - .describe_services() - .cluster(cluster) - .services(&service_name) - .send() - .await; - let is_gone = match response { - Ok(response) => response - .services() - .first() - .map(|service| service.status() == Some("INACTIVE")) - .unwrap_or(true), - Err(error) => { - eprintln!("\n āš  describe_services error (retrying): {error}"); - false + println!(" āœ“ ECS service is already draining; resuming delete cleanup"); + } + + const DRAIN_POLL_ATTEMPTS: u32 = 12; + const DRAIN_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); + eprint!(" ā³ Waiting for drain to complete..."); + for attempt in 0..DRAIN_POLL_ATTEMPTS { + let complete = match ecs.describe_services() + .cluster(&checkpoint.cluster_arn).services(&checkpoint.service_arn) + .send().await { + Ok(response) => { + let reasons = failure_reasons(&response); + let service = single_described_service(&response)?; + if let Some(service) = service { + validate_returned_service(checkpoint, service)?; } - }; - if is_gone { - if attempt == 0 { - eprintln!(" done (immediate)"); - } else { - let elapsed = u64::from(attempt) * DRAIN_POLL_INTERVAL.as_secs(); - eprintln!(" done ({elapsed}s)"); + match classify_service_identity( + true, + service.is_some(), + service.and_then(|service| service.status()), + &reasons, + ) { + Ok(ServiceIdentityState::RetryGone) => true, + Ok(ServiceIdentityState::Live) => false, + Err(error) => { + eprintln!("\n āš  ambiguous DescribeServices response (retrying): {error}"); + false + } } - break; } - if attempt == DRAIN_POLL_ATTEMPTS - 1 { - eprintln!(" timed out (service may still be draining)"); - } else { + Err(error) => { + eprintln!("\n āš  describe_services error (retrying): {error}"); + false + } + }; + match drain_poll_action(complete, attempt, DRAIN_POLL_ATTEMPTS) { + DrainPollAction::Complete => { eprintln!(" done"); return Ok(()); } + DrainPollAction::Retry => { eprint!("."); tokio::time::sleep(DRAIN_POLL_INTERVAL).await; } + DrainPollAction::TimedOut => anyhow::bail!( + "checkpointed ECS service did not reach an unambiguous absent state; dependent cleanup was not started (safe to retry)" + ), } } + unreachable!() +} - if let Err(error) = - crate::ingress::teardown(aws_config, namespace, name, registry_arn.as_deref()).await - { - eprintln!(" āš  ingress teardown skipped: {error}"); - } - if let Err(error) = crate::ingress::delete_api(aws_config, namespace, name).await { - eprintln!(" āš  HTTP API cleanup skipped: {error}"); - } - - let mut cleanup_failures = Vec::new(); +async fn cleanup_s3( + s3: &aws_sdk_s3::Client, bucket: &str, namespace: &str, name: &str, +) -> Result<()> { let manifest_key = format!("manifests/{namespace}/{name}.yaml"); - match s3 - .delete_object() - .bucket(bucket) - .key(&manifest_key) - .send() - .await - { - Ok(_) => println!(" āœ“ Manifest removed from S3"), - Err(error) => cleanup_failures.push(format!( - "failed to delete s3://{bucket}/{manifest_key}: {error}" - )), - } + s3.delete_object().bucket(bucket).key(&manifest_key).send().await + .with_context(|| format!("failed to delete s3://{bucket}/{manifest_key}"))?; + println!(" āœ“ Manifest removed from S3"); let artifact_prefix = format!("artifacts/{namespace}/{name}/"); let mut continuation_token = None; loop { - let response = match s3 - .list_objects_v2() - .bucket(bucket) - .prefix(&artifact_prefix) - .set_continuation_token(continuation_token) - .send() - .await - { - Ok(response) => response, - Err(error) => { - cleanup_failures.push(format!( - "failed to list config artifacts under s3://{bucket}/{artifact_prefix}: {error}" - )); - break; - } - }; + let response = s3.list_objects_v2().bucket(bucket).prefix(&artifact_prefix) + .set_continuation_token(continuation_token).send().await + .with_context(|| format!( + "failed to list config artifacts under s3://{bucket}/{artifact_prefix}" + ))?; for object in response.contents() { if let Some(key) = object.key() { - if let Err(error) = s3 - .delete_object() - .bucket(bucket) - .key(key) - .send() - .await - { - cleanup_failures - .push(format!("failed to delete s3://{bucket}/{key}: {error}")); - } + s3.delete_object().bucket(bucket).key(key).send().await + .with_context(|| format!("failed to delete s3://{bucket}/{key}"))?; } } continuation_token = response.next_continuation_token().map(str::to_owned); - if continuation_token.is_none() { - break; - } + if continuation_token.is_none() { break; } } - if cleanup_failures.is_empty() { - println!(" āœ“ Config artifacts removed from S3"); - } else { - anyhow::bail!( - "post-delete cleanup incomplete (safe to retry): {}", - cleanup_failures.join("; ") - ); + println!(" āœ“ Config artifacts removed from S3"); + Ok(()) +} + +async fn run_with_bucket( + aws_config: &aws_config::SdkConfig, + resource: &str, + name: &str, + cluster: &str, + namespace: &str, + bucket: &str, +) -> Result> { + if resource != "oabservice" { + anyhow::bail!("unknown resource type: {resource}. Use 'oabservice'"); } + let ecs = aws_sdk_ecs::Client::new(aws_config); + let s3 = aws_sdk_s3::Client::new(aws_config); + crate::apply::ensure_no_pending_ingress_teardown(&s3, bucket, namespace, name) + .await?; + println!("Deleting {name}..."); + match prepare_identity( + aws_config, &ecs, &s3, namespace, name, cluster, bucket, + ) + .await? { + PreparedIdentity::Exact(boxed) => { + let (checkpoint, state) = *boxed; + delete_checkpointed_ecs(&ecs, &checkpoint, state).await?; + crate::ingress::delete_exact( + aws_config, + namespace, + name, + checkpoint.api_id.as_deref(), + checkpoint.registry_arn.as_deref(), + &checkpoint.partition, + &checkpoint.account, + &checkpoint.region, + ) + .await?; + cleanup_s3(&s3, bucket, namespace, name).await?; + let key = checkpoint_key(namespace, name); + s3.delete_object().bucket(bucket).key(&key).send().await + .with_context(|| format!( + "cleanup completed but failed to remove s3://{bucket}/{key}" + ))?; + println!(" āœ“ Delete checkpoint removed"); + } + } println!("\nāœ“ {name} deleted"); - Ok(()) + Ok(Vec::new()) } - #[cfg(test)] mod tests { use super::*; + fn test_sdk_config() -> aws_config::SdkConfig { + aws_config::SdkConfig::builder() + .behavior_version(aws_config::BehaviorVersion::latest()) + .build() + } + + fn checkpoint() -> DeleteCheckpoint { + DeleteCheckpoint { + version: DELETE_CHECKPOINT_VERSION, + namespace: "prod".to_string(), + name: "bot".to_string(), + bucket: "control-plane".to_string(), + partition: "aws".to_string(), + account: "123456789012".to_string(), + region: "us-east-1".to_string(), + requested_cluster: "cluster".to_string(), + cluster_arn: "arn:aws:ecs:us-east-1:123456789012:cluster/cluster".to_string(), + service_arn: "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-bot".to_string(), + service_created_at: 1_700_000_000, + registry_arn: Some( + "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-123".to_string(), + ), + api_id: Some("api-123".to_string()), + } + } + + #[tokio::test] + async fn delete_services_rejects_empty_target_set() { + let cfg = test_sdk_config(); + let err = delete_services( + &cfg, + &[], + &DeleteOptions::new("cluster").with_control_plane_bucket("control-plane"), + ) + .await + .unwrap_err(); + assert_eq!(err.kind, DeleteErrorKind::Validation); + assert!(err.to_string().contains("empty target set"), "{err}"); + } + + #[tokio::test] + async fn delete_services_rejects_empty_cluster() { + let cfg = test_sdk_config(); + let targets = [DeleteTarget::new("prod", "bot")]; + let err = delete_services( + &cfg, + &targets, + &DeleteOptions::new("").with_control_plane_bucket("control-plane"), + ) + .await + .unwrap_err(); + assert_eq!(err.kind, DeleteErrorKind::Validation); + assert!(err.to_string().contains("cluster must not be empty"), "{err}"); + } + #[test] - fn delete_phase_requests_delete_only_for_active_service() { + fn delete_options_support_optional_bucket_override() { + let options = DeleteOptions::new("cluster").with_control_plane_bucket("control-plane"); + assert_eq!(options.cluster, "cluster"); + assert_eq!(options.control_plane_bucket.as_deref(), Some("control-plane")); + } + + #[tokio::test] + async fn delete_services_rejects_blank_target_fields() { + let cfg = test_sdk_config(); + let targets = [DeleteTarget::new("prod", " ")]; + let err = delete_services( + &cfg, + &targets, + &DeleteOptions::new("cluster").with_control_plane_bucket("control-plane"), + ) + .await + .unwrap_err(); + assert_eq!(err.kind, DeleteErrorKind::Validation); + } + + #[test] + fn initial_describe_failures_and_missing_service_fail_closed() { + assert!(classify_service_identity(false, false, None, &["MISSING"]).is_err()); + assert!(classify_service_identity(false, false, None, &["ACCESS_DENIED"]).is_err()); + assert!(classify_service_identity(false, false, None, &[]).is_err()); + assert!(classify_service_identity(false, true, Some("INACTIVE"), &[]).is_err()); + assert!(classify_service_identity(false, true, None, &[]).is_err()); + } + + #[test] + fn exact_retry_checkpoint_authorizes_only_unambiguous_missing_identity() { assert_eq!( - ecs_delete_phase(Some("ACTIVE")).unwrap(), - EcsDeletePhase::Delete + classify_service_identity(true, false, None, &["MISSING"]).unwrap(), + ServiceIdentityState::RetryGone ); assert_eq!( - ecs_delete_phase(Some("DRAINING")).unwrap(), - EcsDeletePhase::Drain + classify_service_identity(true, true, Some("INACTIVE"), &[]).unwrap(), + ServiceIdentityState::RetryGone ); + assert!(classify_service_identity(true, false, None, &["ACCESS_DENIED"]).is_err()); + assert!(classify_service_identity( + true, + false, + None, + &["MISSING", "ACCESS_DENIED"] + ) + .is_err()); + assert!(classify_service_identity( + true, + true, + Some("DRAINING"), + &["MISSING"] + ) + .is_err()); + assert!(classify_service_identity( + true, + false, + None, + &["MISSING", "MISSING"] + ) + .is_err()); + assert!(classify_service_identity(true, false, None, &[]).is_err()); + assert!(classify_service_identity(true, true, None, &[]).is_err()); + assert!(validate_service_incarnation(1_700_000_000, Some(1_700_000_001)).is_err()); + assert!(validate_service_incarnation(1_700_000_000, None).is_err()); + assert!(validate_service_incarnation(1_700_000_000, Some(1_700_000_000)).is_ok()); assert_eq!( - ecs_delete_phase(Some("INACTIVE")).unwrap(), - EcsDeletePhase::Cleanup + classify_service_identity(true, true, Some("DRAINING"), &[]).unwrap(), + ServiceIdentityState::Live ); - assert_eq!(ecs_delete_phase(None).unwrap(), EcsDeletePhase::Cleanup); } #[test] - fn delete_phase_rejects_unknown_status() { + fn checkpoint_rejects_boundary_mismatch_and_accepts_exact_retry() { + let checkpoint = checkpoint(); + validate_checkpoint( + &checkpoint, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .unwrap(); + validate_checkpoint( + &checkpoint, + "prod", + "bot", + &checkpoint.cluster_arn, + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .unwrap(); + assert!(validate_checkpoint( + &checkpoint, + "prod", + "bot", + "cluster", + "other-bucket", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + assert!(validate_checkpoint( + &checkpoint, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "999999999999", + "us-east-1", + ) + .is_err()); + + let mut mismatched_cluster = checkpoint.clone(); + mismatched_cluster.requested_cluster = "other-cluster".to_string(); + assert!(validate_checkpoint( + &mismatched_cluster, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + assert!(validate_checkpoint( + &checkpoint, + "prod", + "bot", + "other-cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + + let mut mismatched_partition = checkpoint.clone(); + mismatched_partition.partition = "aws-cn".to_string(); + assert!(validate_checkpoint( + &mismatched_partition, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + + let mut invalid_dependency = checkpoint.clone(); + invalid_dependency.registry_arn = None; + assert!(validate_checkpoint( + &invalid_dependency, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + + invalid_dependency.api_id = None; + invalid_dependency.registry_arn = Some("srv-not-an-arn".to_string()); + assert!(validate_checkpoint( + &invalid_dependency, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + + let mut invalid_incarnation = checkpoint.clone(); + invalid_incarnation.service_created_at = 0; + assert!(validate_checkpoint( + &invalid_incarnation, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + } + + #[test] + fn checkpoint_rejects_arn_boundary_mismatches() { + let cases = [ + ( + "arn:aws:ecs:us-east-1:999999999999:cluster/cluster", + "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-bot", + "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-123", + ), + ( + "arn:aws:ecs:us-west-2:123456789012:cluster/cluster", + "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-bot", + "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-123", + ), + ( + "arn:aws:ecs:us-east-1:123456789012:cluster/cluster", + "arn:aws:ecs:us-east-1:123456789012:service/other/oab-prod-bot", + "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-123", + ), + ( + "arn:aws:ecs:us-east-1:123456789012:cluster/cluster", + "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-other", + "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-123", + ), + ( + "arn:aws:ecs:us-east-1:123456789012:cluster/cluster", + "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-bot", + "arn:aws:servicediscovery:us-east-1:999999999999:service/srv-123", + ), + ( + "arn:aws:ecs:us-east-1:123456789012:cluster/cluster", + "arn:aws:ecs:us-east-1:123456789012:service/cluster/oab-prod-bot", + "arn:aws:servicediscovery:us-west-2:123456789012:service/srv-123", + ), + ( + "arn:aws:ecs:us-east-1:123456789012:cluster/cluster", + "arn:aws-cn:ecs:us-east-1:123456789012:service/cluster/oab-prod-bot", + "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-123", + ), + ]; + + for (cluster_arn, service_arn, registry_arn) in cases { + let mut candidate = checkpoint(); + candidate.cluster_arn = cluster_arn.to_string(); + candidate.service_arn = service_arn.to_string(); + candidate.registry_arn = Some(registry_arn.to_string()); + assert!(validate_checkpoint( + &candidate, + "prod", + "bot", + "cluster", + "control-plane", + "aws", + "123456789012", + "us-east-1", + ) + .is_err()); + } + } + + #[test] + fn delete_target_derives_ecs_service_name() { + assert_eq!( + DeleteTarget::new("prod", "nest-my-oab").ecs_service_name(), + "oab-prod-nest-my-oab" + ); + } + + #[tokio::test] + async fn ambiguous_namespace_delimiter_is_rejected_before_aws() { + let opts = DeleteOptions::new("cluster"); + let accepted = [DeleteTarget::new("prod", "team-bot")]; + let rejected = [DeleteTarget::new("prod-team", "bot")]; + assert_eq!( + accepted[0].ecs_service_name(), + rejected[0].ecs_service_name() + ); + assert!(validate_delete_request(&accepted, &opts).is_ok()); + + let error = delete_services(&test_sdk_config(), &rejected, &opts) + .await + .unwrap_err(); + assert_eq!(error.kind, DeleteErrorKind::Validation); + assert!(error.to_string().contains("must not contain '-'")); + } + + #[test] + fn later_failure_preserves_the_completed_partial_report() { + let first = DeleteTarget::new("prod", "first"); + let second = DeleteTarget::new("prod", "second"); + let report = record_delete_result(DeleteReport::default(), &first, Ok(Vec::new())) + .expect("first target should complete"); + let error = record_delete_result( + report, + &second, + Err(anyhow::anyhow!("synthetic teardown failure")), + ) + .unwrap_err(); + + assert_eq!(error.kind, DeleteErrorKind::Teardown); + assert_eq!(error.failed_service.as_ref(), Some(&second)); + assert_eq!(error.completed.services.len(), 1); + assert_eq!(error.completed.services[0].name, "first"); + } + + #[test] + fn pending_apply_ingress_teardown_blocks_delete_completion() { + assert!(crate::apply::require_no_pending_ingress_teardown(false).is_ok()); + let error = crate::apply::require_no_pending_ingress_teardown(true).unwrap_err(); + assert!(error.to_string().contains("re-run the ingress-free apply")); + } + + #[test] + fn drain_poll_action_requires_completion_before_cleanup() { + assert_eq!(drain_poll_action(true, 0, 12), DrainPollAction::Complete); + assert_eq!(drain_poll_action(false, 0, 12), DrainPollAction::Retry); + assert_eq!(drain_poll_action(false, 11, 12), DrainPollAction::TimedOut); + } + + #[test] + fn delete_phase_requests_delete_only_for_active_service() { + assert_eq!(ecs_delete_phase(Some("ACTIVE")).unwrap(), EcsDeletePhase::Delete); + assert_eq!(ecs_delete_phase(Some("DRAINING")).unwrap(), EcsDeletePhase::Drain); + assert_eq!(ecs_delete_phase(Some("INACTIVE")).unwrap(), EcsDeletePhase::Cleanup); + assert_eq!(ecs_delete_phase(None).unwrap(), EcsDeletePhase::Cleanup); assert!(ecs_delete_phase(Some("UNKNOWN")).is_err()); } } diff --git a/operator/src/ingress.rs b/operator/src/ingress.rs index 57761ea3d..29175bb66 100644 --- a/operator/src/ingress.rs +++ b/operator/src/ingress.rs @@ -26,6 +26,7 @@ use crate::manifest::{Ingress, OABServiceManifest}; use anyhow::{Context, Result}; +use aws_sdk_apigatewayv2::error::ProvideErrorMetadata; use aws_sdk_apigatewayv2::types::{ConnectionType, IntegrationType, ProtocolType}; use aws_sdk_servicediscovery::types::{DnsConfig, DnsRecord, RecordType}; use std::collections::HashMap; @@ -266,11 +267,49 @@ fn has_stage_path_override(request_parameters: Option<&HashMap>) /// Extract the Cloud Map service ID from its ARN /// (`arn:aws:servicediscovery:::service/`). +/// +/// Test-only reference parser: every production path resolves IDs through +/// [`cloud_map_service_id_for_boundary`], which additionally pins the +/// caller's partition/account/region (exact-identity delete contract). +#[cfg(test)] fn cloud_map_service_id_from_arn(arn: &str) -> Option { - arn.rsplit('/') - .next() - .filter(|s| !s.is_empty()) - .map(|s| s.to_string()) + let mut parts = arn.splitn(6, ':'); + if parts.next() != Some("arn") + || parts.next().filter(|part| !part.is_empty()).is_none() + || parts.next() != Some("servicediscovery") + || parts.next().filter(|part| !part.is_empty()).is_none() + || parts.next().filter(|part| !part.is_empty()).is_none() + { + return None; + } + let resource = parts.next()?; + let service_id = resource.strip_prefix("service/")?; + if service_id.is_empty() || service_id.contains('/') { + return None; + } + Some(service_id.to_string()) +} + +fn cloud_map_service_id_for_boundary( + arn: &str, + expected_partition: &str, + expected_account: &str, + expected_region: &str, +) -> Option { + let mut parts = arn.splitn(6, ':'); + if parts.next() != Some("arn") + || parts.next() != Some(expected_partition) + || parts.next() != Some("servicediscovery") + || parts.next() != Some(expected_region) + || parts.next() != Some(expected_account) + { + return None; + } + let service_id = parts.next()?.strip_prefix("service/")?; + if service_id.is_empty() || service_id.contains('/') { + return None; + } + Some(service_id.to_string()) } /// Build the public webhook URL(s) from the API endpoint and paths. @@ -283,204 +322,227 @@ fn webhook_urls(api_endpoint: &str, paths: &[String]) -> Vec { .collect() } -/// Best-effort teardown of the *per-bot ingress wiring* for `namespace/name`: -/// its routes, integration, and stage on the per-bot HTTP API, plus its Cloud -/// Map service. Deliberately does NOT delete the HTTP API resource itself ā€” -/// only what points at the now-gone task ā€” so the API's `api-id` (and thus the -/// public webhook URL's hostname) survives an ECS-service recreate cycle. Use -/// [`delete_api`] separately when the bot is being permanently removed. +/// Exact-identity teardown of per-bot API Gateway wiring for apply-time ingress +/// removal. The ECS registry ARN selects the unique same-name HTTP API whose +/// integration URI equals that ARN. Without an ECS registry ARN this function +/// returns a warning and leaves resources untouched; it never falls back to a +/// name-only match. /// -/// The shared resources (the VPC Link and the security-group inbound rule) are -/// intentionally left in place since other bots may still use them. Safe to -/// call for bots that never had ingress ā€” it simply finds nothing and returns. -/// Errors that prevent teardown are propagated. Degraded cleanup that can be -/// completed manually is returned as warning text so apply can include it in -/// its structured report while the CLI still renders it. +/// The HTTP API resource itself is retained so its endpoint survives if ingress +/// is re-enabled. Cloud Map must be deleted separately with +/// [`delete_cloud_map_exact`] only after ECS has detached the registry. Shared +/// VPC Link and security-group resources are also left in place. pub async fn teardown( config: &aws_config::SdkConfig, namespace: &str, name: &str, known_registry_arn: Option<&str>, ) -> Result> { - let mut warnings = Vec::new(); - let service_name = format!("oab-{namespace}-{name}"); - let api = aws_sdk_apigatewayv2::Client::new(config); + let Some(registry_arn) = known_registry_arn else { + let warning = format!( + "ingress teardown left resources untouched for {namespace}/{name}: ECS returned no exact registry ARN" + ); + eprintln!(" āš  {warning}"); + return Ok(vec![warning]); + }; - // ā”€ā”€ API Gateway: strip routes + integration + stage, keep the API itself ā”€ - if let Some((api_id, _)) = find_api(&api, &api_name(namespace, name)).await? { - // Delete all routes on this API first (integrations can't be deleted - // while a route still targets them). - let mut route_ids = Vec::new(); - let mut next: Option = None; - loop { - let mut req = api.get_routes().api_id(&api_id); - if let Some(t) = &next { - req = req.next_token(t); - } - let resp = req.send().await.context("failed to list routes")?; - for r in resp.items() { - if let Some(id) = r.route_id() { - route_ids.push(id.to_string()); - } - } - match resp.next_token() { - Some(t) => next = Some(t.to_string()), - None => break, - } - } - for route_id in &route_ids { - if let Err(error) = api - .delete_route() - .api_id(&api_id) - .route_id(route_id) - .send() - .await - { - let warning = format!( - "failed to delete ingress route {route_id} from HTTP API {api_id}: {error}" - ); - eprintln!(" āš  {warning}"); - warnings.push(warning); - } + let api_id = resolve_api_id_for_registry(config, namespace, name, registry_arn).await?; + if let Some(api_id) = api_id.as_deref() { + clear_api_wiring(config, api_id).await?; + eprintln!(" āœ“ Cleared exact ingress wiring on HTTP API {api_id}"); + } + Ok(Vec::new()) +} + +async fn clear_api_wiring(config: &aws_config::SdkConfig, api_id: &str) -> Result<()> { + let api = aws_sdk_apigatewayv2::Client::new(config); + let mut route_ids = Vec::new(); + let mut next = None; + loop { + let mut request = api.get_routes().api_id(api_id); + if let Some(token) = &next { request = request.next_token(token); } + let response = request.send().await.context("failed to list exact API routes")?; + route_ids.extend(response.items().iter().filter_map(|route| route.route_id().map(str::to_owned))); + next = response.next_token().map(str::to_owned); + if next.is_none() { break; } + } + for route_id in route_ids { + match api.delete_route().api_id(api_id).route_id(&route_id).send().await { + Ok(_) => {} + Err(error) if error.code() == Some("NotFoundException") => {} + Err(error) => return Err(error).context("failed to delete exact API route"), } + } - // Delete integrations (there's normally just one, but clean up all). - let mut integration_ids = Vec::new(); - let mut next: Option = None; - loop { - let mut req = api.get_integrations().api_id(&api_id); - if let Some(t) = &next { - req = req.next_token(t); - } - let resp = req.send().await.context("failed to list integrations")?; - for i in resp.items() { - if let Some(id) = i.integration_id() { - integration_ids.push(id.to_string()); - } - } - match resp.next_token() { - Some(t) => next = Some(t.to_string()), - None => break, - } + let mut integration_ids = Vec::new(); + let mut next = None; + loop { + let mut request = api.get_integrations().api_id(api_id); + if let Some(token) = &next { request = request.next_token(token); } + let response = request.send().await.context("failed to list exact API integrations")?; + integration_ids.extend(response.items().iter().filter_map(|item| item.integration_id().map(str::to_owned))); + next = response.next_token().map(str::to_owned); + if next.is_none() { break; } + } + for integration_id in integration_ids { + match api.delete_integration().api_id(api_id).integration_id(&integration_id).send().await { + Ok(_) => {} + Err(error) if error.code() == Some("NotFoundException") => {} + Err(error) => return Err(error).context("failed to delete exact API integration"), } - for integration_id in &integration_ids { - if let Err(error) = api - .delete_integration() - .api_id(&api_id) - .integration_id(integration_id) - .send() - .await - { - let warning = format!( - "failed to delete ingress integration {integration_id} from HTTP API {api_id}: {error}" - ); - eprintln!(" āš  {warning}"); - warnings.push(warning); + } + match api.delete_stage().api_id(api_id).stage_name(STAGE_NAME).send().await { + Ok(_) => {} + Err(error) if error.code() == Some("NotFoundException") => {} + Err(error) => return Err(error).context("failed to delete exact API stage"), + } + Ok(()) +} + +async fn delete_cloud_map_by_arn( + config: &aws_config::SdkConfig, + registry_arn: &str, + service_name: &str, + expected_boundary: (&str, &str, &str), +) -> Result<()> { + let (partition, account, region) = expected_boundary; + let service_id = cloud_map_service_id_for_boundary( + registry_arn, + partition, + account, + region, + ) + .context("Cloud Map service ARN is outside the exact delete boundary")?; + let discovery = aws_sdk_servicediscovery::Client::new(config); + let mut last_error = None; + for attempt in 0..6 { + if attempt > 0 { tokio::time::sleep(std::time::Duration::from_secs(5)).await; } + match discovery.delete_service().id(&service_id).send().await { + Ok(_) => { + eprintln!(" āœ“ Deleted Cloud Map service: {service_name} ({service_id})"); + return Ok(()); } + Err(error) if error.code() == Some("ResourceNotFoundException") => return Ok(()), + Err(error) => last_error = Some(error), } + } + Err(anyhow::Error::new(last_error.expect("delete attempt always records an error"))) + .with_context(|| format!("failed to delete exact Cloud Map service {service_id}")) +} - if let Err(error) = api - .delete_stage() - .api_id(&api_id) - .stage_name(STAGE_NAME) - .send() - .await - { - let warning = format!( - "failed to delete ingress stage {STAGE_NAME} from HTTP API {api_id}: {error}" - ); - eprintln!(" āš  {warning}"); - warnings.push(warning); - } - if warnings.is_empty() { - eprintln!( - " āœ“ Cleared ingress wiring on HTTP API {} ({} route(s), {} integration(s)) ā€” API itself kept so its URL survives a recreate", - api_name(namespace, name), - route_ids.len(), - integration_ids.len() - ); - } else { - eprintln!( - " āš  Ingress wiring cleanup on HTTP API {} completed with {} warning(s)", - api_name(namespace, name), - warnings.len() - ); - } +/// Delete the exact Cloud Map service selected by an ECS registry ARN. +/// Apply calls this only after it has observed the registry detached from ECS. +pub(crate) async fn delete_cloud_map_exact( + config: &aws_config::SdkConfig, + registry_arn: &str, + service_name: &str, +) -> Result<()> { + let identity = aws_sdk_sts::Client::new(config) + .get_caller_identity() + .send() + .await + .context("failed to identify Cloud Map delete caller")?; + let account = identity + .account() + .context("STS response missing caller account")?; + let caller_arn = identity.arn().context("STS response missing caller ARN")?; + let mut arn_parts = caller_arn.splitn(3, ':'); + if arn_parts.next() != Some("arn") { + anyhow::bail!("STS returned an invalid caller ARN"); } + let partition = arn_parts + .next() + .filter(|value| !value.is_empty()) + .context("STS caller ARN is missing its partition")?; + let region = config + .region() + .map(|region| region.as_ref().to_string()) + .context("AWS region must be resolved before Cloud Map delete")?; + delete_cloud_map_by_arn( + config, + registry_arn, + service_name, + (partition, account, ®ion), + ) + .await +} +fn integration_uri_matches(actual: Option<&str>, expected: &str) -> bool { + actual == Some(expected) +} - // ā”€ā”€ Cloud Map: delete the per-bot service (needs no live instances) ā”€ā”€ā”€ā”€ā”€ā”€ - // Prefer resolving the exact service from the ECS service's own registry - // ARN (passed by the caller when known) over a name-only account-wide - // scan ā€” two bots with the same namespace/name in different VPCs (e.g. - // staging vs. prod sharing an account) would otherwise collide and the - // wrong one could be deleted. - let sd = aws_sdk_servicediscovery::Client::new(config); - let service_id: Option = if let Some(arn) = known_registry_arn { - cloud_map_service_id_from_arn(arn) - } else { - let mut found: Option = None; - let mut pages = sd.list_services().into_paginator().send(); - 'svc: while let Some(page) = pages.next().await { - let page = page.context("failed to list Cloud Map services")?; - for s in page.services() { - if s.name() == Some(service_name.as_str()) { - found = s.id().map(|x| x.to_string()); - break 'svc; - } - } - } - found +/// Resolve a same-name API only when the name identifies exactly one resource +/// and that API has an integration URI equal to the exact ECS registry ARN. +/// Duplicate names are rejected before inspecting integrations, even if only +/// one duplicate currently points at the registry. +pub(crate) async fn resolve_api_id_for_registry( + config: &aws_config::SdkConfig, + namespace: &str, + name: &str, + registry_arn: &str, +) -> Result> { + let api = aws_sdk_apigatewayv2::Client::new(config); + let expected_name = api_name(namespace, name); + let candidates = find_apis(&api, &expected_name).await?; + let Some((api_id, _)) = select_unique_named_api(&expected_name, &candidates)? else { + return Ok(None); }; - if let Some(service_id) = service_id { - // ECS deregisters the task's Cloud Map instance asynchronously when a - // service scales to 0 / is deleted, so `delete_service` can fail with - // "still has registered instances" for a short window even though the - // task is already gone. Retry briefly instead of giving up on the - // first attempt ā€” this is the common case, not an edge case. - let mut last_err = None; - let mut deleted = false; - for attempt in 0..6 { - if attempt > 0 { - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - } - match sd.delete_service().id(&service_id).send().await { - Ok(_) => { - eprintln!(" āœ“ Deleted Cloud Map service: {service_name}"); - deleted = true; - break; - } - Err(e) => last_err = Some(e), - } + + let mut next = None; + let mut matched = false; + loop { + let mut request = api.get_integrations().api_id(&api_id); + if let Some(token) = &next { + request = request.next_token(token); } - if !deleted { - let warning = format!( - "Cloud Map service '{service_name}' was not deleted after retrying; it still has registered instances. Remove it manually with `aws servicediscovery delete-service --id {service_id}` ({})", - last_err.map(|error| error.to_string()).unwrap_or_default() - ); - eprintln!(" āš  {warning}"); - warnings.push(warning); + let response = request + .send() + .await + .with_context(|| format!("failed to list integrations for API {api_id}"))?; + matched |= response + .items() + .iter() + .any(|integration| integration_uri_matches(integration.integration_uri(), registry_arn)); + next = response.next_token().map(str::to_owned); + if next.is_none() { + break; } } - Ok(warnings) + Ok(matched.then_some(api_id)) } -/// Permanently delete the bot's per-bot HTTP API (`oab-webhook--`), -/// cascading its routes/integration/stage with it. This DESTROYS the `api-id` -/// and therefore the public webhook URL's hostname ā€” only call this when the -/// bot itself is being permanently removed (`oabctl delete`), never from the -/// `apply` recreate path, which relies on the API surviving so its URL stays -/// stable across an ECS-service recreate. -pub async fn delete_api(config: &aws_config::SdkConfig, namespace: &str, name: &str) -> Result<()> { - let api = aws_sdk_apigatewayv2::Client::new(config); - let name_str = api_name(namespace, name); - if let Some((api_id, _)) = find_api(&api, &name_str).await? { - match api.delete_api().api_id(&api_id).send().await { - Ok(_) => eprintln!(" āœ“ Deleted HTTP API: {name_str}"), - Err(e) => eprintln!(" āš  Failed to delete HTTP API {api_id}: {e}"), +/// Permanently delete only checkpointed dependent identities. Exact-ID +/// NotFound is idempotent success; every other failure is fatal so the caller +/// retains its durable checkpoint for retry. +#[allow(clippy::too_many_arguments)] +pub(crate) async fn delete_exact( + config: &aws_config::SdkConfig, + namespace: &str, + name: &str, + api_id: Option<&str>, + registry_arn: Option<&str>, + partition: &str, + account: &str, + region: &str, +) -> Result<()> { + if let Some(api_id) = api_id { + let api = aws_sdk_apigatewayv2::Client::new(config); + match api.delete_api().api_id(api_id).send().await { + Ok(_) => eprintln!(" āœ“ Deleted exact HTTP API: {api_id}"), + Err(error) if error.code() == Some("NotFoundException") => {} + Err(error) => return Err(error).context("failed to delete exact HTTP API"), } } + if let Some(registry_arn) = registry_arn { + delete_cloud_map_by_arn( + config, + registry_arn, + &format!("oab-{namespace}-{name}"), + (partition, account, region), + ) + .await?; + } Ok(()) } @@ -875,12 +937,21 @@ async fn ensure_api( Ok((id, endpoint)) } -/// Find an HTTP API by name, returning `(api_id, api_endpoint)`. +/// Find an HTTP API by name, rejecting duplicate names instead of selecting +/// whichever candidate AWS happens to return first. async fn find_api( api: &aws_sdk_apigatewayv2::Client, api_name: &str, ) -> Result> { - // apigatewayv2 has no smithy paginator for GetApis; page manually. + let candidates = find_apis(api, api_name).await?; + select_unique_named_api(api_name, &candidates) +} + +async fn find_apis( + api: &aws_sdk_apigatewayv2::Client, + api_name: &str, +) -> Result> { + let mut candidates = Vec::new(); let mut next: Option = None; loop { let mut req = api.get_apis(); @@ -888,20 +959,50 @@ async fn find_api( req = req.next_token(t); } let resp = req.send().await.context("failed to list APIs")?; - for a in resp.items() { - if a.name() == Some(api_name) { - let id = a.api_id().context("api missing id")?.to_string(); - let endpoint = a.api_endpoint().unwrap_or_default().to_string(); - return Ok(Some((id, endpoint))); + for candidate in resp.items() { + if is_matching_http_api( + candidate.name(), + candidate.protocol_type().map(ProtocolType::as_str), + api_name, + ) { + let id = candidate.api_id().context("api missing id")?.to_string(); + let endpoint = candidate.api_endpoint().unwrap_or_default().to_string(); + candidates.push((id, endpoint)); } } match resp.next_token() { - Some(t) => next = Some(t.to_string()), - None => return Ok(None), + Some(token) => next = Some(token.to_string()), + None => return Ok(candidates), } } } +fn is_matching_http_api( + candidate_name: Option<&str>, + candidate_protocol: Option<&str>, + expected_name: &str, +) -> bool { + candidate_name == Some(expected_name) && candidate_protocol == Some("HTTP") +} + +fn select_unique_named_api( + api_name: &str, + candidates: &[(String, String)], +) -> Result> { + match candidates { + [] => Ok(None), + [candidate] => Ok(Some(candidate.clone())), + _ => anyhow::bail!( + "multiple HTTP APIs named '{api_name}' exist; refusing first-match selection: {}", + candidates + .iter() + .map(|(id, _)| id.as_str()) + .collect::>() + .join(", ") + ), + } +} + async fn ensure_integration( api: &aws_sdk_apigatewayv2::Client, api_id: &str, @@ -1183,9 +1284,21 @@ mod tests { } #[test] - fn cloud_map_service_id_from_arn_rejects_empty() { + fn cloud_map_service_id_from_arn_rejects_non_service_arns() { assert_eq!(cloud_map_service_id_from_arn(""), None); - assert_eq!(cloud_map_service_id_from_arn("trailing/"), None); + assert_eq!(cloud_map_service_id_from_arn("srv-abc123"), None); + assert_eq!( + cloud_map_service_id_from_arn( + "arn:aws:servicediscovery:us-east-1:123456789012:namespace/ns-123" + ), + None + ); + assert_eq!( + cloud_map_service_id_from_arn( + "arn:aws:servicediscovery:us-east-1:123456789012:service/" + ), + None + ); } #[test] @@ -1236,4 +1349,57 @@ mod tests { vec!["https://abc123.example.com/prod/webhook/telegram"] ); } + + #[test] + fn api_candidates_require_http_protocol() { + let name = "oab-webhook-prod-bot"; + assert!(is_matching_http_api(Some(name), Some("HTTP"), name)); + assert!(!is_matching_http_api(Some(name), Some("WEBSOCKET"), name)); + assert!(!is_matching_http_api(Some("other"), Some("HTTP"), name)); + assert!(!is_matching_http_api(Some(name), None, name)); + } + + #[test] + fn duplicate_named_apis_fail_closed() { + let candidates = vec![ + ("api-a".to_string(), "https://a".to_string()), + ("api-b".to_string(), "https://b".to_string()), + ]; + assert!(select_unique_named_api("oab-webhook-prod-bot", &candidates).is_err()); + assert_eq!( + select_unique_named_api("oab-webhook-prod-bot", &candidates[..1]) + .unwrap() + .unwrap() + .0, + "api-a" + ); + } + + #[test] + fn sole_api_requires_exact_registry_integration_uri() { + let registry = "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-1"; + assert!(integration_uri_matches(Some(registry), registry)); + assert!(!integration_uri_matches( + Some("arn:aws:servicediscovery:us-east-1:123456789012:service/srv-other"), + registry + )); + assert!(!integration_uri_matches(None, registry)); + } + + #[test] + fn cloud_map_arn_requires_exact_boundary() { + let arn = "arn:aws:servicediscovery:us-east-1:123456789012:service/srv-1"; + assert_eq!( + cloud_map_service_id_for_boundary(arn, "aws", "123456789012", "us-east-1"), + Some("srv-1".to_string()) + ); + assert_eq!( + cloud_map_service_id_for_boundary(arn, "aws-cn", "123456789012", "us-east-1"), + None + ); + assert_eq!( + cloud_map_service_id_for_boundary(arn, "aws", "999999999999", "us-east-1"), + None + ); + } } diff --git a/operator/src/lib.rs b/operator/src/lib.rs index ced58b1dc..52ad9e4f6 100644 --- a/operator/src/lib.rs +++ b/operator/src/lib.rs @@ -1,7 +1,7 @@ //! Programmatic OAB manifest validation and ECS reconciliation. //! -//! The public facade intentionally contains only the manifest model and the -//! structured apply API. CLI implementation details and resource-management +//! The public facade contains the manifest model and the structured apply +//! and delete APIs. CLI implementation details and other resource-management //! helpers remain private. //! //! # Example @@ -42,9 +42,29 @@ //! an ARN require the apply caller to have `secretsmanager:DescribeSecret`, so //! oabctl can resolve the name to the full ARN required by ECS. //! -//! Programmatic apply never reads `~/.oabctl/config.toml`. Use -//! [`ApplyOptions::with_control_plane_bucket`] for an explicit bucket; otherwise -//! resolution uses `OAB_CONTROL_PLANE_BUCKET` and then the caller's AWS account. +//! Programmatic apply and delete never read `~/.oabctl/config.toml`. Use +//! [`ApplyOptions::with_control_plane_bucket`] or +//! [`DeleteOptions::with_control_plane_bucket`] for an explicit bucket; +//! otherwise both use the shared `OAB_CONTROL_PLANE_BUCKET` then caller-account +//! resolution chain. +//! +//! [`delete_services`] tears down by `namespace`+`name` ([`DeleteTarget`]). +//! Delete namespaces must not contain `-`; names may contain it, making the +//! existing `oab-{namespace}-{name}` resource identity injective for accepted +//! targets. Before ECS mutation delete durably records the resolved bucket, +//! caller partition/account/region, canonical cluster/service ARNs, the ECS +//! service `createdAt` incarnation, and exact ingress IDs in S3. Initial +//! identity ambiguity fails closed; retries use only checkpointed IDs and +//! remove the delete checkpoint on success. If apply has a pending exact +//! ingress-teardown checkpoint, delete fails before destructive mutation and +//! requires the caller to finish the ingress-free apply first, preserving that +//! cleanup identity. There is no name-only orphan cleanup fallback. +//! +//! Programmatic apply and delete do not provide an internal same-target lock. +//! Callers must serialize mutations for the same AWS account, Region, +//! control-plane bucket, ECS cluster, namespace, and name. After an accidental +//! overlap, stop concurrent writers, inspect any retained checkpoint, and +//! explicitly re-apply the desired state or retry delete. pub mod apply; mod bootstrap; @@ -52,7 +72,7 @@ mod cli; mod config; mod control_plane; mod create; -mod delete; +pub mod delete; mod get; mod ingress; pub mod manifest; @@ -63,6 +83,10 @@ pub use apply::{ apply_manifests, AppliedService, ApplyAction, ApplyError, ApplyErrorKind, ApplyOptions, ApplyReport, ServiceTarget, }; +pub use delete::{ + delete_services, DeleteError, DeleteErrorKind, DeleteOptions, DeleteReport, DeleteTarget, + DeletedService, +}; pub use manifest::{ AgentOverride, EcsNetworking, EcsRuntime, FleetMetadata, FleetSpec, FleetTemplate, Ingress, KubernetesRuntime, Metadata, OABFleetManifest, OABServiceManifest, RawManifest, Resources,