test: cut full-suite runtime from 163s to 44s

crates/ployz/src/cloud_enroll.rs

@@ -697,6 +697,11 @@ mod tests {
             let server = tokio::spawn(async move {
                 let (mut held, _) = listener.accept().await.unwrap();
                 read_http(&mut held).await;
+                // The request has reached the server, so its timeout is armed.
+                // Advance only that wait; use real time again for the retry's I/O.
+                tokio::time::pause();
+                tokio::time::advance(REQUEST_TIMEOUT).await;
+                tokio::time::resume();
                 // Keep the first response open until the retry has succeeded.
                 let (mut retry, _) = listener.accept().await.unwrap();
                 read_http(&mut retry).await;

crates/ployz/src/connect.rs

@@ -837,3 +837,6 @@ fn rpc_error_code(code: tonic::Code) -> RpcErrorCode {
 #[cfg(test)]
 #[path = "connect/tests.rs"]
 mod tests;
+
+#[cfg(test)]
+pub(crate) mod test_support;

crates/ployz/src/connect/test_support.rs

@@ -0,0 +1,79 @@
+//! In-memory RPC transport for tests that advance Tokio time.
+
+use std::{
+    convert::Infallible,
+    future::Future,
+    sync::{Arc, Mutex},
+};
+
+use futures_util::StreamExt;
+use ployz_core::OpaquePayload;
+use tonic::{
+    Request, Response, Status,
+    codec::ProstCodec,
+    transport::{Channel, Server},
+};
+
+use super::{BoxProxyStream, Client, ConnectError, Connector};
+use crate::context::{Connection, ConnectionSource};
+
+struct Connected(Channel);
+
+#[tonic::async_trait]
+impl Connector for Connected {
+    async fn connect(&self, _: &Connection) -> Result<Channel, ConnectError> {
+        Ok(self.0.clone())
+    }
+    async fn dial_proxy(
+        &self,
+        _: &Connection,
+        _: &str,
+        _: &str,
+    ) -> Result<BoxProxyStream, ConnectError> {
+        unreachable!("this RPC fixture does not open a proxy")
+    }
+}
+
+pub(crate) async fn rpc_client<F, Fut>(
+    rpc: F,
+) -> (
+    Client,
+    tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
+)
+where
+    F: Fn(Request<OpaquePayload>) -> Fut + Clone + Send + Sync + 'static,
+    Fut: Future<Output = Result<Response<OpaquePayload>, Status>> + Send + 'static,
+{
+    let rpc = tower::service_fn(rpc);
+    let service = tower::service_fn(move |request: http::Request<tonic::body::Body>| {
+        let rpc = rpc.clone();
+        async move {
+            Ok::<_, Infallible>(
+                tonic::server::Grpc::new(ProstCodec::default())
+                    .unary(rpc, request)
+                    .await,
+            )
+        }
+    });
+    // Both peers run inside Tokio; clock advancement cannot outrun socket I/O.
+    let (client_io, server_io) = tokio::io::duplex(64 * 1024);
+    let incoming =
+        tokio_stream::once(Ok::<_, std::io::Error>(server_io)).chain(tokio_stream::pending());
+    let server = tokio::spawn(Server::builder().serve_with_incoming(service, incoming));
+    let client_io = Arc::new(Mutex::new(Some(client_io)));
+    let channel = Channel::from_static("http://memory.invalid")
+        .connect_with_connector(tower::service_fn(move |_| {
+            std::future::ready(Ok::<_, std::io::Error>(hyper_util::rt::TokioIo::new(
+                client_io.lock().unwrap().take().unwrap(),
+            )))
+        }))
+        .await
+        .unwrap();
+    let client = Client::new(
+        channel.clone(),
+        Connection::tcp("127.0.0.1:1".parse().unwrap()),
+        ConnectionSource::Direct,
+        Arc::new(Connected(channel)),
+    );
+    (client, server)
+}

crates/ployz/src/global_catch_up_tests.rs

@@ -786,3 +786,106 @@ fn container_on(
     })
     .unwrap()
 }
+
+#[tokio::test(start_paused = true)]
+async fn real_catch_up_client_retries_readiness_and_placement_to_their_budget() {
+    use ployz_core::{
+        ContainerCreated, InspectTelemetry, LocalMachinePhase, MachineDetails, OpaquePayload,
+        RpcRequestBody, RpcResponse,
+    };
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
+    use std::time::Duration;
+    use tonic::{Request, Response, Status};
+
+    for placement in [false, true] {
+        for failures in [1, 4] {
+            let target = machine('1', "joiner");
+            let observed = target.clone();
+            let calls = Arc::new(AtomicUsize::new(0));
+            let attempts = calls.clone();
+            let request = EnsureGlobalSlotRequest {
+                project_name: ProjectName::parse("app").unwrap(),
+                resolved_spec: requested(ServiceMode::Global)
+                    .to_resolved(service_id('a'), ResolvedUpdateConfig::default())
+                    .unwrap(),
+            };
+            let expected = request.clone();
+            let (mut client, server) =
+                crate::connect::test_support::rpc_client(move |rpc: Request<OpaquePayload>| {
+                    let target = observed.clone();
+                    let expected = expected.clone();
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        assert_eq!(
+                            rpc.metadata().get(ployz_core::ONE_TARGET_HEADER).unwrap(),
+                            target.id.as_str()
+                        );
+                        #[expect(
+                            clippy::wildcard_enum_match_arm,
+                            reason = "fixture accepts only the two catch-up RPCs under test"
+                        )]
+                        let response = match rpc.into_inner().decode_request().unwrap().body {
+                            RpcRequestBody::Inspect(inspect) => {
+                                assert!(!placement);
+                                assert_eq!(inspect.telemetry, InspectTelemetry::BridgeCapacity);
+                                RpcResponse::from(MachineDetails {
+                                    id: target.id,
+                                    phase: LocalMachinePhase::Participating,
+                                    public_key: target.public_key,
+                                    advertised_endpoints: Vec::new(),
+                                    machine: Some(target),
+                                    store_version: Default::default(),
+                                    rtts: Vec::new(),
+                                    cloud_paired: false,
+                                    telemetry: None,
+                                    storage: None,
+                                })
+                            }
+                            RpcRequestBody::EnsureGlobalSlot(ensure) => {
+                                assert!(placement);
+                                assert_eq!(ensure, expected);
+                                RpcResponse::from(ContainerCreated {
+                                    container_id: container_id('a'),
+                                    display_name: "api".into(),
+                                })
+                            }
+                            other => panic!("unexpected catch-up RPC: {other:?}"),
+                        };
+                        if attempt < failures {
+                            Err(Status::unavailable("transient catch-up failure"))
+                        } else {
+                            Ok(Response::new(response.encode().unwrap()))
+                        }
+                    }
+                })
+                .await;
+            let started = tokio::time::Instant::now();
+            let succeeded = if placement {
+                CatchUpClient::ensure_global_slot(&mut client, &target.id, request)
+                    .await
+                    .is_ok()
+            } else {
+                CatchUpClient::bridge_capacity(&mut client, &target.id)
+                    .await
+                    .is_ok()
+            };
+            assert_eq!(succeeded, failures == 1);
+            assert_eq!(
+                calls.load(Ordering::SeqCst),
+                if failures == 1 { 2 } else { 4 }
+            );
+            assert_eq!(
+                started.elapsed(),
+                if failures == 1 {
+                    Duration::from_millis(500)
+                } else {
+                    Duration::from_secs(6)
+                }
+            );
+            server.abort();
+        }
+    }
+}

