fix(spider-client): Make the gRPC client's request futures Send.

components/spider-client/src/client.rs

@@ -290,3 +290,32 @@ impl SpiderClientBuilder {
 }
 
 const DEFAULT_POOL_SIZE: NonZeroUsize = NonZeroUsize::new(8).unwrap();
+
+/// Compile-time assertion that the public client handles are `Send + Sync`.
+const _: () = {
+    const fn assert_send_sync<T: Send + Sync>() {}
+    assert_send_sync::<SpiderClient>();
+    assert_send_sync::<SpiderClientBuilder>();
+};
+
+/// Compile-time assertion that every public [`SpiderClient`] async method returns a `Send` future.
+///
+/// This function is never called; it exists only to type-check the `Send` bound on each returned
+/// future. The arguments are taken as parameters (never constructed) so no real values are needed.
+#[expect(dead_code, reason = "compile-time-only `Send` assertion; never called")]
+fn assert_client_futures_send(
+    client: &SpiderClient,
+    resource_group_id: ResourceGroupId,
+    job_id: JobId,
+    task_graph: &TaskGraph,
+) {
+    const fn assert_send<FutureType: Send>(_: &FutureType) {}
+    assert_send(&client.submit_job(resource_group_id, task_graph, Vec::new()));
+    assert_send(&client.start_job(job_id));
+    assert_send(&client.cancel_job(job_id));
+    assert_send(&client.get_job_state(job_id));
+    assert_send(&client.get_job_outputs(job_id));
+    assert_send(&client.get_job_error(job_id));
+    assert_send(&client.add_resource_group(String::new(), Vec::new()));
+    assert_send(&client.verify_resource_group(resource_group_id, Vec::new()));
+}

components/spider-client/src/grpc/job.rs

