microvms_app/control/

transport.rs

// SPDX-License-Identifier: Apache-2.0
//! The control plane's rest-json calls, and the seam a transport implements.
//!
//! # Why this is hand-rolled
//!
//! `lambda-microvms` has no `aws-sdk-rust` crate. The alternatives were vendoring
//! smithy-rs codegen — Gradle, JDK 17, Kotlin, and a project that declares its external
//! interfaces unstable — or signing 24 rest-json operations by hand with the first-party
//! low-level crates, which is the flow the AWS SDK maintainers point at for exactly this
//! case. rest-json is also the easiest protocol to hand-roll: plain JSON bodies,
//! URI-bound operations, standard headers, no XML and no query-protocol encoding.
//!
//! # The seam, and what the fake is allowed to know
//!
//! [`Transport`] is a trait with one method. The real implementation signs and sends;
//! the test implementation is a **contract recorder** that asserts on the emitted
//! request in the shape the service model declares.
//!
//! The rule the recorder follows is the lesson from the `buildState` bug: a fake that
//! shares the client's assumptions hides exactly the class of defect the fake exists to
//! catch. So the recorder never asserts on a value the client computed, and every
//! response it hands back is **literal JSON in the model's spelling** — see
//! [`crate::control::ops`] for why that is not a round trip through this crate's own
//! serializer.
//!
//! # Local validation is not optional
//!
//! botocore's `VALIDATED_METADATA_ATTRS` is `{required, min, document, union}`, so
//! `max`, `pattern`, and `enum` violations are serialized, sent, and answered with a
//! `ValidationException` — confirmed empirically for `max` (runHookPayload 4097,
//! maximumDurationInSeconds 28801, ImageName 65 chars, clientToken 129 chars), for
//! `pattern`, and for `enum`. The same holds here with more force: **nothing** in this
//! transport reads the service model at runtime, so every constraint is checked in the
//! caller before a request is built or it is not checked at all.

use crate::error::{Error, ErrorKind};
use crate::region::Region;
use std::time::Duration;

/// An opt-in throttle budget. Other transient errors retain the default budget.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RetryPolicy {
    max_attempts: u32,
    max_elapsed: Duration,
}

impl RetryPolicy {
    /// Includes the initial request. Elapsed time includes requests and sleeps;
    /// an in-flight request is allowed to finish, but no retry starts after the deadline.
    pub fn new(max_attempts: u32, max_elapsed: Duration) -> Result<Self, Error> {
        if max_attempts == 0 || max_elapsed.is_zero() {
            return Err(Error::invalid_arg(
                "retry max_attempts and max_elapsed must be positive",
            ));
        }
        Ok(Self {
            max_attempts,
            max_elapsed,
        })
    }
}

/// Cancels pending retries without discarding an in-flight launch response.
#[derive(Clone, Debug)]
pub struct RetryCancellation(tokio::sync::watch::Sender<bool>);

impl Default for RetryCancellation {
    fn default() -> Self {
        Self(tokio::sync::watch::Sender::new(false))
    }
}

impl RetryCancellation {
    /// Stop before the next request. An accepted launch is still recorded.
    pub fn cancel(&self) {
        self.0.send_replace(true);
    }

    fn is_cancelled(&self) -> bool {
        *self.0.borrow()
    }

    pub(crate) async fn cancelled(&self) {
        let mut receiver = self.0.subscribe();
        let _ = receiver.wait_for(|cancelled| *cancelled).await;
    }
}

/// Work done by one control-plane call, independent of concurrent callers.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct RetryDetails {
    pub attempt_count: u32,
    pub backoff: Duration,
}

fn retry_delays() -> backon::ExponentialBuilder {
    backon::ExponentialBuilder::default()
        .with_jitter()
        .with_min_delay(Duration::from_millis(200))
        .with_max_delay(Duration::from_secs(20))
        .with_max_times(5)
}

/// Sends with an optional throttle budget, measuring and sleeping on the plane's clock.
/// The call is cloned unchanged, including its idempotency token, on each attempt.
pub(crate) async fn send_configured(
    transport: &dyn Transport,
    call: Call,
    accept: &[u16],
    policy: Option<RetryPolicy>,
    clock: &dyn crate::clock::Clock,
) -> Result<(Reply, RetryDetails), Error> {
    send_cancellable(transport, call, accept, policy, clock, None).await
}

pub(crate) async fn send_cancellable(
    transport: &dyn Transport,
    call: Call,
    accept: &[u16],
    policy: Option<RetryPolicy>,
    clock: &dyn crate::clock::Clock,
    cancellation: Option<&RetryCancellation>,
) -> Result<(Reply, RetryDetails), Error> {
    use backon::BackoffBuilder;
    let started = clock.elapsed();
    let mut details = RetryDetails::default();
    let mut delays = retry_delays().without_max_times().build();
    let mut ordinary_retries = 0;
    loop {
        if cancellation.is_some_and(RetryCancellation::is_cancelled) {
            return Err(Error::new(
                ErrorKind::Interrupted,
                "launch retries cancelled before a VM was assigned",
            )
            .with_retry_details(details.attempt_count, details.backoff));
        }
        details.attempt_count += 1;
        let error = match transport.send(call.clone()).await {
            Ok(reply) if reply.is_success() || accept.contains(&reply.status) => {
                return Ok((reply, details));
            }
            Ok(reply) => classify_failure_in_region(call.operation, &reply, transport.region()),
            Err(error) => error,
        };
        let delay = delays.next().expect("unbounded backoff");
        let elapsed = clock.elapsed().saturating_sub(started);
        let allowed = if error.throttled()
            && let Some(policy) = policy
        {
            details.attempt_count < policy.max_attempts
                && elapsed.saturating_add(delay) < policy.max_elapsed
        } else {
            ordinary_retries += 1;
            ordinary_retries <= 5
        };
        if !error.retryable() || !allowed {
            return Err(error.with_retry_details(details.attempt_count, details.backoff));
        }
        if let Some(cancellation) = cancellation {
            tokio::select! {
                _ = cancellation.cancelled() => continue,
                _ = clock.sleep(delay) => {}
            }
        } else {
            clock.sleep(delay).await;
        }
        details.backoff += delay;
        // A scheduler delay can carry the sleep past the deadline.
        if error.throttled()
            && policy.is_some_and(|p| clock.elapsed().saturating_sub(started) >= p.max_elapsed)
        {
            return Err(error.with_retry_details(details.attempt_count, details.backoff));
        }
    }
}

/// The pinned API version, which is also the first path segment of every operation.
///
/// Read from [`crate::constants::MODEL_API_VERSION`] rather than written again, so the
/// drift gate's version check and the URLs cannot disagree: a client signing
/// `/2025-09-09/...` while claiming to implement a different model is a client whose
/// constraint checks were verified against the wrong thing.
const API_PATH_VERSION: &str = crate::constants::MODEL_API_VERSION;

/// The SigV4 signing name, from the model's `signingName`.
///
/// `lambda`, not `lambda-microvms`. The `endpointPrefix` is also `lambda`, which is why
/// an endpoint resolves for every AWS region and a client constructs happily for a region
/// that does not run MicroVMs (TRAP-6, `crate::region`).
pub const SIGNING_NAME: &str = "lambda";

/// One control-plane call: method, path, optional JSON body.
///
/// A struct rather than a long parameter list because the recorder asserts on it as a
/// whole, and because `method` and `path` next to each other is how a reader checks a
/// path against the model's `http` trait.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Call {
    /// The operation name, for diagnostics and for the recorder's ledger.
    ///
    /// Carried rather than derived from the path: the recorder's TRAP-11 assertion is
    /// "an exec lifecycle emits no call named `CreateMicrovmShellAuthToken`", and
    /// matching that on a path is a substring test that a renamed route would slip past.
    pub operation: &'static str,
    /// `GET`, `POST`, `PATCH`, or `DELETE` — the four the model's `http` traits use for the
    /// operations this client implements.
    pub method: Method,
    /// The path **after** the host, with the API version and every URI parameter already
    /// percent-encoded. Includes the query string when the operation has one.
    pub path: String,
    /// The rest-json body, or `None` for an operation the model gives no body members.
    pub body: Option<Vec<u8>>,
}

/// The HTTP methods the model's operations use.
///
/// An enum so a call site cannot pass `"post"` and have the canonical request silently
/// disagree with the wire method.
///
/// # `Patch` is one operation's, and only one
///
/// The model uses `PATCH` for exactly `UpdateMicrovmImageVersion` and `PUT` for exactly
/// `UpdateMicrovmImage`. `Put` is deliberately absent: this client does not implement that
/// operation, and a variant nothing constructs is a variant that suggests it does.
/// `PATCH` is not interchangeable with `POST` here — a `POST` to the version path is a route
/// the service does not declare, and the answer is a 404 that reads as a missing version
/// rather than as a wrong method.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Method {
    Get,
    Post,
    Patch,
    Delete,
}

