#[cfg(test)]
use std::{
collections::{HashMap, HashSet},
sync::Mutex,
};
use anyhow::{Context, Result, ensure};
use async_trait::async_trait;
use subtle::ConstantTimeEq;
use crate::{actor::ActorScope, postgres::PostgresDatabase};
use super::{ActorJwtIssuer, issuer::IssuedActorToken};
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct HostLaunchSpec {
pub namespace_id: String,
pub code_revision: String,
pub image_ref: String,
pub working_directory: String,
pub actor_entrypoint: Option<String>,
}
impl HostLaunchSpec {
pub(crate) fn validate(&self) -> Result<()> {
ActorScope {
namespace_id: self.namespace_id.clone(),
}
.validate()?;
validate_component("code revision", &self.code_revision, 128)?;
ensure!(
!self.image_ref.is_empty() && self.image_ref.len() <= 255,
"sandbox image reference must contain between 1 and 255 bytes"
);
ensure!(
self.working_directory.starts_with('/') && self.working_directory.len() <= 1024,
"host working directory must be an absolute path of at most 1024 bytes"
);
if let Some(entrypoint) = &self.actor_entrypoint {
ensure!(
!entrypoint.is_empty() && entrypoint.len() <= 1024,
"actor entrypoint must contain between 1 and 1024 bytes"
);
}
Ok(())
}
}
#[async_trait]
pub(crate) trait AdminRegistry: Send + Sync {
async fn ensure_namespace_and_register_deployment(&self, spec: &HostLaunchSpec)
-> Result<bool>;
async fn launch_spec(&self, namespace_id: &str) -> Result<Option<HostLaunchSpec>>;
}
#[derive(Clone)]
pub(crate) struct AdminService {
token: String,
registry: std::sync::Arc<dyn AdminRegistry>,
issuer: ActorJwtIssuer,
}
impl AdminService {
pub(crate) fn new(
token: String,
registry: std::sync::Arc<dyn AdminRegistry>,
issuer: ActorJwtIssuer,
) -> Result<Self> {
ensure!(
!token.is_empty() && token.trim() == token,
"admin token is invalid"
);
Ok(Self {
token,
registry,
issuer,
})
}
pub(crate) fn authenticate(&self, authorization: &str) -> Result<()> {
let token = authorization
.strip_prefix("Bearer ")
.context("admin credential must use Bearer authentication")?;
ensure!(
!token.is_empty()
&& token.trim() == token
&& bool::from(token.as_bytes().ct_eq(self.token.as_bytes())),
"admin credential is invalid"
);
Ok(())
}
pub(crate) async fn ensure_namespace_and_register_deployment(
&self,
spec: &HostLaunchSpec,
) -> Result<bool> {
self.registry
.ensure_namespace_and_register_deployment(spec)
.await
}
pub(crate) async fn deployment_exists(&self, namespace_id: &str) -> Result<bool> {
Ok(self.registry.launch_spec(namespace_id).await?.is_some())
}
pub(crate) fn issue_workflow_token(
&self,
namespace_id: &str,
execution_id: &str,
storage_region: &str,
deadline_unix_ms: i64,
) -> Result<IssuedActorToken> {
self.issuer
.issue_workflow(namespace_id, execution_id, storage_region, deadline_unix_ms)
}
pub(crate) fn jwks_json(&self) -> Result<Vec<u8>> {
self.issuer.jwks_json()
}
}
#[cfg(test)]
#[derive(Default)]
pub(crate) struct LocalAdminRegistry {
state: Mutex<LocalAdminState>,
}
#[cfg(test)]
#[derive(Default)]
struct LocalAdminState {
namespaces: HashSet<String>,
launch_specs: HashMap<String, HostLaunchSpec>,
}
#[cfg(test)]
#[async_trait]
impl AdminRegistry for LocalAdminRegistry {
async fn ensure_namespace_and_register_deployment(
&self,
spec: &HostLaunchSpec,
) -> Result<bool> {
spec.validate()?;
let mut state = self
.state
.lock()
.map_err(|_| anyhow::anyhow!("admin registry lock poisoned"))?;
state.namespaces.insert(spec.namespace_id.clone());
let changed = state.launch_specs.get(&spec.namespace_id) != Some(spec);
state
.launch_specs
.insert(spec.namespace_id.clone(), spec.clone());
Ok(changed)
}
async fn launch_spec(&self, namespace_id: &str) -> Result<Option<HostLaunchSpec>> {
validate_namespace(namespace_id)?;
Ok(self
.state
.lock()
.map_err(|_| anyhow::anyhow!("admin registry lock poisoned"))?
.launch_specs
.get(namespace_id)
.cloned())
}
}
pub(crate) struct PostgresAdminRegistry {
database: PostgresDatabase,
}
impl PostgresAdminRegistry {
pub(crate) fn from_database(database: PostgresDatabase) -> Self {
Self { database }
}
}
#[async_trait]
impl AdminRegistry for PostgresAdminRegistry {
async fn ensure_namespace_and_register_deployment(
&self,
spec: &HostLaunchSpec,
) -> Result<bool> {
spec.validate()?;
Ok(self
.database
.client()
.execute(
"WITH ensured_namespace AS ( \
INSERT INTO durable_object_namespaces (namespace_id) VALUES ($1) \
ON CONFLICT (namespace_id) DO UPDATE SET namespace_id = EXCLUDED.namespace_id \
RETURNING namespace_id \
) \
INSERT INTO durable_object_project_specs \
(namespace_id, code_revision, image_ref, working_directory, actor_entrypoint) \
SELECT namespace_id, $2, $3, $4, $5 FROM ensured_namespace \
ON CONFLICT (namespace_id) DO UPDATE SET \
code_revision = EXCLUDED.code_revision, image_ref = EXCLUDED.image_ref, \
working_directory = EXCLUDED.working_directory, actor_entrypoint = EXCLUDED.actor_entrypoint, \
updated_at = clock_timestamp() \
WHERE (durable_object_project_specs.code_revision, \
durable_object_project_specs.image_ref, \
durable_object_project_specs.working_directory, \
durable_object_project_specs.actor_entrypoint) \
IS DISTINCT FROM \
(EXCLUDED.code_revision, EXCLUDED.image_ref, \
EXCLUDED.working_directory, EXCLUDED.actor_entrypoint)",
&[
&spec.namespace_id,
&spec.code_revision,
&spec.image_ref,
&spec.working_directory,
&spec.actor_entrypoint,
],
)
.await
.context("ensure PostgreSQL namespace and register project deployment")?
== 1)
}
async fn launch_spec(&self, namespace_id: &str) -> Result<Option<HostLaunchSpec>> {
validate_namespace(namespace_id)?;
Ok(self
.database
.client()
.query_opt(
"SELECT code_revision, image_ref, working_directory, actor_entrypoint \
FROM durable_object_project_specs WHERE namespace_id = $1",
&[&namespace_id],
)
.await
.context("load PostgreSQL host launch spec")?
.map(|row| HostLaunchSpec {
namespace_id: namespace_id.to_owned(),
code_revision: row.get(0),
image_ref: row.get(1),
working_directory: row.get(2),
actor_entrypoint: row.get(3),
}))
}
}
fn validate_namespace(namespace_id: &str) -> Result<()> {
ActorScope {
namespace_id: namespace_id.to_owned(),
}
.validate()
}
fn validate_component(name: &str, value: &str, maximum: usize) -> Result<()> {
ensure!(!value.is_empty(), "{name} must not be empty");
ensure!(
value.len() <= maximum,
"{name} must be at most {maximum} bytes"
);
ensure!(
value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')),
"{name} contains unsupported characters"
);
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn registration_replaces_the_projects_active_revision() -> Result<()> {
let registry = LocalAdminRegistry::default();
assert!(
registry
.ensure_namespace_and_register_deployment(&spec("im-1"))
.await?
);
assert!(
!registry
.ensure_namespace_and_register_deployment(&spec("im-1"))
.await?
);
let mut replacement = spec("im-2");
replacement.code_revision = "revision-2".into();
assert!(
registry
.ensure_namespace_and_register_deployment(&replacement)
.await?
);
assert_eq!(registry.launch_spec("project-1").await?, Some(replacement));
assert!(
registry
.state
.lock()
.map_err(|_| anyhow::anyhow!("admin registry lock poisoned"))?
.namespaces
.contains("project-1")
);
Ok(())
}
#[tokio::test]
async fn postgres_registration_ensures_the_namespace_and_deployment_atomically() -> Result<()> {
let Some(url) = std::env::var("DURABLE_OBJECT_TEST_POSTGRES_URL").ok() else {
return Ok(());
};
let database = PostgresDatabase::connect(&url).await?;
let registry = PostgresAdminRegistry::from_database(database.clone());
let mut deployment = spec("image-1");
deployment.namespace_id = format!("project-{}", uuid::Uuid::new_v4());
assert!(
registry
.ensure_namespace_and_register_deployment(&deployment)
.await?
);
assert_eq!(
registry.launch_spec(&deployment.namespace_id).await?,
Some(deployment.clone())
);
let namespace_exists = database
.client()
.query_one(
"SELECT EXISTS(SELECT 1 FROM durable_object_namespaces WHERE namespace_id = $1)",
&[&deployment.namespace_id],
)
.await?
.get::<_, bool>(0);
assert!(namespace_exists);
Ok(())
}
fn spec(image: &str) -> HostLaunchSpec {
HostLaunchSpec {
namespace_id: "project-1".into(),
code_revision: "revision-1".into(),
image_ref: image.into(),
working_directory: "/workspace".into(),
actor_entrypoint: Some("src/durable-objects.ts".into()),
}
}
}