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}