impl Method {
    pub fn as_str(self) -> &'static str {
        match self {
            Method::Get => "GET",
            Method::Post => "POST",
            Method::Patch => "PATCH",
            Method::Delete => "DELETE",
        }
    }
}

impl Call {
    /// A `GET` with no body.
    pub fn get(operation: &'static str, path: impl Into<String>) -> Self {
        Self {
            operation,
            method: Method::Get,
            path: path.into(),
            body: None,
        }
    }

    /// A `POST` carrying a serialized rest-json body.
    pub fn post_json<T: serde::Serialize>(
        operation: &'static str,
        path: impl Into<String>,
        body: &T,
    ) -> Result<Self, Error> {
        let body = serde_json::to_vec(body).map_err(|error| {
            Error::new(
                ErrorKind::Unexpected,
                format!("could not serialize the {operation} request body: {error}"),
            )
        })?;
        Ok(Self {
            operation,
            method: Method::Post,
            path: path.into(),
            body: Some(body),
        })
    }

    /// A `PATCH` carrying a serialized rest-json body.
    ///
    /// The same shape as [`Call::post_json`] and written beside it rather than folded into a
    /// method-taking constructor, for the reason [`Method`]'s docs give: the method is part of
    /// the route the model declares, and a `with_method(Method::Patch)` builder is one a call
    /// site can forget to call. `UpdateMicrovmImageVersion` is the only operation that reaches
    /// this.
    pub fn patch_json<T: serde::Serialize>(
        operation: &'static str,
        path: impl Into<String>,
        body: &T,
    ) -> Result<Self, Error> {
        let body = serde_json::to_vec(body).map_err(|error| {
            Error::new(
                ErrorKind::Unexpected,
                format!("could not serialize the {operation} request body: {error}"),
            )
        })?;
        Ok(Self {
            operation,
            method: Method::Patch,
            path: path.into(),
            body: Some(body),
        })
    }

    /// A `POST` with no body members. `SuspendMicrovm` and `ResumeMicrovm` are both this
    /// shape: their only member is the URI-located identifier.
    ///
    /// The body is `Some(b"{}")` rather than `None`, because an empty POST and a POST of
    /// `{}` are different canonical requests and rest-json services answer the former
    /// inconsistently.
    pub fn post_empty(operation: &'static str, path: impl Into<String>) -> Self {
        Self {
            operation,
            method: Method::Post,
            path: path.into(),
            body: Some(b"{}".to_vec()),
        }
    }

    /// A `DELETE` with no body.
    pub fn delete(operation: &'static str, path: impl Into<String>) -> Self {
        Self {
            operation,
            method: Method::Delete,
            path: path.into(),
            body: None,
        }
    }

    /// The body as bytes, empty when there is none — which is what gets signed.
    pub fn body_bytes(&self) -> &[u8] {
        self.body.as_deref().unwrap_or(&[])
    }
}

/// What the control plane answered: a status, a raw body, and the error type the service
/// named, if it named one.
///
/// Raw bytes rather than a deserialized value because the error path has to read a body
/// the success types cannot hold, and because TRAP-6's null `message` field is only
/// visible to something that did not already require a `String` there.
#[derive(Clone, Debug)]
pub struct Reply {
    pub status: u16,
    pub body: Vec<u8>,
    /// The `x-amzn-ErrorType` header's error name, cut down by [`error_type_name`], or
    /// `None` when the response carried no such header. [`classify_failure`] reads it
    /// before the status, because the service has sent a throttle as HTTP 400.
    pub error_type: Option<String>,
}

impl Reply {
    /// Deserializes the body as `T`, or reports the failure with the body attached.
    ///
    /// The body goes in the message because a rest-json parse failure with the text
    /// withheld is unactionable: the reader cannot tell a renamed member from a truncated
    /// response. Truncated to 512 bytes, since a list response can be large and the first
    /// 512 characters always contain the member that failed.
    pub fn json<T: serde::de::DeserializeOwned>(&self, operation: &str) -> Result<T, Error> {
        serde_json::from_slice(&self.body).map_err(|error| {
            let text = String::from_utf8_lossy(&self.body);
            let shown: String = text.chars().take(512).collect();
            Error::new(
                ErrorKind::Platform,
                format!(
                    "could not read the {operation} response ({error}). The service model for \
                     {API_PATH_VERSION} is what this client's shapes are transcribed from, so a \
                     member that has moved is a drift-gate failure rather than a transport bug. \
                     Body: {shown}"
                ),
            )
            .with_source(error)
        })
    }

    /// Whether the status is a success. rest-json operations here answer 200 or 201.
    fn is_success(&self) -> bool {
        (200..300).contains(&self.status)
    }
}

/// The one seam every operation goes through.
///
/// One method, so a fake is a few lines and there is no partially-overridden transport
/// whose behaviour is half real. `&self` rather than `&mut self` so the recorder can be
/// shared behind an `Arc` while a lifecycle runs.
pub trait Transport: Send + Sync {
    /// The service region, when known, for diagnostics about ambiguous denials.
    /// Custom transports may leave this unknown without changing their behavior.
    fn region(&self) -> Option<&Region> {
        None
    }

    /// Performs `call`, returning whatever the far side answered.
    ///
    /// A transport-level failure (no status at all) is an `Err`; a 4xx or 5xx is an
    /// `Ok(Reply)` carrying the status, because classifying a status is
    /// [`classify_failure`]'s job and a transport that classified would be a second
    /// place the taxonomy lives.
    fn send(
        &self,
        call: Call,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Reply, Error>> + Send + '_>>;

    /// Resolves the credentials the next [`Transport::send`] would sign with, sending nothing
    /// to the service (BIND-15's credentials check).
    ///
    /// Resolving may reach an SSO, credential-process, or instance-metadata source, none of
    /// which bills. The default answers `Ok`: a transport that does not sign (every scripted
    /// one) has nothing to resolve.
    fn resolve_credentials(
        &self,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), Error>> + Send + '_>> {
        Box::pin(async { Ok(()) })
    }
}

/// Sends `call` through `transport`, retrying the retryable statuses, and turns a failure
/// status into a classified [`Error`].
///
/// # Which failures retry
///
/// 429 (`ThrottlingException`), 500 (`InternalServerException`), and any other 5xx, plus
/// a transport error that produced no status. AWS standard-mode semantics, matched by
/// hand because the smithy retry pieces are coupled to an orchestrator this client does
/// not have.
///
/// # Why retrying the mutating operations is safe
///
/// Every mutating operation this client sends carries a `clientToken`, so a retry is
/// idempotent by the service's own contract — which is the same property that makes the
/// token dangerous to derive (TRAP-1). `SuspendMicrovm`/`ResumeMicrovm`/`TerminateMicrovm`
/// take no token and are retried anyway: they are state transitions whose second
/// application is a no-op or a `ConflictException`, and a conflict is not retried.
pub async fn send_with_retry(transport: &dyn Transport, call: Call) -> Result<Reply, Error> {
    use backon::Retryable;

    let operation = call.operation;
    let attempt = || {
        let call = call.clone();
        async move {
            let reply = transport.send(call).await?;
            if reply.is_success() {
                return Ok(reply);
            }
            Err(classify_failure_in_region(
                operation,
                &reply,
                transport.region(),
            ))
        }
    };

    let mut details = RetryDetails {
        attempt_count: 1,
        backoff: Duration::ZERO,
    };
    let result = attempt
        .retry(retry_delays())
        .when(Error::retryable)
        .notify(|_, delay| {
            details.attempt_count += 1;
            details.backoff += delay;
        })
        .await;
    result.map_err(|error| error.with_retry_details(details.attempt_count, details.backoff))
}

/// [`send_with_retry`], answering `Ok` for the statuses in `accept` as well as for a
/// success.
///
/// For a call whose failure status is an answer rather than an error: `GetMicrovmImage`'s 404
/// is "there is no image under this name", which `ensure_image` reads as a state
/// ([`crate::control::ensure`]). Retries are unchanged — a throttle or a 5xx is still
/// retried — and every other status is still classified by [`classify_failure`].
pub async fn send_accepting(
    transport: &dyn Transport,
    call: Call,
    accept: &[u16],
) -> Result<Reply, Error> {
    use backon::Retryable;

    let operation = call.operation;
    let attempt = || {
        let call = call.clone();
        async move {
            let reply = transport.send(call).await?;
            if reply.is_success() || accept.contains(&reply.status) {
                return Ok(reply);
            }
            Err(classify_failure_in_region(
                operation,
                &reply,
                transport.region(),
            ))
        }
    };

    let mut details = RetryDetails {
        attempt_count: 1,
        backoff: Duration::ZERO,
    };
    let result = attempt
        .retry(retry_delays())
        .when(Error::retryable)
        .notify(|_, delay| {
            details.attempt_count += 1;
            details.backoff += delay;
        })
        .await;
    result.map_err(|error| error.with_retry_details(details.attempt_count, details.backoff))
}

