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