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#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
58#[serde(rename_all = "camelCase")]
59pub struct WarmImageRequest {
60 pub namespace_id: String,
61 pub code_revision: String,
62 pub canonical_region: String,
63 pub image_ref: String,
64}
65
66#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
67#[serde(rename_all = "camelCase")]
68pub struct ImageWarmup {
69 pub provider: String,
70 pub resource_id: String,
71 pub total_ms: u64,
72}
73
74#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
75#[serde(rename_all = "camelCase")]
76pub struct TerminateHostsRequest {
77 pub namespace_id: String,
78 pub code_revision: String,
79 pub canonical_regions: Vec<String>,
80}
81
82#[derive(Clone, Debug, PartialEq, Eq, Deserialize)]
83#[serde(rename_all = "camelCase")]
84pub struct HostTermination {
85 pub provider: String,
86 pub resource_ids: Vec<String>,
87}
88
89#[async_trait]
90pub trait SandboxProvider: Send + Sync {
91 async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle>;
92 async fn warm_image(&self, request: &WarmImageRequest) -> Result<ImageWarmup>;
93 async fn terminate_hosts(&self, request: &TerminateHostsRequest) -> Result<HostTermination>;
94}
95
96#[derive(Clone)]
97pub struct HostSandboxRuntimeConfig {
98 pub control_plane_url: String,
99 pub jwt_issuer: String,
100 pub invocation_jwt_audience: String,
101 pub actor_idle_timeout_ms: u64,
102 pub host_idle_timeout_ms: u64,
103}
104
105pub struct CommandSandboxProvider {
106 provider_name: String,
107 command: String,
108 environment: HashMap<String, String>,
109}
110
111impl CommandSandboxProvider {
112 pub fn new(
113 provider_name: String,
114 command: String,
115 mut environment: HashMap<String, String>,
116 ) -> Result<Self> {
117 ensure!(
118 !provider_name.is_empty() && provider_name.trim() == provider_name,
119 "sandbox provider name must be non-empty without surrounding whitespace"
120 );
121 ensure!(
122 !command.is_empty() && command.trim() == command,
123 "DURABLE_OBJECT_SANDBOX_COMMAND must be non-empty without surrounding whitespace"
124 );
125 if let Ok(path) = std::env::var("PATH") {
126 environment.entry("PATH".into()).or_insert(path);
127 }
128 Ok(Self {
129 provider_name,
130 command,
131 environment,
132 })
133 }
134}
135
136#[async_trait]
137impl SandboxProvider for CommandSandboxProvider {
138 async fn ensure_host(&self, request: &EnsureHostRequest) -> Result<ActorHostHandle> {
139 let response: ActorHostHandle = self.execute("ensure_host", request).await?;
140 ensure!(
141 response.canonical_region == request.canonical_region,
142 "{} sandbox command returned a host in the wrong canonical region",
143 self.provider_name
144 );
145 ensure!(
146 !response.host_id.as_str().is_empty() && !response.route.is_empty(),
147 "{} sandbox command returned an invalid host",
148 self.provider_name
149 );
150 Ok(response)
151 }
152
153 async fn warm_image(&self, request: &WarmImageRequest) -> Result<ImageWarmup> {
154 self.execute("warm_image", request).await
155 }
156
157 async fn terminate_hosts(&self, request: &TerminateHostsRequest) -> Result<HostTermination> {
158 self.execute("terminate_hosts", request).await
159 }
160}
161
162impl CommandSandboxProvider {
163 async fn execute<Request: Serialize, Reply: for<'de> Deserialize<'de>>(
164 &self,
165 operation: &str,
166 request: &Request,
167 ) -> Result<Reply> {
168 let mut child = Command::new(&self.command)
169 .env_clear()
170 .envs(&self.environment)
171 .stdin(Stdio::piped())
172 .stdout(Stdio::piped())
173 .stderr(Stdio::piped())
174 .kill_on_drop(true)
175 .spawn()
176 .with_context(|| {
177 format!(
178 "start {} sandbox command {:?}",
179 self.provider_name, self.command
180 )
181 })?;
182 let document = serde_json::to_vec(&ProviderCommand { operation, request })?;
183 let mut stdin = child
184 .stdin
185 .take()
186 .with_context(|| format!("open {} sandbox command stdin", self.provider_name))?;
187 stdin
188 .write_all(&document)
189 .await
190 .with_context(|| format!("write {} sandbox command request", self.provider_name))?;
191 stdin
192 .shutdown()
193 .await
194 .with_context(|| format!("close {} sandbox command stdin", self.provider_name))?;
195 drop(stdin);
196 let output = tokio::time::timeout(PROVIDER_REQUEST_TIMEOUT, child.wait_with_output())
197 .await
198 .with_context(|| format!("{} sandbox command timed out", self.provider_name))??;
199 ensure!(
200 output.status.success(),
201 "{} sandbox command failed with {}: {}",
202 self.provider_name,
203 output.status,
204 String::from_utf8_lossy(&output.stderr).trim()
205 );
206 serde_json::from_slice(&output.stdout)
207 .with_context(|| format!("decode {} sandbox command response", self.provider_name))
208 }
209}
210
211#[derive(Serialize)]
212struct ProviderCommand<'a, Request> {
213 operation: &'a str,
214 request: &'a Request,
215}
216
217#[cfg(test)]
218mod tests {
219 use super::*;
220
221 #[test]
222 fn decodes_provider_provisioning_timings() {
223 let handle: ActorHostHandle = serde_json::from_value(serde_json::json!({
224 "hostId": "host.v1.namespace.revision.session",
225 "route": "https://host.example.com",
226 "canonicalRegion": "north-america-east",
227 "provisioning": {
228 "provider": "modal",
229 "resourceId": "sb-actor",
230 "reused": false,
231 "resourceLookupMs": 12,
232 "existingLookupMs": 34,
233 "createMs": 56,
234 "placementMs": 78,
235 "tunnelMs": 90,
236 "readyMs": 123,
237 "metadataMs": 4,
238 "totalMs": 397
239 }
240 }))
241 .expect("actor host handle");
242
243 let provisioning = handle.provisioning.expect("provisioning timings");
244 assert_eq!(provisioning.resource_id, "sb-actor");
245 assert_eq!(provisioning.create_ms, 56);
246 assert_eq!(provisioning.total_ms, 397);
247 }
248
249 #[test]
250 fn rejects_ambiguous_command_configuration() {
251 assert!(CommandSandboxProvider::new("".into(), "modal".into(), HashMap::new()).is_err());
252 assert!(CommandSandboxProvider::new("modal".into(), "".into(), HashMap::new()).is_err());
253 }
254}