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}
39
40#[async_trait]
41pub trait SandboxProvider: Send + Sync {
42    async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle>;
43}
44
45#[derive(Clone)]
46pub struct HostSandboxRuntimeConfig {
47    pub control_plane_url: String,
48    pub jwt_issuer: String,
49    pub invocation_jwt_audience: String,
50    pub actor_idle_timeout_ms: u64,
51    pub host_idle_timeout_ms: u64,
52}
53
54pub struct CommandSandboxProvider {
55    provider_name: String,
56    command: String,
57    environment: HashMap<String, String>,
58}
59
60impl CommandSandboxProvider {
61    pub fn new(
62        provider_name: String,
63        command: String,
64        mut environment: HashMap<String, String>,
65    ) -> Result<Self> {
66        ensure!(
67            !provider_name.is_empty() && provider_name.trim() == provider_name,
68            "sandbox provider name must be non-empty without surrounding whitespace"
69        );
70        ensure!(
71            !command.is_empty() && command.trim() == command,
72            "DURABLE_OBJECT_SANDBOX_COMMAND must be non-empty without surrounding whitespace"
73        );
74        if let Ok(path) = std::env::var("PATH") {
75            environment.entry("PATH".into()).or_insert(path);
76        }
77        Ok(Self {
78            provider_name,
79            command,
80            environment,
81        })
82    }
83}
84
85#[async_trait]
86impl SandboxProvider for CommandSandboxProvider {
87    async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle> {
88        let response: ActorHostHandle = self.execute("ensure_host", request).await?;
89        ensure!(
90            response.canonical_region == request.canonical_region,
91            "{} sandbox command returned a host in the wrong canonical region",
92            self.provider_name
93        );
94        ensure!(
95            !response.host_id.as_str().is_empty() && !response.route.is_empty(),
96            "{} sandbox command returned an invalid host",
97            self.provider_name
98        );
99        Ok(response)
100    }
101}
102
103impl CommandSandboxProvider {
104    async fn execute<Request: Serialize, Reply: for<'de> Deserialize<'de>>(
105        &self,
106        operation: &str,
107        request: &Request,
108    ) -> Result<Reply> {
109        let mut child = Command::new(&self.command)
110            .env_clear()
111            .envs(&self.environment)
112            .stdin(Stdio::piped())
113            .stdout(Stdio::piped())
114            .stderr(Stdio::piped())
115            .kill_on_drop(true)
116            .spawn()
117            .with_context(|| {
118                format!(
119                    "start {} sandbox command {:?}",
120                    self.provider_name, self.command
121                )
122            })?;
123        let document = serde_json::to_vec(&ProviderCommand { operation, request })?;
124        let mut stdin = child
125            .stdin
126            .take()
127            .with_context(|| format!("open {} sandbox command stdin", self.provider_name))?;
128        stdin
129            .write_all(&document)
130            .await
131            .with_context(|| format!("write {} sandbox command request", self.provider_name))?;
132        stdin
133            .shutdown()
134            .await
135            .with_context(|| format!("close {} sandbox command stdin", self.provider_name))?;
136        drop(stdin);
137        let output = tokio::time::timeout(PROVIDER_REQUEST_TIMEOUT, child.wait_with_output())
138            .await
139            .with_context(|| format!("{} sandbox command timed out", self.provider_name))??;
140        ensure!(
141            output.status.success(),
142            "{} sandbox command failed with {}: {}",
143            self.provider_name,
144            output.status,
145            String::from_utf8_lossy(&output.stderr).trim()
146        );
147        serde_json::from_slice(&output.stdout)
148            .with_context(|| format!("decode {} sandbox command response", self.provider_name))
149    }
150}
151
152#[derive(Serialize)]
153struct ProviderCommand<'a, Request> {
154    operation: &'a str,
155    request: &'a Request,
156}
157
158#[cfg(test)]
159mod tests {
160    use super::*;
161
162    #[test]
163    fn rejects_ambiguous_command_configuration() {
164        assert!(CommandSandboxProvider::new("".into(), "modal".into(), HashMap::new()).is_err());
165        assert!(CommandSandboxProvider::new("modal".into(), "".into(), HashMap::new()).is_err());
166    }
167}
little_durable_objects/sandbox.rs

little_durable_objects/
sandbox.rs
