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