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