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