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