diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs index a046304140..36f472d0b0 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/adapter_characterization_tests/pooled_adoption.rs @@ -31,6 +31,10 @@ const WAKE: &str = "WorkerApiInternal/WakeSandbox"; /// The per-sandbox setup steps, in the order adoption applies them. const SETUP_STEPS: [&str; 3] = [UPDATE, SET_NETWORK_POLICY, SET_AUTO_LIFECYCLE]; +/// Zero-hint rate limits that spend the node-api client's whole retry count. +const RATE_LIMITS: usize = + pplx_space_common_grpc::DEFAULT_TCP_GRPC_CONNECT_RETRY_POLICY.max_retries; + /// The status the node-api client returns when its own budget for `command` /// runs out. fn client_deadline(command: &str) -> tonic::Status { @@ -255,6 +259,9 @@ async fn pooled_adoption_applies_per_sandbox_setup_before_route_adoption() -> Te enum SetupFailure { Update(tonic::Status), NetworkPolicy(tonic::Status), + /// The node rate-limits the policy update with a zero retry hint until + /// the client's retry count is spent, then fails it with this status. + NetworkPolicyAfterRateLimits(tonic::Status), /// The gateway confirmed a policy other than the request-resolved one. NetworkPolicyMismatch, AutoLifecycle(tonic::Status), @@ -275,8 +282,11 @@ enum SetupFailure { StatusCode::SERVICE_UNAVAILABLE, true )] +// What node-api relays when its sandbox-manager connection closes under a +// queued request. Without its local "transport error" source the retrying +// client does not replay it. #[case::update_cancelled( - SetupFailure::Update(tonic::Status::cancelled("Timeout expired")), + SetupFailure::Update(tonic::Status::cancelled("operation was canceled")), StatusCode::REQUEST_TIMEOUT, false )] @@ -295,6 +305,16 @@ enum SetupFailure { StatusCode::INTERNAL_SERVER_ERROR, false )] +// Zero-hint rate limits replay immediately but count against the client's +// retry limit, so the `Unavailable` that follows them comes back unreplayed +// well inside the setup budget. +#[case::policy_unavailable_after_rate_limits( + SetupFailure::NetworkPolicyAfterRateLimits(tonic::Status::unavailable( + "synchronous network policy update requires a live net-gateway" + )), + StatusCode::SERVICE_UNAVAILABLE, + true +)] #[case::policy_mismatch( SetupFailure::NetworkPolicyMismatch, StatusCode::SERVICE_UNAVAILABLE, @@ -325,6 +345,16 @@ async fn pooled_adoption_stops_at_a_failed_setup_step( SetupFailure::NetworkPolicy(status) => { (SET_NETWORK_POLICY, ScriptedReply::Status(status.clone())) } + SetupFailure::NetworkPolicyAfterRateLimits(status) => { + let rate_limited = pplx_space_common_grpc::rate_limited_status( + "UDS gRPC request-rate limit exceeded; retry_after_ms=0", + Some(std::time::Duration::ZERO), + ); + let replies = std::iter::repeat_n(ScriptedReply::Status(rate_limited), RATE_LIMITS) + .chain([ScriptedReply::Status(status.clone())]) + .collect(); + (SET_NETWORK_POLICY, ScriptedReply::Sequence(replies)) + } SetupFailure::NetworkPolicyMismatch => ( SET_NETWORK_POLICY, network_policy_confirmed(&NetworkPolicyV1::allow_all()), @@ -354,6 +384,13 @@ async fn pooled_adoption_stops_at_a_failed_setup_step( !worker.received(failed_step).is_empty(), "{failure:?}: the failing step must have been reached" ); + if let SetupFailure::NetworkPolicyAfterRateLimits(_) = &failure { + assert_eq!( + worker.received(SET_NETWORK_POLICY).len(), + RATE_LIMITS + 1, + "every rate limit was replayed and the final status was not" + ); + } let later_steps = SETUP_STEPS .iter() .skip_while(|step| **step != failed_step) diff --git a/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs b/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs index a92e065ea1..f3c49180c8 100644 --- a/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs +++ b/crates/libs/api/src/controllers/substrate/kubernetes/scripted_test_doubles.rs @@ -145,12 +145,25 @@ type CallLog = Arc>>; pub(super) enum ScriptedReply { Status(tonic::Status), Message(Bytes), + /// Answers the n-th call to its method with the n-th reply and every + /// later call with the last one. An empty sequence scripts nothing. + Sequence(Arc<[ScriptedReply]>), } impl ScriptedReply { pub(super) fn message(message: &impl prost::Message) -> Self { Self::Message(Bytes::from(message.encode_to_vec())) } + + /// The reply to the call numbered `attempt`, counting from zero. + fn for_attempt(&self, attempt: usize) -> &Self { + match self { + Self::Sequence(replies) => replies + .get(attempt.min(replies.len().saturating_sub(1))) + .map_or(self, |reply| reply.for_attempt(attempt)), + reply => reply, + } + } } /// Runs when a call to its method arrives, before the call is answered, so a @@ -206,18 +219,23 @@ where .await .map(|body| body.to_bytes()) .unwrap_or_default(); - // Strip the 5-byte gRPC length-prefix frame header. - calls.lock().unwrap().push(GrpcCall { - method, - message: framed.slice(5.min(framed.len())..), - }); + let attempt = { + let mut calls = calls.lock().unwrap(); + let attempt = calls.iter().filter(|call| call.method == method).count(); + // Strip the 5-byte gRPC length-prefix frame header. + calls.push(GrpcCall { + method, + message: framed.slice(5.min(framed.len())..), + }); + attempt + }; if let Some(observer) = observer { observer().await; } - match reply { - Some(ScriptedReply::Status(status)) => Ok(status.into_http()), - Some(ScriptedReply::Message(message)) => Ok(grpc_unary_ok(message)), - None => { + match reply.as_ref().map(|reply| reply.for_attempt(attempt)) { + Some(ScriptedReply::Status(status)) => Ok(status.clone().into_http()), + Some(ScriptedReply::Message(message)) => Ok(grpc_unary_ok(message.clone())), + Some(ScriptedReply::Sequence(_)) | None => { let request = http::Request::from_parts( parts, tonic::body::Body::new(http_body_util::Full::new(framed)),