1use std::{collections::HashMap, env, future::Future, net::SocketAddr, sync::Arc, time::Duration};
2
3use anyhow::{Context, Result, ensure};
4use tonic::transport::Server;
5use tracing::info;
6
7use crate::{
8 host_leases::PostgresHostLeaseStore,
9 placement::PostgresObjectPlacementStore,
10 postgres::PostgresDatabase,
11 sandbox::{CommandSandboxProvider, HostSandboxRuntimeConfig},
12 storage_urls::{GcsStorageUrlSigner, validate_buckets},
13};
14
15use super::{ActorJwtVerifier, ControlPlaneService};
16
17const DEFAULT_JWT_ISSUER: &str = "durable-object-control-plane";
18const DEFAULT_AUTHORITY_AUDIENCE: &str = "durable-object-authority";
19const DEFAULT_INVOCATION_AUDIENCE: &str = "durable-object-invoke";
20const DEFAULT_JWT_TTL_SECONDS: u64 = 1_800;
21const DEFAULT_ACTOR_IDLE_TIMEOUT_MS: u64 = 60_000;
22const DEFAULT_HOST_IDLE_TIMEOUT_MS: u64 = 300_000;
23const MAX_IDLE_TIMEOUT_MS: u64 = 86_400_000;
24
25pub struct ControlPlaneProcessConfig {
26 pub bind: SocketAddr,
27 pub jwt_signing_key: String,
28 pub jwt_key_id: String,
29 pub jwt_issuer: String,
30 pub authority_audience: String,
31 pub invocation_audience: String,
32 pub jwt_max_lifetime: Duration,
33 pub admin_token: String,
34 pub storage: ControlPlaneStorageConfig,
35 pub sandbox_provider: Option<SandboxProviderConfig>,
36 pub socket_event_sink: Option<SocketEventSinkConfig>,
37 pub socket_authenticator: Option<SocketAuthenticatorConfig>,
38}
39
40pub struct ControlPlaneStorageConfig {
41 pub postgres_url: String,
42 pub standard_buckets: HashMap<String, String>,
43}
44
45pub struct SandboxProviderConfig {
46 pub provider_name: String,
47 pub command: String,
48 pub environment: HashMap<String, String>,
49 pub runtime: HostSandboxRuntimeConfig,
50}
51
52pub struct SocketEventSinkConfig {
53 pub url: String,
54 pub token: String,
55}
56
57pub struct SocketAuthenticatorConfig {
58 pub url: String,
59 pub token: String,
60}
61
62impl ControlPlaneProcessConfig {
63 pub fn from_env() -> Result<Self> {
64 Self::from_lookup(|name| env::var(name).ok())
65 }
66}
67
68pub async fn serve_control_plane(
69 config: ControlPlaneProcessConfig,
70 shutdown: impl Future<Output = ()> + Send + 'static,
71) -> Result<()> {
72 let bind = config.bind;
73 let routes = control_plane_routes(config).await?;
74 info!(bind = %bind, "durable-object control plane is ready");
75 Server::builder()
76 .accept_http1(true)
77 .add_routes(routes)
78 .serve_with_shutdown(bind, shutdown)
79 .await
80 .context("serve durable-object control plane")
81}
82
83async fn control_plane_routes(config: ControlPlaneProcessConfig) -> Result<tonic::service::Routes> {
84 let issuer = super::ActorJwtIssuer::from_base64_pkcs8(
85 &config.jwt_signing_key,
86 config.jwt_key_id,
87 config.jwt_issuer.clone(),
88 config.authority_audience.clone(),
89 config.invocation_audience,
90 config.jwt_max_lifetime,
91 )?;
92 let auth = ActorJwtVerifier::for_scope(
93 issuer.verifier_keys_json()?,
94 config.jwt_issuer,
95 config.authority_audience,
96 super::ActorTokenPurpose::ControlPlane,
97 config.jwt_max_lifetime,
98 )?;
99 let database = PostgresDatabase::connect(&config.storage.postgres_url).await?;
100 let leases = Arc::new(PostgresHostLeaseStore::from_database(database.clone()));
101 let placements = Arc::new(PostgresObjectPlacementStore::from_database(
102 database.clone(),
103 ));
104 let registry = Arc::new(super::PostgresAdminRegistry::from_database(database));
105 let storage_urls = Arc::new(GcsStorageUrlSigner::from_adc(
106 config.storage.standard_buckets,
107 )?);
108 let provisioner = sandbox_provisioner(config.sandbox_provider, &issuer, &leases)?;
109 let socket_events = config
110 .socket_event_sink
111 .map(|sink| super::event_sink::HttpSocketMessageEventSink::new(sink.url, sink.token))
112 .transpose()?
113 .map(|sink| Arc::new(sink) as Arc<dyn super::event_sink::SocketMessageEventSink>);
114 let socket_authenticator = config
115 .socket_authenticator
116 .map(|auth| super::socket_auth::HttpSocketAuthenticator::new(auth.url, auth.token))
117 .transpose()?
118 .map(|auth| Arc::new(auth) as Arc<dyn super::socket_auth::SocketAuthenticator>);
119 let service = ControlPlaneService::new(leases, placements, storage_urls, auth)
120 .with_routing(registry.clone(), issuer.clone(), provisioner)
121 .with_socket_event_sink(socket_events)
122 .with_socket_authenticator(socket_authenticator);
123 let admin = super::admin::AdminService::new(config.admin_token, registry, issuer)?;
124 let public_api = super::public_api::router(service.clone(), admin);
125 let internal_api = service.into_internal_service();
126 Ok(tonic::service::Routes::from(public_api).add_service(internal_api))
127}
128
129fn sandbox_provisioner(
130 config: Option<SandboxProviderConfig>,
131 issuer: &super::ActorJwtIssuer,
132 leases: &Arc<PostgresHostLeaseStore>,
133) -> Result<Option<Arc<dyn super::service::HostProvisioner>>> {
134 config
135 .map(
136 |config| -> Result<Arc<dyn super::service::HostProvisioner>> {
137 let provider = Arc::new(CommandSandboxProvider::new(
138 config.provider_name,
139 config.command,
140 config.environment,
141 )?);
142 Ok(Arc::new(super::service::SandboxHostProvisioner::new(
143 provider,
144 config.runtime,
145 issuer.clone(),
146 leases.clone(),
147 )))
148 },
149 )
150 .transpose()
151}
152
153impl ControlPlaneProcessConfig {
154 fn from_lookup(mut get: impl FnMut(&str) -> Option<String>) -> Result<Self> {
155 let bind = get("DURABLE_OBJECT_CONTROL_PLANE_BIND")
156 .unwrap_or_else(|| "127.0.0.1:7100".into())
157 .parse()
158 .context("DURABLE_OBJECT_CONTROL_PLANE_BIND must be a socket address")?;
159 let jwt_signing_key = required(&mut get, "DURABLE_OBJECT_JWT_SIGNING_KEY")?;
160 let jwt_key_id = get("DURABLE_OBJECT_JWT_KEY_ID").unwrap_or_else(|| "primary".into());
161 let jwt_issuer =
162 get("DURABLE_OBJECT_JWT_ISSUER").unwrap_or_else(|| DEFAULT_JWT_ISSUER.into());
163 let authority_audience = get("DURABLE_OBJECT_AUTHORITY_JWT_AUDIENCE")
164 .unwrap_or_else(|| DEFAULT_AUTHORITY_AUDIENCE.into());
165 let invocation_audience = get("DURABLE_OBJECT_INVOKE_JWT_AUDIENCE")
166 .unwrap_or_else(|| DEFAULT_INVOCATION_AUDIENCE.into());
167 let jwt_max_lifetime = Duration::from_secs(
168 get("DURABLE_OBJECT_JWT_MAX_TTL_SECONDS")
169 .map(|value| value.parse())
170 .transpose()
171 .context("DURABLE_OBJECT_JWT_MAX_TTL_SECONDS must be an integer")?
172 .unwrap_or(DEFAULT_JWT_TTL_SECONDS),
173 );
174 ensure!(
175 !jwt_max_lifetime.is_zero(),
176 "DURABLE_OBJECT_JWT_MAX_TTL_SECONDS must be positive"
177 );
178 let admin_token = required(&mut get, "DURABLE_OBJECT_ADMIN_TOKEN")?;
179 ensure!(
180 admin_token.trim() == admin_token,
181 "DURABLE_OBJECT_ADMIN_TOKEN has surrounding whitespace"
182 );
183 let standard_buckets: HashMap<String, String> =
184 serde_json::from_str(&required(&mut get, "DURABLE_OBJECT_STANDARD_BUCKETS")?)
185 .context("DURABLE_OBJECT_STANDARD_BUCKETS must be a JSON region-to-bucket map")?;
186 validate_buckets(&standard_buckets)?;
187 let storage = ControlPlaneStorageConfig {
188 postgres_url: required(&mut get, "DURABLE_OBJECT_POSTGRES_URL")?,
189 standard_buckets,
190 };
191 let sandbox_provider =
192 sandbox_provider_config(&mut get, &jwt_issuer, &invocation_audience)?;
193 let socket_event_sink = socket_event_sink_config(&mut get)?;
194 let socket_authenticator = socket_authenticator_config(&mut get)?;
195 Ok(Self {
196 bind,
197 jwt_signing_key,
198 jwt_key_id,
199 jwt_issuer,
200 authority_audience,
201 invocation_audience,
202 jwt_max_lifetime,
203 admin_token,
204 storage,
205 sandbox_provider,
206 socket_event_sink,
207 socket_authenticator,
208 })
209 }
210}
211
212fn socket_authenticator_config(
213 get: &mut impl FnMut(&str) -> Option<String>,
214) -> Result<Option<SocketAuthenticatorConfig>> {
215 let url = get("DURABLE_OBJECT_SOCKET_AUTH_URL");
216 let token = get("DURABLE_OBJECT_SOCKET_AUTH_TOKEN");
217 match (url, token) {
218 (None, None) => Ok(None),
219 (Some(url), Some(token)) => Ok(Some(SocketAuthenticatorConfig {
220 url: validated_http_url(&url, "DURABLE_OBJECT_SOCKET_AUTH_URL")?,
221 token,
222 })),
223 _ => anyhow::bail!(
224 "DURABLE_OBJECT_SOCKET_AUTH_URL and DURABLE_OBJECT_SOCKET_AUTH_TOKEN must be configured together"
225 ),
226 }
227}
228
229fn socket_event_sink_config(
230 get: &mut impl FnMut(&str) -> Option<String>,
231) -> Result<Option<SocketEventSinkConfig>> {
232 let url = get("DURABLE_OBJECT_SOCKET_EVENT_URL");
233 let token = get("DURABLE_OBJECT_SOCKET_EVENT_TOKEN");
234 match (url, token) {
235 (None, None) => Ok(None),
236 (Some(url), Some(token)) => Ok(Some(SocketEventSinkConfig {
237 url: validated_http_url(&url, "DURABLE_OBJECT_SOCKET_EVENT_URL")?,
238 token,
239 })),
240 _ => anyhow::bail!(
241 "DURABLE_OBJECT_SOCKET_EVENT_URL and DURABLE_OBJECT_SOCKET_EVENT_TOKEN must be configured together"
242 ),
243 }
244}
245
246fn sandbox_provider_config(
247 get: &mut impl FnMut(&str) -> Option<String>,
248 jwt_issuer: &str,
249 invocation_audience: &str,
250) -> Result<Option<SandboxProviderConfig>> {
251 let Some(provider_name) = get("DURABLE_OBJECT_SANDBOX_PROVIDER") else {
252 ensure!(
253 get("DURABLE_OBJECT_SANDBOX_COMMAND").is_none(),
254 "sandbox command requires a provider"
255 );
256 return Ok(None);
257 };
258 ensure!(
259 provider_name == "modal",
260 "unsupported sandbox provider {provider_name:?}"
261 );
262 let environment = HashMap::from([
263 (
264 "MODAL_TOKEN_ID".into(),
265 provider_credential(get, "MODAL_TOKEN_ID")?,
266 ),
267 (
268 "MODAL_TOKEN_SECRET".into(),
269 provider_credential(get, "MODAL_TOKEN_SECRET")?,
270 ),
271 ]);
272 let control_plane_url = validated_http_url(
273 &required(get, "DURABLE_OBJECT_CONTROL_PLANE_URL")?,
274 "DURABLE_OBJECT_CONTROL_PLANE_URL",
275 )?;
276 Ok(Some(SandboxProviderConfig {
277 provider_name,
278 command: get("DURABLE_OBJECT_SANDBOX_COMMAND")
279 .unwrap_or_else(|| "little-durable-objects-modal".into()),
280 environment,
281 runtime: HostSandboxRuntimeConfig {
282 control_plane_url,
283 jwt_issuer: jwt_issuer.into(),
284 invocation_jwt_audience: invocation_audience.into(),
285 actor_idle_timeout_ms: idle_timeout(
286 get,
287 "DURABLE_OBJECT_ACTOR_IDLE_TIMEOUT_MS",
288 DEFAULT_ACTOR_IDLE_TIMEOUT_MS,
289 )?,
290 host_idle_timeout_ms: idle_timeout(
291 get,
292 "DURABLE_OBJECT_HOST_IDLE_TIMEOUT_MS",
293 DEFAULT_HOST_IDLE_TIMEOUT_MS,
294 )?,
295 },
296 }))
297}
298
299fn provider_credential(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
300 let value = required(get, name)?;
301 ensure!(value.trim() == value, "{name} has surrounding whitespace");
302 Ok(value)
303}
304
305fn required(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
306 let value = get(name).with_context(|| format!("{name} is required"))?;
307 ensure!(!value.is_empty(), "{name} must not be empty");
308 Ok(value)
309}
310
311fn validated_http_url(value: &str, name: &str) -> Result<String> {
312 let url = reqwest::Url::parse(value).with_context(|| format!("{name} must be a URL"))?;
313 ensure!(
314 matches!(url.scheme(), "http" | "https") && url.host_str().is_some(),
315 "{name} must be HTTP or HTTPS"
316 );
317 Ok(url.to_string())
318}
319
320fn idle_timeout(
321 get: &mut impl FnMut(&str) -> Option<String>,
322 name: &str,
323 default: u64,
324) -> Result<u64> {
325 let value = get(name)
326 .map(|value| value.parse())
327 .transpose()
328 .with_context(|| format!("{name} must be an integer"))?
329 .unwrap_or(default);
330 ensure!(
331 (1..=MAX_IDLE_TIMEOUT_MS).contains(&value),
332 "{name} is outside the supported range"
333 );
334 Ok(value)
335}
336
337#[cfg(test)]
338mod tests {
339 use super::*;
340
341 #[test]
342 fn parses_the_minimal_storage_configuration() -> Result<()> {
343 let values = HashMap::from([
344 ("DURABLE_OBJECT_JWT_SIGNING_KEY", "c2lnbmluZw=="),
345 ("DURABLE_OBJECT_ADMIN_TOKEN", "admin-token"),
346 (
347 "DURABLE_OBJECT_POSTGRES_URL",
348 "postgresql://localhost/actors",
349 ),
350 (
351 "DURABLE_OBJECT_STANDARD_BUCKETS",
352 "{\"us-east\":\"actor-state-test\"}",
353 ),
354 ]);
355 let config = ControlPlaneProcessConfig::from_lookup(|name| {
356 values.get(name).map(|value| (*value).into())
357 })?;
358 assert_eq!(
359 config.storage.standard_buckets["us-east"],
360 "actor-state-test"
361 );
362 Ok(())
363 }
364
365 #[test]
366 fn parses_socket_event_sink_only_when_url_and_token_are_present() -> Result<()> {
367 let mut complete = HashMap::from([
368 (
369 "DURABLE_OBJECT_SOCKET_EVENT_URL",
370 "https://api.example.com/events",
371 ),
372 ("DURABLE_OBJECT_SOCKET_EVENT_TOKEN", "event-token"),
373 ]);
374 let sink =
375 socket_event_sink_config(&mut |name| complete.get(name).map(|value| (*value).into()))?
376 .context("socket event sink was not configured")?;
377 assert_eq!(sink.url, "https://api.example.com/events");
378 assert_eq!(sink.token, "event-token");
379
380 complete.remove("DURABLE_OBJECT_SOCKET_EVENT_TOKEN");
381 assert!(
382 socket_event_sink_config(&mut |name| complete.get(name).map(|value| (*value).into()))
383 .is_err()
384 );
385 Ok(())
386 }
387
388 #[test]
389 fn parses_socket_authenticator_only_when_url_and_token_are_present() -> Result<()> {
390 let mut complete = HashMap::from([
391 (
392 "DURABLE_OBJECT_SOCKET_AUTH_URL",
393 "https://api.example.com/authorize",
394 ),
395 ("DURABLE_OBJECT_SOCKET_AUTH_TOKEN", "auth-token"),
396 ]);
397 let auth = socket_authenticator_config(&mut |name| {
398 complete.get(name).map(|value| (*value).into())
399 })?
400 .context("socket authenticator was not configured")?;
401 assert_eq!(auth.url, "https://api.example.com/authorize");
402 assert_eq!(auth.token, "auth-token");
403
404 complete.remove("DURABLE_OBJECT_SOCKET_AUTH_TOKEN");
405 assert!(
406 socket_authenticator_config(&mut |name| complete
407 .get(name)
408 .map(|value| (*value).into()))
409 .is_err()
410 );
411 Ok(())
412 }
413}