little_durable_objects/control_plane/
process.rs

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}
37
38pub struct ControlPlaneStorageConfig {
39    pub postgres_url: String,
40    pub standard_buckets: HashMap<String, String>,
41}
42
43pub struct SandboxProviderConfig {
44    pub provider_name: String,
45    pub command: String,
46    pub environment: HashMap<String, String>,
47    pub runtime: HostSandboxRuntimeConfig,
48}
49
50impl ControlPlaneProcessConfig {
51    pub fn from_env() -> Result<Self> {
52        Self::from_lookup(|name| env::var(name).ok())
53    }
54}
55
56pub async fn serve_control_plane(
57    config: ControlPlaneProcessConfig,
58    shutdown: impl Future<Output = ()> + Send + 'static,
59) -> Result<()> {
60    let bind = config.bind;
61    let routes = control_plane_routes(config).await?;
62    info!(bind = %bind, "durable-object control plane is ready");
63    Server::builder()
64        .accept_http1(true)
65        .add_routes(routes)
66        .serve_with_shutdown(bind, shutdown)
67        .await
68        .context("serve durable-object control plane")
69}
70
71async fn control_plane_routes(config: ControlPlaneProcessConfig) -> Result<tonic::service::Routes> {
72    let issuer = super::ActorJwtIssuer::from_base64_pkcs8(
73        &config.jwt_signing_key,
74        config.jwt_key_id,
75        config.jwt_issuer.clone(),
76        config.authority_audience.clone(),
77        config.invocation_audience,
78        config.jwt_max_lifetime,
79    )?;
80    let auth = ActorJwtVerifier::for_scope(
81        issuer.verifier_keys_json()?,
82        config.jwt_issuer,
83        config.authority_audience,
84        super::ActorTokenPurpose::ControlPlane,
85        config.jwt_max_lifetime,
86    )?;
87    let database = PostgresDatabase::connect(&config.storage.postgres_url).await?;
88    let leases = Arc::new(PostgresHostLeaseStore::from_database(database.clone()));
89    let placements = Arc::new(PostgresObjectPlacementStore::from_database(
90        database.clone(),
91    ));
92    let registry = Arc::new(super::PostgresAdminRegistry::from_database(database));
93    let storage_urls = Arc::new(GcsStorageUrlSigner::from_adc(
94        config.storage.standard_buckets,
95    )?);
96    let provisioner = sandbox_provisioner(config.sandbox_provider, &issuer, &leases)?;
97    let service = ControlPlaneService::new(leases, placements, storage_urls, auth).with_routing(
98        registry.clone(),
99        issuer.clone(),
100        provisioner,
101    );
102    let admin = super::admin::AdminService::new(config.admin_token, registry, issuer)?;
103    let public_api = super::public_api::router(service.clone(), admin);
104    let internal_api = service.into_internal_service();
105    Ok(tonic::service::Routes::from(public_api).add_service(internal_api))
106}
107
108fn sandbox_provisioner(
109    config: Option<SandboxProviderConfig>,
110    issuer: &super::ActorJwtIssuer,
111    leases: &Arc<PostgresHostLeaseStore>,
112) -> Result<Option<Arc<dyn super::service::HostProvisioner>>> {
113    config
114        .map(
115            |config| -> Result<Arc<dyn super::service::HostProvisioner>> {
116                let provider = Arc::new(CommandSandboxProvider::new(
117                    config.provider_name,
118                    config.command,
119                    config.environment,
120                )?);
121                Ok(Arc::new(super::service::SandboxHostProvisioner::new(
122                    provider,
123                    config.runtime,
124                    issuer.clone(),
125                    leases.clone(),
126                )))
127            },
128        )
129        .transpose()
130}
131
132impl ControlPlaneProcessConfig {
133    fn from_lookup(mut get: impl FnMut(&str) -> Option<String>) -> Result<Self> {
134        let bind = get("DURABLE_OBJECT_CONTROL_PLANE_BIND")
135            .unwrap_or_else(|| "127.0.0.1:7100".into())
136            .parse()
137            .context("DURABLE_OBJECT_CONTROL_PLANE_BIND must be a socket address")?;
138        let jwt_signing_key = required(&mut get, "DURABLE_OBJECT_JWT_SIGNING_KEY")?;
139        let jwt_key_id = get("DURABLE_OBJECT_JWT_KEY_ID").unwrap_or_else(|| "primary".into());
140        let jwt_issuer =
141            get("DURABLE_OBJECT_JWT_ISSUER").unwrap_or_else(|| DEFAULT_JWT_ISSUER.into());
142        let authority_audience = get("DURABLE_OBJECT_AUTHORITY_JWT_AUDIENCE")
143            .unwrap_or_else(|| DEFAULT_AUTHORITY_AUDIENCE.into());
144        let invocation_audience = get("DURABLE_OBJECT_INVOKE_JWT_AUDIENCE")
145            .unwrap_or_else(|| DEFAULT_INVOCATION_AUDIENCE.into());
146        let jwt_max_lifetime = Duration::from_secs(
147            get("DURABLE_OBJECT_JWT_MAX_TTL_SECONDS")
148                .map(|value| value.parse())
149                .transpose()
150                .context("DURABLE_OBJECT_JWT_MAX_TTL_SECONDS must be an integer")?
151                .unwrap_or(DEFAULT_JWT_TTL_SECONDS),
152        );
153        ensure!(
154            !jwt_max_lifetime.is_zero(),
155            "DURABLE_OBJECT_JWT_MAX_TTL_SECONDS must be positive"
156        );
157        let admin_token = required(&mut get, "DURABLE_OBJECT_ADMIN_TOKEN")?;
158        ensure!(
159            admin_token.trim() == admin_token,
160            "DURABLE_OBJECT_ADMIN_TOKEN has surrounding whitespace"
161        );
162        let standard_buckets: HashMap<String, String> =
163            serde_json::from_str(&required(&mut get, "DURABLE_OBJECT_STANDARD_BUCKETS")?)
164                .context("DURABLE_OBJECT_STANDARD_BUCKETS must be a JSON region-to-bucket map")?;
165        validate_buckets(&standard_buckets)?;
166        let storage = ControlPlaneStorageConfig {
167            postgres_url: required(&mut get, "DURABLE_OBJECT_POSTGRES_URL")?,
168            standard_buckets,
169        };
170        let sandbox_provider =
171            sandbox_provider_config(&mut get, &jwt_issuer, &invocation_audience)?;
172        Ok(Self {
173            bind,
174            jwt_signing_key,
175            jwt_key_id,
176            jwt_issuer,
177            authority_audience,
178            invocation_audience,
179            jwt_max_lifetime,
180            admin_token,
181            storage,
182            sandbox_provider,
183        })
184    }
185}
186
187fn sandbox_provider_config(
188    get: &mut impl FnMut(&str) -> Option<String>,
189    jwt_issuer: &str,
190    invocation_audience: &str,
191) -> Result<Option<SandboxProviderConfig>> {
192    let Some(provider_name) = get("DURABLE_OBJECT_SANDBOX_PROVIDER") else {
193        ensure!(
194            get("DURABLE_OBJECT_SANDBOX_COMMAND").is_none(),
195            "sandbox command requires a provider"
196        );
197        return Ok(None);
198    };
199    ensure!(
200        provider_name == "modal",
201        "unsupported sandbox provider {provider_name:?}"
202    );
203    let environment = HashMap::from([
204        (
205            "MODAL_TOKEN_ID".into(),
206            provider_credential(get, "MODAL_TOKEN_ID")?,
207        ),
208        (
209            "MODAL_TOKEN_SECRET".into(),
210            provider_credential(get, "MODAL_TOKEN_SECRET")?,
211        ),
212    ]);
213    let control_plane_url = validated_http_url(
214        &required(get, "DURABLE_OBJECT_CONTROL_PLANE_URL")?,
215        "DURABLE_OBJECT_CONTROL_PLANE_URL",
216    )?;
217    Ok(Some(SandboxProviderConfig {
218        provider_name,
219        command: get("DURABLE_OBJECT_SANDBOX_COMMAND")
220            .unwrap_or_else(|| "little-durable-objects-modal".into()),
221        environment,
222        runtime: HostSandboxRuntimeConfig {
223            control_plane_url,
224            jwt_issuer: jwt_issuer.into(),
225            invocation_jwt_audience: invocation_audience.into(),
226            actor_idle_timeout_ms: idle_timeout(
227                get,
228                "DURABLE_OBJECT_ACTOR_IDLE_TIMEOUT_MS",
229                DEFAULT_ACTOR_IDLE_TIMEOUT_MS,
230            )?,
231            host_idle_timeout_ms: idle_timeout(
232                get,
233                "DURABLE_OBJECT_HOST_IDLE_TIMEOUT_MS",
234                DEFAULT_HOST_IDLE_TIMEOUT_MS,
235            )?,
236        },
237    }))
238}
239
240fn provider_credential(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
241    let value = required(get, name)?;
242    ensure!(value.trim() == value, "{name} has surrounding whitespace");
243    Ok(value)
244}
245
246fn required(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
247    let value = get(name).with_context(|| format!("{name} is required"))?;
248    ensure!(!value.is_empty(), "{name} must not be empty");
249    Ok(value)
250}
251
252fn validated_http_url(value: &str, name: &str) -> Result<String> {
253    let url = reqwest::Url::parse(value).with_context(|| format!("{name} must be a URL"))?;
254    ensure!(
255        matches!(url.scheme(), "http" | "https") && url.host_str().is_some(),
256        "{name} must be HTTP or HTTPS"
257    );
258    Ok(url.to_string())
259}
260
261fn idle_timeout(
262    get: &mut impl FnMut(&str) -> Option<String>,
263    name: &str,
264    default: u64,
265) -> Result<u64> {
266    let value = get(name)
267        .map(|value| value.parse())
268        .transpose()
269        .with_context(|| format!("{name} must be an integer"))?
270        .unwrap_or(default);
271    ensure!(
272        (1..=MAX_IDLE_TIMEOUT_MS).contains(&value),
273        "{name} is outside the supported range"
274    );
275    Ok(value)
276}
277
278#[cfg(test)]
279mod tests {
280    use super::*;
281
282    #[test]
283    fn parses_the_minimal_storage_configuration() -> Result<()> {
284        let values = HashMap::from([
285            ("DURABLE_OBJECT_JWT_SIGNING_KEY", "c2lnbmluZw=="),
286            ("DURABLE_OBJECT_ADMIN_TOKEN", "admin-token"),
287            (
288                "DURABLE_OBJECT_POSTGRES_URL",
289                "postgresql://localhost/actors",
290            ),
291            (
292                "DURABLE_OBJECT_STANDARD_BUCKETS",
293                "{\"us-east\":\"actor-state-test\"}",
294            ),
295        ]);
296        let config = ControlPlaneProcessConfig::from_lookup(|name| {
297            values.get(name).map(|value| (*value).into())
298        })?;
299        assert_eq!(
300            config.storage.standard_buckets["us-east"],
301            "actor-state-test"
302        );
303        Ok(())
304    }
305}
little_durable_objects/control_plane/process.rs

little_durable_objects/control_plane/
process.rs
