1use std::{
2 collections::HashMap,
3 time::{Duration, Instant},
4};
5
6use anyhow::{Context, Result, ensure};
7use async_trait::async_trait;
8use serde::{Deserialize, Serialize};
9
10use crate::host::HostId;
11
12mod command_process;
13
14const PROVIDER_REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
15const MAX_PROVIDER_OUTPUT_BYTES: usize = 1024 * 1024;
16
17#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
18#[serde(rename_all = "camelCase")]
19pub struct EnsureHostRequest {
20 pub namespace_id: String,
21 pub code_revision: String,
22 pub canonical_region: String,
23 pub host_id: HostId,
24 pub session_id: String,
25 pub host_token: String,
26 pub jwt_public_keys: String,
27 pub control_plane_url: String,
28 pub jwt_issuer: String,
29 pub invocation_jwt_audience: String,
30 pub image_ref: String,
31 pub working_directory: String,
32 pub actor_entrypoint: Option<String>,
33 pub actor_idle_timeout_ms: u64,
34 pub host_idle_timeout_ms: u64,
35}
36
37#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
38#[serde(rename_all = "camelCase")]
39pub struct ActorHostHandle {
40 pub host_id: HostId,
41 pub route: String,
42 pub canonical_region: String,
43 pub provisioning: Option<ActorHostProvisioning>,
44}
45
46#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
47#[serde(rename_all = "camelCase")]
48pub struct ActorHostProvisioning {
49 pub provider: String,
50 pub resource_id: String,
51 pub reused: bool,
52 pub started_at_ms: u64,
53 pub input_parsed_at_ms: Option<u64>,
54 pub sdk_loaded_at_ms: Option<u64>,
55 pub resources_resolved_at_ms: Option<u64>,
56 pub existing_host_checked_at_ms: Option<u64>,
57 pub sandbox_scheduled_at_ms: Option<u64>,
58 pub host_ready_observed_at_ms: Option<u64>,
59 pub route_read_at_ms: Option<u64>,
60 pub metadata_written_at_ms: Option<u64>,
61 pub completed_at_ms: u64,
62 #[serde(default)]
63 pub command_spawned_at_ms: Option<u64>,
64 #[serde(default)]
65 pub request_written_at_ms: Option<u64>,
66 #[serde(default)]
67 pub process_completed_at_ms: Option<u64>,
68 #[serde(default)]
69 pub response_decoded_at_ms: Option<u64>,
70}
71
72#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
73#[serde(rename_all = "camelCase")]
74pub struct WarmImageRequest {
75 pub namespace_id: String,
76 pub code_revision: String,
77 pub canonical_region: String,
78 pub image_ref: String,
79}
80
81#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
82#[serde(rename_all = "camelCase")]
83pub struct ImageWarmup {
84 pub provider: String,
85 pub resource_id: String,
86 pub total_ms: u64,
87}
88
89#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
90#[serde(rename_all = "camelCase")]
91pub struct TerminateHostsRequest {
92 pub namespace_id: String,
93 pub code_revision: String,
94 pub canonical_regions: Vec<String>,
95}
96
97#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
98#[serde(rename_all = "camelCase")]
99pub struct HostTermination {
100 pub provider: String,
101 pub resource_ids: Vec<String>,
102}
103
104#[async_trait]
105pub trait SandboxProvider: Send + Sync {
106 async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle>;
107 async fn warm_image(&self, request: &WarmImageRequest) -> Result<ImageWarmup>;
108 async fn terminate_hosts(&self, request: &TerminateHostsRequest) -> Result<HostTermination>;
109}
110
111#[derive(Clone)]
112pub struct HostSandboxRuntimeConfig {
113 pub control_plane_url: String,
114 pub jwt_issuer: String,
115 pub invocation_jwt_audience: String,
116 pub actor_idle_timeout_ms: u64,
117 pub host_idle_timeout_ms: u64,
118}
119
120pub struct CommandSandboxProvider {
121 provider_name: String,
122 command: String,
123 environment: HashMap<String, String>,
124}
125
126impl CommandSandboxProvider {
127 pub fn new(
128 provider_name: String,
129 command: String,
130 mut environment: HashMap<String, String>,
131 ) -> Result<Self> {
132 ensure!(
133 !provider_name.is_empty() && provider_name.trim() == provider_name,
134 "sandbox provider name must be non-empty without surrounding whitespace"
135 );
136 ensure!(
137 !command.is_empty() && command.trim() == command,
138 "DURABLE_OBJECT_SANDBOX_COMMAND must be non-empty without surrounding whitespace"
139 );
140 if let Ok(path) = std::env::var("PATH") {
141 environment.entry("PATH".into()).or_insert(path);
142 }
143 Ok(Self {
144 provider_name,
145 command,
146 environment,
147 })
148 }
149}
150
151#[async_trait]
152impl SandboxProvider for CommandSandboxProvider {
153 async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle> {
154 let (mut response, command): (ActorHostHandle, _) =
155 self.execute_timed("ensure_host", request).await?;
156 if let Some(provisioning) = &mut response.provisioning {
157 provisioning.command_spawned_at_ms = command.spawned_at_ms;
158 provisioning.request_written_at_ms = command.request_written_at_ms;
159 provisioning.process_completed_at_ms = command.process_completed_at_ms;
160 provisioning.response_decoded_at_ms = command.response_decoded_at_ms;
161 }
162 ensure!(
163 response.canonical_region == request.canonical_region,
164 "{} sandbox command returned a host in the wrong canonical region",
165 self.provider_name
166 );
167 ensure!(
168 !response.host_id.as_str().is_empty() && !response.route.is_empty(),
169 "{} sandbox command returned an invalid host",
170 self.provider_name
171 );
172 Ok(response)
173 }
174
175 async fn warm_image(&self, request: &WarmImageRequest) -> Result<ImageWarmup> {
176 self.execute("warm_image", request).await
177 }
178
179 async fn terminate_hosts(&self, request: &TerminateHostsRequest) -> Result<HostTermination> {
180 self.execute("terminate_hosts", request).await
181 }
182}
183
184impl CommandSandboxProvider {
185 async fn execute<Request: Serialize, Reply: for<'de> Deserialize<'de>>(
186 &self,
187 operation: &str,
188 request: &Request,
189 ) -> Result<Reply> {
190 Ok(self.execute_timed(operation, request).await?.0)
191 }
192
193 async fn execute_timed<Request: Serialize, Reply: for<'de> Deserialize<'de>>(
194 &self,
195 operation: &str,
196 request: &Request,
197 ) -> Result<(Reply, ProviderCommandTimings)> {
198 let started_at = Instant::now();
199 let mut timings = ProviderCommandTimings::default();
200 match self
201 .execute_timed_inner(operation, request, started_at, &mut timings)
202 .await
203 {
204 Ok(response) => Ok((response, timings)),
205 Err(source) => Err(ProviderCommandFailure { source, timings }.into()),
206 }
207 }
208
209 async fn execute_timed_inner<Request: Serialize, Reply: for<'de> Deserialize<'de>>(
210 &self,
211 operation: &str,
212 request: &Request,
213 started_at: Instant,
214 timings: &mut ProviderCommandTimings,
215 ) -> Result<Reply> {
216 let command = ProviderCommand { operation, request };
217 let execution = command_process::exchange(
218 &self.command,
219 &self.environment,
220 &command,
221 started_at,
222 timings,
223 );
224 tokio::time::timeout(PROVIDER_REQUEST_TIMEOUT, execution)
225 .await
226 .context("sandbox provider command timed out; outcome may be unknown")?
227 }
228}
229
230#[derive(Debug, Default)]
231struct ProviderCommandTimings {
232 spawned_at_ms: Option<u64>,
233 request_written_at_ms: Option<u64>,
234 process_completed_at_ms: Option<u64>,
235 response_decoded_at_ms: Option<u64>,
236}
237
238#[derive(Debug)]
239pub(crate) struct ProviderCommandFailure {
240 source: anyhow::Error,
241 timings: ProviderCommandTimings,
242}
243
244impl ProviderCommandFailure {
245 pub(crate) fn spawned_at_ms(&self) -> Option<u64> {
246 self.timings.spawned_at_ms
247 }
248
249 pub(crate) fn request_written_at_ms(&self) -> Option<u64> {
250 self.timings.request_written_at_ms
251 }
252
253 pub(crate) fn process_completed_at_ms(&self) -> Option<u64> {
254 self.timings.process_completed_at_ms
255 }
256
257 pub(crate) fn response_decoded_at_ms(&self) -> Option<u64> {
258 self.timings.response_decoded_at_ms
259 }
260}
261
262impl std::fmt::Display for ProviderCommandFailure {
263 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
264 self.source.fmt(formatter)
265 }
266}
267
268impl std::error::Error for ProviderCommandFailure {
269 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
270 self.source.source()
271 }
272}
273
274fn elapsed_ms(started_at: Instant) -> u64 {
275 u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX)
276}
277
278#[derive(Serialize)]
279struct ProviderCommand<'a, Request> {
280 operation: &'a str,
281 request: &'a Request,
282}
283
284#[cfg(test)]
285mod tests {
286 use super::*;
287
288 #[test]
289 fn decodes_provider_provisioning_timings() {
290 let handle: ActorHostHandle = serde_json::from_value(serde_json::json!({
291 "hostId": "host.v1.namespace.revision.session",
292 "route": "https://host.example.com",
293 "canonicalRegion": "north-america-east",
294 "provisioning": {
295 "provider": "modal",
296 "resourceId": "sb-actor",
297 "reused": false,
298 "startedAtMs": 0,
299 "resourcesResolvedAtMs": 12,
300 "existingHostCheckedAtMs": 34,
301 "sandboxScheduledAtMs": 56,
302 "hostReadyObservedAtMs": 123,
303 "routeReadAtMs": 125,
304 "metadataWrittenAtMs": 129,
305 "completedAtMs": 130
306 }
307 }))
308 .expect("actor host handle");
309
310 let provisioning = handle.provisioning.expect("provisioning timings");
311 assert_eq!(provisioning.resource_id, "sb-actor");
312 assert_eq!(provisioning.sandbox_scheduled_at_ms, Some(56));
313 assert_eq!(provisioning.completed_at_ms, 130);
314 }
315
316 #[test]
317 fn rejects_ambiguous_command_configuration() {
318 assert!(CommandSandboxProvider::new("".into(), "modal".into(), HashMap::new()).is_err());
319 assert!(CommandSandboxProvider::new("modal".into(), "".into(), HashMap::new()).is_err());
320 }
321
322 #[tokio::test]
323 async fn provider_calls_use_independent_processes() -> Result<()> {
324 let (directory, provider) = test_provider()?;
325 let requests = (0..5)
326 .map(|index| {
327 serde_json::json!({
328 "index": index,
329 "barrier": directory.path(),
330 })
331 })
332 .collect::<Vec<_>>();
333 let replies = tokio::time::timeout(
334 Duration::from_secs(3),
335 futures_util::future::try_join_all(
336 requests
337 .iter()
338 .map(|request| provider.execute::<_, serde_json::Value>("test", request)),
339 ),
340 )
341 .await
342 .expect("all five independent processes must start before any replies")?;
343 let pids = replies
344 .iter()
345 .map(|reply| reply["pid"].as_u64().unwrap())
346 .collect::<std::collections::HashSet<_>>();
347 assert_eq!(pids.len(), 5);
348 for (index, reply) in replies.iter().enumerate() {
349 assert_eq!(reply["index"], index);
350 }
351 Ok(())
352 }
353
354 #[tokio::test]
355 async fn provider_failures_do_not_affect_other_calls() -> Result<()> {
356 let (_directory, provider) = test_provider()?;
357 for (request, message) in [
358 (serde_json::json!({"fail": true}), "test failure"),
359 (serde_json::json!({"oversized": true}), "stdout exceeds"),
360 (
361 serde_json::json!({"malformed": true}),
362 "decode provider response",
363 ),
364 (serde_json::json!({"exit": true}), "exited"),
365 ] {
366 let healthy_request = serde_json::json!({"index": 42});
367 let (failed, healthy) = tokio::join!(
368 provider.execute::<_, serde_json::Value>("test", &request),
369 provider.execute::<_, serde_json::Value>("test", &healthy_request),
370 );
371 assert!(failed.unwrap_err().to_string().contains(message));
372 assert_eq!(healthy?["index"], 42);
373 }
374 Ok(())
375 }
376
377 #[tokio::test]
378 async fn cancelling_one_provider_call_terminates_only_its_process() -> Result<()> {
379 let (directory, provider) = test_provider()?;
380 let marker = directory.path().join("cancelled.pid");
381 let request = serde_json::json!({"marker": marker, "delay": 30_000});
382 let mut cancelled = Box::pin(provider.execute::<_, serde_json::Value>("test", &request));
383 let wait_for_start = async {
384 while !marker.exists() {
385 tokio::time::sleep(Duration::from_millis(5)).await;
386 }
387 };
388 tokio::select! {
389 result = &mut cancelled => panic!("provider should still be waiting: {result:?}"),
390 started = tokio::time::timeout(Duration::from_secs(3), wait_for_start) => started?,
391 }
392 let pid: u32 = std::fs::read_to_string(marker)?.parse()?;
393 drop(cancelled);
394 let healthy: serde_json::Value = provider
395 .execute("test", &serde_json::json!({"index": 42}))
396 .await?;
397 assert_eq!(healthy["index"], 42);
398 tokio::time::timeout(Duration::from_secs(3), async {
399 while tokio::process::Command::new("kill")
400 .args(["-0", &pid.to_string()])
401 .stderr(std::process::Stdio::null())
402 .status()
403 .await?
404 .success()
405 {
406 tokio::time::sleep(Duration::from_millis(5)).await;
407 }
408 anyhow::Ok(())
409 })
410 .await??;
411 Ok(())
412 }
413
414 fn test_provider() -> Result<(tempfile::TempDir, CommandSandboxProvider)> {
415 use std::os::unix::fs::PermissionsExt;
416 let directory = tempfile::tempdir()?;
417 let path = directory.path().join("custom-provider");
418 std::fs::write(
419 &path,
420 r#"#!/usr/bin/env node
421const fs = require('node:fs');
422const command = JSON.parse(fs.readFileSync(0, 'utf8'));
423const request = command.request;
424const reply = result => process.stdout.write(JSON.stringify({status: 'success', result}) + '\n');
425if (request.barrier) {
426 fs.writeFileSync(request.barrier + '/' + process.pid + '.started', '');
427 const timer = setInterval(() => {
428 if (fs.readdirSync(request.barrier).filter(name => name.endsWith('.started')).length === 5) {
429 clearInterval(timer);
430 reply({pid: process.pid, index: request.index});
431 }
432 }, 5);
433} else if (request.fail) {
434 process.stdout.write(JSON.stringify({status: 'failure', error: 'test failure'}) + '\n');
435} else if (request.oversized) {
436 process.stdout.write('x'.repeat(1024 * 1024 + 1));
437} else if (request.malformed) {
438 process.stdout.write('not json\n');
439} else if (request.exit) {
440 process.exitCode = 1;
441} else {
442 if (request.marker) fs.writeFileSync(request.marker, String(process.pid));
443 setTimeout(() => reply({pid: process.pid, index: request.index}), request.delay ?? 0);
444}
445"#,
446 )?;
447 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;
448 let provider = CommandSandboxProvider::new(
449 "modal".into(),
450 path.display().to_string(),
451 HashMap::new(),
452 )?;
453 Ok((directory, provider))
454 }
455}