1use crate::error::{Error, ErrorKind};
37use crate::region::Region;
38use std::time::Duration;
39
40#[derive(Clone, Copy, Debug, PartialEq, Eq)]
42pub struct RetryPolicy {
43 max_attempts: u32,
44 max_elapsed: Duration,
45}
46
47impl RetryPolicy {
48 pub fn new(max_attempts: u32, max_elapsed: Duration) -> Result<Self, Error> {
51 if max_attempts == 0 || max_elapsed.is_zero() {
52 return Err(Error::invalid_arg(
53 "retry max_attempts and max_elapsed must be positive",
54 ));
55 }
56 Ok(Self {
57 max_attempts,
58 max_elapsed,
59 })
60 }
61}
62
63#[derive(Clone, Debug)]
65pub struct RetryCancellation(tokio::sync::watch::Sender<bool>);
66
67impl Default for RetryCancellation {
68 fn default() -> Self {
69 Self(tokio::sync::watch::Sender::new(false))
70 }
71}
72
73impl RetryCancellation {
74 pub fn cancel(&self) {
76 self.0.send_replace(true);
77 }
78
79 fn is_cancelled(&self) -> bool {
80 *self.0.borrow()
81 }
82
83 pub(crate) async fn cancelled(&self) {
84 let mut receiver = self.0.subscribe();
85 let _ = receiver.wait_for(|cancelled| *cancelled).await;
86 }
87}
88
89#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
91pub struct RetryDetails {
92 pub attempt_count: u32,
93 pub backoff: Duration,
94}
95
96fn retry_delays() -> backon::ExponentialBuilder {
97 backon::ExponentialBuilder::default()
98 .with_jitter()
99 .with_min_delay(Duration::from_millis(200))
100 .with_max_delay(Duration::from_secs(20))
101 .with_max_times(5)
102}
103
104pub(crate) async fn send_configured(
107 transport: &dyn Transport,
108 call: Call,
109 accept: &[u16],
110 policy: Option<RetryPolicy>,
111 clock: &dyn crate::clock::Clock,
112) -> Result<(Reply, RetryDetails), Error> {
113 send_cancellable(transport, call, accept, policy, clock, None).await
114}
115
116pub(crate) async fn send_cancellable(
117 transport: &dyn Transport,
118 call: Call,
119 accept: &[u16],
120 policy: Option<RetryPolicy>,
121 clock: &dyn crate::clock::Clock,
122 cancellation: Option<&RetryCancellation>,
123) -> Result<(Reply, RetryDetails), Error> {
124 use backon::BackoffBuilder;
125 let started = clock.elapsed();
126 let mut details = RetryDetails::default();
127 let mut delays = retry_delays().without_max_times().build();
128 let mut ordinary_retries = 0;
129 loop {
130 if cancellation.is_some_and(RetryCancellation::is_cancelled) {
131 return Err(Error::new(
132 ErrorKind::Interrupted,
133 "launch retries cancelled before a VM was assigned",
134 )
135 .with_retry_details(details.attempt_count, details.backoff));
136 }
137 details.attempt_count += 1;
138 let error = match transport.send(call.clone()).await {
139 Ok(reply) if reply.is_success() || accept.contains(&reply.status) => {
140 return Ok((reply, details));
141 }
142 Ok(reply) => classify_failure_in_region(call.operation, &reply, transport.region()),
143 Err(error) => error,
144 };
145 let delay = delays.next().expect("unbounded backoff");
146 let elapsed = clock.elapsed().saturating_sub(started);
147 let allowed = if error.throttled()
148 && let Some(policy) = policy
149 {
150 details.attempt_count < policy.max_attempts
151 && elapsed.saturating_add(delay) < policy.max_elapsed
152 } else {
153 ordinary_retries += 1;
154 ordinary_retries <= 5
155 };
156 if !error.retryable() || !allowed {
157 return Err(error.with_retry_details(details.attempt_count, details.backoff));
158 }
159 if let Some(cancellation) = cancellation {
160 tokio::select! {
161 _ = cancellation.cancelled() => continue,
162 _ = clock.sleep(delay) => {}
163 }
164 } else {
165 clock.sleep(delay).await;
166 }
167 details.backoff += delay;
168 if error.throttled()
170 && policy.is_some_and(|p| clock.elapsed().saturating_sub(started) >= p.max_elapsed)
171 {
172 return Err(error.with_retry_details(details.attempt_count, details.backoff));
173 }
174 }
175}
176
177const API_PATH_VERSION: &str = crate::constants::MODEL_API_VERSION;
184
185pub const SIGNING_NAME: &str = "lambda";
191
192#[derive(Clone, Debug, Eq, PartialEq)]
198pub struct Call {
199 pub operation: &'static str,
205 pub method: Method,
208 pub path: String,
211 pub body: Option<Vec<u8>>,
213}
214
215#[derive(Clone, Copy, Debug, Eq, PartialEq)]
229pub enum Method {
230 Get,
231 Post,
232 Patch,
233 Delete,
234}
235
236impl Method {
237 pub fn as_str(self) -> &'static str {
238 match self {
239 Method::Get => "GET",
240 Method::Post => "POST",
241 Method::Patch => "PATCH",
242 Method::Delete => "DELETE",
243 }
244 }
245}
246
247impl Call {
248 pub fn get(operation: &'static str, path: impl Into<String>) -> Self {
250 Self {
251 operation,
252 method: Method::Get,
253 path: path.into(),
254 body: None,
255 }
256 }
257
258 pub fn post_json<T: serde::Serialize>(
260 operation: &'static str,
261 path: impl Into<String>,
262 body: &T,
263 ) -> Result<Self, Error> {
264 let body = serde_json::to_vec(body).map_err(|error| {
265 Error::new(
266 ErrorKind::Unexpected,
267 format!("could not serialize the {operation} request body: {error}"),
268 )
269 })?;
270 Ok(Self {
271 operation,
272 method: Method::Post,
273 path: path.into(),
274 body: Some(body),
275 })
276 }
277
278 pub fn patch_json<T: serde::Serialize>(
286 operation: &'static str,
287 path: impl Into<String>,
288 body: &T,
289 ) -> Result<Self, Error> {
290 let body = serde_json::to_vec(body).map_err(|error| {
291 Error::new(
292 ErrorKind::Unexpected,
293 format!("could not serialize the {operation} request body: {error}"),
294 )
295 })?;
296 Ok(Self {
297 operation,
298 method: Method::Patch,
299 path: path.into(),
300 body: Some(body),
301 })
302 }
303
304 pub fn post_empty(operation: &'static str, path: impl Into<String>) -> Self {
311 Self {
312 operation,
313 method: Method::Post,
314 path: path.into(),
315 body: Some(b"{}".to_vec()),
316 }
317 }
318
319 pub fn delete(operation: &'static str, path: impl Into<String>) -> Self {
321 Self {
322 operation,
323 method: Method::Delete,
324 path: path.into(),
325 body: None,
326 }
327 }
328
329 pub fn body_bytes(&self) -> &[u8] {
331 self.body.as_deref().unwrap_or(&[])
332 }
333}
334
335#[derive(Clone, Debug)]
342pub struct Reply {
343 pub status: u16,
344 pub body: Vec<u8>,
345 pub error_type: Option<String>,
349}
350
351impl Reply {
352 pub fn json<T: serde::de::DeserializeOwned>(&self, operation: &str) -> Result<T, Error> {
359 serde_json::from_slice(&self.body).map_err(|error| {
360 let text = String::from_utf8_lossy(&self.body);
361 let shown: String = text.chars().take(512).collect();
362 Error::new(
363 ErrorKind::Platform,
364 format!(
365 "could not read the {operation} response ({error}). The service model for \
366 {API_PATH_VERSION} is what this client's shapes are transcribed from, so a \
367 member that has moved is a drift-gate failure rather than a transport bug. \
368 Body: {shown}"
369 ),
370 )
371 .with_source(error)
372 })
373 }
374
375 fn is_success(&self) -> bool {
377 (200..300).contains(&self.status)
378 }
379}
380
381pub trait Transport: Send + Sync {
387 fn region(&self) -> Option<&Region> {
390 None
391 }
392
393 fn send(
400 &self,
401 call: Call,
402 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Reply, Error>> + Send + '_>>;
403
404 fn resolve_credentials(
411 &self,
412 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), Error>> + Send + '_>> {
413 Box::pin(async { Ok(()) })
414 }
415}
416
417pub async fn send_with_retry(transport: &dyn Transport, call: Call) -> Result<Reply, Error> {
435 use backon::Retryable;
436
437 let operation = call.operation;
438 let attempt = || {
439 let call = call.clone();
440 async move {
441 let reply = transport.send(call).await?;
442 if reply.is_success() {
443 return Ok(reply);
444 }
445 Err(classify_failure_in_region(
446 operation,
447 &reply,
448 transport.region(),
449 ))
450 }
451 };
452
453 let mut details = RetryDetails {
454 attempt_count: 1,
455 backoff: Duration::ZERO,
456 };
457 let result = attempt
458 .retry(retry_delays())
459 .when(Error::retryable)
460 .notify(|_, delay| {
461 details.attempt_count += 1;
462 details.backoff += delay;
463 })
464 .await;
465 result.map_err(|error| error.with_retry_details(details.attempt_count, details.backoff))
466}
467
468pub async fn send_accepting(
476 transport: &dyn Transport,
477 call: Call,
478 accept: &[u16],
479) -> Result<Reply, Error> {
480 use backon::Retryable;
481
482 let operation = call.operation;
483 let attempt = || {
484 let call = call.clone();
485 async move {
486 let reply = transport.send(call).await?;
487 if reply.is_success() || accept.contains(&reply.status) {
488 return Ok(reply);
489 }
490 Err(classify_failure_in_region(
491 operation,
492 &reply,
493 transport.region(),
494 ))
495 }
496 };
497
498 let mut details = RetryDetails {
499 attempt_count: 1,
500 backoff: Duration::ZERO,
501 };
502 let result = attempt
503 .retry(retry_delays())
504 .when(Error::retryable)
505 .notify(|_, delay| {
506 details.attempt_count += 1;
507 details.backoff += delay;
508 })
509 .await;
510 result.map_err(|error| error.with_retry_details(details.attempt_count, details.backoff))
511}
512
513pub fn classify_failure(operation: &str, reply: &Reply) -> Error {
533 classify_failure_in_region(operation, reply, None)
534}
535
536fn classify_failure_in_region(operation: &str, reply: &Reply, region: Option<&Region>) -> Error {
537 let parsed: crate::control::ops::ServiceErrorWire =
538 serde_json::from_slice(&reply.body).unwrap_or_default();
539 let detail = parsed.message.as_deref().unwrap_or("").trim().to_string();
540
541 let (kind, explanation) = match reply.status {
542 status if reply.error_type.as_deref() == Some("ThrottlingException") => (
547 ErrorKind::Retryable,
548 format!(
549 "ThrottlingException, named by the x-amzn-ErrorType header on an HTTP {status} \
550 — retry the identical request."
551 ),
552 ),
553 400 => (
554 ErrorKind::Platform,
555 "ValidationException — the service refused a value. Every constraint the pinned \
556 service model states on a member this client sends is checked locally before the \
557 call — min as well as max, pattern, and enum, because nothing else validates a \
558 request here: this client signs with aws-sigv4 and sends with reqwest, so botocore's \
559 VALIDATED_METADATA_ATTRS never applies to it. So reaching this status means a \
560 constraint this client does not yet check, and closing it is a guard in \
561 crates/microvms-app/src/control plus a line in tools/check-model-drift.py."
562 .to_string(),
563 ),
564 402 => (
565 ErrorKind::Platform,
566 "ServiceQuotaExceededException — an account limit, not a request defect.".to_string(),
567 ),
568 403 if detail.is_empty() => (
569 ErrorKind::Credentials,
570 null_message_denial(operation, region),
571 ),
572 403 => (
573 ErrorKind::Credentials,
574 "AccessDeniedException, with a message — the service reported a genuine denial."
575 .to_string(),
576 ),
577 404 => (
578 ErrorKind::Platform,
579 "ResourceNotFoundException — the image or MicroVM does not exist.".to_string(),
580 ),
581 409 => (
582 ErrorKind::Platform,
583 "ConflictException — the resource is in a state that forbids this call. An image in \
584 CREATING refuses deletion, which is why teardown retries."
585 .to_string(),
586 ),
587 429 => (
588 ErrorKind::Retryable,
589 "ThrottlingException — retry the identical request.".to_string(),
590 ),
591 status if status >= 500 => (
592 ErrorKind::Retryable,
593 "InternalServerException — a service fault, retry the identical request.".to_string(),
594 ),
595 _ => (
596 ErrorKind::Platform,
597 "an unmodeled status: the service model declares 400, 402, 403, 404, 409, 429, and \
598 500 for these operations."
599 .to_string(),
600 ),
601 };
602
603 let said = if detail.is_empty() {
604 "the service sent no message".to_string()
605 } else {
606 format!("the service said {detail:?}")
607 };
608 Error::new(
609 kind,
610 format!(
611 "{operation} failed with HTTP {}: {said}. {explanation}",
612 reply.status
613 ),
614 )
615 .with_throttled(
616 reply.status == 429 || reply.error_type.as_deref() == Some("ThrottlingException"),
617 )
618}
619
620fn null_message_denial(operation: &str, region: Option<&Region>) -> String {
622 let availability = match region {
623 Some(region) if region.is_supported() => format!(
624 "{} carries MicroVMs; an IAM policy is the likely cause.",
625 region.as_str()
626 ),
627 Some(region) => format!(
628 "{} is not in the known MicroVMs region list; verify service availability as well as the IAM policy.",
629 region.as_str()
630 ),
631 None => "The cause may be an unsupported region or an IAM denial.".to_string(),
632 };
633 let region_arg = region
634 .map(|region| format!(" --region {}", region.as_str()))
635 .unwrap_or_default();
636 format!(
637 "AccessDeniedException with an empty or null message field. {availability} \
638 A create call with tags can require the implied lambda:TagResource action. \
639 Check the full denial in CloudTrail: aws cloudtrail lookup-events{region_arg} \
640 --lookup-attributes AttributeKey=EventName,AttributeValue={operation} \
641 --query 'Events[].CloudTrailEvent' --output text. Inspect errorCode and errorMessage \
642 for the denied action (docs/PLATFORM.md)."
643 )
644}
645
646pub fn error_type_name(header: &str) -> &str {
653 let name = header.split(':').next().unwrap_or(header);
654 name.split_once('#').map_or(name, |(_, after)| after).trim()
655}
656
657pub fn endpoint_for(region: &Region) -> String {
662 format!("https://{SIGNING_NAME}.{}.amazonaws.com", region.as_str())
663}
664
665pub mod paths {
672 use super::API_PATH_VERSION;
673
674 pub fn encode_segment(segment: &str) -> String {
682 use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};
683 const ESCAPED: &AsciiSet = &NON_ALPHANUMERIC
684 .remove(b'-')
685 .remove(b'_')
686 .remove(b'.')
687 .remove(b'~');
688 utf8_percent_encode(segment, ESCAPED).to_string()
689 }
690
691 pub fn microvm_images() -> String {
693 format!("/{API_PATH_VERSION}/microvm-images")
694 }
695
696 pub fn microvm_images_list(name_filter: Option<&str>, next_token: Option<&str>) -> String {
705 let mut query: Vec<String> = Vec::new();
706 if let Some(filter) = name_filter {
707 query.push(format!("nameFilter={}", encode_segment(filter)));
708 }
709 if let Some(token) = next_token {
710 query.push(format!("nextToken={}", encode_segment(token)));
711 }
712 if query.is_empty() {
713 return microvm_images();
714 }
715 format!("{}?{}", microvm_images(), query.join("&"))
716 }
717
718 pub fn microvm_image(image: &str) -> String {
720 format!(
721 "/{API_PATH_VERSION}/microvm-images/{}",
722 encode_segment(image)
723 )
724 }
725
726 fn with_next_token(path: String, next_token: Option<&str>) -> String {
735 match next_token {
736 Some(token) => format!("{path}?nextToken={}", encode_segment(token)),
737 None => path,
738 }
739 }
740
741 pub fn image_versions(image: &str, next_token: Option<&str>) -> String {
749 image_versions_paged(image, next_token, None)
750 }
751
752 pub fn image_versions_paged(
781 image: &str,
782 next_token: Option<&str>,
783 max_results: Option<u32>,
784 ) -> String {
785 let mut query: Vec<String> = Vec::new();
786 if let Some(size) = max_results {
787 query.push(format!("maxResults={}", size.clamp(1, 50)));
788 }
789 if let Some(token) = next_token {
790 query.push(format!("nextToken={}", encode_segment(token)));
791 }
792 let path = format!("{}/versions", microvm_image(image));
793 if query.is_empty() {
794 return path;
795 }
796 format!("{path}?{}", query.join("&"))
797 }
798
799 pub fn image_version(image: &str, version: &str) -> String {
810 format!(
811 "{}/versions/{}",
812 microvm_image(image),
813 encode_segment(version)
814 )
815 }
816
817 pub fn image_build(image: &str, version: &str, build_id: &str) -> String {
823 format!(
824 "{}/builds/{}",
825 image_version(image, version),
826 encode_segment(build_id)
827 )
828 }
829
830 pub fn managed_microvm_images(next_token: Option<&str>) -> String {
837 with_next_token(
838 format!("/{API_PATH_VERSION}/managed-microvm-images"),
839 next_token,
840 )
841 }
842
843 pub fn managed_image_versions(image: &str, next_token: Option<&str>) -> String {
851 with_next_token(
852 format!(
853 "/{API_PATH_VERSION}/managed-microvm-images/{}/versions",
854 encode_segment(image)
855 ),
856 next_token,
857 )
858 }
859
860 pub fn image_builds(image: &str, version: &str, next_token: Option<&str>) -> String {
863 with_next_token(
864 format!("{}/builds", image_version(image, version)),
865 next_token,
866 )
867 }
868
869 pub fn microvms() -> String {
876 format!("/{API_PATH_VERSION}/microvms")
877 }
878
879 pub fn microvms_list(next_token: Option<&str>) -> String {
881 with_next_token(microvms(), next_token)
882 }
883
884 pub fn microvms_list_matching(
888 image_identifier: Option<&str>,
889 image_version: Option<&str>,
890 next_token: Option<&str>,
891 ) -> String {
892 let query: Vec<String> = [
893 ("imageIdentifier", image_identifier),
894 ("imageVersion", image_version),
895 ("nextToken", next_token),
896 ]
897 .into_iter()
898 .filter_map(|(key, value)| value.map(|value| format!("{key}={}", encode_segment(value))))
899 .collect();
900 if query.is_empty() {
901 return microvms();
902 }
903 format!("{}?{}", microvms(), query.join("&"))
904 }
905
906 pub fn microvm(id: &str) -> String {
908 format!("/{API_PATH_VERSION}/microvms/{}", encode_segment(id))
909 }
910
911 pub fn suspend(id: &str) -> String {
913 format!("{}/suspend", microvm(id))
914 }
915
916 pub fn resume(id: &str) -> String {
918 format!("{}/resume", microvm(id))
919 }
920
921 pub fn auth_token(id: &str) -> String {
923 format!("{}/auth-token", microvm(id))
924 }
925
926 pub fn shell_auth_token(id: &str) -> String {
933 format!("{}/shell-auth-token", microvm(id))
934 }
935}
936
937#[cfg(test)]
938mod tests {
939 use super::*;
940
941 #[tokio::test]
942 async fn throttle_budget_counts_attempts_bounds_elapsed_and_keeps_other_failures_default() {
943 use crate::clock::Clock;
944 use crate::control::fake::{Answer, FakeControlPlane, TestClock};
945 for (status, policy, expected) in [
946 (429, None, 6),
947 (
948 429,
949 Some(RetryPolicy::new(3, Duration::from_secs(3600)).unwrap()),
950 3,
951 ),
952 (
953 429,
954 Some(RetryPolicy::new(9, Duration::from_millis(1)).unwrap()),
955 1,
956 ),
957 (
958 503,
959 Some(RetryPolicy::new(9, Duration::from_secs(3600)).unwrap()),
960 6,
961 ),
962 (
963 403,
964 Some(RetryPolicy::new(9, Duration::from_secs(3600)).unwrap()),
965 1,
966 ),
967 ] {
968 let transport = FakeControlPlane::new();
969 for _ in 0..expected {
970 transport.answer("RunMicrovm", Answer::failure(status, "failure"));
971 }
972 transport.answer("RunMicrovm", Answer::ok("{}"));
975 let clock = TestClock::new();
976 let error = send_configured(
977 &transport,
978 Call::get("RunMicrovm", "/run"),
979 &[],
980 policy,
981 &clock,
982 )
983 .await
984 .unwrap_err();
985 assert_eq!(transport.call_count("RunMicrovm"), expected);
986 assert_eq!(error.attempt_count(), expected as u32);
987 assert_eq!(error.throttled(), status == 429);
988 assert_eq!(error.backoff(), clock.elapsed());
989 if expected == 1 {
990 assert_eq!(clock.elapsed(), Duration::ZERO);
991 }
992 }
993 assert!(RetryPolicy::new(0, Duration::from_secs(1)).is_err());
994 assert!(RetryPolicy::new(1, Duration::ZERO).is_err());
995 }
996
997 #[tokio::test]
998 async fn a_retry_at_the_elapsed_budget_boundary_does_not_sleep() {
999 use crate::clock::Clock;
1000 use crate::control::fake::{Answer, FakeControlPlane, TestClock};
1001 let transport = FakeControlPlane::new();
1002 transport.answer("RunMicrovm", Answer::throttled_as_400());
1003 let gate = transport.hold_responses();
1004 let clock = TestClock::new();
1005 clock.set_unix_now(Duration::ZERO);
1006 let pending = send_configured(
1007 &transport,
1008 Call::get("RunMicrovm", "/run"),
1009 &[],
1010 Some(RetryPolicy::new(2, Duration::MAX).unwrap()),
1011 &clock,
1012 );
1013 tokio::pin!(pending);
1014 tokio::select! {
1015 biased;
1016 _ = &mut pending => panic!("response must wait for the gate"),
1017 _ = std::future::ready(()) => {}
1018 }
1019 assert_eq!(transport.call_count("RunMicrovm"), 1);
1020 clock.advance(Duration::MAX);
1023 gate.notify_one();
1024 let error = tokio::select! {
1027 biased;
1028 result = &mut pending => result.unwrap_err(),
1029 _ = std::future::ready(()) => panic!("elapsed budget must stop before another request"),
1030 };
1031 assert!(error.throttled());
1032 assert_eq!(error.attempt_count(), 1);
1033 assert_eq!(transport.call_count("RunMicrovm"), 1);
1034 assert_eq!(error.backoff(), Duration::ZERO);
1035 assert_eq!(clock.elapsed(), Duration::MAX);
1036 }
1037
1038 #[tokio::test]
1039 async fn configured_retries_recover_header_throttles_with_the_identical_call() {
1040 use crate::clock::Clock;
1041 use crate::control::fake::{Answer, FakeControlPlane, TestClock};
1042 let transport = FakeControlPlane::new();
1043 for _ in 0..7 {
1044 transport.answer("RunMicrovm", Answer::throttled_as_400());
1045 }
1046 transport.answer("RunMicrovm", Answer::ok("{}"));
1047 let clock = TestClock::new();
1048 let call = Call::post_json(
1049 "RunMicrovm",
1050 "/run",
1051 &serde_json::json!({"clientToken":"same-launch"}),
1052 )
1053 .unwrap();
1054 let (_, details) = send_configured(
1055 &transport,
1056 call.clone(),
1057 &[],
1058 Some(RetryPolicy::new(8, Duration::from_secs(3600)).unwrap()),
1059 &clock,
1060 )
1061 .await
1062 .unwrap();
1063 assert_eq!(details.attempt_count, 8);
1064 assert_eq!(details.backoff, clock.elapsed());
1065 assert!(details.backoff > Duration::ZERO);
1066 assert!(transport.calls().iter().all(|sent| sent == &call));
1067 transport.answer("GetMicrovm", Answer::failure(404, "missing"));
1068 let (reply, details) = send_configured(
1069 &transport,
1070 Call::get("GetMicrovm", "/missing"),
1071 &[404],
1072 None,
1073 &clock,
1074 )
1075 .await
1076 .unwrap();
1077 assert_eq!(reply.status, 404);
1078 assert_eq!(details.attempt_count, 1);
1079 assert_eq!(details.backoff, Duration::ZERO);
1080 }
1081
1082 #[tokio::test(start_paused = true)]
1083 async fn exhausted_public_retry_reports_throttle_attempts_and_backoff() {
1084 use crate::control::fake::{Answer, FakeControlPlane};
1085 for accepts_missing in [false, true] {
1086 let transport = FakeControlPlane::new();
1087 for _ in 0..6 {
1088 transport.answer("ListMicrovms", Answer::throttled_as_400());
1089 }
1090 transport.answer("ListMicrovms", Answer::ok("{}"));
1091 let call = Call::get("ListMicrovms", "/list");
1092 let error = if accepts_missing {
1093 send_accepting(&transport, call, &[404]).await
1094 } else {
1095 send_with_retry(&transport, call).await
1096 }
1097 .unwrap_err();
1098 assert_eq!(transport.call_count("ListMicrovms"), 6);
1099 assert!(error.throttled());
1100 assert_eq!(error.attempt_count(), 6);
1101 assert!(error.backoff() >= Duration::from_millis(6200));
1102 assert!(error.backoff() < Duration::from_millis(12400));
1103 }
1104 for status in [200, 404] {
1105 let transport = FakeControlPlane::new();
1106 transport.answer("GetMicrovm", Answer::failure(status, "accepted"));
1107 let reply = send_accepting(&transport, Call::get("GetMicrovm", "/missing"), &[404])
1108 .await
1109 .unwrap();
1110 assert_eq!(reply.status, status);
1111 assert_eq!(transport.call_count("GetMicrovm"), 1);
1112 }
1113 }
1114
1115 #[tokio::test]
1116 async fn a_cancel_signal_wakes_backoff_before_another_request() {
1117 use crate::clock::testing::YieldingClock;
1118 use crate::control::fake::{Answer, FakeControlPlane};
1119 let transport = FakeControlPlane::new();
1120 transport.answer("RunMicrovm", Answer::throttled_as_400());
1121 let clock = YieldingClock::new();
1122 let cancellation = RetryCancellation::default();
1123 let call = Call::get("RunMicrovm", "/run");
1124 let pending = send_cancellable(
1125 &transport,
1126 call,
1127 &[],
1128 Some(RetryPolicy::new(100, Duration::from_secs(3600)).unwrap()),
1129 &clock,
1130 Some(&cancellation),
1131 );
1132 tokio::pin!(pending);
1133 tokio::select! { biased; _ = &mut pending => panic!("backoff should yield"), _ = std::future::ready(()) => {} }
1134 cancellation.cancel();
1135 let error = pending.await.unwrap_err();
1136 assert_eq!(error.kind(), ErrorKind::Interrupted);
1137 assert_eq!(transport.call_count("RunMicrovm"), 1);
1138 }
1139
1140 #[tokio::test]
1141 async fn dropping_a_retry_future_prevents_another_request() {
1142 use crate::clock::testing::YieldingClock;
1143 use crate::control::fake::{Answer, FakeControlPlane};
1144 let transport = FakeControlPlane::new();
1145 transport.answer("RunMicrovm", Answer::throttled_as_400());
1146 let clock = YieldingClock::new();
1147 let mut future = Box::pin(send_configured(
1148 &transport,
1149 Call::get("RunMicrovm", "/run"),
1150 &[],
1151 Some(RetryPolicy::new(100, Duration::from_secs(3600)).unwrap()),
1152 &clock,
1153 ));
1154 tokio::select! { biased; _ = &mut future => panic!("retry sleep should yield"), _ = std::future::ready(()) => {} }
1155 drop(future);
1156 clock.advance(Duration::from_secs(3600));
1158 tokio::task::yield_now().await;
1159 tokio::task::yield_now().await;
1160 assert_eq!(transport.call_count("RunMicrovm"), 1);
1161 }
1162
1163 #[test]
1167 fn the_endpoint_uses_the_models_lambda_prefix_for_every_region() {
1168 assert_eq!(
1169 endpoint_for(&Region::UsEast1),
1170 "https://lambda.us-east-1.amazonaws.com"
1171 );
1172 assert_eq!(
1173 endpoint_for(&Region::ApNortheast1),
1174 "https://lambda.ap-northeast-1.amazonaws.com"
1175 );
1176 assert_eq!(
1178 endpoint_for(&Region::unlisted("eu-central-1")),
1179 "https://lambda.eu-central-1.amazonaws.com"
1180 );
1181 }
1182
1183 #[test]
1187 fn every_path_matches_the_models_request_uri() {
1188 assert_eq!(paths::microvm_images(), "/2025-09-09/microvm-images");
1189 assert_eq!(
1190 paths::microvm_image("img"),
1191 "/2025-09-09/microvm-images/img"
1192 );
1193 assert_eq!(
1194 paths::image_versions("img", None),
1195 "/2025-09-09/microvm-images/img/versions"
1196 );
1197 assert_eq!(
1198 paths::image_version("img", "1"),
1199 "/2025-09-09/microvm-images/img/versions/1"
1200 );
1201 assert_eq!(
1202 paths::image_builds("img", "1", None),
1203 "/2025-09-09/microvm-images/img/versions/1/builds"
1204 );
1205 assert_eq!(
1206 paths::image_build("img", "1", "build-abc"),
1207 "/2025-09-09/microvm-images/img/versions/1/builds/build-abc"
1208 );
1209 assert_eq!(
1210 paths::managed_microvm_images(None),
1211 "/2025-09-09/managed-microvm-images"
1212 );
1213 assert_eq!(
1214 paths::managed_image_versions(
1215 "arn:aws:lambda:us-east-1:aws:microvm-image:al2023-1",
1216 None
1217 ),
1218 "/2025-09-09/managed-microvm-images/\
1219 arn%3Aaws%3Alambda%3Aus-east-1%3Aaws%3Amicrovm-image%3Aal2023-1/versions"
1220 );
1221 assert_eq!(paths::microvms(), "/2025-09-09/microvms");
1222 assert_eq!(paths::microvms_list(None), "/2025-09-09/microvms");
1223 assert_eq!(paths::microvm("mvm-1"), "/2025-09-09/microvms/mvm-1");
1224 assert_eq!(
1225 paths::suspend("mvm-1"),
1226 "/2025-09-09/microvms/mvm-1/suspend"
1227 );
1228 assert_eq!(paths::resume("mvm-1"), "/2025-09-09/microvms/mvm-1/resume");
1229 assert_eq!(
1230 paths::auth_token("mvm-1"),
1231 "/2025-09-09/microvms/mvm-1/auth-token"
1232 );
1233 assert_eq!(
1234 paths::shell_auth_token("mvm-1"),
1235 "/2025-09-09/microvms/mvm-1/shell-auth-token"
1236 );
1237 }
1238
1239 #[test]
1246 fn the_image_listing_path_encodes_its_query_members() {
1247 assert_eq!(
1248 paths::microvm_images_list(None, None),
1249 "/2025-09-09/microvm-images"
1250 );
1251 assert_eq!(
1252 paths::microvm_images_list(Some("my-image"), None),
1253 "/2025-09-09/microvm-images?nameFilter=my-image"
1254 );
1255 assert_eq!(
1256 paths::microvm_images_list(Some("my-image"), Some("a+b/c=")),
1257 "/2025-09-09/microvm-images?nameFilter=my-image&nextToken=a%2Bb%2Fc%3D"
1258 );
1259 assert_eq!(
1260 paths::microvm_images_list(None, Some("token")),
1261 "/2025-09-09/microvm-images?nextToken=token"
1262 );
1263 }
1264
1265 #[test]
1276 fn the_other_paginated_listings_encode_their_cursor_too() {
1277 assert_eq!(
1278 paths::image_versions("img", Some("a+b/c=")),
1279 "/2025-09-09/microvm-images/img/versions?nextToken=a%2Bb%2Fc%3D"
1280 );
1281 assert_eq!(
1282 paths::image_builds("img", "1", Some("a+b/c=")),
1283 "/2025-09-09/microvm-images/img/versions/1/builds?nextToken=a%2Bb%2Fc%3D"
1284 );
1285 assert_eq!(
1286 paths::microvms_list(Some("a+b/c=")),
1287 "/2025-09-09/microvms?nextToken=a%2Bb%2Fc%3D"
1288 );
1289
1290 for path in [
1293 paths::image_versions("img", None),
1294 paths::image_builds("img", "1", None),
1295 paths::microvms_list(None),
1296 ] {
1297 assert!(
1298 !path.contains('?'),
1299 "an absent cursor emits no query: {path}"
1300 );
1301 }
1302 }
1303
1304 #[test]
1315 fn the_version_listings_explicit_page_size_is_ordered_and_clamped() {
1316 assert_eq!(
1317 paths::image_versions_paged("img", None, Some(1)),
1318 "/2025-09-09/microvm-images/img/versions?maxResults=1"
1319 );
1320 assert_eq!(
1323 paths::image_versions_paged("img", Some("a+b/c="), Some(2)),
1324 "/2025-09-09/microvm-images/img/versions?maxResults=2&nextToken=a%2Bb%2Fc%3D"
1325 );
1326 assert_eq!(
1327 paths::image_versions_paged("img", None, Some(0)),
1328 "/2025-09-09/microvm-images/img/versions?maxResults=1",
1329 "the model's min is 1; a 0 would be refused"
1330 );
1331 assert_eq!(
1332 paths::image_versions_paged("img", None, Some(999)),
1333 "/2025-09-09/microvm-images/img/versions?maxResults=50",
1334 "the model's max is 50; a larger value would be refused"
1335 );
1336
1337 assert_eq!(
1340 paths::image_versions_paged("img", None, None),
1341 paths::image_versions("img", None)
1342 );
1343 assert!(
1344 !paths::image_versions("img", Some("t")).contains("maxResults"),
1345 "no production path names a page size: {}",
1346 paths::image_versions("img", Some("t"))
1347 );
1348 }
1349
1350 #[test]
1363 fn the_managed_routes_are_their_own_collection_and_encode_their_cursor() {
1364 assert!(
1365 !paths::managed_microvm_images(None).contains('?'),
1366 "an absent cursor emits no query member"
1367 );
1368 assert_eq!(
1369 paths::managed_microvm_images(Some("a+b/c=")),
1370 "/2025-09-09/managed-microvm-images?nextToken=a%2Bb%2Fc%3D"
1371 );
1372 assert_eq!(
1373 paths::managed_image_versions("al2023-1", Some("a+b/c=")),
1374 "/2025-09-09/managed-microvm-images/al2023-1/versions?nextToken=a%2Bb%2Fc%3D"
1375 );
1376
1377 assert_ne!(
1380 paths::managed_microvm_images(None),
1381 paths::microvm_images_list(None, None)
1382 );
1383 assert!(
1384 !paths::microvm_images().contains("managed"),
1385 "the account's own collection must not acquire the managed prefix"
1386 );
1387 }
1388
1389 #[test]
1396 fn one_builds_path_is_the_listing_plus_an_encoded_segment() {
1397 let path = paths::image_build("img", "1", "4a4c5e30-811f-47fa-9893-260ea6a37a8f");
1398 assert_eq!(
1399 path,
1400 "/2025-09-09/microvm-images/img/versions/1/builds/4a4c5e30-811f-47fa-9893-260ea6a37a8f"
1401 );
1402 assert!(!path.contains('?'), "{path}");
1403 assert!(
1404 path.starts_with(&paths::image_builds("img", "1", None)),
1405 "the get path must extend the list path rather than be a second route: {path}"
1406 );
1407 assert_eq!(
1409 paths::image_build("img", "1", "a/b"),
1410 "/2025-09-09/microvm-images/img/versions/1/builds/a%2Fb"
1411 );
1412 }
1413
1414 #[test]
1418 fn deleting_one_version_carries_no_pagination_cursor() {
1419 let path = paths::image_version("img", "1");
1420 assert_eq!(path, "/2025-09-09/microvm-images/img/versions/1");
1421 assert!(!path.contains('?'), "{path}");
1422 assert!(!path.contains("nextToken"), "{path}");
1423 }
1424
1425 #[test]
1436 fn a_real_arn_identifier_is_percent_encoded_colons_and_all() {
1437 let arn = "arn:aws:lambda:us-east-1:123456789012:microvm-image:img";
1438 let path = paths::microvm_image(arn);
1439 assert_eq!(
1440 path,
1441 "/2025-09-09/microvm-images/arn%3Aaws%3Alambda%3Aus-east-1%3A123456789012%3Amicrovm-image%3Aimg"
1442 );
1443 assert!(
1444 !path["/2025-09-09/microvm-images/".len()..].contains(':'),
1445 "an unencoded colon in a path segment is a signature mismatch: {path}"
1446 );
1447 }
1448
1449 #[test]
1458 fn the_segment_encoder_encodes_slashes_wherever_they_come_from() {
1459 assert_eq!(paths::encode_segment("a/b"), "a%2Fb");
1460 let path = paths::microvm_image("has/a/slash");
1461 assert_eq!(path, "/2025-09-09/microvm-images/has%2Fa%2Fslash");
1462 assert!(
1463 !path["/2025-09-09/microvm-images/".len()..].contains('/'),
1464 "an encoded identifier contributes no path separators: {path}"
1465 );
1466 }
1467
1468 #[test]
1472 fn the_path_version_is_the_pinned_model_version() {
1473 assert_eq!(API_PATH_VERSION, crate::constants::MODEL_API_VERSION);
1474 assert!(paths::microvms().starts_with("/2025-09-09/"));
1475 }
1476
1477 #[test]
1480 fn the_signing_name_is_lambda_not_lambda_microvms() {
1481 assert_eq!(SIGNING_NAME, "lambda");
1482 }
1483
1484 #[test]
1486 fn a_null_message_access_denied_names_both_possible_causes() {
1487 let reply = Reply {
1488 status: 403,
1489 body: br#"{"message": null}"#.to_vec(),
1490 error_type: None,
1491 };
1492 let error = classify_failure("ListMicrovms", &reply);
1493 assert_eq!(error.kind(), ErrorKind::Credentials);
1494 let message = error.to_string();
1495 assert!(message.contains("null"), "{message}");
1496 assert!(message.contains("region"), "{message}");
1497 assert!(
1498 message.contains("unsupported region or an IAM denial"),
1499 "{message}"
1500 );
1501 assert!(message.contains("lambda:TagResource"), "{message}");
1502 assert!(message.contains("cloudtrail lookup-events"), "{message}");
1503 }
1504
1505 #[tokio::test]
1507 async fn supported_region_null_denials_point_to_iam_in_both_send_paths() {
1508 use crate::control::fake::{Answer, FakeControlPlane};
1509 let transport = FakeControlPlane::in_region(Region::UsEast1);
1510 transport.answer(
1511 "CreateMicrovmImage",
1512 Answer {
1513 status: 403,
1514 body: r#"{"message":null}"#.to_string(),
1515 error_type: Some("AccessDeniedException".to_string()),
1516 },
1517 );
1518 for accept in [false, true] {
1519 let call = Call::get("CreateMicrovmImage", "/images");
1520 let error = if accept {
1521 send_accepting(&transport, call, &[404]).await.unwrap_err()
1522 } else {
1523 send_with_retry(&transport, call).await.unwrap_err()
1524 };
1525 assert_eq!(error.kind(), ErrorKind::Credentials);
1526 assert!(!error.retryable());
1527 let message = error.to_string();
1528 for detail in [
1529 "us-east-1 carries MicroVMs",
1530 "policy is the likely cause",
1531 "lambda:TagResource",
1532 "cloudtrail lookup-events --region us-east-1",
1533 "AttributeValue=CreateMicrovmImage",
1534 "errorMessage",
1535 ] {
1536 assert!(message.contains(detail), "{message}");
1537 }
1538 assert!(!message.contains("Check the region before the policy"));
1539 }
1540 assert_eq!(transport.call_count("CreateMicrovmImage"), 2);
1541 }
1542
1543 #[test]
1545 fn unlisted_region_null_denials_keep_both_causes_visible() {
1546 let reply = Reply {
1547 status: 403,
1548 body: b"{}".to_vec(),
1549 error_type: None,
1550 };
1551 let error = classify_failure_in_region(
1552 "ListMicrovms",
1553 &reply,
1554 Some(&Region::unlisted("eu-central-1")),
1555 );
1556 let message = error.to_string();
1557 assert!(message.contains("verify service availability as well as the IAM policy"));
1558 assert!(message.contains("--region eu-central-1"));
1559 assert!(!message.contains("policy is the likely cause"));
1560 }
1561
1562 #[test]
1566 fn a_403_with_a_message_is_reported_as_a_genuine_denial() {
1567 let reply = Reply {
1568 status: 403,
1569 body: br#"{"message": "User: arn:aws:iam::1:user/u is not authorized"}"#.to_vec(),
1570 error_type: None,
1571 };
1572 let error = classify_failure("RunMicrovm", &reply);
1573 assert_eq!(error.kind(), ErrorKind::Credentials);
1574 let message = error.to_string();
1575 assert!(message.contains("not authorized"), "{message}");
1576 assert!(
1577 message.contains("genuine denial"),
1578 "must distinguish itself from the null-message case: {message}"
1579 );
1580 assert!(
1581 !message.contains("does not carry MicroVMs"),
1582 "must not blame the region when the service named a principal: {message}"
1583 );
1584 }
1585
1586 #[test]
1590 fn a_400_names_the_validation_refusal_and_the_local_checks_it_passed() {
1591 let reply = Reply {
1592 status: 400,
1593 body: br#"{"message": "1 validation error detected"}"#.to_vec(),
1594 error_type: None,
1595 };
1596 let error = classify_failure("RunMicrovm", &reply);
1597 assert_eq!(error.kind(), ErrorKind::Platform);
1598 let message = error.to_string();
1599 assert!(message.contains("ValidationException"), "{message}");
1600 assert!(message.contains("checked locally"), "{message}");
1601 assert!(!message.contains("unmodeled status"), "{message}");
1602 }
1603
1604 #[test]
1614 fn a_throttle_named_by_the_error_type_header_is_retryable_whatever_the_status() {
1615 let reply = Reply {
1616 status: 400,
1617 body: br#"{"message":null,"throttlingReasons":null}"#.to_vec(),
1618 error_type: Some("ThrottlingException".to_string()),
1619 };
1620 let error = classify_failure("ListMicrovms", &reply);
1621 assert!(error.retryable(), "{error}");
1622 assert!(error.throttled(), "the header identifies the throttle");
1623 assert_eq!(error.kind(), ErrorKind::Retryable);
1624 let message = error.to_string();
1625 assert!(message.contains("ThrottlingException"), "{message}");
1626 assert!(message.contains("HTTP 400"), "{message}");
1627 assert!(!message.contains("ValidationException"), "{message}");
1628 }
1629
1630 #[test]
1633 fn a_400_naming_another_error_type_is_still_a_validation_refusal() {
1634 let reply = Reply {
1635 status: 400,
1636 body: br#"{"message": "1 validation error detected"}"#.to_vec(),
1637 error_type: Some("ValidationException".to_string()),
1638 };
1639 let error = classify_failure("RunMicrovm", &reply);
1640 assert!(!error.retryable(), "{error}");
1641 assert!(error.to_string().contains("checked locally"), "{error}");
1642 }
1643
1644 #[test]
1651 fn the_error_type_name_drops_the_uri_suffix_and_the_namespace() {
1652 let cases = [
1653 (
1654 "ThrottlingException:http://internal.amazon.com/coral/com.amazon.coral.availability/",
1655 "ThrottlingException",
1656 ),
1657 (
1658 "aws.protocoltests.restjson#FooError:http://internal.amazon.com/coral/com.amazon.coral.validate/",
1659 "FooError",
1660 ),
1661 (
1662 "com.amazonaws.lambda#ResourceNotFoundException",
1663 "ResourceNotFoundException",
1664 ),
1665 ("ConflictException", "ConflictException"),
1666 ];
1667 for (header, name) in cases {
1668 assert_eq!(error_type_name(header), name, "{header}");
1669 }
1670 }
1671
1672 #[test]
1676 fn only_throttles_and_server_faults_are_retryable() {
1677 let cases = [
1678 (400, false),
1679 (402, false),
1680 (403, false),
1681 (404, false),
1682 (409, false),
1683 (429, true),
1684 (500, true),
1685 (502, true),
1686 (503, true),
1687 ];
1688 for (status, retryable) in cases {
1689 let reply = Reply {
1690 status,
1691 body: br#"{"message": "detail"}"#.to_vec(),
1692 error_type: None,
1693 };
1694 let error = classify_failure("GetMicrovm", &reply);
1695 assert_eq!(error.retryable(), retryable, "HTTP {status}");
1696 assert_eq!(error.throttled(), status == 429, "HTTP {status}");
1697 }
1698 }
1699
1700 #[test]
1704 fn a_control_plane_failure_carries_no_daemon_wire_kind() {
1705 for status in [400, 403, 404, 409, 429, 500] {
1706 let reply = Reply {
1707 status,
1708 body: Vec::new(),
1709 error_type: None,
1710 };
1711 assert_eq!(
1712 classify_failure("GetMicrovm", &reply).wire_kind(),
1713 None,
1714 "HTTP {status} never reached the daemon"
1715 );
1716 }
1717 }
1718
1719 #[test]
1723 fn an_unparseable_error_body_still_classifies_by_status() {
1724 let reply = Reply {
1725 status: 429,
1726 body: b"<html>Too Many Requests</html>".to_vec(),
1727 error_type: None,
1728 };
1729 let error = classify_failure("RunMicrovm", &reply);
1730 assert!(error.retryable(), "a throttle is a throttle");
1731 assert!(error.to_string().contains("no message"), "{error}");
1732 }
1733
1734 #[test]
1738 fn a_response_that_does_not_parse_reports_the_body() {
1739 let reply = Reply {
1740 status: 200,
1741 body: br#"{"microvmId": 42}"#.to_vec(),
1742 error_type: None,
1743 };
1744 let error = reply
1745 .json::<crate::control::ops::MicrovmResponseWire>("GetMicrovm")
1746 .expect_err("microvmId is a string shape");
1747 let message = error.to_string();
1748 assert!(message.contains("GetMicrovm"), "{message}");
1749 assert!(message.contains("microvmId"), "{message}");
1750 assert_eq!(error.kind(), ErrorKind::Platform);
1751 }
1752
1753 #[test]
1757 fn a_bodyless_post_sends_an_empty_json_object() {
1758 let call = Call::post_empty("SuspendMicrovm", paths::suspend("mvm-1"));
1759 assert_eq!(call.body.as_deref(), Some(&b"{}"[..]));
1760 assert_eq!(call.method, Method::Post);
1761 }
1762
1763 #[test]
1766 fn a_get_and_a_delete_carry_no_body() {
1767 let get = Call::get("GetMicrovm", paths::microvm("mvm-1"));
1768 assert_eq!(get.body, None);
1769 assert_eq!(get.body_bytes(), b"");
1770
1771 let delete = Call::delete("TerminateMicrovm", paths::microvm("mvm-1"));
1772 assert_eq!(delete.body, None);
1773 assert_eq!(delete.method, Method::Delete);
1774 }
1775
1776 #[test]
1778 fn the_methods_are_the_ones_the_model_uses_spelled_uppercase() {
1779 assert_eq!(Method::Get.as_str(), "GET");
1780 assert_eq!(Method::Post.as_str(), "POST");
1781 assert_eq!(Method::Patch.as_str(), "PATCH");
1782 assert_eq!(Method::Delete.as_str(), "DELETE");
1783 }
1784
1785 #[test]
1797 fn a_patch_carries_its_body_and_is_not_a_post() {
1798 let call = Call::patch_json(
1799 "UpdateMicrovmImageVersion",
1800 paths::image_version("img", "2.0"),
1801 &crate::control::ops::UpdateImageVersionWire {
1802 status: crate::control::ops::VersionStatus::Inactive,
1803 },
1804 )
1805 .expect("serialises");
1806
1807 assert_eq!(call.method, Method::Patch);
1808 assert_ne!(
1809 call.method,
1810 Method::Post,
1811 "a POST to the version path is a route the model does not declare"
1812 );
1813 assert_eq!(call.path, "/2025-09-09/microvm-images/img/versions/2.0");
1814 assert_eq!(
1815 call.body.as_deref(),
1816 Some(&br#"{"status":"INACTIVE"}"#[..]),
1817 "the one member the request has must reach the body"
1818 );
1819 assert_eq!(call.operation, "UpdateMicrovmImageVersion");
1820 }
1821}