/// Turns a failure status and body into an [`Error`] of the right kind.
///
/// # The statuses are the control plane's, not the daemon's
///
/// [`crate::error::WireKind`] is the *daemon's* status discipline and deliberately has no
/// generic 4xx fallback, because the daemon's 400-versus-404 distinction is load-bearing.
/// None of that applies here: these seven statuses come from the modeled exception shapes
/// (`ValidationException` 400, `ServiceQuotaExceededException` 402,
/// `AccessDeniedException` 403, `ResourceNotFoundException` 404, `ConflictException` 409,
/// `ThrottlingException` 429, `InternalServerException` 500). So a control-plane failure
/// carries an [`ErrorKind`] and **no** `WireKind` — which is what `error.rs` already
/// documents: "`None` for every local reject and every control-plane failure — those never
/// reached the in-VM daemon".
///
/// # 403 and the null message
///
/// A 403 whose `message` is null is ambiguous (TRAP-6): both a region without MicroVMs and
/// a denied implied action can produce it. The diagnostic leads with the region when it is
/// outside [`crate::region::MEASURED_REGIONS`], names both causes, and points to CloudTrail
/// for the service's full denial.
pub fn classify_failure(operation: &str, reply: &Reply) -> Error {
    classify_failure_in_region(operation, reply, None)
}

fn classify_failure_in_region(operation: &str, reply: &Reply, region: Option<&Region>) -> Error {
    let parsed: crate::control::ops::ServiceErrorWire =
        serde_json::from_slice(&reply.body).unwrap_or_default();
    let detail = parsed.message.as_deref().unwrap_or("").trim().to_string();

    let (kind, explanation) = match reply.status {
        // The error type outranks the status. A throttled `ListMicrovms` answered HTTP 400
        // with `x-amzn-ErrorType: ThrottlingException`, where the model says 429
        // (docs/PLATFORM.md, measured 2026-10-02), and read by status alone it was a
        // validation refusal that nothing retried.
        status if reply.error_type.as_deref() == Some("ThrottlingException") => (
            ErrorKind::Retryable,
            format!(
                "ThrottlingException, named by the x-amzn-ErrorType header on an HTTP {status} \
                 — retry the identical request."
            ),
        ),
        400 => (
            ErrorKind::Platform,
            "ValidationException — the service refused a value. Every constraint the pinned \
             service model states on a member this client sends is checked locally before the \
             call — min as well as max, pattern, and enum, because nothing else validates a \
             request here: this client signs with aws-sigv4 and sends with reqwest, so botocore's \
             VALIDATED_METADATA_ATTRS never applies to it. So reaching this status means a \
             constraint this client does not yet check, and closing it is a guard in \
             crates/microvms-app/src/control plus a line in tools/check-model-drift.py."
                .to_string(),
        ),
        402 => (
            ErrorKind::Platform,
            "ServiceQuotaExceededException — an account limit, not a request defect.".to_string(),
        ),
        403 if detail.is_empty() => (
            ErrorKind::Credentials,
            null_message_denial(operation, region),
        ),
        403 => (
            ErrorKind::Credentials,
            "AccessDeniedException, with a message — the service reported a genuine denial."
                .to_string(),
        ),
        404 => (
            ErrorKind::Platform,
            "ResourceNotFoundException — the image or MicroVM does not exist.".to_string(),
        ),
        409 => (
            ErrorKind::Platform,
            "ConflictException — the resource is in a state that forbids this call. An image in \
             CREATING refuses deletion, which is why teardown retries."
                .to_string(),
        ),
        429 => (
            ErrorKind::Retryable,
            "ThrottlingException — retry the identical request.".to_string(),
        ),
        status if status >= 500 => (
            ErrorKind::Retryable,
            "InternalServerException — a service fault, retry the identical request.".to_string(),
        ),
        _ => (
            ErrorKind::Platform,
            "an unmodeled status: the service model declares 400, 402, 403, 404, 409, 429, and \
             500 for these operations."
                .to_string(),
        ),
    };

    let said = if detail.is_empty() {
        "the service sent no message".to_string()
    } else {
        format!("the service said {detail:?}")
    };
    Error::new(
        kind,
        format!(
            "{operation} failed with HTTP {}: {said}. {explanation}",
            reply.status
        ),
    )
    .with_throttled(
        reply.status == 429 || reply.error_type.as_deref() == Some("ThrottlingException"),
    )
}

/// The service can omit its denial detail for both regional availability and IAM errors.
fn null_message_denial(operation: &str, region: Option<&Region>) -> String {
    let availability = match region {
        Some(region) if region.is_measured() => format!(
            "{} carries MicroVMs; an IAM policy is the likely cause.",
            region.as_str()
        ),
        Some(region) => format!(
            "{} is not in the known MicroVMs region list; verify service availability as well as the IAM policy.",
            region.as_str()
        ),
        None => "The cause may be an unsupported region or an IAM denial.".to_string(),
    };
    let region_arg = region
        .map(|region| format!(" --region {}", region.as_str()))
        .unwrap_or_default();
    format!(
        "AccessDeniedException with an empty or null message field. {availability} \
         A create call with tags can require the implied lambda:TagResource action. \
         Check the full denial in CloudTrail: aws cloudtrail lookup-events{region_arg} \
         --lookup-attributes AttributeKey=EventName,AttributeValue={operation} \
         --query 'Events[].CloudTrailEvent' --output text. Inspect errorCode and errorMessage \
         for the denied action (docs/PLATFORM.md)."
    )
}

/// The error name in an `x-amzn-ErrorType` header value: what precedes the first `:` and
/// follows the first `#`.
///
/// That is the rest-json protocol's rule, and both parts occur here: the service's throttle
/// arrived as `ThrottlingException:http://internal.amazon.com/coral/com.amazon.coral.availability/`,
/// and other services send a `namespace#Name` form.
pub fn error_type_name(header: &str) -> &str {
    let name = header.split(':').next().unwrap_or(header);
    name.split_once('#').map_or(name, |(_, after)| after).trim()
}

/// `https://lambda.<region>.amazonaws.com`, from the model's `endpointPrefix`.
///
/// The prefix is `lambda` rather than `lambda-microvms`, which is why this resolves for
/// every AWS region including the ones that answer the null-message denial.
pub fn endpoint_for(region: &Region) -> String {
    format!("https://{SIGNING_NAME}.{}.amazonaws.com", region.as_str())
}

/// The operation paths, one function per `http.requestUri` in the model.
///
/// Here rather than inlined at each call site so a path is written once and can be read
/// against the model in one place. Every URI parameter is percent-encoded, because an
/// image identifier may be an ARN — which contains `:` and `/` — and an unencoded ARN in a
/// path segment is a different resource.
pub mod paths {
    use super::API_PATH_VERSION;

