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