crates/ployz/src/handlers/machine/helpers.rs

@@ -334,3 +334,7 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+#[path = "setup_tests.rs"]
+mod setup_tests;

crates/ployz/src/handlers/machine/setup_tests.rs

@@ -0,0 +1,129 @@
+//! Real setup RPC deadlines and initialization recovery under virtual time.
+
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+
+use crate::connect::Client;
+use ployz_core::{
+    InitializeRequest, InspectRequest, LocalMachinePhase, Machine, MachineDetails, MachineId,
+    MachineName, OpaquePayload, RpcRequestBody, RpcResponse, WireGuardPublicKey, op,
+};
+use tokio::time::Instant;
+use tonic::{Request, Response, Status};
+
+async fn starting_machine() -> (
+    Client,
+    Machine,
+    Arc<AtomicUsize>,
+    tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
+) {
+    let machine = Machine {
+        id: MachineId::random(),
+        name: MachineName::parse("founder").unwrap(),
+        subnet: "10.210.0.0/24".parse().unwrap(),
+        public_key: WireGuardPublicKey([2; 32]),
+        public_ip: None,
+        advertised_endpoints: Vec::new(),
+        runtime: Default::default(),
+    };
+    let initialized = Arc::new(AtomicUsize::new(0));
+    let calls = initialized.clone();
+    let observed = machine.clone();
+    let ready = Instant::now() + Duration::from_secs(70);
+    let rpc = move |request: Request<OpaquePayload>| {
+        let calls = calls.clone();
+        let machine = observed.clone();
+        async move {
+            #[expect(
+                clippy::wildcard_enum_match_arm,
+                reason = "this fixture rejects every RPC except Initialize and Inspect"
+            )]
+            match request.into_inner().decode_request().unwrap().body {
+                RpcRequestBody::Initialize(_) => {
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    std::future::pending::<Result<Response<OpaquePayload>, Status>>().await
+                }
+                RpcRequestBody::Inspect(_) => {
+                    let started = calls.load(Ordering::SeqCst) > 0;
+                    if started {
+                        tokio::time::sleep_until(ready).await;
+                    }
+                    let details = MachineDetails {
+                        id: machine.id,
+                        phase: if started {
+                            LocalMachinePhase::Participating
+                        } else {
+                            LocalMachinePhase::Uninitialized
+                        },
+                        public_key: machine.public_key,
+                        advertised_endpoints: Vec::new(),
+                        machine: started.then_some(machine),
+                        store_version: Default::default(),
+                        rtts: Vec::new(),
+                        cloud_paired: false,
+                        telemetry: None,
+                        storage: None,
+                    };
+                    Ok(Response::new(RpcResponse::from(details).encode().unwrap()))
+                }
+                request => panic!("unexpected setup request: {request:?}"),
+            }
+        }
+    };
+    let (client, server) = crate::connect::test_support::rpc_client(rpc).await;
+    (client, machine, initialized, server)
+}
+
+fn initialize_request(machine: &Machine) -> InitializeRequest {
+    InitializeRequest {
+        name: machine.name.clone(),
+        cluster_network: "10.210.0.0/16".parse().unwrap(),
+        public_ip: None,
+        advertised_endpoints: Vec::new(),
+        wireguard_mtu: None,
+        cloud_pairing: None,
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn initialize_recovers_beyond_the_read_budget_without_replaying_the_mutation() {
+    let started = Instant::now();
+    let (mut client, machine, calls, server) = starting_machine().await;
+    let result = super::initialize(&mut client, initialize_request(&machine))
+        .await
+        .unwrap();
+    assert_eq!(result.machine, machine);
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    assert_eq!(Instant::now() - started, Duration::from_secs(70));
+    server.abort();
+}
+
+#[tokio::test(start_paused = true)]
+async fn held_setup_rpcs_keep_the_mutation_and_read_deadlines() {
+    let (mut client, machine, calls, server) = starting_machine().await;
+    let started = Instant::now();
+    let error = client
+        .call_unretried::<op::Initialize>(initialize_request(&machine), None)
+        .await
+        .unwrap_err();
+    assert_eq!(Instant::now() - started, Duration::from_secs(5));
+    assert!(
+        error
+            .to_string()
+            .contains("Machine setup mutation reply timed out")
+    );
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    let started = Instant::now();
+    let error = client
+        .call_repeatable::<op::Inspect>(InspectRequest::default(), None)
+        .await
+        .unwrap_err();
+    assert_eq!(Instant::now() - started, Duration::from_secs(60));
+    assert!(error.to_string().contains("Machine setup read timed out"));
+    server.abort();
+}