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 public_route_file: Option<PathBuf>,
48 pub private_hostname: Option<String>,
49 pub route_file: Option<PathBuf>,
50 pub jwt_issuer: String,
51 pub invocation_jwt_audience: String,
52 pub jwt_max_lifetime: Duration,
53 pub lease_duration: Duration,
54 pub renew_every: Duration,
55 pub host_idle_timeout: Duration,
56 metadata: Option<HostMetadataFile>,
57 startup_started_at: Instant,
58 configuration_loaded_at_ms: f64,
59}
60
61struct HostMetadataFile {
62 path: PathBuf,
63 canonical_region: String,
64}
65
66impl ActorHostConfig {
67 pub fn from_env() -> Result<Self> {
68 Self::from_lookup(|name| env::var(name).ok())
69 }
70}
71
72pub async fn serve_actor_host<F>(config: ActorHostConfig, shutdown: F) -> Result<()>
73where
74 F: Future<Output = ()> + Send + 'static,
75{
76 let mut timings = HostStartupTimings::new(&config);
77 let prepared = match prepare_actor_host(&config, &mut timings).await {
78 Ok(prepared) => prepared,
79 Err(error) => {
80 log_startup(&config, &timings, "failed", Some(&error));
81 return Err(error);
82 }
83 };
84 let PreparedActorHost {
85 invocation_auth,
86 listener,
87 route,
88 executor_connection,
89 mut javascript,
90 host,
91 lease,
92 renewal,
93 } = prepared;
94 let mut lease_lost = renewal.lease_lost();
95 let mut activity = host.activity();
96
97 let service = ActorHostGrpcService::new(host.clone(), invocation_auth).into_service();
98 if let Err(error) = executor_connection.mark_ready().await {
99 log_startup(&config, &timings, "failed", Some(&error));
100 return Err(error);
101 }
102 timings.executor_notified_at_ms = Some(timings.elapsed_ms());
103 log_startup(&config, &timings, "ready", None);
104 let stop = CancellationToken::new();
105 let server_stop = stop.clone();
106 let mut grpc_server = Box::pin(
107 Server::builder()
108 .add_service(service)
109 .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async move {
110 server_stop.cancelled().await
111 }),
112 );
113 let mut server =
114 Box::pin(async move { grpc_server.as_mut().await.context("serve actor host gRPC") });
115 let mut executor_task = Box::pin(executor_connection.run(stop.clone()));
116 tokio::pin!(shutdown);
117
118 info!(host_id = %config.host_id, namespace_id = %config.namespace_id, route, "durable-object host is ready");
119 let stop_result = wait_for_host_stop(
120 server.as_mut(),
121 executor_task.as_mut(),
122 &mut javascript,
123 shutdown.as_mut(),
124 &mut lease_lost,
125 &mut activity,
126 config.host_idle_timeout,
127 )
128 .await;
129 stop_host_tasks(&host, &stop, server, executor_task).await;
130 drop(javascript);
131 let renewal_result = renewal.shutdown().await;
132 let unregister_result = lease.unregister().await;
133 info!(host_id = %config.host_id, "durable-object host stopped");
134 stop_result?;
135 renewal_result?;
136 unregister_result
137}
138
139impl ActorHostConfig {
140 fn from_lookup(mut get: impl FnMut(&str) -> Option<String>) -> Result<Self> {
141 let startup_started_at = Instant::now();
142 let control_plane_url = required(&mut get, "DURABLE_OBJECT_CONTROL_PLANE_URL")?;
143 let host_token = required(&mut get, "DURABLE_OBJECT_HOST_TOKEN")?;
144 let jwt_public_keys = required(&mut get, "DURABLE_OBJECT_JWT_PUBLIC_KEYS")?;
145 let namespace_id = required(&mut get, "DURABLE_OBJECT_NAMESPACE_ID")?;
146 ActorScope {
147 namespace_id: namespace_id.clone(),
148 }
149 .validate()?;
150 let host_id = super::HostId::new(required(&mut get, "DURABLE_OBJECT_HOST_ID")?);
151 ensure!(
152 host_id
153 .as_str()
154 .starts_with(&format!("host.v1.{namespace_id}.")),
155 "DURABLE_OBJECT_HOST_ID does not belong to DURABLE_OBJECT_NAMESPACE_ID"
156 );
157 let session_id = required(&mut get, "DURABLE_OBJECT_SESSION_ID")?;
158 uuid::Uuid::parse_str(&session_id).context("DURABLE_OBJECT_SESSION_ID must be a UUID")?;
159 let executor_socket = get("DURABLE_OBJECT_EXECUTOR_SOCKET")
160 .map(PathBuf::from)
161 .unwrap_or_else(|| PathBuf::from("/tmp/durable-object-executor.sock"));
162 let host_route = get("DURABLE_OBJECT_HOST_ROUTE");
163 if let Some(route) = &host_route {
164 tonic::transport::Endpoint::from_shared(route.clone())
165 .context("DURABLE_OBJECT_HOST_ROUTE must be a valid HTTP or HTTPS URI")?;
166 }
167 let private_hostname = get("DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME");
168 if let Some(hostname) = &private_hostname {
169 ensure!(
170 !hostname.is_empty() && hostname.trim() == hostname,
171 "DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME must be non-empty without surrounding whitespace"
172 );
173 }
174 ensure!(
175 host_route.is_none() || private_hostname.is_none(),
176 "DURABLE_OBJECT_HOST_ROUTE and DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME are mutually exclusive"
177 );
178 let route_file = get("DURABLE_OBJECT_HOST_ROUTE_FILE").map(PathBuf::from);
179 let public_route_file = get("DURABLE_OBJECT_HOST_PUBLIC_ROUTE_FILE").map(PathBuf::from);
180 let metadata = HostMetadataFile::from_lookup(&mut get)?;
181 ensure!(
182 public_route_file.is_none()
183 || (host_route.is_none() && private_hostname.is_none() && route_file.is_none()),
184 "DURABLE_OBJECT_HOST_PUBLIC_ROUTE_FILE cannot be combined with other host route settings"
185 );
186 let host_bind = get("DURABLE_OBJECT_HOST_BIND")
187 .unwrap_or_else(|| {
188 if private_hostname.is_some() {
189 "[::]:7101"
190 } else if host_route.is_some() || public_route_file.is_some() {
191 "0.0.0.0:7101"
192 } else {
193 "127.0.0.1:0"
194 }
195 .into()
196 })
197 .parse()
198 .context("DURABLE_OBJECT_HOST_BIND must be a socket address")?;
199 let jwt_issuer = get("DURABLE_OBJECT_JWT_ISSUER")
200 .unwrap_or_else(|| "durable-object-control-plane".into());
201 let invocation_jwt_audience = get("DURABLE_OBJECT_INVOKE_JWT_AUDIENCE")
202 .unwrap_or_else(|| "durable-object-invoke".into());
203 let jwt_max_lifetime =
204 duration_seconds(&mut get, "DURABLE_OBJECT_JWT_MAX_TTL_SECONDS", 1_800)?;
205 let lease_duration = duration_ms(&mut get, "DURABLE_OBJECT_LEASE_MS", 30_000)?;
206 let renew_every = duration_ms(&mut get, "DURABLE_OBJECT_RENEW_MS", 10_000)?;
207 let host_idle_timeout = duration_ms(
208 &mut get,
209 "DURABLE_OBJECT_HOST_IDLE_TIMEOUT_MS",
210 DEFAULT_HOST_IDLE_TIMEOUT_MS,
211 )?;
212 ensure!(
213 host_idle_timeout.as_millis() <= u128::from(MAX_IDLE_TIMEOUT_MS),
214 "DURABLE_OBJECT_HOST_IDLE_TIMEOUT_MS is too large"
215 );
216 ensure!(
217 lease_duration.as_millis() <= u128::from(MAX_HOST_LEASE_DURATION_MS),
218 "DURABLE_OBJECT_LEASE_MS is too large"
219 );
220 ensure!(
221 renew_every < lease_duration,
222 "DURABLE_OBJECT_RENEW_MS must be shorter than DURABLE_OBJECT_LEASE_MS"
223 );
224 Ok(Self {
225 control_plane_url,
226 host_token,
227 jwt_public_keys,
228 namespace_id,
229 host_id,
230 session_id,
231 executor_socket,
232 host_bind,
233 host_route,
234 public_route_file,
235 private_hostname,
236 route_file,
237 jwt_issuer,
238 invocation_jwt_audience,
239 jwt_max_lifetime,
240 lease_duration,
241 renew_every,
242 host_idle_timeout,
243 metadata,
244 configuration_loaded_at_ms: startup_started_at.elapsed().as_secs_f64() * 1_000.0,
245 startup_started_at,
246 })
247 }
248}
249
250impl HostMetadataFile {
251 fn from_lookup(get: &mut impl FnMut(&str) -> Option<String>) -> Result<Option<Self>> {
252 let Some(path) = get("DURABLE_OBJECT_HOST_METADATA_FILE") else {
253 return Ok(None);
254 };
255 ensure!(
256 !path.is_empty(),
257 "DURABLE_OBJECT_HOST_METADATA_FILE must not be empty"
258 );
259 let canonical_region = required(get, "DURABLE_OBJECT_REGION")?;
260 crate::placement::validate_region(&canonical_region)?;
261 Ok(Some(Self {
262 path: path.into(),
263 canonical_region,
264 }))
265 }
266}
267
268struct PreparedActorHost {
269 invocation_auth: ActorJwtVerifier,
270 listener: TcpListener,
271 route: String,
272 executor_connection: ActorExecutorConnection,
273 javascript: tokio::process::Child,
274 host: Arc<ActorHost>,
275 lease: Arc<HostLeaseMaintainer>,
276 renewal: LeaseRenewalTask,
277}
278
279async fn prepare_actor_host(
280 config: &ActorHostConfig,
281 timings: &mut HostStartupTimings,
282) -> Result<PreparedActorHost> {
283 let invocation_auth = invocation_auth(config)?;
284 timings.authentication_ready_at_ms = Some(timings.elapsed_ms());
285 let started_at = timings.started_at;
286 let (control_plane, (listener, route, endpoint), (executor_connection, javascript)) =
287 connect_host_dependencies(
288 timed_connection(
289 started_at,
290 &mut timings.control_plane_connected_at_ms,
291 ControlPlaneClient::connect(&config.control_plane_url, &config.host_token),
292 ),
293 bind_host(
294 config,
295 started_at,
296 &mut timings.listener_bound_at_ms,
297 &mut timings.private_route_resolved_at_ms,
298 ),
299 timed_connection(
300 started_at,
301 &mut timings.executor_attached_at_ms,
302 connect_executor(
303 &config.executor_socket,
304 started_at,
305 &mut timings.javascript_spawned_at_ms,
306 ),
307 ),
308 )
309 .await?;
310 let control_plane = Arc::new(control_plane);
311 let host = Arc::new(ActorHost::new(
312 endpoint.clone(),
313 config.namespace_id.clone(),
314 executor_connection.executor(),
315 control_plane.clone(),
316 Arc::new(HttpStateTransport::new()),
317 ));
318 let lease = Arc::new(HostLeaseMaintainer::new(
319 endpoint,
320 config.session_id.clone(),
321 control_plane as Arc<dyn HostLeaseRegistry>,
322 Arc::new(SystemClock),
323 config.lease_duration,
324 config.renew_every,
325 )?);
326 let renewal = lease.clone().start().await?;
327 timings.lease_registered_at_ms = Some(timings.elapsed_ms());
328 Ok(PreparedActorHost {
329 invocation_auth,
330 listener,
331 route,
332 executor_connection,
333 javascript,
334 host,
335 lease,
336 renewal,
337 })
338}
339
340fn invocation_auth(config: &ActorHostConfig) -> Result<ActorJwtVerifier> {
341 ActorJwtVerifier::for_scope(
342 &config.jwt_public_keys,
343 config.jwt_issuer.clone(),
344 config.invocation_jwt_audience.clone(),
345 ActorTokenPurpose::Invocation,
346 config.jwt_max_lifetime,
347 )
348}
349
350async fn connect_host_dependencies<C, L, E>(
351 control_plane: impl Future<Output = Result<C>>,
352 listener: impl Future<Output = Result<L>>,
353 executor: impl Future<Output = Result<E>>,
354) -> Result<(C, L, E)> {
355 tokio::try_join!(control_plane, listener, executor)
356}
357
358async fn timed_connection<T>(
359 started_at: Instant,
360 milestone: &mut Option<f64>,
361 operation: impl Future<Output = Result<T>>,
362) -> Result<T> {
363 let result = operation.await?;
364 *milestone = Some(started_at.elapsed().as_secs_f64() * 1_000.0);
365 Ok(result)
366}
367
368async fn bind_host(
369 config: &ActorHostConfig,
370 started_at: Instant,
371 listener_bound_at_ms: &mut Option<f64>,
372 route_resolved_at_ms: &mut Option<f64>,
373) -> Result<(TcpListener, String, HostEndpoint)> {
374 let listener = TcpListener::bind(config.host_bind)
375 .await
376 .with_context(|| format!("bind actor host at {}", config.host_bind))?;
377 *listener_bound_at_ms = Some(started_at.elapsed().as_secs_f64() * 1_000.0);
378 let route = advertised_route(config, listener.local_addr()?).await?;
379 if let Some(path) = &config.route_file {
380 tokio::fs::write(path, &route)
381 .await
382 .with_context(|| format!("write actor host route to {}", path.display()))?;
383 }
384 write_host_metadata(config, &route).await?;
385 *route_resolved_at_ms = Some(started_at.elapsed().as_secs_f64() * 1_000.0);
386 let endpoint = HostEndpoint {
387 id: config.host_id.clone(),
388 route: route.clone(),
389 };
390 Ok((listener, route, endpoint))
391}
392
393async fn write_host_metadata(config: &ActorHostConfig, route: &str) -> Result<()> {
394 let Some(metadata) = &config.metadata else {
395 return Ok(());
396 };
397 let document = serde_json::to_vec(&serde_json::json!({
398 "hostId": config.host_id,
399 "route": route,
400 "canonicalRegion": metadata.canonical_region,
401 }))?;
402 let temporary = metadata
403 .path
404 .with_extension(format!("{}.tmp", uuid::Uuid::new_v4()));
405 tokio::fs::write(&temporary, document)
406 .await
407 .context("write actor host metadata")?;
408 tokio::fs::rename(&temporary, &metadata.path)
409 .await
410 .context("publish actor host metadata")
411}
412
413struct HostStartupTimings {
414 started_at: Instant,
415 configuration_loaded_at_ms: f64,
416 authentication_ready_at_ms: Option<f64>,
417 control_plane_connected_at_ms: Option<f64>,
418 listener_bound_at_ms: Option<f64>,
419 private_route_resolved_at_ms: Option<f64>,
420 executor_attached_at_ms: Option<f64>,
421 javascript_spawned_at_ms: Option<f64>,
422 lease_registered_at_ms: Option<f64>,
423 executor_notified_at_ms: Option<f64>,
424}
425
426impl HostStartupTimings {
427 fn new(config: &ActorHostConfig) -> Self {
428 Self {
429 started_at: config.startup_started_at,
430 configuration_loaded_at_ms: config.configuration_loaded_at_ms,
431 authentication_ready_at_ms: None,
432 control_plane_connected_at_ms: None,
433 listener_bound_at_ms: None,
434 private_route_resolved_at_ms: None,
435 executor_attached_at_ms: None,
436 javascript_spawned_at_ms: None,
437 lease_registered_at_ms: None,
438 executor_notified_at_ms: None,
439 }
440 }
441
442 fn elapsed_ms(&self) -> f64 {
443 self.started_at.elapsed().as_secs_f64() * 1_000.0
444 }
445}
446
447fn log_startup(
448 config: &ActorHostConfig,
449 timings: &HostStartupTimings,
450 outcome: &str,
451 error: Option<&anyhow::Error>,
452) {
453 info!(
454 event = "actor_host_startup",
455 namespace_id = %config.namespace_id,
456 host_id = %config.host_id,
457 started_at_ms = 0,
458 configuration_loaded_at_ms = timings.configuration_loaded_at_ms,
459 authentication_ready_at_ms = timings.authentication_ready_at_ms,
460 control_plane_connected_at_ms = timings.control_plane_connected_at_ms,
461 listener_bound_at_ms = timings.listener_bound_at_ms,
462 private_route_resolved_at_ms = timings.private_route_resolved_at_ms,
463 executor_attached_at_ms = timings.executor_attached_at_ms,
464 javascript_spawned_at_ms = timings.javascript_spawned_at_ms,
465 lease_registered_at_ms = timings.lease_registered_at_ms,
466 executor_notified_at_ms = timings.executor_notified_at_ms,
467 completed_at_ms = timings.elapsed_ms(),
468 outcome,
469 error = error.map(|error| format!("{error:#}")),
470 "actor host startup completed"
471 );
472}
473
474async fn advertised_route(config: &ActorHostConfig, bound: SocketAddr) -> Result<String> {
475 if let Some(route) = &config.host_route {
476 return Ok(route.clone());
477 }
478 if let Some(path) = &config.public_route_file {
479 return tokio::time::timeout(Duration::from_secs(60), read_public_route(path))
480 .await
481 .context("public host route was not published within 60 seconds")?;
482 }
483 let Some(hostname) = &config.private_hostname else {
484 return Ok(format!("http://{bound}"));
485 };
486 let address = lookup_host((hostname.as_str(), bound.port()))
487 .await
488 .with_context(|| format!("resolve actor host private hostname {hostname}"))?
489 .find(SocketAddr::is_ipv6)
490 .with_context(|| format!("actor host private hostname {hostname} has no IPv6 address"))?;
491 Ok(format!("http://{address}"))
492}
493
494async fn read_public_route(path: &std::path::Path) -> Result<String> {
495 loop {
496 match tokio::fs::read_to_string(path).await {
497 Ok(route) if !route.trim().is_empty() => {
498 let route = route.trim().to_owned();
499 let endpoint = tonic::transport::Endpoint::from_shared(route.clone())
500 .context("public host route file must contain a valid HTTPS URI")?;
501 ensure!(
502 endpoint.uri().scheme_str() == Some("https") && endpoint.uri().host().is_some(),
503 "public host route must use HTTPS"
504 );
505 return Ok(route);
506 }
507 Ok(_) => {}
508 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
509 Err(error) => return Err(error).context("read public host route"),
510 }
511 tokio::time::sleep(Duration::from_millis(50)).await;
512 }
513}
514
515async fn connect_executor(
516 socket: &std::path::Path,
517 started_at: Instant,
518 javascript_spawned_at_ms: &mut Option<f64>,
519) -> Result<(ActorExecutorConnection, tokio::process::Child)> {
520 let listener = ActorExecutorListener::bind(socket).await?;
521 let javascript = spawn_javascript_process()?;
522 *javascript_spawned_at_ms = Some(started_at.elapsed().as_secs_f64() * 1_000.0);
523 Ok((listener.accept().await?, javascript))
524}
525
526async fn wait_for_host_stop<ServerFuture, ExecutorFuture, ShutdownFuture>(
527 mut server: std::pin::Pin<&mut ServerFuture>,
528 mut executor: std::pin::Pin<&mut ExecutorFuture>,
529 javascript: &mut tokio::process::Child,
530 mut shutdown: std::pin::Pin<&mut ShutdownFuture>,
531 lease_lost: &mut tokio::sync::watch::Receiver<bool>,
532 activity: &mut tokio::sync::watch::Receiver<usize>,
533 idle_timeout: Duration,
534) -> Result<()>
535where
536 ServerFuture: Future<Output = Result<()>> + ?Sized,
537 ExecutorFuture: Future<Output = Result<()>> + ?Sized,
538 ShutdownFuture: Future<Output = ()> + ?Sized,
539{
540 let mut idle_deadline = tokio::time::Instant::now() + idle_timeout;
541 loop {
542 tokio::select! {
543 result = server.as_mut() => break result.context("serve actor host network endpoints"),
544 result = executor.as_mut() => break result.context("run JavaScript actor executor"),
545 result = javascript.wait() => break Err(anyhow::anyhow!("JavaScript actor executor exited with {}", result?)),
546 () = shutdown.as_mut() => break Ok(()),
547 changed = lease_lost.changed() => {
548 if changed.is_err() || *lease_lost.borrow() {
549 break Err(anyhow::anyhow!("host lease expired; host self-fenced"));
550 }
551 }
552 changed = activity.changed() => {
553 if changed.is_err() { break Err(anyhow::anyhow!("actor activity tracker stopped")); }
554 if *activity.borrow() == 0 {
555 idle_deadline = tokio::time::Instant::now() + idle_timeout;
556 }
557 }
558 () = tokio::time::sleep_until(idle_deadline), if *activity.borrow() == 0 => break Ok(()),
559 }
560 }
561}
562
563async fn stop_host_tasks(
564 host: &ActorHost,
565 stop: &CancellationToken,
566 server: impl Future,
567 executor: impl Future,
568) {
569 if let Err(error) = host.drain(HOST_ACTOR_DRAIN_TIMEOUT).await {
570 error!(error = %format!("{error:#}"), "actor invocations did not drain cleanly");
571 }
572 stop.cancel();
573 let _ = tokio::time::timeout(HOST_TASK_SHUTDOWN_TIMEOUT, async {
574 let _ = tokio::join!(server, executor);
575 })
576 .await;
577}
578
579fn spawn_javascript_process() -> Result<tokio::process::Child> {
580 Command::new("node")
581 .args([
582 "--eval",
583 "import(\"little-durable-objects/host\").then(module => module.runDurableObjectHost())",
584 ])
585 .stdin(Stdio::null())
586 .stdout(Stdio::inherit())
587 .stderr(Stdio::inherit())
588 .kill_on_drop(true)
589 .spawn()
590 .context("start JavaScript actor executor")
591}
592
593fn required(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
594 let value = get(name).with_context(|| format!("{name} is required"))?;
595 ensure!(!value.is_empty(), "{name} must not be empty");
596 Ok(value)
597}
598
599fn duration_ms(
600 get: &mut impl FnMut(&str) -> Option<String>,
601 name: &str,
602 default: u64,
603) -> Result<Duration> {
604 let value = get(name)
605 .map(|value| value.parse::<u64>())
606 .transpose()
607 .with_context(|| format!("{name} must be an integer number of milliseconds"))?
608 .unwrap_or(default);
609 ensure!(value > 0, "{name} must be positive");
610 Ok(Duration::from_millis(value))
611}
612
613fn duration_seconds(
614 get: &mut impl FnMut(&str) -> Option<String>,
615 name: &str,
616 default: u64,
617) -> Result<Duration> {
618 let value = get(name)
619 .map(|value| value.parse::<u64>())
620 .transpose()
621 .with_context(|| format!("{name} must be an integer number of seconds"))?
622 .unwrap_or(default);
623 ensure!(value > 0, "{name} must be positive");
624 Ok(Duration::from_secs(value))
625}
626
627#[cfg(test)]
628mod tests {
629 use super::*;
630 use std::collections::HashMap;
631
632 #[test]
633 fn host_needs_no_local_state_directory() -> Result<()> {
634 let values = values();
635 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned())?;
636 assert_eq!(
637 config.executor_socket,
638 PathBuf::from("/tmp/durable-object-executor.sock")
639 );
640 assert_eq!(config.host_idle_timeout, Duration::from_secs(300));
641 Ok(())
642 }
643
644 #[tokio::test]
645 async fn host_publishes_complete_metadata_before_dependencies_are_ready() -> Result<()> {
646 let directory = tempfile::tempdir()?;
647 let path = directory.path().join("host.json");
648 let mut values = values();
649 values.insert("DURABLE_OBJECT_HOST_BIND".into(), "127.0.0.1:0".into());
650 values.insert(
651 "DURABLE_OBJECT_HOST_METADATA_FILE".into(),
652 path.display().to_string(),
653 );
654 values.insert("DURABLE_OBJECT_REGION".into(), "north-america-east".into());
655 values.insert(
656 "DURABLE_OBJECT_HOST_ROUTE".into(),
657 "https://host.example.com".into(),
658 );
659 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned())?;
660 let (_, route, _) = bind_host(&config, Instant::now(), &mut None, &mut None).await?;
661 let metadata: serde_json::Value = serde_json::from_slice(&tokio::fs::read(&path).await?)?;
662 assert_eq!(
663 metadata,
664 serde_json::json!({
665 "hostId": config.host_id,
666 "route": route,
667 "canonicalRegion": "north-america-east",
668 })
669 );
670 let handle: crate::sandbox::ActorHostHandle = serde_json::from_value(metadata)?;
671 assert_eq!(handle.host_id, config.host_id);
672 assert_eq!(std::fs::read_dir(directory.path())?.count(), 1);
673 Ok(())
674 }
675
676 #[tokio::test]
677 async fn metadata_publication_failure_prevents_host_readiness() -> Result<()> {
678 let directory = tempfile::tempdir()?;
679 let mut values = values();
680 values.insert(
681 "DURABLE_OBJECT_HOST_METADATA_FILE".into(),
682 directory
683 .path()
684 .join("missing/host.json")
685 .display()
686 .to_string(),
687 );
688 values.insert("DURABLE_OBJECT_REGION".into(), "north-america-east".into());
689 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned())?;
690 assert!(
691 bind_host(&config, Instant::now(), &mut None, &mut None)
692 .await
693 .is_err()
694 );
695 Ok(())
696 }
697
698 #[test]
699 fn host_metadata_requires_a_valid_region() {
700 for region in [None, Some(""), Some("bad/region")] {
701 let mut values = values();
702 values.insert(
703 "DURABLE_OBJECT_HOST_METADATA_FILE".into(),
704 "/tmp/host.json".into(),
705 );
706 if let Some(region) = region {
707 values.insert("DURABLE_OBJECT_REGION".into(), region.into());
708 }
709 assert!(ActorHostConfig::from_lookup(|name| values.get(name).cloned()).is_err());
710 }
711 }
712
713 #[tokio::test]
714 async fn host_connections_start_without_waiting_for_each_other() -> Result<()> {
715 let barrier = tokio::sync::Barrier::new(3);
716 let connect = || async {
717 barrier.wait().await;
718 Ok(())
719 };
720 tokio::time::timeout(
721 Duration::from_millis(100),
722 connect_host_dependencies(connect(), connect(), connect()),
723 )
724 .await
725 .context("host dependencies ran sequentially")??;
726 Ok(())
727 }
728
729 #[test]
730 fn private_network_hosts_bind_ipv6_and_publish_their_route() -> Result<()> {
731 let mut values = values();
732 values.insert(
733 "DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME".into(),
734 "i6pn.modal.local".into(),
735 );
736 values.insert(
737 "DURABLE_OBJECT_HOST_ROUTE_FILE".into(),
738 "/tmp/durable-object-route".into(),
739 );
740
741 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned())?;
742
743 assert_eq!(config.host_bind, "[::]:7101".parse()?);
744 assert_eq!(config.private_hostname.as_deref(), Some("i6pn.modal.local"));
745 assert_eq!(
746 config.route_file,
747 Some(PathBuf::from("/tmp/durable-object-route"))
748 );
749 Ok(())
750 }
751
752 #[tokio::test]
753 async fn public_route_can_arrive_after_the_host_process_starts() -> Result<()> {
754 let directory = tempfile::tempdir()?;
755 let path = directory.path().join("route");
756 let mut values = values();
757 values.insert(
758 "DURABLE_OBJECT_HOST_PUBLIC_ROUTE_FILE".into(),
759 path.display().to_string(),
760 );
761 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned())?;
762 assert_eq!(config.host_bind, "0.0.0.0:7101".parse()?);
763 let publish = async {
764 tokio::time::sleep(Duration::from_millis(10)).await;
765 tokio::fs::write(path, "https://host.example.com").await?;
766 anyhow::Ok(())
767 };
768 let (route, ()) = tokio::try_join!(advertised_route(&config, config.host_bind), publish)?;
769 assert_eq!(route, "https://host.example.com");
770 Ok(())
771 }
772
773 #[test]
774 fn public_route_file_cannot_be_combined_with_other_route_settings() {
775 for conflict in [
776 "DURABLE_OBJECT_HOST_ROUTE",
777 "DURABLE_OBJECT_HOST_PRIVATE_HOSTNAME",
778 "DURABLE_OBJECT_HOST_ROUTE_FILE",
779 ] {
780 let mut values = values();
781 values.insert(
782 "DURABLE_OBJECT_HOST_PUBLIC_ROUTE_FILE".into(),
783 "/tmp/input-route".into(),
784 );
785 values.insert(conflict.into(), "https://host.example.com".into());
786 assert!(
787 ActorHostConfig::from_lookup(|name| values.get(name).cloned()).is_err(),
788 "{conflict}"
789 );
790 }
791 }
792
793 #[test]
794 fn startup_timings_begin_with_only_configuration_loaded() {
795 let values = values();
796 let config = ActorHostConfig::from_lookup(|name| values.get(name).cloned()).unwrap();
797 let timings = HostStartupTimings::new(&config);
798
799 assert!(timings.configuration_loaded_at_ms <= timings.elapsed_ms());
800 assert!(timings.control_plane_connected_at_ms.is_none());
801 assert!(timings.javascript_spawned_at_ms.is_none());
802 assert!(timings.executor_notified_at_ms.is_none());
803 }
804
805 fn values() -> HashMap<String, String> {
806 HashMap::from([
807 (
808 "DURABLE_OBJECT_CONTROL_PLANE_URL".into(),
809 "http://127.0.0.1:7100".into(),
810 ),
811 ("DURABLE_OBJECT_HOST_TOKEN".into(), "host-jwt".into()),
812 ("DURABLE_OBJECT_JWT_PUBLIC_KEYS".into(), "{}".into()),
813 ("DURABLE_OBJECT_NAMESPACE_ID".into(), "project-1".into()),
814 (
815 "DURABLE_OBJECT_HOST_ID".into(),
816 "host.v1.project-1.revision-1.host-1".into(),
817 ),
818 (
819 "DURABLE_OBJECT_SESSION_ID".into(),
820 "00000000-0000-4000-8000-000000000001".into(),
821 ),
822 ])
823 }
824}