little_durable_objects/
sandbox.rs

1use std::{collections::HashMap, process::Stdio, time::Duration};
2
3use anyhow::{Context, Result, ensure};
4use async_trait::async_trait;
5use serde::{Deserialize, Serialize};
6use tokio::{io::AsyncWriteExt, process::Command};
7
8use crate::host::HostId;
9
10const PROVIDER_REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
11
12#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
13#[serde(rename_all = "camelCase")]
14pub struct EnsureHostRequest {
15    pub namespace_id: String,
16    pub code_revision: String,
17    pub canonical_region: String,
18    pub host_id: HostId,
19    pub session_id: String,
20    pub host_token: String,
21    pub jwt_public_keys: String,
22    pub control_plane_url: String,
23    pub jwt_issuer: String,
24    pub invocation_jwt_audience: String,
25    pub image_ref: String,
26    pub working_directory: String,
27    pub actor_entrypoint: Option<String>,
28    pub actor_idle_timeout_ms: u64,
29    pub host_idle_timeout_ms: u64,
30}
31
32#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
33#[serde(rename_all = "camelCase")]
34pub struct ActorHostHandle {
35    pub host_id: HostId,
36    pub route: String,
37    pub canonical_region: String,
38    pub provisioning: Option<ActorHostProvisioning>,
39}
40
41#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
42#[serde(rename_all = "camelCase")]
43pub struct ActorHostProvisioning {
44    pub provider: String,
45    pub resource_id: String,
46    pub reused: bool,
47    pub resource_lookup_ms: u64,
48    pub existing_lookup_ms: u64,
49    pub create_ms: u64,
50    pub placement_ms: u64,
51    pub tunnel_ms: u64,
52    pub ready_ms: u64,
53    pub metadata_ms: u64,
54    pub total_ms: u64,
55}
56
57#[async_trait]
58pub trait SandboxProvider: Send + Sync {
59    async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle>;
60}
61
62#[derive(Clone)]
63pub struct HostSandboxRuntimeConfig {
64    pub control_plane_url: String,
65    pub jwt_issuer: String,
66    pub invocation_jwt_audience: String,
67    pub actor_idle_timeout_ms: u64,
68    pub host_idle_timeout_ms: u64,
69}
70
71pub struct CommandSandboxProvider {
72    provider_name: String,
73    command: String,
74    environment: HashMap<String, String>,
75}
76
77impl CommandSandboxProvider {
78    pub fn new(
79        provider_name: String,
80        command: String,
81        mut environment: HashMap<String, String>,
82    ) -> Result<Self> {
83        ensure!(
84            !provider_name.is_empty() && provider_name.trim() == provider_name,
85            "sandbox provider name must be non-empty without surrounding whitespace"
86        );
87        ensure!(
88            !command.is_empty() && command.trim() == command,
89            "DURABLE_OBJECT_SANDBOX_COMMAND must be non-empty without surrounding whitespace"
90        );
91        if let Ok(path) = std::env::var("PATH") {
92            environment.entry("PATH".into()).or_insert(path);
93        }
94        Ok(Self {
95            provider_name,
96            command,
97            environment,
98        })
99    }
100}
101
102#[async_trait]
103impl SandboxProvider for CommandSandboxProvider {
104    async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle> {
105        let response: ActorHostHandle = self.execute("ensure_host", request).await?;
106        ensure!(
107            response.canonical_region == request.canonical_region,
108            "{} sandbox command returned a host in the wrong canonical region",
109            self.provider_name
110        );
111        ensure!(
112            !response.host_id.as_str().is_empty() && !response.route.is_empty(),
113            "{} sandbox command returned an invalid host",
114            self.provider_name
115        );
116        Ok(response)
117    }
118}
119
120impl CommandSandboxProvider {
121    async fn execute<Request: Serialize, Reply: for<'de> Deserialize<'de>>(
122        &self,
123        operation: &str,
124        request: &Request,
125    ) -> Result<Reply> {
126        let mut child = Command::new(&self.command)
127            .env_clear()
128            .envs(&self.environment)
129            .stdin(Stdio::piped())
130            .stdout(Stdio::piped())
131            .stderr(Stdio::piped())
132            .kill_on_drop(true)
133            .spawn()
134            .with_context(|| {
135                format!(
136                    "start {} sandbox command {:?}",
137                    self.provider_name, self.command
138                )
139            })?;
140        let document = serde_json::to_vec(&ProviderCommand { operation, request })?;
141        let mut stdin = child
142            .stdin
143            .take()
144            .with_context(|| format!("open {} sandbox command stdin", self.provider_name))?;
145        stdin
146            .write_all(&document)
147            .await
148            .with_context(|| format!("write {} sandbox command request", self.provider_name))?;
149        stdin
150            .shutdown()
151            .await
152            .with_context(|| format!("close {} sandbox command stdin", self.provider_name))?;
153        drop(stdin);
154        let output = tokio::time::timeout(PROVIDER_REQUEST_TIMEOUT, child.wait_with_output())
155            .await
156            .with_context(|| format!("{} sandbox command timed out", self.provider_name))??;
157        ensure!(
158            output.status.success(),
159            "{} sandbox command failed with {}: {}",
160            self.provider_name,
161            output.status,
162            String::from_utf8_lossy(&output.stderr).trim()
163        );
164        serde_json::from_slice(&output.stdout)
165            .with_context(|| format!("decode {} sandbox command response", self.provider_name))
166    }
167}
168
169#[derive(Serialize)]
170struct ProviderCommand<'a, Request> {
171    operation: &'a str,
172    request: &'a Request,
173}
174
175#[cfg(test)]
176mod tests {
177    use super::*;
178
179    #[test]
180    fn decodes_provider_provisioning_timings() {
181        let handle: ActorHostHandle = serde_json::from_value(serde_json::json!({
182            "hostId": "host.v1.namespace.revision.session",
183            "route": "https://host.example.com",
184            "canonicalRegion": "north-america-east",
185            "provisioning": {
186                "provider": "modal",
187                "resourceId": "sb-actor",
188                "reused": false,
189                "resourceLookupMs": 12,
190                "existingLookupMs": 34,
191                "createMs": 56,
192                "placementMs": 78,
193                "tunnelMs": 90,
194                "readyMs": 123,
195                "metadataMs": 4,
196                "totalMs": 397
197            }
198        }))
199        .expect("actor host handle");
200
201        let provisioning = handle.provisioning.expect("provisioning timings");
202        assert_eq!(provisioning.resource_id, "sb-actor");
203        assert_eq!(provisioning.create_ms, 56);
204        assert_eq!(provisioning.total_ms, 397);
205    }
206
207    #[test]
208    fn rejects_ambiguous_command_configuration() {
209        assert!(CommandSandboxProvider::new("".into(), "modal".into(), HashMap::new()).is_err());
210        assert!(CommandSandboxProvider::new("modal".into(), "".into(), HashMap::new()).is_err());
211    }
212}
little_durable_objects/sandbox.rs

little_durable_objects/
sandbox.rs