    /// Percent-encodes a URI path segment.
    ///
    /// Everything outside RFC 3986's unreserved set is escaped. **`/` is encoded**,
    /// which is the whole point: an image ARN's slashes would otherwise split into
    /// extra path segments and address something else. `:` is encoded too — legal
    /// unencoded in a path per the RFC, but AWS's own SDKs encode it in labels and
    /// the signature must match whatever is sent.
    pub fn encode_segment(segment: &str) -> String {
        use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};
        const ESCAPED: &AsciiSet = &NON_ALPHANUMERIC
            .remove(b'-')
            .remove(b'_')
            .remove(b'.')
            .remove(b'~');
        utf8_percent_encode(segment, ESCAPED).to_string()
    }

    /// `POST /2025-09-09/microvm-images`
    pub fn microvm_images() -> String {
        format!("/{API_PATH_VERSION}/microvm-images")
    }

    /// `GET /2025-09-09/microvm-images`, with the listing's query members.
    ///
    /// `nameFilter` is the model's server-side **substring** filter — it narrows the
    /// listing, it does not answer "the image named X", so an exact-match comparison
    /// still happens in the caller. `nextToken` is the pagination cursor and is opaque,
    /// which is why both values go through [`encode_segment`]: an opaque token may carry
    /// `+` or `=`, and an unencoded one desynchronises the SigV4 canonical query from
    /// the query actually sent.
    pub fn microvm_images_list(name_filter: Option<&str>, next_token: Option<&str>) -> String {
        let mut query: Vec<String> = Vec::new();
        if let Some(filter) = name_filter {
            query.push(format!("nameFilter={}", encode_segment(filter)));
        }
        if let Some(token) = next_token {
            query.push(format!("nextToken={}", encode_segment(token)));
        }
        if query.is_empty() {
            return microvm_images();
        }
        format!("{}?{}", microvm_images(), query.join("&"))
    }

    /// `GET|DELETE /2025-09-09/microvm-images/{imageIdentifier}`
    pub fn microvm_image(image: &str) -> String {
        format!(
            "/{API_PATH_VERSION}/microvm-images/{}",
            encode_segment(image)
        )
    }

    /// One `nextToken=` query member, appended to `path`, or `path` unchanged.
    ///
    /// Shared by the three listings that take nothing else, so the encoding decision is
    /// made once. The value goes through [`encode_segment`] for the reason
    /// [`microvm_images_list`] states: a pagination token is **opaque**, so it may carry
    /// `+`, `/`, or `=`, and an unencoded one desynchronises the SigV4 canonical query
    /// from the query actually sent — a rejection that reads like bad credentials rather
    /// than like a malformed URL.
    fn with_next_token(path: String, next_token: Option<&str>) -> String {
        match next_token {
            Some(token) => format!("{path}?nextToken={}", encode_segment(token)),
            None => path,
        }
    }

    /// `GET /2025-09-09/microvm-images/{imageIdentifier}/versions`, with the pagination
    /// cursor when there is one.
    ///
    /// `maxResults` is not sent: the model caps it at 50 and the service's own default is
    /// what a caller reading every page wants, so naming a page size would only be a
    /// second number to keep right. [`image_versions_paged`] is the one caller that does
    /// name one, and says why.
    pub fn image_versions(image: &str, next_token: Option<&str>) -> String {
        image_versions_paged(image, next_token, None)
    }

    /// [`image_versions`] with an explicit `maxResults`, which **no production path sends**.
    ///
    /// # Why a page size exists at all when nothing ships one
    ///
    /// Because without it the cursor encoding is unfalsifiable against the real service,
    /// and that encoding is the one place in this module where being wrong reads as a
    /// *credentials* failure rather than a URL one. A real `nextToken` is opaque base64
    /// carrying `+`, `/`, and `=`; an unencoded one desynchronises the SigV4 canonical
    /// query and the service answers 403. Every test that could catch it went through an
    /// injected transport, so the fake minted the token and the signer never saw it.
    ///
    /// Forcing a real cursor needs a listing with more items than one page, and the
    /// service's default page is larger than any image's version count here — so the only
    /// way to make the real service mint a real cursor for a resource that already exists
    /// is to ask for a smaller page. That is what this is for, and
    /// `crates/microvms-core/tests/live_pagination.rs` is its caller.
    ///
    /// # Why it is not simply a parameter on [`image_versions`]
    ///
    /// A page size on the production path is a second number to keep right, and a wrong
    /// one is invisible: reading every page with `maxResults=1` is correct and slow, which
    /// is the kind of defect that survives. Keeping it to a separately named function means
    /// the production call sites cannot acquire one by accident, and a reader of
    /// [`image_versions`] sees no page size to wonder about.
    ///
    /// `max_results` is clamped to the model's `1..=50`, because a value outside it is a
    /// `ValidationException` and a caller of this function is trying to observe pagination
    /// rather than to discover the bound.
    pub fn image_versions_paged(
        image: &str,
        next_token: Option<&str>,
        max_results: Option<u32>,
    ) -> String {
        let mut query: Vec<String> = Vec::new();
        if let Some(size) = max_results {
            query.push(format!("maxResults={}", size.clamp(1, 50)));
        }
        if let Some(token) = next_token {
            query.push(format!("nextToken={}", encode_segment(token)));
        }
        let path = format!("{}/versions", microvm_image(image));
        if query.is_empty() {
            return path;
        }
        format!("{path}?{}", query.join("&"))
    }

    /// `GET|PATCH|DELETE /2025-09-09/microvm-images/{imageIdentifier}/versions/{imageVersion}`
    ///
    /// One path for three operations, because the model gives them one `requestUri` and the
    /// method is what distinguishes them: `GetMicrovmImageVersion` reads it,
    /// `UpdateMicrovmImageVersion` patches it, `DeleteMicrovmImageVersion` deletes it. The
    /// method lives on the [`Call`] rather than in three path functions, so a reader checking
    /// this against `service-2.json` compares one string.
    ///
    /// Built from [`microvm_image`] rather than from [`image_versions`], because that one
    /// now takes a cursor and none of the three may carry a query member.
    pub fn image_version(image: &str, version: &str) -> String {
        format!(
            "{}/versions/{}",
            microvm_image(image),
            encode_segment(version)
        )
    }

    /// `GET /2025-09-09/microvm-images/{imageIdentifier}/versions/{imageVersion}/builds/{buildId}`
    ///
    /// The list path plus one segment. `buildId` is percent-encoded like every other URI
    /// parameter even though a real one is a UUID: the model types it `NonBlankString`, so it
    /// is not the encoder's business to assume the shape of a value the service mints.
    pub fn image_build(image: &str, version: &str, build_id: &str) -> String {
        format!(
            "{}/builds/{}",
            image_version(image, version),
            encode_segment(build_id)
        )
    }

    /// `GET /2025-09-09/managed-microvm-images`, with the pagination cursor when there is one.
    ///
    /// A **different route** from [`microvm_images`], not a filter on it: the managed bases
    /// live under `managed-microvm-images` and their summaries carry a different shape
    /// (`ManagedMicrovmImageSummary`, three members). Reading the account's own image listing
    /// would never return one.
    pub fn managed_microvm_images(next_token: Option<&str>) -> String {
        with_next_token(
            format!("/{API_PATH_VERSION}/managed-microvm-images"),
            next_token,
        )
    }

    /// `GET /2025-09-09/managed-microvm-images/{imageIdentifier}/versions`, with the cursor.
    ///
    /// `imageIdentifier` here is the managed base's **full ARN**, not its name: measured
    /// 2026-08-16, `al2023-1` alone answers `ValidationException: Invalid ARN format:
    /// al2023-1`. That is why [`crate::control::BaseImage::arn`] is what the caller passes —
    /// and why the wrapper refuses a non-ARN locally, since the rejection names the value
    /// without saying which member wanted an ARN.
    pub fn managed_image_versions(image: &str, next_token: Option<&str>) -> String {
        with_next_token(
            format!(
                "/{API_PATH_VERSION}/managed-microvm-images/{}/versions",
                encode_segment(image)
            ),
            next_token,
        )
    }

    /// `GET /2025-09-09/microvm-images/{imageIdentifier}/versions/{imageVersion}/builds`,
    /// with the pagination cursor when there is one.
    pub fn image_builds(image: &str, version: &str, next_token: Option<&str>) -> String {
        with_next_token(
            format!("{}/builds", image_version(image, version)),
            next_token,
        )
    }

    /// `POST /2025-09-09/microvms`
    ///
    /// The bare collection path, which is what `RunMicrovm` posts to. `ListMicrovms` reads
    /// the same route and lives in [`microvms_list`] rather than here, so a launch cannot
    /// accidentally carry a `nextToken` — the same split [`microvm_images`] and
    /// [`microvm_images_list`] already make.
    pub fn microvms() -> String {
        format!("/{API_PATH_VERSION}/microvms")
    }

    /// `GET /2025-09-09/microvms`, with the pagination cursor when there is one.
    pub fn microvms_list(next_token: Option<&str>) -> String {
        with_next_token(microvms(), next_token)
    }

    /// [`microvms_list`] with the model's `imageIdentifier` and `imageVersion` query filters.
    ///
    /// Keys are emitted in sorted order, which is also SigV4's canonical order.
    pub fn microvms_list_matching(
        image_identifier: Option<&str>,
        image_version: Option<&str>,
        next_token: Option<&str>,
    ) -> String {
        let query: Vec<String> = [
            ("imageIdentifier", image_identifier),
            ("imageVersion", image_version),
            ("nextToken", next_token),
        ]
        .into_iter()
        .filter_map(|(key, value)| value.map(|value| format!("{key}={}", encode_segment(value))))
        .collect();
        if query.is_empty() {
            return microvms();
        }
        format!("{}?{}", microvms(), query.join("&"))
    }

    /// `GET|DELETE /2025-09-09/microvms/{microvmIdentifier}`
    pub fn microvm(id: &str) -> String {
        format!("/{API_PATH_VERSION}/microvms/{}", encode_segment(id))
    }

    /// `POST /2025-09-09/microvms/{microvmIdentifier}/suspend`
    pub fn suspend(id: &str) -> String {
        format!("{}/suspend", microvm(id))
    }

    /// `POST /2025-09-09/microvms/{microvmIdentifier}/resume`
    pub fn resume(id: &str) -> String {
        format!("{}/resume", microvm(id))
    }

    /// `POST /2025-09-09/microvms/{microvmIdentifier}/auth-token`
    pub fn auth_token(id: &str) -> String {
        format!("{}/auth-token", microvm(id))
    }

    /// `POST /2025-09-09/microvms/{microvmIdentifier}/shell-auth-token`
    ///
    /// The shell token's route, its own path rather than a parameter on
    /// [`auth_token`]: the two operations take different bodies (`allowedPorts` exists
    /// only on the ordinary one — the shell is not a port), and the exec path must never
    /// be one boolean away from minting a shell credential (TRAP-11).
    pub fn shell_auth_token(id: &str) -> String {
        format!("{}/shell-auth-token", microvm(id))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn throttle_budget_counts_attempts_bounds_elapsed_and_keeps_other_failures_default() {
        use crate::clock::Clock;
        use crate::control::fake::{Answer, FakeControlPlane, TestClock};
        for (status, policy, expected) in [
            (429, None, 6),
            (
                429,
                Some(RetryPolicy::new(3, Duration::from_secs(3600)).unwrap()),
                3,
            ),
            (
                429,
                Some(RetryPolicy::new(9, Duration::from_millis(1)).unwrap()),
                1,
            ),
            (
                503,
                Some(RetryPolicy::new(9, Duration::from_secs(3600)).unwrap()),
                6,
            ),
            (
                403,
                Some(RetryPolicy::new(9, Duration::from_secs(3600)).unwrap()),
                1,
            ),
        ] {
            let transport = FakeControlPlane::new();
            for _ in 0..expected {
                transport.answer("RunMicrovm", Answer::failure(status, "failure"));
            }
            // An extra retry reaches success, so a broken budget fails the verdict
            // assertion instead of repeating a transient answer forever.
            transport.answer("RunMicrovm", Answer::ok("{}"));
            let clock = TestClock::new();
            let error = send_configured(
                &transport,
                Call::get("RunMicrovm", "/run"),
                &[],
                policy,
                &clock,
            )
            .await
            .unwrap_err();
            assert_eq!(transport.call_count("RunMicrovm"), expected);
            assert_eq!(error.attempt_count(), expected as u32);
            assert_eq!(error.throttled(), status == 429);
            assert_eq!(error.backoff(), clock.elapsed());
            if expected == 1 {
                assert_eq!(clock.elapsed(), Duration::ZERO);
            }
        }
        assert!(RetryPolicy::new(0, Duration::from_secs(1)).is_err());
        assert!(RetryPolicy::new(1, Duration::ZERO).is_err());
    }

    #[tokio::test]
    async fn a_retry_at_the_elapsed_budget_boundary_does_not_sleep() {
        use crate::clock::Clock;
        use crate::control::fake::{Answer, FakeControlPlane, TestClock};
        let transport = FakeControlPlane::new();
        transport.answer("RunMicrovm", Answer::throttled_as_400());
        let gate = transport.hold_responses();
        let clock = TestClock::new();
        clock.set_unix_now(Duration::ZERO);
        let pending = send_configured(
            &transport,
            Call::get("RunMicrovm", "/run"),
            &[],
            Some(RetryPolicy::new(2, Duration::MAX).unwrap()),
            &clock,
        );
        tokio::pin!(pending);
        tokio::select! {
            biased;
            _ = &mut pending => panic!("response must wait for the gate"),
            _ = std::future::ready(()) => {}
        }
        assert_eq!(transport.call_count("RunMicrovm"), 1);
        // Saturating elapsed + jittered delay equals MAX for every possible delay.
        // Advancing during the request also proves request time uses this budget.
        clock.advance(Duration::MAX);
        gate.notify_one();
        // The delivered response must finish this call immediately. A second request
        // would wait on the closed gate, so fail that retry here rather than hanging.
        let error = tokio::select! {
            biased;
            result = &mut pending => result.unwrap_err(),
            _ = std::future::ready(()) => panic!("elapsed budget must stop before another request"),
        };
        assert!(error.throttled());
        assert_eq!(error.attempt_count(), 1);
        assert_eq!(transport.call_count("RunMicrovm"), 1);
        assert_eq!(error.backoff(), Duration::ZERO);
        assert_eq!(clock.elapsed(), Duration::MAX);
    }

    #[tokio::test]
    async fn configured_retries_recover_header_throttles_with_the_identical_call() {
        use crate::clock::Clock;
        use crate::control::fake::{Answer, FakeControlPlane, TestClock};
        let transport = FakeControlPlane::new();
        for _ in 0..7 {
            transport.answer("RunMicrovm", Answer::throttled_as_400());
        }
        transport.answer("RunMicrovm", Answer::ok("{}"));
        let clock = TestClock::new();
        let call = Call::post_json(
            "RunMicrovm",
            "/run",
            &serde_json::json!({"clientToken":"same-launch"}),
        )
        .unwrap();
        let (_, details) = send_configured(
            &transport,
            call.clone(),
            &[],
            Some(RetryPolicy::new(8, Duration::from_secs(3600)).unwrap()),
            &clock,
        )
        .await
        .unwrap();
        assert_eq!(details.attempt_count, 8);
        assert_eq!(details.backoff, clock.elapsed());
        assert!(details.backoff > Duration::ZERO);
        assert!(transport.calls().iter().all(|sent| sent == &call));
        transport.answer("GetMicrovm", Answer::failure(404, "missing"));
        let (reply, details) = send_configured(
            &transport,
            Call::get("GetMicrovm", "/missing"),
            &[404],
            None,
            &clock,
        )
        .await
        .unwrap();
        assert_eq!(reply.status, 404);
        assert_eq!(details.attempt_count, 1);
        assert_eq!(details.backoff, Duration::ZERO);
    }

    #[tokio::test(start_paused = true)]
    async fn exhausted_public_retry_reports_throttle_attempts_and_backoff() {
        use crate::control::fake::{Answer, FakeControlPlane};
        for accepts_missing in [false, true] {
            let transport = FakeControlPlane::new();
            for _ in 0..6 {
                transport.answer("ListMicrovms", Answer::throttled_as_400());
            }
            transport.answer("ListMicrovms", Answer::ok("{}"));
            let call = Call::get("ListMicrovms", "/list");
            let error = if accepts_missing {
                send_accepting(&transport, call, &[404]).await
            } else {
                send_with_retry(&transport, call).await
            }
            .unwrap_err();
            assert_eq!(transport.call_count("ListMicrovms"), 6);
            assert!(error.throttled());
            assert_eq!(error.attempt_count(), 6);
            assert!(error.backoff() >= Duration::from_millis(6200));
            assert!(error.backoff() < Duration::from_millis(12400));
        }
        for status in [200, 404] {
            let transport = FakeControlPlane::new();
            transport.answer("GetMicrovm", Answer::failure(status, "accepted"));
            let reply = send_accepting(&transport, Call::get("GetMicrovm", "/missing"), &[404])
                .await
                .unwrap();
            assert_eq!(reply.status, status);
            assert_eq!(transport.call_count("GetMicrovm"), 1);
        }
    }

    #[tokio::test]
    async fn a_cancel_signal_wakes_backoff_before_another_request() {
        use crate::clock::testing::YieldingClock;
        use crate::control::fake::{Answer, FakeControlPlane};
        let transport = FakeControlPlane::new();
        transport.answer("RunMicrovm", Answer::throttled_as_400());
        let clock = YieldingClock::new();
        let cancellation = RetryCancellation::default();
        let call = Call::get("RunMicrovm", "/run");
        let pending = send_cancellable(
            &transport,
            call,
            &[],
            Some(RetryPolicy::new(100, Duration::from_secs(3600)).unwrap()),
            &clock,
            Some(&cancellation),
        );
        tokio::pin!(pending);
        tokio::select! { biased; _ = &mut pending => panic!("backoff should yield"), _ = std::future::ready(()) => {} }
        cancellation.cancel();
        let error = pending.await.unwrap_err();
        assert_eq!(error.kind(), ErrorKind::Interrupted);
        assert_eq!(transport.call_count("RunMicrovm"), 1);
    }

    #[tokio::test]
    async fn dropping_a_retry_future_prevents_another_request() {
        use crate::clock::testing::YieldingClock;
        use crate::control::fake::{Answer, FakeControlPlane};
        let transport = FakeControlPlane::new();
        transport.answer("RunMicrovm", Answer::throttled_as_400());
        let clock = YieldingClock::new();
        let mut future = Box::pin(send_configured(
            &transport,
            Call::get("RunMicrovm", "/run"),
            &[],
            Some(RetryPolicy::new(100, Duration::from_secs(3600)).unwrap()),
            &clock,
        ));
        tokio::select! { biased; _ = &mut future => panic!("retry sleep should yield"), _ = std::future::ready(()) => {} }
        drop(future);
        // Let any accidentally detached retry run after its backoff before checking.
        clock.advance(Duration::from_secs(3600));
        tokio::task::yield_now().await;
        tokio::task::yield_now().await;
        assert_eq!(transport.call_count("RunMicrovm"), 1);
    }

    /// The endpoint, from the model's `endpointPrefix`. `lambda` and not
    /// `lambda-microvms` — which is the fact that makes TRAP-6 possible, since this
    /// resolves for every AWS region.
    #[test]
    fn the_endpoint_uses_the_models_lambda_prefix_for_every_region() {
        assert_eq!(
            endpoint_for(&Region::US_EAST_1),
            "https://lambda.us-east-1.amazonaws.com"
        );
        assert_eq!(
            endpoint_for(&Region::AP_NORTHEAST_1),
            "https://lambda.ap-northeast-1.amazonaws.com"
        );
        // Including one AWS does not run MicroVMs in, which is the point: us-west-1 answered
        // `ListMicrovms` with the null-message denial when measured (2026-10-24).
        assert_eq!(
            endpoint_for(&"us-west-1".parse().expect("a region name")),
            "https://lambda.us-west-1.amazonaws.com"
        );
    }

    /// Every path against the model's `http.requestUri`, as literals. Transcribed from
    /// `service-2.json` rather than built from a template, because a template shared with
    /// the code under test would agree with a wrong template.
    #[test]
    fn every_path_matches_the_models_request_uri() {
        assert_eq!(paths::microvm_images(), "/2025-09-09/microvm-images");
        assert_eq!(
            paths::microvm_image("img"),
            "/2025-09-09/microvm-images/img"
        );
        assert_eq!(
            paths::image_versions("img", None),
            "/2025-09-09/microvm-images/img/versions"
        );
        assert_eq!(
            paths::image_version("img", "1"),
            "/2025-09-09/microvm-images/img/versions/1"
        );
        assert_eq!(
            paths::image_builds("img", "1", None),
            "/2025-09-09/microvm-images/img/versions/1/builds"
        );
        assert_eq!(
            paths::image_build("img", "1", "build-abc"),
            "/2025-09-09/microvm-images/img/versions/1/builds/build-abc"
        );
        assert_eq!(
            paths::managed_microvm_images(None),
            "/2025-09-09/managed-microvm-images"
        );
        assert_eq!(
            paths::managed_image_versions(
                "arn:aws:lambda:us-east-1:aws:microvm-image:al2023-1",
                None
            ),
            "/2025-09-09/managed-microvm-images/\
             arn%3Aaws%3Alambda%3Aus-east-1%3Aaws%3Amicrovm-image%3Aal2023-1/versions"
        );
        assert_eq!(paths::microvms(), "/2025-09-09/microvms");
        assert_eq!(paths::microvms_list(None), "/2025-09-09/microvms");
        assert_eq!(paths::microvm("mvm-1"), "/2025-09-09/microvms/mvm-1");
        assert_eq!(
            paths::suspend("mvm-1"),
            "/2025-09-09/microvms/mvm-1/suspend"
        );
        assert_eq!(paths::resume("mvm-1"), "/2025-09-09/microvms/mvm-1/resume");
        assert_eq!(
            paths::auth_token("mvm-1"),
            "/2025-09-09/microvms/mvm-1/auth-token"
        );
        assert_eq!(
            paths::shell_auth_token("mvm-1"),
            "/2025-09-09/microvms/mvm-1/shell-auth-token"
        );
    }

    /// The listing path carries its query members in the model's spelling, percent-encoded,
    /// and a bare listing is the plain collection path with no `?`.
    ///
    /// An opaque `nextToken` may carry `+`, `/`, or `=`; unencoded, those desynchronise the
    /// SigV4 canonical query from the query actually sent, and the rejection reads like bad
    /// credentials.
    #[test]
    fn the_image_listing_path_encodes_its_query_members() {
        assert_eq!(
            paths::microvm_images_list(None, None),
            "/2025-09-09/microvm-images"
        );
        assert_eq!(
            paths::microvm_images_list(Some("my-image"), None),
            "/2025-09-09/microvm-images?nameFilter=my-image"
        );
        assert_eq!(
            paths::microvm_images_list(Some("my-image"), Some("a+b/c=")),
            "/2025-09-09/microvm-images?nameFilter=my-image&nextToken=a%2Bb%2Fc%3D"
        );
        assert_eq!(
            paths::microvm_images_list(None, Some("token")),
            "/2025-09-09/microvm-images?nextToken=token"
        );
    }

    /// The other three paginated listings encode their cursor the same way, and emit no
    /// `?` at all without one.
    ///
    /// Written as literals per listing rather than as a loop over a helper, because the
    /// bug this closes is a path that forgets the cursor entirely — and a loop over the
    /// helper would pass against three call sites that never pass one.
    ///
    /// **Falsification** — drop the [`paths::encode_segment`] call from
    /// `with_next_token` and every `%2B`/`%2F`/`%3D` assertion below goes red, which is
    /// the SigV4 canonical-query desynchronisation this encoder exists to prevent.
    #[test]
    fn the_other_paginated_listings_encode_their_cursor_too() {
        assert_eq!(
            paths::image_versions("img", Some("a+b/c=")),
            "/2025-09-09/microvm-images/img/versions?nextToken=a%2Bb%2Fc%3D"
        );
        assert_eq!(
            paths::image_builds("img", "1", Some("a+b/c=")),
            "/2025-09-09/microvm-images/img/versions/1/builds?nextToken=a%2Bb%2Fc%3D"
        );
        assert_eq!(
            paths::microvms_list(Some("a+b/c=")),
            "/2025-09-09/microvms?nextToken=a%2Bb%2Fc%3D"
        );

        // No cursor means no query string, not an empty one: `?nextToken=` is a member
        // present and blank, which is a different request from a member absent.
        for path in [
            paths::image_versions("img", None),
            paths::image_builds("img", "1", None),
            paths::microvms_list(None),
        ] {
            assert!(
                !path.contains('?'),
                "an absent cursor emits no query: {path}"
            );
        }
    }

    /// The version listing's explicit page size: ordered before the cursor, clamped to the
    /// model's `1..=50`, and absent from the path when nothing asks for one.
    ///
    /// The clamp is asserted at both ends because a value outside the bound is a
    /// `ValidationException`, and the only caller of this function
    /// (`tests/live_pagination.rs`) is trying to *observe* a real cursor — a run that
    /// failed on the page size instead would prove nothing about the encoding it came for.
    ///
    /// **Falsification** — remove the `.clamp(1, 50)` and the `0` and `999` cases go red
    /// with `maxResults=0` and `maxResults=999`, both of which the service refuses.
    #[test]
    fn the_version_listings_explicit_page_size_is_ordered_and_clamped() {
        assert_eq!(
            paths::image_versions_paged("img", None, Some(1)),
            "/2025-09-09/microvm-images/img/versions?maxResults=1"
        );
        // Both members, and `maxResults` first — the order the query is built in, which is
        // asserted so a reordering is a visible change rather than a silent one.
        assert_eq!(
            paths::image_versions_paged("img", Some("a+b/c="), Some(2)),
            "/2025-09-09/microvm-images/img/versions?maxResults=2&nextToken=a%2Bb%2Fc%3D"
        );
        assert_eq!(
            paths::image_versions_paged("img", None, Some(0)),
            "/2025-09-09/microvm-images/img/versions?maxResults=1",
            "the model's min is 1; a 0 would be refused"
        );
        assert_eq!(
            paths::image_versions_paged("img", None, Some(999)),
            "/2025-09-09/microvm-images/img/versions?maxResults=50",
            "the model's max is 50; a larger value would be refused"
        );

        // And the production spelling is unchanged by the new one existing: `image_versions`
        // delegates here with no page size, so it must still emit exactly what it did.
        assert_eq!(
            paths::image_versions_paged("img", None, None),
            paths::image_versions("img", None)
        );
        assert!(
            !paths::image_versions("img", Some("t")).contains("maxResults"),
            "no production path names a page size: {}",
            paths::image_versions("img", Some("t"))
        );
    }

    /// The **managed** routes are a different collection from the account's own, and their
    /// cursor is encoded the same way.
    ///
    /// Asserted as its own case because the plausible mistake is treating the managed listing
    /// as a filter on `microvm-images` — it is a separate `requestUri` in the model with a
    /// different response shape, so a path built from [`paths::microvm_images`] would address
    /// the account's images and answer 200 with the wrong items.
    ///
    /// **Falsification** — drop the `managed-` prefix from either function and the two
    /// literals below go red while every request still succeeds against the real service,
    /// which is exactly why these are literals transcribed from the model rather than a
    /// shared template.
    #[test]
    fn the_managed_routes_are_their_own_collection_and_encode_their_cursor() {
        assert!(
            !paths::managed_microvm_images(None).contains('?'),
            "an absent cursor emits no query member"
        );
        assert_eq!(
            paths::managed_microvm_images(Some("a+b/c=")),
            "/2025-09-09/managed-microvm-images?nextToken=a%2Bb%2Fc%3D"
        );
        assert_eq!(
            paths::managed_image_versions("al2023-1", Some("a+b/c=")),
            "/2025-09-09/managed-microvm-images/al2023-1/versions?nextToken=a%2Bb%2Fc%3D"
        );

        // Not a filter on the account's own listing: the two collections are different path
        // segments, and reading one for the other answers 200 with the wrong items.
        assert_ne!(
            paths::managed_microvm_images(None),
            paths::microvm_images_list(None, None)
        );
        assert!(
            !paths::microvm_images().contains("managed"),
            "the account's own collection must not acquire the managed prefix"
        );
    }

    /// A `GET` of one build is the list path plus one encoded segment, and it carries no query
    /// member.
    ///
    /// The `buildId` is percent-encoded even though a real one is a UUID: the model types it
    /// `NonBlankString`, so the encoder does not get to assume the shape of a value the
    /// service mints.
    #[test]
    fn one_builds_path_is_the_listing_plus_an_encoded_segment() {
        let path = paths::image_build("img", "1", "4a4c5e30-811f-47fa-9893-260ea6a37a8f");
        assert_eq!(
            path,
            "/2025-09-09/microvm-images/img/versions/1/builds/4a4c5e30-811f-47fa-9893-260ea6a37a8f"
        );
        assert!(!path.contains('?'), "{path}");
        assert!(
            path.starts_with(&paths::image_builds("img", "1", None)),
            "the get path must extend the list path rather than be a second route: {path}"
        );
        // And an identifier with a separator in it cannot split into extra segments.
        assert_eq!(
            paths::image_build("img", "1", "a/b"),
            "/2025-09-09/microvm-images/img/versions/1/builds/a%2Fb"
        );
    }

    /// A `DELETE` of one version carries no query member, even though its sibling
    /// [`paths::image_versions`] now takes a cursor. A version path with `?nextToken=`
    /// appended addresses nothing.
    #[test]
    fn deleting_one_version_carries_no_pagination_cursor() {
        let path = paths::image_version("img", "1");
        assert_eq!(path, "/2025-09-09/microvm-images/img/versions/1");
        assert!(!path.contains('?'), "{path}");
        assert!(!path.contains("nextToken"), "{path}");
    }

    /// A **real** image ARN in a URI parameter is percent-encoded, every colon included.
    /// `MicrovmImageIdentifier` is documented as "ARN or ID", so this is the ordinary case
    /// rather than an edge one.
    ///
    /// The literal is the shape the service actually returns, measured 2026-08-15 in
    /// us-east-1: `microvm-image:<name>` with a **colon**, which is also the only form the
    /// model's `TaggableResource` pattern admits. This test used to hold
    /// `microvm-image/img`, and the slash was not a harmless stand-in — `GetMicrovmImage`
    /// on that form answers `AccessDeniedException`, because IAM evaluates a malformed ARN
    /// as a resource with no matching policy.
    #[test]
    fn a_real_arn_identifier_is_percent_encoded_colons_and_all() {
        let arn = "arn:aws:lambda:us-east-1:123456789012:microvm-image:img";
        let path = paths::microvm_image(arn);
        assert_eq!(
            path,
            "/2025-09-09/microvm-images/arn%3Aaws%3Alambda%3Aus-east-1%3A123456789012%3Amicrovm-image%3Aimg"
        );
        assert!(
            !path["/2025-09-09/microvm-images/".len()..].contains(':'),
            "an unencoded colon in a path segment is a signature mismatch: {path}"
        );
    }

    /// The encoder still encodes `/`, which is the property that matters even though a real
    /// image ARN carries none.
    ///
    /// Kept as its own case rather than folded into the test above, because the input here
    /// is a *shape the service does not use* and labelling it "an ARN" is what let the slash
    /// form look load-bearing for twelve fakes. An unencoded slash would split into extra
    /// path segments addressing something else, and other identifiers this encoder sees
    /// (`imageVersion`, a `nextToken` cursor) genuinely can contain one.
    #[test]
    fn the_segment_encoder_encodes_slashes_wherever_they_come_from() {
        assert_eq!(paths::encode_segment("a/b"), "a%2Fb");
        let path = paths::microvm_image("has/a/slash");
        assert_eq!(path, "/2025-09-09/microvm-images/has%2Fa%2Fslash");
        assert!(
            !path["/2025-09-09/microvm-images/".len()..].contains('/'),
            "an encoded identifier contributes no path separators: {path}"
        );
    }

    /// The API version in the path is the constant the drift gate checks, not a second
    /// literal. Two copies of `2025-09-09` are two things to update, and the one that gets
    /// missed is the one nothing compiles against.
    #[test]
    fn the_path_version_is_the_pinned_model_version() {
        assert_eq!(API_PATH_VERSION, crate::constants::MODEL_API_VERSION);
        assert!(paths::microvms().starts_with("/2025-09-09/"));
    }

    /// The signing name is `lambda`, from the model's `signingName`. A signature computed
    /// for `lambda-microvms` is rejected in a way that reads like bad credentials.
    #[test]
    fn the_signing_name_is_lambda_not_lambda_microvms() {
        assert_eq!(SIGNING_NAME, "lambda");
    }

    /// TRAP-6: a missing message cannot distinguish service availability from IAM.
    #[test]
    fn a_null_message_access_denied_names_both_possible_causes() {
        let reply = Reply {
            status: 403,
            body: br#"{"message": null}"#.to_vec(),
            error_type: None,
        };
        let error = classify_failure("ListMicrovms", &reply);
        assert_eq!(error.kind(), ErrorKind::Credentials);
        let message = error.to_string();
        assert!(message.contains("null"), "{message}");
        assert!(message.contains("region"), "{message}");
        assert!(
            message.contains("unsupported region or an IAM denial"),
            "{message}"
        );
        assert!(message.contains("lambda:TagResource"), "{message}");
        assert!(message.contains("cloudtrail lookup-events"), "{message}");
    }

    /// TRAP-6: known service regions prioritize IAM and preserve the event name for lookup.
    #[tokio::test]
    async fn supported_region_null_denials_point_to_iam_in_both_send_paths() {
        use crate::control::fake::{Answer, FakeControlPlane};
        let transport = FakeControlPlane::in_region(Region::US_EAST_1);
        transport.answer(
            "CreateMicrovmImage",
            Answer {
                status: 403,
                body: r#"{"message":null}"#.to_string(),
                error_type: Some("AccessDeniedException".to_string()),
            },
        );
        for accept in [false, true] {
            let call = Call::get("CreateMicrovmImage", "/images");
            let error = if accept {
                send_accepting(&transport, call, &[404]).await.unwrap_err()
            } else {
                send_with_retry(&transport, call).await.unwrap_err()
            };
            assert_eq!(error.kind(), ErrorKind::Credentials);
            assert!(!error.retryable());
            let message = error.to_string();
            for detail in [
                "us-east-1 carries MicroVMs",
                "policy is the likely cause",
                "lambda:TagResource",
                "cloudtrail lookup-events --region us-east-1",
                "AttributeValue=CreateMicrovmImage",
                "errorMessage",
            ] {
                assert!(message.contains(detail), "{message}");
            }
            assert!(!message.contains("Check the region before the policy"));
        }
        assert_eq!(transport.call_count("CreateMicrovmImage"), 2);
    }

    /// TRAP-6: a region outside the measured list leads with service availability and does
    /// not rule out an IAM denial.
    #[test]
    fn unlisted_region_null_denials_keep_both_causes_visible() {
        let reply = Reply {
            status: 403,
            body: b"{}".to_vec(),
            error_type: None,
        };
        let error = classify_failure_in_region(
            "ListMicrovms",
            &reply,
            Some(&"us-west-1".parse().expect("a region name")),
        );
        let message = error.to_string();
        assert!(message.contains("verify service availability as well as the IAM policy"));
        assert!(message.contains("--region us-west-1"));
        assert!(!message.contains("policy is the likely cause"));
    }

    /// A 403 *with* a message is a genuine denial and says so, so the two are not
    /// conflated. A classifier that named the region trap on every 403 would be as
    /// misleading in the other direction.
    #[test]
    fn a_403_with_a_message_is_reported_as_a_genuine_denial() {
        let reply = Reply {
            status: 403,
            body: br#"{"message": "User: arn:aws:iam::1:user/u is not authorized"}"#.to_vec(),
            error_type: None,
        };
        let error = classify_failure("RunMicrovm", &reply);
        assert_eq!(error.kind(), ErrorKind::Credentials);
        let message = error.to_string();
        assert!(message.contains("not authorized"), "{message}");
        assert!(
            message.contains("genuine denial"),
            "must distinguish itself from the null-message case: {message}"
        );
        assert!(
            !message.contains("does not carry MicroVMs"),
            "must not blame the region when the service named a principal: {message}"
        );
    }

    /// A 400 is the service refusing a value this client should have refused first, so the
    /// message names the validation refusal and the local checks it got past, not the
    /// catch-all for a status the model doesn't declare.
    #[test]
    fn a_400_names_the_validation_refusal_and_the_local_checks_it_passed() {
        let reply = Reply {
            status: 400,
            body: br#"{"message": "1 validation error detected"}"#.to_vec(),
            error_type: None,
        };
        let error = classify_failure("RunMicrovm", &reply);
        assert_eq!(error.kind(), ErrorKind::Platform);
        let message = error.to_string();
        assert!(message.contains("ValidationException"), "{message}");
        assert!(message.contains("checked locally"), "{message}");
        assert!(!message.contains("unmodeled status"), "{message}");
    }

    /// The service's throttle as it arrived from `ListMicrovms` on 2026-10-02: HTTP 400,
    /// named `ThrottlingException` only by the `x-amzn-ErrorType` header, with a null
    /// message. Classified by status it read as a validation refusal and was never
    /// retried, which failed every listing past the account's page rate.
    ///
    /// **Falsification** — run 2026-10-02. Guard the classifier's header arm with `false`, so
    /// the status alone decides again, and this fails on the retryability assertion with the
    /// `ValidationException` explanation (`guards:fire -- --only
    /// control-throttle-classified-by-status`). Restored.
    #[test]
    fn a_throttle_named_by_the_error_type_header_is_retryable_whatever_the_status() {
        let reply = Reply {
            status: 400,
            body: br#"{"message":null,"throttlingReasons":null}"#.to_vec(),
            error_type: Some("ThrottlingException".to_string()),
        };
        let error = classify_failure("ListMicrovms", &reply);
        assert!(error.retryable(), "{error}");
        assert!(error.throttled(), "the header identifies the throttle");
        assert_eq!(error.kind(), ErrorKind::Retryable);
        let message = error.to_string();
        assert!(message.contains("ThrottlingException"), "{message}");
        assert!(message.contains("HTTP 400"), "{message}");
        assert!(!message.contains("ValidationException"), "{message}");
    }

    /// Another error name on a 400 leaves the status rule alone: the header decides only
    /// for the throttle, so a genuine validation refusal still isn't retried.
    #[test]
    fn a_400_naming_another_error_type_is_still_a_validation_refusal() {
        let reply = Reply {
            status: 400,
            body: br#"{"message": "1 validation error detected"}"#.to_vec(),
            error_type: Some("ValidationException".to_string()),
        };
        let error = classify_failure("RunMicrovm", &reply);
        assert!(!error.retryable(), "{error}");
        assert!(error.to_string().contains("checked locally"), "{error}");
    }

    /// The header's name is what precedes the first `:` and follows the first `#`, the
    /// rest-json rule, on the value the service sent and on the namespaced form.
    ///
    /// **Falsification** — run 2026-10-02. Keep the whole header instead of what precedes the
    /// `:`, and the service's own value comes back with its URI suffix and fails the first
    /// case (`guards:fire -- --only control-error-type-keeps-the-uri`). Restored.
    #[test]
    fn the_error_type_name_drops_the_uri_suffix_and_the_namespace() {
        let cases = [
            (
                "ThrottlingException:http://internal.amazon.com/coral/com.amazon.coral.availability/",
                "ThrottlingException",
            ),
            (
                "aws.protocoltests.restjson#FooError:http://internal.amazon.com/coral/com.amazon.coral.validate/",
                "FooError",
            ),
            (
                "com.amazonaws.lambda#ResourceNotFoundException",
                "ResourceNotFoundException",
            ),
            ("ConflictException", "ConflictException"),
        ];
        for (header, name) in cases {
            assert_eq!(error_type_name(header), name, "{header}");
        }
    }

    /// Only the throttle and the 5xx family are retryable. A retried 409 would spin
    /// against an image in CREATING, and a retried 400 would send the same rejected value
    /// five more times.
    #[test]
    fn only_throttles_and_server_faults_are_retryable() {
        let cases = [
            (400, false),
            (402, false),
            (403, false),
            (404, false),
            (409, false),
            (429, true),
            (500, true),
            (502, true),
            (503, true),
        ];
        for (status, retryable) in cases {
            let reply = Reply {
                status,
                body: br#"{"message": "detail"}"#.to_vec(),
                error_type: None,
            };
            let error = classify_failure("GetMicrovm", &reply);
            assert_eq!(error.retryable(), retryable, "HTTP {status}");
            assert_eq!(error.throttled(), status == 429, "HTTP {status}");
        }
    }

    /// A control-plane failure carries **no** `WireKind`: nothing reached the in-VM
    /// daemon, so there is no daemon-chosen status to report and the CLI's `data.kind`
    /// must be absent rather than invented.
    #[test]
    fn a_control_plane_failure_carries_no_daemon_wire_kind() {
        for status in [400, 403, 404, 409, 429, 500] {
            let reply = Reply {
                status,
                body: Vec::new(),
                error_type: None,
            };
            assert_eq!(
                classify_failure("GetMicrovm", &reply).wire_kind(),
                None,
                "HTTP {status} never reached the daemon"
            );
        }
    }

    /// An unparseable error body still classifies by status rather than failing. A service
    /// answering an HTML error page or a truncated body must not turn a 429 into an
    /// unexpected error, which would stop the retry.
    #[test]
    fn an_unparseable_error_body_still_classifies_by_status() {
        let reply = Reply {
            status: 429,
            body: b"<html>Too Many Requests</html>".to_vec(),
            error_type: None,
        };
        let error = classify_failure("RunMicrovm", &reply);
        assert!(error.retryable(), "a throttle is a throttle");
        assert!(error.to_string().contains("no message"), "{error}");
    }

    /// A body-parse failure names the operation and shows the body, because a rest-json
    /// parse error with the text withheld cannot distinguish a renamed member from a
    /// truncated response.
    #[test]
    fn a_response_that_does_not_parse_reports_the_body() {
        let reply = Reply {
            status: 200,
            body: br#"{"microvmId": 42}"#.to_vec(),
            error_type: None,
        };
        let error = reply
            .json::<crate::control::ops::MicrovmResponseWire>("GetMicrovm")
            .expect_err("microvmId is a string shape");
        let message = error.to_string();
        assert!(message.contains("GetMicrovm"), "{message}");
        assert!(message.contains("microvmId"), "{message}");
        assert_eq!(error.kind(), ErrorKind::Platform);
    }

    /// A `POST` with no body members sends `{}` rather than nothing: an empty POST and a
    /// POST of `{}` are different canonical requests, and rest-json services answer the
    /// former inconsistently.
    #[test]
    fn a_bodyless_post_sends_an_empty_json_object() {
        let call = Call::post_empty("SuspendMicrovm", paths::suspend("mvm-1"));
        assert_eq!(call.body.as_deref(), Some(&b"{}"[..]));
        assert_eq!(call.method, Method::Post);
    }

    /// A `GET` and a `DELETE` carry no body at all, and the signed empty payload is what
    /// `body_bytes` yields.
    #[test]
    fn a_get_and_a_delete_carry_no_body() {
        let get = Call::get("GetMicrovm", paths::microvm("mvm-1"));
        assert_eq!(get.body, None);
        assert_eq!(get.body_bytes(), b"");

        let delete = Call::delete("TerminateMicrovm", paths::microvm("mvm-1"));
        assert_eq!(delete.body, None);
        assert_eq!(delete.method, Method::Delete);
    }

    /// The methods, spelled as the canonical request needs them: uppercase.
    #[test]
    fn the_methods_are_the_ones_the_model_uses_spelled_uppercase() {
        assert_eq!(Method::Get.as_str(), "GET");
        assert_eq!(Method::Post.as_str(), "POST");
        assert_eq!(Method::Patch.as_str(), "PATCH");
        assert_eq!(Method::Delete.as_str(), "DELETE");
    }

    /// A `PATCH` carries its serialized body and is **not** a `POST`.
    ///
    /// The distinction is load-bearing rather than cosmetic: `UpdateMicrovmImageVersion` is the
    /// model's only `PATCH`, and a `POST` to the version path is a route the service does not
    /// declare — so the failure would be a 404 that reads as a missing version rather than as a
    /// wrong method. The body assertion is here too, because a `PATCH` whose body was dropped
    /// would be a request with no `status` member at all.
    ///
    /// **Falsification** — set `method: Method::Post` in `patch_json` and the method assertion
    /// goes red; the path and body assertions would still pass, which is why the method is
    /// asserted separately.
    #[test]
    fn a_patch_carries_its_body_and_is_not_a_post() {
        let call = Call::patch_json(
            "UpdateMicrovmImageVersion",
            paths::image_version("img", "2.0"),
            &crate::control::ops::UpdateImageVersionWire {
                status: crate::control::ops::VersionStatus::Inactive,
            },
        )
        .expect("serialises");

        assert_eq!(call.method, Method::Patch);
        assert_ne!(
            call.method,
            Method::Post,
            "a POST to the version path is a route the model does not declare"
        );
        assert_eq!(call.path, "/2025-09-09/microvm-images/img/versions/2.0");
        assert_eq!(
            call.body.as_deref(),
            Some(&br#"{"status":"INACTIVE"}"#[..]),
            "the one member the request has must reach the body"
        );
        assert_eq!(call.operation, "UpdateMicrovmImageVersion");
    }
}