1use std::{
2 env, future::Future, net::SocketAddr, path::PathBuf, process::Stdio, sync::Arc, time::Duration,
3};
4
5use anyhow::{Context, Result, ensure};
6use tokio::{
7 net::{TcpListener, lookup_host},
8 process::Command,
9};
10use tokio_stream::wrappers::TcpListenerStream;
11use tokio_util::sync::CancellationToken;
12use tonic::transport::Server;
13use tracing::{error, info};
14
15use crate::{
16 actor::{ActorExecutorConnection, ActorExecutorListener, ActorScope},
17 clock::SystemClock,
18 control_plane::{ActorJwtVerifier, ActorTokenPurpose, ControlPlaneClient},
19 grpc::ActorHostGrpcService,
20 host_leases::{HostLeaseRegistry, MAX_HOST_LEASE_DURATION_MS},
21 state_transport::HttpStateTransport,
22};
23
24use super::{ActorHost, HostEndpoint, HostLeaseMaintainer, LeaseRenewalTask};
25
26const HOST_TASK_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2);
27const HOST_ACTOR_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
28const DEFAULT_HOST_IDLE_TIMEOUT_MS: u64 = 300_000;
29const MAX_IDLE_TIMEOUT_MS: u64 = 86_400_000;
30
31pub struct ActorHostConfig {
32 pub control_plane_url: String,
33 pub host_token: String,
34 pub jwt_public_keys: String,
35 pub namespace_id: String,
36 pub host_id: super::HostId,
37 pub session_id: String,
38 pub executor_socket: PathBuf,
39 pub host_bind: SocketAddr,
40 pub host_route: Option<String>,
41 pub private_hostname: Option<String>,
42 pub route_file: Option<PathBuf>,
43 pub jwt_issuer: String,
44 pub invocation_jwt_audience: String,
45 pub jwt_max_lifetime: Duration,
46 pub lease_duration: Duration,
47 pub renew_every: Duration,
48 pub host_idle_timeout: Duration,
49}
50
51impl ActorHostConfig {
52 pub fn from_env() -> Result<Self> {
53 Self::from_lookup(|name| env::var(name).ok())
54 }
55}
56
57pub async fn serve_actor_host<F>(config: ActorHostConfig, shutdown: F) -> Result<()>
58where
59 F: Future<Output = ()> + Send + 'static,
60{
61 let PreparedActorHost {
62 invocation_auth,
63 listener,
64 route,
65 executor_connection,
66 mut javascript,
67 host,
68 lease,
69 renewal,
70 } = prepare_actor_host(&config).await?;
71 let mut lease_lost = renewal.lease_lost();
72 let mut activity = host.activity();
73
74 let service = ActorHostGrpcService::new(host.clone(), invocation_auth).into_service();
75 executor_connection.mark_ready().await?;
76 let stop = CancellationToken::new();
77 let server_stop = stop.clone();
78 let mut server = Box::pin(
79 Server::builder()
80 .add_service(service)
81 .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async move {
82 server_stop.cancelled().await
83 }),
84 );
85 let mut executor_task = Box::pin(executor_connection.run(stop.clone()));
86 tokio::pin!(shutdown);
87
88 info!(host_id = %config.host_id, namespace_id = %config.namespace_id, route, "durable-object host is ready");
89 let stop_result = wait_for_host_stop(
90 server.as_mut(),
91 executor_task.as_mut(),
92 &mut javascript,
93 shutdown.as_mut(),
94 &mut lease_lost,
95 &mut activity,
96 config.host_idle_timeout,
97 )
98 .await;
99 stop_host_tasks(&host, &stop, server, executor_task).await;
100 drop(javascript);
101 let renewal_result = renewal.shutdown().await;
102 let unregister_result = lease.unregister().await;
103 info!(host_id = %config.host_id, "durable-object host stopped");
104 stop_result?;
105 renewal_result?;
106 unregister_result
107}
108
109impl ActorHostConfig {
110 fn from_lookup(mut get: impl FnMut(&str) -> Option<String>) -> Result<Self> {
111 let control_plane_url = required(&mut get, "DURABLE_OBJECT_CONTROL_PLANE_URL")?;
112 let host_token = required(&mut get, "DURABLE_OBJECT_HOST_TOKEN")?;
113 let jwt_public_keys = required(&mut get, "DURABLE_OBJECT_JWT_PUBLIC_KEYS")?;
114 let namespace_id = required(&mut get, "DURABLE_OBJECT_NAMESPACE_ID")?;
115 ActorScope {
116 namespace_id: namespace_id.clone(),
117 }
118 .validate()?;
119 let host_id = super::HostId::new(required(&mut get, "DURABLE_OBJECT_HOST_ID")?);
120 ensure!(
121 host_id
122 .as_str()
123 .starts_with(&format!("host.v1.{namespace_id}.")),
124 "DURABLE_OBJECT_HOST_ID does not belong to DURABLE_OBJECT_NAMESPACE_ID"
125 );
126 let session_id = required(&mut get, "DURABLE_OBJECT_SESSION_ID")?;
127 uuid::Uuid::parse_str(&session_id).context("DURABLE_OBJECT_SESSION_ID must be a UUID")?;
128 let executor_socket = get("DURABLE_OBJECT_EXECUTOR_SOCKET")
129 .map(PathBuf::from)
130 .unwrap_or_else(|| PathBuf::from("/tmp/durable-object-executor.sock"));
131 let host_route = get("DURABLE_OBJECT_HOST_ROUTE");
132 if let Some(route) = &host_route {
133 tonic::transport::Endpoint::from_shared(route.clone())
134 .context("DURABLE_OBJECT_HOST_ROUTE must be a valid HTTP or HTTPS URI")?;
135 }
136 let private_hostname = get("DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME");
137 if let Some(hostname) = &private_hostname {
138 ensure!(
139 !hostname.is_empty() && hostname.trim() == hostname,
140 "DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME must be non-empty without surrounding whitespace"
141 );
142 }
143 ensure!(
144 host_route.is_none() || private_hostname.is_none(),
145 "DURABLE_OBJECT_HOST_ROUTE and DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME are mutually exclusive"
146 );
147 let route_file = get("DURABLE_OBJECT_HOST_ROUTE_FILE").map(PathBuf::from);
148 let host_bind = get("DURABLE_OBJECT_HOST_BIND")
149 .unwrap_or_else(|| {
150 if private_hostname.is_some() {
151 "[::]:7101"
152 } else if host_route.is_some() {
153 "0.0.0.0:7101"
154 } else {
155 "127.0.0.1:0"
156 }
157 .into()
158 })
159 .parse()
160 .context("DURABLE_OBJECT_HOST_BIND must be a socket address")?;
161 let jwt_issuer = get("DURABLE_OBJECT_JWT_ISSUER")
162 .unwrap_or_else(|| "durable-object-control-plane".into());
163 let invocation_jwt_audience = get("DURABLE_OBJECT_INVOKE_JWT_AUDIENCE")
164 .unwrap_or_else(|| "durable-object-invoke".into());
165 let jwt_max_lifetime =
166 duration_seconds(&mut get, "DURABLE_OBJECT_JWT_MAX_TTL_SECONDS", 1_800)?;
167 let lease_duration = duration_ms(&mut get, "DURABLE_OBJECT_LEASE_MS", 30_000)?;
168 let renew_every = duration_ms(&mut get, "DURABLE_OBJECT_RENEW_MS", 10_000)?;
169 let host_idle_timeout = duration_ms(
170 &mut get,
171 "DURABLE_OBJECT_HOST_IDLE_TIMEOUT_MS",
172 DEFAULT_HOST_IDLE_TIMEOUT_MS,
173 )?;
174 ensure!(
175 host_idle_timeout.as_millis() <= u128::from(MAX_IDLE_TIMEOUT_MS),
176 "DURABLE_OBJECT_HOST_IDLE_TIMEOUT_MS is too large"
177 );
178 ensure!(
179 lease_duration.as_millis() <= u128::from(MAX_HOST_LEASE_DURATION_MS),
180 "DURABLE_OBJECT_LEASE_MS is too large"
181 );
182 ensure!(
183 renew_every < lease_duration,
184 "DURABLE_OBJECT_RENEW_MS must be shorter than DURABLE_OBJECT_LEASE_MS"
185 );
186 Ok(Self {
187 control_plane_url,
188 host_token,
189 jwt_public_keys,
190 namespace_id,
191 host_id,
192 session_id,
193 executor_socket,
194 host_bind,
195 host_route,
196 private_hostname,
197 route_file,
198 jwt_issuer,
199 invocation_jwt_audience,
200 jwt_max_lifetime,
201 lease_duration,
202 renew_every,
203 host_idle_timeout,
204 })
205 }
206}
207
208struct PreparedActorHost {
209 invocation_auth: ActorJwtVerifier,
210 listener: TcpListener,
211 route: String,
212 executor_connection: ActorExecutorConnection,
213 javascript: tokio::process::Child,
214 host: Arc<ActorHost>,
215 lease: Arc<HostLeaseMaintainer>,
216 renewal: LeaseRenewalTask,
217}
218
219async fn prepare_actor_host(config: &ActorHostConfig) -> Result<PreparedActorHost> {
220 let invocation_auth = invocation_auth(config)?;
221 let control_plane =
222 Arc::new(ControlPlaneClient::connect(&config.control_plane_url, &config.host_token).await?);
223 let (listener, route, endpoint) = bind_host(config).await?;
224 let (executor_connection, javascript) = connect_executor(&config.executor_socket).await?;
225 let host = Arc::new(ActorHost::new(
226 endpoint.clone(),
227 config.namespace_id.clone(),
228 executor_connection.executor(),
229 control_plane.clone(),
230 Arc::new(HttpStateTransport::new()),
231 ));
232 let lease = Arc::new(HostLeaseMaintainer::new(
233 endpoint,
234 config.session_id.clone(),
235 control_plane as Arc<dyn HostLeaseRegistry>,
236 Arc::new(SystemClock),
237 config.lease_duration,
238 config.renew_every,
239 )?);
240 let renewal = lease.clone().start().await?;
241 Ok(PreparedActorHost {
242 invocation_auth,
243 listener,
244 route,
245 executor_connection,
246 javascript,
247 host,
248 lease,
249 renewal,
250 })
251}
252
253fn invocation_auth(config: &ActorHostConfig) -> Result<ActorJwtVerifier> {
254 ActorJwtVerifier::for_scope(
255 &config.jwt_public_keys,
256 config.jwt_issuer.clone(),
257 config.invocation_jwt_audience.clone(),
258 ActorTokenPurpose::Invocation,
259 config.jwt_max_lifetime,
260 )
261}
262
263async fn bind_host(config: &ActorHostConfig) -> Result<(TcpListener, String, HostEndpoint)> {
264 let listener = TcpListener::bind(config.host_bind)
265 .await
266 .with_context(|| format!("bind actor host at {}", config.host_bind))?;
267 let route = advertised_route(config, listener.local_addr()?).await?;
268 if let Some(path) = &config.route_file {
269 tokio::fs::write(path, &route)
270 .await
271 .with_context(|| format!("write actor host route to {}", path.display()))?;
272 }
273 let endpoint = HostEndpoint {
274 id: config.host_id.clone(),
275 route: route.clone(),
276 };
277 Ok((listener, route, endpoint))
278}
279
280async fn advertised_route(config: &ActorHostConfig, bound: SocketAddr) -> Result<String> {
281 if let Some(route) = &config.host_route {
282 return Ok(route.clone());
283 }
284 let Some(hostname) = &config.private_hostname else {
285 return Ok(format!("http://{bound}"));
286 };
287 let address = lookup_host((hostname.as_str(), bound.port()))
288 .await
289 .with_context(|| format!("resolve actor host private hostname {hostname}"))?
290 .find(SocketAddr::is_ipv6)
291 .with_context(|| format!("actor host private hostname {hostname} has no IPv6 address"))?;
292 Ok(format!("http://{address}"))
293}
294
295async fn connect_executor(
296 socket: &std::path::Path,
297) -> Result<(ActorExecutorConnection, tokio::process::Child)> {
298 let listener = ActorExecutorListener::bind(socket).await?;
299 let javascript = spawn_javascript_process()?;
300 Ok((listener.accept().await?, javascript))
301}
302
303async fn wait_for_host_stop<ServerFuture, ExecutorFuture, ShutdownFuture>(
304 mut server: std::pin::Pin<&mut ServerFuture>,
305 mut executor: std::pin::Pin<&mut ExecutorFuture>,
306 javascript: &mut tokio::process::Child,
307 mut shutdown: std::pin::Pin<&mut ShutdownFuture>,
308 lease_lost: &mut tokio::sync::watch::Receiver<bool>,
309 activity: &mut tokio::sync::watch::Receiver<usize>,
310 idle_timeout: Duration,
311) -> Result<()>
312where
313 ServerFuture: Future<Output = std::result::Result<(), tonic::transport::Error>> + ?Sized,
314 ExecutorFuture: Future<Output = Result<()>> + ?Sized,
315 ShutdownFuture: Future<Output = ()> + ?Sized,
316{
317 let mut idle_deadline = tokio::time::Instant::now() + idle_timeout;
318 loop {
319 tokio::select! {
320 result = server.as_mut() => break result.context("serve actor host gRPC"),
321 result = executor.as_mut() => break result.context("run JavaScript actor executor"),
322 result = javascript.wait() => break Err(anyhow::anyhow!("JavaScript actor executor exited with {}", result?)),
323 () = shutdown.as_mut() => break Ok(()),
324 changed = lease_lost.changed() => {
325 if changed.is_err() || *lease_lost.borrow() {
326 break Err(anyhow::anyhow!("host lease expired; host self-fenced"));
327 }
328 }
329 changed = activity.changed() => {
330 if changed.is_err() { break Err(anyhow::anyhow!("actor activity tracker stopped")); }
331 if *activity.borrow() == 0 {
332 idle_deadline = tokio::time::Instant::now() + idle_timeout;
333 }
334 }
335 () = tokio::time::sleep_until(idle_deadline), if *activity.borrow() == 0 => break Ok(()),
336 }
337 }
338}
339
340async fn stop_host_tasks(
341 host: &ActorHost,
342 stop: &CancellationToken,
343 server: impl Future,
344 executor: impl Future,
345) {
346 if let Err(error) = host.drain(HOST_ACTOR_DRAIN_TIMEOUT).await {
347 error!(error = %format!("{error:#}"), "actor invocations did not drain cleanly");
348 }
349 stop.cancel();
350 let _ = tokio::time::timeout(HOST_TASK_SHUTDOWN_TIMEOUT, async {
351 let _ = tokio::join!(server, executor);
352 })
353 .await;
354}
355
356fn spawn_javascript_process() -> Result<tokio::process::Child> {
357 Command::new("node")
358 .args([
359 "--eval",
360 "import(\"little-durable-objects/host\").then(module => module.runDurableObjectHost())",
361 ])
362 .stdin(Stdio::null())
363 .stdout(Stdio::inherit())
364 .stderr(Stdio::inherit())
365 .kill_on_drop(true)
366 .spawn()
367 .context("start JavaScript actor executor")
368}
369
370fn required(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
371 let value = get(name).with_context(|| format!("{name} is required"))?;
372 ensure!(!value.is_empty(), "{name} must not be empty");
373 Ok(value)
374}
375
376fn duration_ms(
377 get: &mut impl FnMut(&str) -> Option<String>,
378 name: &str,
379 default: u64,
380) -> Result<Duration> {
381 let value = get(name)
382 .map(|value| value.parse::<u64>())
383 .transpose()
384 .with_context(|| format!("{name} must be an integer number of milliseconds"))?
385 .unwrap_or(default);
386 ensure!(value > 0, "{name} must be positive");
387 Ok(Duration::from_millis(value))
388}
389
390fn duration_seconds(
391 get: &mut impl FnMut(&str) -> Option<String>,
392 name: &str,
393 default: u64,
394) -> Result<Duration> {
395 let value = get(name)
396 .map(|value| value.parse::<u64>())
397 .transpose()
398 .with_context(|| format!("{name} must be an integer number of seconds"))?
399 .unwrap_or(default);
400 ensure!(value > 0, "{name} must be positive");
401 Ok(Duration::from_secs(value))
402}
403
404#[cfg(test)]
405mod tests {
406 use super::*;
407 use std::collections::HashMap;
408
409 #[test]
410 fn host_needs_no_local_state_directory() -> Result<()> {
411 let values = values();
412 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned())?;
413 assert_eq!(
414 config.executor_socket,
415 PathBuf::from("/tmp/durable-object-executor.sock")
416 );
417 assert_eq!(config.host_idle_timeout, Duration::from_secs(300));
418 Ok(())
419 }
420
421 #[test]
422 fn private_network_hosts_bind_ipv6_and_publish_their_route() -> Result<()> {
423 let mut values = values();
424 values.insert(
425 "DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME".into(),
426 "i6pn.modal.local".into(),
427 );
428 values.insert(
429 "DURABLE_OBJECT_HOST_ROUTE_FILE".into(),
430 "/tmp/durable-object-route".into(),
431 );
432
433 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned())?;
434
435 assert_eq!(config.host_bind, "[::]:7101".parse()?);
436 assert_eq!(config.private_hostname.as_deref(), Some("i6pn.modal.local"));
437 assert_eq!(
438 config.route_file,
439 Some(PathBuf::from("/tmp/durable-object-route"))
440 );
441 Ok(())
442 }
443
444 fn values() -> HashMap<String, String> {
445 HashMap::from([
446 (
447 "DURABLE_OBJECT_CONTROL_PLANE_URL".into(),
448 "http://127.0.0.1:7100".into(),
449 ),
450 ("DURABLE_OBJECT_HOST_TOKEN".into(), "host-jwt".into()),
451 ("DURABLE_OBJECT_JWT_PUBLIC_KEYS".into(), "{}".into()),
452 ("DURABLE_OBJECT_NAMESPACE_ID".into(), "project-1".into()),
453 (
454 "DURABLE_OBJECT_HOST_ID".into(),
455 "host.v1.project-1.revision-1.host-1".into(),
456 ),
457 (
458 "DURABLE_OBJECT_SESSION_ID".into(),
459 "00000000-0000-4000-8000-000000000001".into(),
460 ),
461 ])
462 }
463}