@@ -85,16 +85,15 @@ impl JobOrchestrationClient {
             .to_zstd_compressed_json()
             .map_err(|error| ClientError::Serialization(error.to_string()))?;
         let compressed_serialized_inputs = serialize_inputs(inputs)?;
-        let request = storage::RegisterJobRequest {
-            resource_group_id: resource_group_id.get(),
-            compressed_serialized_task_graph,
-            compressed_serialized_inputs,
-        };
-        let response = call_with_retry(self.retry_config, async || {
-            self.connection_pool
-                .get_client()
-                .register_job(request.clone())
-                .await
+        let pool = self.connection_pool.clone();
+        let response = call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::RegisterJobRequest {
+                resource_group_id: resource_group_id.get(),
+                compressed_serialized_task_graph: compressed_serialized_task_graph.clone(),
+                compressed_serialized_inputs: compressed_serialized_inputs.clone(),
+            };
+            async move { client.register_job(request).await }
         })
         .await
         .map_err(|status| job_status_to_error(&status))?
@@ -116,11 +115,13 @@ impl JobOrchestrationClient {
     /// * Forwards [`JobOrchestrationServiceClient::start_job`]'s status on failure.
     /// * Forwards [`job_state_response_to_result`]'s return values on failure.
     pub async fn start_job(&self, job_id: JobId) -> Result<JobState, ClientError> {
-        let request = storage::JobIdRequest {
-            job_id: job_id.get(),
-        };
-        let response = call_with_retry(self.retry_config, async || {
-            self.connection_pool.get_client().start_job(request).await
+        let pool = self.connection_pool.clone();
+        let response = call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::JobIdRequest {
+                job_id: job_id.get(),
+            };
+            async move { client.start_job(request).await }
         })
         .await
         .map_err(|status| job_status_to_error(&status))?
@@ -142,11 +143,13 @@ impl JobOrchestrationClient {
     /// * Forwards [`JobOrchestrationServiceClient::cancel_job`]'s status on failure.
     /// * Forwards [`job_state_response_to_result`]'s return values on failure.
     pub async fn cancel_job(&self, job_id: JobId) -> Result<JobState, ClientError> {
-        let request = storage::JobIdRequest {
-            job_id: job_id.get(),
-        };
-        let response = call_with_retry(self.retry_config, async || {
-            self.connection_pool.get_client().cancel_job(request).await
+        let pool = self.connection_pool.clone();
+        let response = call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::JobIdRequest {
+                job_id: job_id.get(),
+            };
+            async move { client.cancel_job(request).await }
         })
         .await
         .map_err(|status| job_status_to_error(&status))?
@@ -168,14 +171,13 @@ impl JobOrchestrationClient {
     /// * Forwards [`JobOrchestrationServiceClient::get_job_state`]'s status on failure.
     /// * Forwards [`job_state_response_to_result`]'s return values on failure.
     pub async fn get_job_state(&self, job_id: JobId) -> Result<JobState, ClientError> {
-        let request = storage::JobIdRequest {
-            job_id: job_id.get(),
-        };
-        let response = call_with_retry(self.retry_config, async || {
-            self.connection_pool
-                .get_client()
-                .get_job_state(request)
-                .await
+        let pool = self.connection_pool.clone();
+        let response = call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::JobIdRequest {
+                job_id: job_id.get(),
+            };
+            async move { client.get_job_state(request).await }
         })
         .await
         .map_err(|status| job_status_to_error(&status))?
@@ -199,14 +201,13 @@ impl JobOrchestrationClient {
     ///   [`ClientError::Deserialization`].
     /// * Forwards [`JobOrchestrationServiceClient::get_job_outputs`]'s status on failure.
     pub async fn get_job_outputs(&self, job_id: JobId) -> Result<Vec<TaskOutput>, ClientError> {
-        let request = storage::JobIdRequest {
-            job_id: job_id.get(),
-        };
-        let response = call_with_retry(self.retry_config, async || {
-            self.connection_pool
-                .get_client()
-                .get_job_outputs(request)
-                .await
+        let pool = self.connection_pool.clone();
+        let response = call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::JobIdRequest {
+                job_id: job_id.get(),
+            };
+            async move { client.get_job_outputs(request).await }
         })
         .await
         .map_err(|status| job_status_to_error(&status))?
@@ -228,14 +229,13 @@ impl JobOrchestrationClient {
     ///
     /// * Forwards [`JobOrchestrationServiceClient::get_job_error`]'s status on failure.
     pub async fn get_job_error(&self, job_id: JobId) -> Result<String, ClientError> {
-        let request = storage::JobIdRequest {
-            job_id: job_id.get(),
-        };
-        let response = call_with_retry(self.retry_config, async || {
-            self.connection_pool
-                .get_client()
-                .get_job_error(request)
-                .await
+        let pool = self.connection_pool.clone();
+        let response = call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::JobIdRequest {
+                job_id: job_id.get(),
+            };
+            async move { client.get_job_error(request).await }
         })
         .await
         .map_err(|status| job_status_to_error(&status))?

components/spider-client/src/grpc/resource_group.rs

@@ -69,15 +69,14 @@ impl ResourceGroupManagementClient {
         external_resource_group_id: String,
         password: Vec<u8>,
     ) -> Result<ResourceGroupId, ClientError> {
-        let request = storage::AddResourceGroupRequest {
-            external_resource_group_id,
-            password,
-        };
-        let response = call_with_retry(self.retry_config, async || {
-            self.connection_pool
-                .get_client()
-                .add_resource_group(request.clone())
-                .await
+        let pool = self.connection_pool.clone();
+        let response = call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::AddResourceGroupRequest {
+                external_resource_group_id: external_resource_group_id.clone(),
+                password: password.clone(),
+            };
+            async move { client.add_resource_group(request).await }
         })
         .await
         .map_err(|status| resource_group_status_to_error(&status))?
@@ -103,15 +102,14 @@ impl ResourceGroupManagementClient {
         resource_group_id: ResourceGroupId,
         password: Vec<u8>,
     ) -> Result<(), ClientError> {
-        let request = storage::VerifyResourceGroupRequest {
-            resource_group_id: resource_group_id.get(),
-            password,
-        };
-        call_with_retry(self.retry_config, async || {
-            self.connection_pool
-                .get_client()
-                .verify_resource_group(request.clone())
-                .await
+        let pool = self.connection_pool.clone();
+        call_with_retry(self.retry_config, move || {
+            let mut client = pool.get_client();
+            let request = storage::VerifyResourceGroupRequest {
+                resource_group_id: resource_group_id.get(),
+                password: password.clone(),
+            };
+            async move { client.verify_resource_group(request).await }
         })
         .await
         .map_err(|status| resource_group_status_to_error(&status))?;