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