use std::future::Future;
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use axum::body::Body;
use axum::extract::{ConnectInfo, Path, Query, Request, State};
use axum::http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{any, get, post, put};
use axum::{Extension, Json, Router};
use boatramp_core::access::{AccessConfig, BasicAuth};
use boatramp_core::authz::{GrantedRole, TokenMeta};
use boatramp_core::config::{DeployConfig, SiteConfig};
use boatramp_core::cose::{self, Claims, Signer};
use boatramp_core::deploy::{
DeployMetaInput, DeployStore, FileEntry, GcOptions, GcReport, Manifest,
};
use boatramp_core::matcher::Pattern;
use boatramp_core::route::{self, Outcome};
use boatramp_core::{DeployError, StorageError};
use futures::StreamExt;
use serde::{Deserialize, Serialize};
mod admin_api;
pub mod sql_shim;
#[cfg(feature = "oidc")]
pub(crate) use admin_api::auth_exchange;
pub(crate) use admin_api::{
activate_deployment, cert_status, compute_dns, compute_dns_resolve, compute_exec, compute_ipam,
compute_netdiag, compute_reconcile, compute_restart, compute_set_health, compute_status,
create_deployment, current_deployment, delete_compute, delete_compute_volume,
delete_project_tenancy, delete_site, get_compute, get_daemon_config, get_deployment,
get_project_tenancy, get_site_config, invalidate_cache, list_aliases, list_compute,
list_compute_volumes, list_deployments, list_sites, prune_delete, prune_report, put_blob,
put_compute, put_daemon_config, put_project_tenancy, put_site_config, remove_alias,
rollback_daemon_config, scrub_blobs, set_alias, sql_exec, sql_ping, sql_query,
};
#[cfg(feature = "handlers")]
pub(crate) use admin_api::{
delete_graphql_safelist, delete_graphql_subgraph, get_graphql_supergraph,
list_graphql_safelist, put_graphql_function_subgraph, put_graphql_sql_subgraph,
put_graphql_subgraph, register_graphql_safelist,
};
#[cfg(feature = "admin")]
mod admin_controller;
mod auth;
#[cfg(feature = "console")]
pub mod console;
mod content;
mod control_api;
#[cfg(feature = "email")]
mod email_spool;
#[cfg(feature = "session")]
mod session_driver;
#[cfg(feature = "session")]
mod session_store;
#[cfg(feature = "admin")]
pub use admin_controller::ServerAdminController;
#[cfg(feature = "compression")]
pub(crate) use content::maybe_compress;
pub(crate) use content::multipart_byteranges;
pub(crate) use content::{
negotiate_encoding, parse_ranges, response_headers, set_content_encoding, MAX_RANGES,
};
pub(crate) use control_api::{
add_root_anchor, auth_whoami, bootstrap_token, cluster_join, cluster_members, cluster_promote,
cluster_revoke, cluster_rotate_key, create_join_token, create_token, delete_email_profile,
delete_secret, get_authz_policy, list_email_profiles, list_root_anchors, list_secrets,
list_tokens, put_authz_policy, remove_root_anchor, revoke_token, set_email_profile, set_secret,
show_email_profile,
};
#[cfg(all(test, feature = "handlers"))]
use control_api::{BootstrapRequest, CreateJoinTokenRequest, JoinRequest};
#[cfg(feature = "email")]
pub use email_spool::NodeEmailSpool;
mod domain_verify;
pub use domain_verify::{spawn_domain_verify_reconcile, verification_pending_page};
pub mod envelope;
#[cfg(feature = "handlers")]
mod graphql_apq;
#[cfg(feature = "handlers")]
mod graphql_cache;
#[cfg(feature = "handlers")]
mod graphql_data;
#[cfg(feature = "handlers")]
mod graphql_federation;
#[cfg(feature = "handlers")]
mod graphql_gateway;
#[cfg(feature = "handlers")]
mod graphql_graphiql;
#[cfg(feature = "handlers")]
mod graphql_guard;
#[cfg(feature = "handlers")]
mod graphql_plan;
#[cfg(feature = "handlers")]
mod graphql_registry;
#[cfg(feature = "handlers")]
mod graphql_subscription;
#[cfg(feature = "handlers")]
mod handler_cache;
#[cfg(feature = "handlers")]
mod handler_dispatch;
#[cfg(feature = "handlers")]
pub(crate) use handler_dispatch::{
build_bindings, dispatch_consumer_batch, dispatch_handler, precheck_component, read_blob_bytes,
read_blob_fully, resolve_secret_env,
};
#[cfg(all(feature = "handlers", test))]
use handler_dispatch::{resolve_env, set_forwarded_headers};
mod function_api;
pub(crate) use function_api::{
alias_function, deploy_function, list_functions, remove_function, rollback_function,
};
#[cfg(feature = "handlers")]
pub use function_api::{
component_requires, host_capability_features, host_capability_features_detailed, unmet_requires,
};
#[cfg(all(test, feature = "handlers"))]
use function_api::{AliasBody, DeployFunctionQuery, FunctionUpsert, RollbackBody};
pub use function_api::{CapabilityFeature, Lifecycle};
mod gateway;
mod host;
pub(crate) use host::{is_local_host, parse_deploy_host, strip_port};
#[cfg(feature = "http3")]
mod http3;
mod limits;
#[cfg(feature = "handlers")]
mod logs;
#[cfg(feature = "handlers")]
mod metrics;
#[cfg(feature = "oidc")]
mod oidc;
mod operator;
pub(crate) use operator::prometheus_metrics;
#[cfg(feature = "handlers")]
pub(crate) use operator::{
operator_dlq, operator_function_logs, operator_function_logs_stream, operator_handler_stats,
operator_logs, operator_logs_stream,
};
mod proxy;
pub use proxy::spawn_compute_reconcile;
pub(crate) use proxy::{
await_warm, compute_endpoint_regions, compute_endpoints, dispatch_gateway, has_parked_replica,
proxy, COMPUTE_WAKE_TIMEOUT,
};
mod splice;
mod http_serve;
pub use http_serve::{
alpn_h1_h2, serve_plaintext, serve_plaintext_listener, serve_router_conn, serve_tls,
serve_tls_listener, ReloadableTls, ServeInput,
};
#[cfg(feature = "handlers")]
pub(crate) use proxy::is_upgrade_request;
#[cfg(all(test, feature = "handlers"))]
use proxy::{gateway_addr_allowed, CLOUD_METADATA_IPV4};
mod project_api;
pub(crate) use project_api::{create_project, delete_project, get_project, list_projects};
mod project_scope;
pub(crate) use project_scope::{project_scope, OriginalPath, ProjectContext};
mod ratelimit;
mod routes;
pub use routes::{router, router_with, router_with_fast};
#[cfg(feature = "mcp")]
mod mcp_http;
#[cfg(feature = "handlers")]
mod scheduler;
mod serve_pipeline;
#[cfg(feature = "handlers")]
mod tenant_resolve;
pub use serve_pipeline::{http_redirect_router, FastServe};
#[cfg(test)]
mod hotpath_test;
#[cfg(all(test, feature = "handlers"))]
use serve_pipeline::{apply_vary, parse_cookie_header, parse_query_string};
pub(crate) use serve_pipeline::{
serve_bootstrap_identity, serve_by_host, serve_domain_challenge, serve_preview, serve_sites,
BootstrapAttestation,
};
pub mod signer;
mod srvmetrics;
#[cfg(all(feature = "handlers", test))]
use scheduler::run_scheduler_tick;
#[cfg(feature = "handlers")]
pub(crate) use scheduler::{
acquire_site_permit, effective_limits, handler_error_response, handler_unavailable, CronNow,
};
#[cfg(feature = "handlers")]
use scheduler::{CONSUMER_BATCH, CONSUMER_LEASE, CONSUMER_MAX_ATTEMPTS};
#[cfg(feature = "handlers")]
mod function_runtime;
#[cfg(feature = "handlers")]
pub(crate) use function_runtime::{
b64_decode, b64_encode, blob_storage_prefix, capture_response, delete_trigger_handler,
dispatch_function_triggers, drain_function_invocations, execute_function, get_function_usage,
get_invocation_record, invoke_function, list_triggers_handler, new_invocation_id,
put_trigger_handler, webhook_ingress,
};
#[cfg(feature = "session")]
mod session_serve;
#[cfg(feature = "handlers")]
mod stream;
#[cfg(feature = "handlers")]
mod workflow;
pub use auth::{require_auth, Auth};
#[cfg(feature = "http3")]
pub use http3::{
advertise_http3, http3_endpoint, quinn_server_config, serve_http3, serve_http3_endpoint,
Http3Error,
};
pub use limits::{ServerLimits, UploadGuard};
#[cfg(feature = "oidc")]
pub use oidc::{OidcConfig, OidcError, OidcVerifier};
use ratelimit::{KvRateLimiter, RateLimitStore, RateLimiter};
#[cfg(feature = "handlers")]
pub(crate) use stream::{route_matches, serve_stream, serve_ws_stream};
#[cfg(feature = "handlers")]
pub(crate) use workflow::{
define_workflow, delete_workflow_handler, drain_workflow_runs, get_workflow_handler,
get_workflow_run_handler, list_workflows_handler, start_workflow_run,
};
pub use srvmetrics::{server_metrics, ServerMetrics};
#[derive(Clone, Default)]
pub struct HandlerRuntime {
#[cfg(feature = "handlers")]
inner: Option<Arc<HandlerRuntimeInner>>,
}
#[cfg(feature = "handlers")]
struct HandlerRuntimeInner {
engine: boatramp_handlers::HandlerEngine,
async_drain_gate: Arc<tokio::sync::Semaphore>,
kv: Arc<dyn boatramp_core::kv::KvStore>,
storage: Arc<dyn boatramp_core::Storage>,
sql: Option<Arc<dyn boatramp_core::sql::SqlBackends>>,
messaging: Option<Arc<dyn boatramp_core::messaging::Messaging>>,
site_semaphores:
std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Semaphore>>>,
stream_semaphores:
std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Semaphore>>>,
stream_ip_counts: Arc<std::sync::Mutex<std::collections::HashMap<(String, IpAddr), u32>>>,
metrics: metrics::Metrics,
logs: Arc<logs::LogStore>,
#[cfg(feature = "handlers")]
graphql_cache: graphql_cache::GraphqlCache,
cron_leader_gate: std::sync::OnceLock<CronLeaderGate>,
max_blob_bytes: std::sync::OnceLock<u64>,
max_component_bytes: std::sync::OnceLock<u64>,
allow_env_secret_refs: std::sync::OnceLock<bool>,
require_tenancy_declaration: std::sync::OnceLock<bool>,
allow_cross_tenant_db: std::sync::OnceLock<bool>,
secret_store: std::sync::OnceLock<Arc<boatramp_core::secret_store::SecretStore>>,
function_meter_locks:
std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
function_semaphores:
std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Semaphore>>>,
watch_provider: std::sync::OnceLock<Arc<dyn boatramp_core::blob_provision::WatchProvider>>,
provision_tier: std::sync::OnceLock<boatramp_core::blob_notify::ProvisionTier>,
invoker: std::sync::OnceLock<Arc<function_runtime::FunctionInvoker>>,
federation_runner: std::sync::OnceLock<Arc<graphql_gateway::FederationRunner>>,
#[cfg(feature = "email")]
email_profile_store: std::sync::OnceLock<Arc<boatramp_core::email_config::EmailProfileStore>>,
#[cfg(feature = "email")]
email_spool: std::sync::OnceLock<Arc<dyn boatramp_handlers::EmailSpool>>,
#[cfg(feature = "admin")]
admin_controller: std::sync::OnceLock<Arc<admin_controller::ServerAdminController>>,
#[cfg(feature = "admin")]
admin_surfaces:
std::sync::OnceLock<std::collections::BTreeSet<boatramp_handlers::AdminSurface>>,
#[cfg(feature = "session")]
session_store: std::sync::OnceLock<session_store::SessionStore>,
session_signer: std::sync::OnceLock<Arc<dyn Signer>>,
#[cfg(feature = "capability")]
capability_max_ttl_secs: std::sync::OnceLock<u64>,
#[cfg(feature = "handlers")]
tenancy_posture_overrides: std::sync::OnceLock<
Arc<std::collections::BTreeMap<String, boatramp_core::security::ResolvedProjectTenancy>>,
>,
}
#[cfg(feature = "handlers")]
impl HandlerRuntimeInner {
pub(crate) fn project_tenancy_knobs(
&self,
project: &str,
) -> boatramp_core::security::ResolvedProjectTenancy {
if let Some(map) = self.tenancy_posture_overrides.get() {
if let Some(knobs) = map.get(project) {
return *knobs;
}
}
boatramp_core::security::ResolvedProjectTenancy {
require_tenancy_declaration: self
.require_tenancy_declaration
.get()
.copied()
.unwrap_or(true),
allow_cross_tenant_db: self.allow_cross_tenant_db.get().copied().unwrap_or(false),
#[cfg(feature = "capability")]
capability_max_ttl_secs: self.capability_max_ttl_secs.get().copied(),
#[cfg(not(feature = "capability"))]
capability_max_ttl_secs: None,
}
}
}
pub type CronLeaderGate = Arc<dyn Fn() -> bool + Send + Sync>;
impl HandlerRuntime {
pub fn disabled() -> Self {
Self::default()
}
#[cfg(feature = "handlers")]
pub fn new(
engine: boatramp_handlers::HandlerEngine,
kv: Arc<dyn boatramp_core::kv::KvStore>,
storage: Arc<dyn boatramp_core::Storage>,
sql: Option<Arc<dyn boatramp_core::sql::SqlBackends>>,
messaging: Option<Arc<dyn boatramp_core::messaging::Messaging>>,
) -> Self {
let async_drain_slots = engine.async_max_concurrency().max(1);
Self {
inner: Some(Arc::new(HandlerRuntimeInner {
engine,
async_drain_gate: Arc::new(tokio::sync::Semaphore::new(async_drain_slots)),
kv,
storage,
sql,
messaging,
site_semaphores: std::sync::Mutex::new(std::collections::HashMap::new()),
stream_semaphores: std::sync::Mutex::new(std::collections::HashMap::new()),
stream_ip_counts: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
metrics: metrics::Metrics::default(),
logs: Arc::new(logs::LogStore::default()),
#[cfg(feature = "handlers")]
graphql_cache: graphql_cache::GraphqlCache::default(),
cron_leader_gate: std::sync::OnceLock::new(),
max_blob_bytes: std::sync::OnceLock::new(),
max_component_bytes: std::sync::OnceLock::new(),
allow_env_secret_refs: std::sync::OnceLock::new(),
require_tenancy_declaration: std::sync::OnceLock::new(),
allow_cross_tenant_db: std::sync::OnceLock::new(),
secret_store: std::sync::OnceLock::new(),
function_meter_locks: std::sync::Mutex::new(std::collections::HashMap::new()),
function_semaphores: std::sync::Mutex::new(std::collections::HashMap::new()),
watch_provider: std::sync::OnceLock::new(),
provision_tier: std::sync::OnceLock::new(),
invoker: std::sync::OnceLock::new(),
federation_runner: std::sync::OnceLock::new(),
#[cfg(feature = "email")]
email_profile_store: std::sync::OnceLock::new(),
#[cfg(feature = "email")]
email_spool: std::sync::OnceLock::new(),
#[cfg(feature = "admin")]
admin_controller: std::sync::OnceLock::new(),
#[cfg(feature = "admin")]
admin_surfaces: std::sync::OnceLock::new(),
#[cfg(feature = "session")]
session_store: std::sync::OnceLock::new(),
session_signer: std::sync::OnceLock::new(),
#[cfg(feature = "capability")]
capability_max_ttl_secs: std::sync::OnceLock::new(),
#[cfg(feature = "handlers")]
tenancy_posture_overrides: std::sync::OnceLock::new(),
})),
}
}
#[cfg(feature = "handlers")]
pub(crate) fn sql_provider(&self) -> Option<Arc<dyn boatramp_core::sql::SqlBackends>> {
self.inner.as_ref().and_then(|inner| inner.sql.clone())
}
#[cfg(feature = "handlers")]
pub(crate) fn invoker(&self) -> Option<Arc<function_runtime::FunctionInvoker>> {
self.inner
.as_ref()
.and_then(|inner| inner.invoker.get().cloned())
}
#[cfg(feature = "handlers")]
pub(crate) async fn introspect_subgraph_sdl(
&self,
deploy: &DeployStore,
project: boatramp_core::project::ProjectRef<'_>,
function: &boatramp_core::function::Function,
component: &str,
) -> Result<String, function_runtime::SubgraphSdlError> {
match self.inner.as_ref() {
Some(inner) => {
function_runtime::introspect_service_sdl(
inner, deploy, project, function, component,
)
.await
}
None => Err(function_runtime::SubgraphSdlError::Unavailable),
}
}
#[cfg(feature = "handlers")]
pub fn set_invoker(&self, deploy: DeployStore) {
if let Some(inner) = self.inner.as_ref() {
let invoker = Arc::new(function_runtime::FunctionInvoker::new(
deploy,
Arc::downgrade(inner),
));
let _ = inner.invoker.set(invoker);
let runner = Arc::new(graphql_gateway::FederationRunner::new(Arc::downgrade(
inner,
)));
let _ = inner.federation_runner.set(runner);
}
}
#[cfg(feature = "handlers")]
pub fn sql_backends(&self) -> Option<Arc<dyn boatramp_core::sql::SqlBackends>> {
self.inner.as_ref().and_then(|inner| inner.sql.clone())
}
#[cfg(feature = "handlers")]
pub fn set_watch_provider(
&self,
provider: Arc<dyn boatramp_core::blob_provision::WatchProvider>,
) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.watch_provider.set(provider);
}
}
#[cfg(feature = "handlers")]
pub fn set_provision_tier(&self, tier: boatramp_core::blob_notify::ProvisionTier) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.provision_tier.set(tier);
}
}
#[cfg(feature = "handlers")]
pub fn set_max_blob_bytes(&self, max_bytes: u64) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.max_blob_bytes.set(max_bytes);
}
}
#[cfg(feature = "handlers")]
pub fn set_max_component_bytes(&self, max_bytes: u64) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.max_component_bytes.set(max_bytes);
}
}
#[cfg(feature = "handlers")]
pub fn set_allow_env_secret_refs(&self, allow: bool) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.allow_env_secret_refs.set(allow);
}
}
#[cfg(feature = "handlers")]
pub fn set_tenancy_posture(&self, require_declaration: bool, allow_cross_tenant: bool) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.require_tenancy_declaration.set(require_declaration);
let _ = inner.allow_cross_tenant_db.set(allow_cross_tenant);
}
}
#[cfg(feature = "handlers")]
pub fn set_project_tenancy_overrides(
&self,
overrides: std::collections::BTreeMap<
String,
boatramp_core::security::ResolvedProjectTenancy,
>,
) {
if overrides.is_empty() {
return;
}
if let Some(inner) = self.inner.as_ref() {
let _ = inner.tenancy_posture_overrides.set(Arc::new(overrides));
}
}
#[cfg(feature = "handlers")]
pub fn set_session_signer(&self, signer: Arc<dyn Signer>) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.session_signer.set(signer);
}
}
#[cfg(feature = "capability")]
pub fn set_capability_minting(&self, max_ttl_secs: u64) {
if max_ttl_secs == 0 {
return;
}
if let Some(inner) = self.inner.as_ref() {
let _ = inner.capability_max_ttl_secs.set(max_ttl_secs);
}
}
#[cfg(feature = "handlers")]
pub fn set_secret_store(&self, store: Arc<boatramp_core::secret_store::SecretStore>) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.secret_store.set(store);
}
}
#[cfg(feature = "email")]
pub fn set_email_profile_store(
&self,
store: Arc<boatramp_core::email_config::EmailProfileStore>,
) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.email_profile_store.set(store);
}
}
#[cfg(feature = "email")]
pub fn set_email_spool(&self, spool: Arc<dyn boatramp_handlers::EmailSpool>) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.email_spool.set(spool);
}
}
#[cfg(feature = "admin")]
pub fn set_admin(
&self,
controller: Arc<admin_controller::ServerAdminController>,
surfaces: std::collections::BTreeSet<boatramp_handlers::AdminSurface>,
) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.admin_controller.set(controller);
let _ = inner.admin_surfaces.set(surfaces);
}
}
#[cfg(feature = "handlers")]
pub(crate) fn allow_env_secret_refs(&self) -> bool {
self.inner
.as_ref()
.and_then(|inner| inner.allow_env_secret_refs.get().copied())
.unwrap_or(false)
}
#[cfg(feature = "handlers")]
pub fn set_cron_leader_gate(&self, gate: CronLeaderGate) {
if let Some(inner) = self.inner.as_ref() {
let _ = inner.cron_leader_gate.set(gate);
}
}
#[cfg(feature = "handlers")]
async fn precheck_activation(
&self,
deploy: &DeployStore,
manifest: &Manifest,
site_config: Option<&SiteConfig>,
) -> Result<(), String> {
let Some(inner) = self.inner.as_ref() else {
return Ok(());
};
if manifest.config.handlers.is_empty() && manifest.config.consumers.is_empty() {
return Ok(());
}
let site_handlers = site_config
.and_then(|c| c.handlers.as_ref())
.filter(|h| h.enabled)
.ok_or_else(|| {
"deployment ships handlers/consumers but the site has them disabled".to_string()
})?;
let max_component = inner.max_component_bytes.get().copied().unwrap_or(0);
let allow_env_secret_refs = inner.allow_env_secret_refs.get().copied().unwrap_or(false);
crate::handler_dispatch::admit_secret_refs(&site_handlers.secrets, allow_env_secret_refs)
.map_err(|err| format!("handler secrets: {err}"))?;
let sync_ceiling = inner.engine.sync_timeout_ms();
let async_ceiling = inner.engine.async_timeout_ms();
if let Some(ms) = site_handlers.max_timeout_ms {
if u64::from(ms) > sync_ceiling {
tracing::warn!(
"site max_timeout_ms={ms} exceeds sync_max_timeout_ms={sync_ceiling}: \
synchronous HTTP handlers are capped at {sync_ceiling}ms; the extra time \
applies only to async calls (?mode=async / triggers), capped at \
async_max_timeout_ms={async_ceiling}"
);
}
}
for handler in &manifest.config.handlers {
if let Some(ms) = handler.limits.as_ref().and_then(|l| l.timeout_ms) {
if u64::from(ms) > sync_ceiling {
let route = &handler.route;
tracing::warn!(
"route {route:?} declares limits.timeout_ms={ms}, above \
sync_max_timeout_ms={sync_ceiling}: synchronous HTTP calls to this route \
are capped at {sync_ceiling}ms; the {ms}ms only applies to async calls \
(?mode=async / a queue trigger / a #[consumer]), capped at \
async_max_timeout_ms={async_ceiling}. If you need {ms}ms synchronously, \
that isn't possible — move the work to the async lane"
);
}
}
if !handler.streaming {
if let Some(entry) = manifest.files.get(&handler.component) {
if let Ok(bytes) = read_blob_bytes(deploy, &entry.hash).await {
if crate::function_api::component_declares_streaming_route(
&bytes,
&handler.route,
) {
let route = &handler.route;
tracing::warn!(
"route {route:?} is a streaming handler (#[handler(stream)]) but \
its config lacks streaming = true: it will run on the sync request \
lane and be cut at sync_max_timeout_ms={sync_ceiling}ms. Set \
streaming = true so it serves on the dedicated streaming lane (its \
own concurrency budget + a much larger wall-clock)."
);
}
}
}
}
if let Some(entry) = manifest.files.get(&handler.component) {
if let Ok(bytes) = read_blob_bytes(deploy, &entry.hash).await {
let unmet = crate::function_api::unmet_requires(&bytes);
if !unmet.is_empty() {
return Err(format!(
"route {:?} [{}] requires capabilities this host does not implement: \
{}. Upgrade boatramp or enable those features — see `boatramp \
capabilities`.",
handler.route,
handler.methods.join(","),
unmet.join(", ")
));
}
}
}
precheck_component(
deploy,
manifest,
site_handlers,
inner,
max_component,
&handler.imports,
&handler.component,
&format!("route {:?} [{}]", handler.route, handler.methods.join(",")),
false,
)
.await?;
}
for consumer in &manifest.config.consumers {
precheck_component(
deploy,
manifest,
site_handlers,
inner,
max_component,
&consumer.imports,
&consumer.component,
&format!("consumer {:?}", consumer.topic),
true,
)
.await?;
}
Ok(())
}
#[cfg(not(feature = "handlers"))]
async fn precheck_activation(
&self,
_deploy: &DeployStore,
_manifest: &Manifest,
_site_config: Option<&SiteConfig>,
) -> Result<(), String> {
Ok(())
}
}
#[derive(Default, Clone)]
pub struct ServerOptions {
pub limits: ServerLimits,
pub probe: Option<Arc<dyn boatramp_core::domain_verify::DomainProbe>>,
pub default_site: Option<String>,
pub implicit_routing: bool,
pub protect_previews: bool,
pub cluster_rate_limit_kv: Option<Arc<dyn boatramp_core::kv::KvStore>>,
pub issuer: Option<Arc<dyn Signer>>,
pub bootstrap_secret: Option<String>,
pub bootstrap_attestation: Option<String>,
pub mesh_control: Option<Arc<dyn MeshControl>>,
pub cors_allowed_origins: Vec<String>,
#[cfg(feature = "oidc")]
pub oidc_verifier: Option<Arc<oidc::OidcVerifier>>,
pub posture: boatramp_core::security::SecurityPosture,
pub served_over_tls: bool,
pub pop_origin: Option<String>,
pub daemon_runtime: Option<Arc<DaemonRuntime>>,
pub operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>>,
pub tenant_deprovisioner: Option<Arc<dyn boatramp_core::sql::TenantDeprovisioner>>,
pub compute_exec: Option<Arc<dyn boatramp_core::compute::ComputeExec>>,
pub compute_volumes: Option<Arc<dyn boatramp_core::compute::ComputeVolumes>>,
pub compute_control: Option<Arc<dyn boatramp_core::compute::ComputeControl>>,
pub secret_store: Option<Arc<boatramp_core::secret_store::SecretStore>>,
pub email_profile_store: Option<Arc<boatramp_core::email_config::EmailProfileStore>>,
#[cfg(feature = "console")]
pub console: Option<console::ConsoleMount>,
}
#[derive(Clone, Copy)]
struct ServedOverTls(bool);
#[derive(Clone, Copy, Default)]
struct ImplicitRouting(bool);
const DAEMON_RELOAD_BACKSTOP: std::time::Duration = std::time::Duration::from_secs(300);
pub struct DaemonRuntime {
baseline: boatramp_core::daemon_config::ConfigBaseline,
state: std::sync::RwLock<DaemonState>,
reload: tokio::sync::Notify,
}
struct DaemonState {
effective: Arc<boatramp_core::daemon_config::EffectiveConfig>,
generation: Option<String>,
}
pub fn config_baseline(options: &ServerOptions) -> boatramp_core::daemon_config::ConfigBaseline {
#[cfg(feature = "console")]
let (console_enabled, console_host, console_path) = match options.console.as_ref() {
Some(m) => (true, Some(m.host.clone()), Some(m.path.clone())),
None => (false, None, None),
};
#[cfg(not(feature = "console"))]
let (console_enabled, console_host, console_path) = (false, None, None);
boatramp_core::daemon_config::ConfigBaseline {
default_site: options.default_site.clone(),
protect_previews: options.protect_previews,
max_upload_bytes: options.limits.max_upload_bytes.unwrap_or(0),
upload_idle_timeout_secs: options.limits.upload_idle_timeout.map(|d| d.as_secs()),
max_concurrent_uploads: options.limits.max_concurrent_uploads.map(|n| n as u64),
cluster_rate_limit: options.cluster_rate_limit_kv.is_some(),
compute_vcpus: 0,
compute_mem_mib: 0,
console_enabled,
console_host,
console_path,
max_upload_ceiling: options.posture.max_upload_bytes,
max_concurrent_uploads_ceiling: None,
posture: options.posture,
}
}
impl DaemonRuntime {
pub fn new(baseline: boatramp_core::daemon_config::ConfigBaseline) -> Self {
let effective =
Arc::new(boatramp_core::daemon_config::DaemonConfig::default().resolve(&baseline));
Self {
baseline,
state: std::sync::RwLock::new(DaemonState {
effective,
generation: None,
}),
reload: tokio::sync::Notify::new(),
}
}
pub fn notify_reload(&self) {
self.reload.notify_one();
}
pub fn effective(&self) -> Arc<boatramp_core::daemon_config::EffectiveConfig> {
self.state
.read()
.expect("daemon config lock")
.effective
.clone()
}
pub fn generation(&self) -> Option<String> {
self.state
.read()
.expect("daemon config lock")
.generation
.clone()
}
pub fn baseline(&self) -> &boatramp_core::daemon_config::ConfigBaseline {
&self.baseline
}
pub async fn reload(&self, deploy: &DeployStore) -> Result<(), DeployError> {
let cfg = deploy.get_daemon_config().await?.unwrap_or_default();
let generation = deploy.daemon_config_generation().await?;
let effective = Arc::new(cfg.resolve(&self.baseline));
*self.state.write().expect("daemon config lock") = DaemonState {
effective,
generation,
};
Ok(())
}
}
#[derive(Clone, Copy, Default)]
struct PreviewPolicy {
protect: bool,
}
#[derive(Clone, Default)]
struct Issuer(Option<Arc<dyn Signer>>);
#[derive(Clone, Default)]
struct BootstrapGate(Option<Arc<BootstrapInner>>);
struct BootstrapInner {
secret_hash: String,
lock: tokio::sync::Mutex<()>,
}
impl BootstrapGate {
fn new(secret: Option<&str>) -> Self {
Self(secret.filter(|s| !s.is_empty()).map(|s| {
Arc::new(BootstrapInner {
secret_hash: boatramp_core::deploy::sha256_hex(s.as_bytes()),
lock: tokio::sync::Mutex::new(()),
})
}))
}
}
#[async_trait::async_trait]
pub trait MeshControl: Send + Sync {
async fn admit(
&self,
mesh_pubkey_hex: &str,
jti: &str,
possession_proof: &[u8],
proof_iat: u64,
now: u64,
advertise_addr: Option<&str>,
) -> Result<JoinOutcome, String>;
async fn rotate_key(&self) -> Result<String, String>;
async fn revoke(&self, node: u64) -> Result<(), String>;
async fn members(&self) -> Result<Vec<MeshMember>, String>;
async fn promote(&self, node: u64) -> Result<(), String>;
}
pub enum JoinOutcome {
Admitted {
members: Vec<String>,
addrs: std::collections::BTreeMap<u64, String>,
},
TokenSpent,
ProofInvalid,
Revoked,
}
#[derive(Debug, Clone, Serialize)]
pub struct MeshMember {
pub node: u64,
pub voter: bool,
pub caught_up: bool,
pub leader: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub addr: Option<String>,
}
#[derive(Clone, Default)]
struct MeshControlHandle(Option<Arc<dyn MeshControl>>);
#[cfg(feature = "oidc")]
#[derive(Clone, Default)]
struct OidcState(Option<Arc<oidc::OidcVerifier>>);
#[cfg(feature = "oidc")]
const EXCHANGE_TTL_SECS: u64 = 3600;
use boatramp_core::time::now_unix;
#[derive(Clone)]
struct CorsState(Arc<Vec<String>>);
const CORS_ALLOW_METHODS: &str = "GET, POST, PUT, DELETE, OPTIONS";
const CORS_ALLOW_HEADERS: &str = "authorization, content-type";
const CORS_MAX_AGE: &str = "600";
fn cors_origin_allowed(allowed: &[String], origin: &str) -> bool {
allowed.iter().any(|a| a == "*" || a == origin)
}
async fn cors(
State(allowed): State<CorsState>,
request: Request,
next: axum::middleware::Next,
) -> Response {
let origin = request
.headers()
.get(header::ORIGIN)
.and_then(|v| v.to_str().ok())
.filter(|o| cors_origin_allowed(&allowed.0, o))
.map(str::to_string);
let is_preflight = request.method() == Method::OPTIONS
&& request
.headers()
.contains_key(header::ACCESS_CONTROL_REQUEST_METHOD);
if is_preflight {
let allow_headers = request
.headers()
.get(header::ACCESS_CONTROL_REQUEST_HEADERS)
.and_then(|v| v.to_str().ok())
.map(str::to_string)
.unwrap_or_else(|| CORS_ALLOW_HEADERS.to_string());
let mut response = Response::new(Body::empty());
*response.status_mut() = StatusCode::NO_CONTENT;
if let Some(origin) = origin {
let headers = response.headers_mut();
set_header(headers, header::ACCESS_CONTROL_ALLOW_ORIGIN, &origin);
set_header(headers, header::VARY, "Origin");
set_header(
headers,
header::ACCESS_CONTROL_ALLOW_METHODS,
CORS_ALLOW_METHODS,
);
set_header(
headers,
header::ACCESS_CONTROL_ALLOW_HEADERS,
&allow_headers,
);
set_header(headers, header::ACCESS_CONTROL_MAX_AGE, CORS_MAX_AGE);
}
return response;
}
let mut response = next.run(request).await;
if let Some(origin) = origin {
let headers = response.headers_mut();
set_header(headers, header::ACCESS_CONTROL_ALLOW_ORIGIN, &origin);
if let Ok(value) = HeaderValue::from_str("Origin") {
headers.append(header::VARY, value);
}
}
response
}
const DRAIN_DEADLINE: Duration = Duration::from_secs(30);
#[derive(Debug, thiserror::Error)]
pub enum ServeError {
#[error("server I/O: {0}")]
Io(#[from] std::io::Error),
}
pub async fn serve(
addr: SocketAddr,
deploy: DeployStore,
auth: Auth,
handlers: HandlerRuntime,
) -> Result<(), ServeError> {
serve_with(addr, deploy, auth, handlers, ServerOptions::default()).await
}
pub(crate) fn disable_nagle(stream: &mut tokio::net::TcpStream) {
if let Err(err) = stream.set_nodelay(true) {
tracing::debug!(%err, "failed to set TCP_NODELAY on an accepted connection");
}
}
pub async fn serve_with(
addr: SocketAddr,
deploy: DeployStore,
auth: Auth,
handlers: HandlerRuntime,
options: ServerOptions,
) -> Result<(), ServeError> {
let tcp = tokio::net::TcpListener::bind(addr).await?;
tracing::info!(%addr, auth = !auth.is_disabled(), "boatramp server listening");
let splice_ctx = splice::SpliceCtx {
deploy: deploy.clone(),
posture: options.posture,
daemon: options.daemon_runtime.clone(),
};
#[cfg(feature = "handlers")]
let scheduler = handlers.spawn_scheduler(deploy.clone());
let gateway_prober = gateway::spawn_active_health_prober();
let (router, fast) = router_with_fast(deploy, auth, handlers, options);
let (signalled_tx, signalled_rx) = tokio::sync::watch::channel(false);
let server = splice::serve(tcp, splice_ctx, (router, fast), async move {
shutdown_signal().await;
let _ = signalled_tx.send(true);
});
let signalled = {
let mut rx = signalled_rx;
async move {
let _ = rx.wait_for(|fired| *fired).await;
}
};
let result = serve_with_drain_deadline(
async move { server.await.map_err(ServeError::from) },
signalled,
DRAIN_DEADLINE,
)
.await;
#[cfg(feature = "handlers")]
if let Some(handle) = scheduler {
handle.abort();
}
gateway_prober.abort();
result
}
async fn serve_with_drain_deadline<Srv, Sig>(
server: Srv,
signalled: Sig,
deadline: Duration,
) -> Result<(), ServeError>
where
Srv: Future<Output = Result<(), ServeError>>,
Sig: Future<Output = ()>,
{
tokio::pin!(server);
let drain_cap = async move {
signalled.await;
tokio::time::sleep(deadline).await;
};
tokio::select! {
result = &mut server => result,
_ = drain_cap => {
tracing::warn!(
deadline_s = deadline.as_secs(),
"drain deadline exceeded; forcing shutdown with requests still in flight"
);
Ok(())
}
}
}
pub async fn shutdown_signal() {
let ctrl_c = async {
let _ = tokio::signal::ctrl_c().await;
};
#[cfg(unix)]
let terminate = async {
if let Ok(mut sig) =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
{
sig.recv().await;
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {}
_ = terminate => {}
}
tracing::info!("shutdown signal received; draining");
}
async fn healthz(Extension(daemon): Extension<Arc<DaemonRuntime>>) -> String {
match daemon.generation() {
Some(gen) => format!("ok gen={gen}"),
None => "ok".to_string(),
}
}
async fn readyz(State(deploy): State<DeployStore>) -> Response {
match deploy.ready().await {
Ok(()) => (StatusCode::OK, "ready\n").into_response(),
Err(err) => {
tracing::warn!(error = %err, "readiness probe failed");
(StatusCode::SERVICE_UNAVAILABLE, "not ready\n").into_response()
}
}
}
#[derive(Clone)]
pub struct RequestId(pub String);
#[cfg(feature = "handlers")]
#[derive(Clone)]
pub struct DomainContext(pub String);
fn request_id_for(headers: &HeaderMap) -> String {
if let Some(id) = headers
.get("x-request-id")
.and_then(|v| v.to_str().ok())
.map(str::trim)
.filter(|s| !s.is_empty())
{
return id.chars().filter(|c| !c.is_control()).take(128).collect();
}
use std::sync::atomic::{AtomicU64, Ordering};
static SEQ: AtomicU64 = AtomicU64::new(0);
let n = SEQ.fetch_add(1, Ordering::Relaxed);
format!("{:x}-{:x}", boatramp_core::time::now_unix_ms(), n)
}
struct AccessLog {
request_id: String,
method: Method,
path: String,
host: String,
client: String,
status: u16,
encoding: String,
start: std::time::Instant,
bytes: std::sync::atomic::AtomicU64,
}
impl Drop for AccessLog {
fn drop(&mut self) {
let bytes = self.bytes.load(std::sync::atomic::Ordering::Relaxed);
srvmetrics::server_metrics().record_request(self.status, bytes);
tracing::info!(
target: "boatramp::access",
request_id = %self.request_id,
method = %self.method,
path = %self.path,
host = %self.host,
client = %self.client,
status = self.status,
bytes = bytes,
encoding = %self.encoding,
cache_result = srvmetrics::cache_result(self.status),
elapsed_ms = self.start.elapsed().as_millis() as u64,
"request"
);
}
}
pub(crate) fn assign_request_id(request: &mut axum::extract::Request) -> String {
let request_id = request_id_for(request.headers());
request
.extensions_mut()
.insert(RequestId(request_id.clone()));
request_id
}
pub(crate) struct AccessLogCtx {
request_id: String,
method: Method,
path: String,
host: String,
client: String,
start: std::time::Instant,
}
impl AccessLogCtx {
pub(crate) fn capture(request: &axum::extract::Request, request_id: String) -> Option<Self> {
if !tracing::enabled!(target: "boatramp::access", tracing::Level::INFO) {
return None;
}
Some(Self {
request_id,
method: request.method().clone(),
path: request.uri().path().to_string(),
host: request
.headers()
.get(header::HOST)
.and_then(|value| value.to_str().ok())
.or_else(|| request.uri().host()) .unwrap_or("-")
.to_string(),
client: request
.extensions()
.get::<axum::extract::ConnectInfo<SocketAddr>>()
.map(|info| info.0.ip().to_string())
.unwrap_or_else(|| "-".to_string()),
start: std::time::Instant::now(),
})
}
pub(crate) fn finish(self, response: Response) -> Response {
let encoding = response
.headers()
.get(header::CONTENT_ENCODING)
.and_then(|v| v.to_str().ok())
.unwrap_or("identity")
.to_string();
let log = AccessLog {
request_id: self.request_id,
method: self.method,
path: self.path,
host: self.host,
client: self.client,
status: response.status().as_u16(),
encoding,
start: self.start,
bytes: std::sync::atomic::AtomicU64::new(0),
};
let (parts, body) = response.into_parts();
let counted = body.into_data_stream().map(move |chunk| {
if let Ok(bytes) = &chunk {
log.bytes
.fetch_add(bytes.len() as u64, std::sync::atomic::Ordering::Relaxed);
}
chunk
});
Response::from_parts(parts, Body::from_stream(counted))
}
}
async fn access_log(mut request: axum::extract::Request, next: axum::middleware::Next) -> Response {
let request_id = assign_request_id(&mut request);
match AccessLogCtx::capture(&request, request_id) {
None => next.run(request).await,
Some(ctx) => ctx.finish(next.run(request).await),
}
}
fn if_none_match(req_headers: &HeaderMap, etag: &str) -> bool {
req_headers
.get(header::IF_NONE_MATCH)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| {
value
.split(',')
.map(str::trim)
.any(|tag| tag == "*" || tag == etag || tag.trim_start_matches("W/") == etag)
})
}
fn set_header(headers: &mut HeaderMap, name: header::HeaderName, value: &str) {
if let Ok(value) = HeaderValue::from_str(value) {
headers.insert(name, value);
}
}
fn not_found() -> Response {
(StatusCode::NOT_FOUND, "not found\n").into_response()
}
fn redirect(status: u16, location: &str) -> Response {
let status = StatusCode::from_u16(status).unwrap_or(StatusCode::FOUND);
match HeaderValue::from_str(location) {
Ok(location) => {
let mut headers = HeaderMap::new();
headers.insert(header::LOCATION, location);
(status, headers).into_response()
}
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "bad redirect target\n").into_response(),
}
}
fn deploy_error_response(err: DeployError) -> Response {
let status = match &err {
DeployError::NotFound(_) | DeployError::Storage(StorageError::NotFound(_)) => {
StatusCode::NOT_FOUND
}
DeployError::HashMismatch { .. } => StatusCode::BAD_REQUEST,
DeployError::Incomplete(_) => StatusCode::CONFLICT,
DeployError::Conflict(_) => StatusCode::CONFLICT,
DeployError::Ambiguous(_) => StatusCode::NOT_FOUND,
DeployError::Invalid(_) => StatusCode::BAD_REQUEST,
_ => StatusCode::INTERNAL_SERVER_ERROR,
};
tracing::warn!(error = %err, "request failed");
(status, format!("{err}\n")).into_response()
}
fn reject_invalid_name(kind: &'static str, value: &str) -> Option<Response> {
boatramp_core::project::validate_resource_name(kind, value)
.err()
.map(|err| (StatusCode::UNPROCESSABLE_ENTITY, format!("{err}\n")).into_response())
}
#[cfg(test)]
mod drain_tests {
use super::*;
#[tokio::test]
async fn deadline_forces_shutdown_after_signal() {
let server = std::future::pending::<Result<(), ServeError>>();
let signalled = async {}; let result = serve_with_drain_deadline(server, signalled, Duration::from_millis(20)).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn server_finishing_first_wins() {
let server = async { Ok(()) };
let signalled = std::future::pending::<()>();
let result = serve_with_drain_deadline(server, signalled, Duration::from_secs(30)).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn deadline_does_not_trip_before_signal() {
let server = async {
tokio::time::sleep(Duration::from_millis(40)).await;
Err(ServeError::Io(std::io::Error::other("server error")))
};
let signalled = std::future::pending::<()>();
let result = serve_with_drain_deadline(server, signalled, Duration::from_millis(10)).await;
assert!(result.is_err());
}
}
#[cfg(all(test, feature = "handlers"))]
mod tests {
use super::*;
use boatramp_core::cose::{LocalSigner, TokenAlg};
use boatramp_core::project::ProjectRef;
#[test]
fn query_string_parses_and_url_decodes() {
let q = parse_query_string("lang=fr&city=S%C3%A3o+Paulo&flag&dup=1&dup=2");
assert_eq!(q.get("lang").map(String::as_str), Some("fr"));
assert_eq!(q.get("city").map(String::as_str), Some("São Paulo")); assert_eq!(q.get("flag").map(String::as_str), Some("")); assert_eq!(q.get("dup").map(String::as_str), Some("1")); }
#[test]
fn cookie_header_parses_pairs() {
let c = parse_cookie_header("beta=1; sid = abc ; empty=");
assert_eq!(c.get("beta").map(String::as_str), Some("1"));
assert_eq!(c.get("sid").map(String::as_str), Some("abc"));
assert_eq!(c.get("empty").map(String::as_str), Some(""));
}
#[test]
fn apply_vary_merges_without_duplicates() {
let base = (StatusCode::OK, "x").into_response();
let r = apply_vary(base, &["accept-language".into()]);
assert_eq!(r.headers().get(header::VARY).unwrap(), "accept-language");
let r = apply_vary(r, &["cookie".into(), "accept-language".into()]);
let v = r.headers().get(header::VARY).unwrap().to_str().unwrap();
assert!(v.contains("accept-language") && v.contains("cookie"));
assert_eq!(v.matches("accept-language").count(), 1);
let plain = apply_vary((StatusCode::OK, "y").into_response(), &[]);
assert!(plain.headers().get(header::VARY).is_none());
}
#[tokio::test]
async fn join_token_endpoint_mints_a_verifiable_bearer_token() {
let keys: Arc<dyn Signer> = Arc::new(LocalSigner::generate(TokenAlg::Es256));
let public = keys.public_key();
let resp = create_join_token(
Extension(Issuer(Some(keys.clone()))),
Json(CreateJoinTokenRequest {
ttl_secs: Some(600),
}),
)
.await;
assert_eq!(resp.status(), StatusCode::CREATED);
let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let parsed: serde_json::Value = serde_json::from_slice(&body).unwrap();
let token = parsed["token"].as_str().unwrap();
let jti = cose::verify_join(token, &public, now_unix()).unwrap();
assert!(!jti.is_empty());
let no_issuer = create_join_token(
Extension(Issuer(None)),
Json(CreateJoinTokenRequest { ttl_secs: None }),
)
.await;
assert_eq!(no_issuer.status(), StatusCode::NOT_IMPLEMENTED);
}
#[tokio::test]
async fn function_write_path_deploy_rollback_alias_remove() {
use boatramp_core::function::Lifecycle;
use boatramp_core::kv::MemoryKv;
use boatramp_core::{ByteStream, GetObject, ObjectMeta, PutMeta, Storage, StorageError};
struct FakeStorage {
present: bool,
}
#[async_trait::async_trait]
impl Storage for FakeStorage {
async fn get(&self, _: &str) -> Result<GetObject, StorageError> {
Err(StorageError::NotFound(String::new()))
}
async fn get_range(
&self,
_: &str,
_: u64,
_: Option<u64>,
) -> Result<GetObject, StorageError> {
Err(StorageError::NotFound(String::new()))
}
async fn put(
&self,
_: &str,
_: ByteStream,
_: PutMeta,
) -> Result<ObjectMeta, StorageError> {
Err(StorageError::unsupported("fake"))
}
async fn head(&self, key: &str) -> Result<ObjectMeta, StorageError> {
if self.present {
Ok(ObjectMeta {
key: key.to_string(),
..Default::default()
})
} else {
Err(StorageError::NotFound(key.to_string()))
}
}
async fn delete(&self, _: &str) -> Result<(), StorageError> {
Ok(())
}
async fn list(&self, _: &str) -> Result<Vec<ObjectMeta>, StorageError> {
Ok(Vec::new())
}
}
async fn body_json(resp: Response) -> (StatusCode, serde_json::Value) {
let status = resp.status();
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let value = if bytes.is_empty() {
serde_json::Value::Null
} else {
serde_json::from_slice(&bytes).unwrap()
};
(status, value)
}
let deploy = DeployStore::new(
Arc::new(FakeStorage { present: true }),
Arc::new(MemoryKv::new()),
);
let v1 = "a".repeat(64);
let v2 = "b".repeat(64);
let (st, body) = body_json(
deploy_function(
State(deploy.clone()),
axum::extract::Extension(crate::ProjectContext::default()),
axum::extract::Extension(Arc::new(crate::HandlerRuntime::disabled())),
axum::extract::Query(DeployFunctionQuery::default()),
Path("greeter".to_string()),
Json(FunctionUpsert {
component: v1.clone(),
config: Default::default(),
lifecycle: Lifecycle::Independent,
}),
)
.await,
)
.await;
assert_eq!(st, StatusCode::OK);
assert_eq!(body["active"], v1);
let (_, body) = body_json(
deploy_function(
State(deploy.clone()),
axum::extract::Extension(crate::ProjectContext::default()),
axum::extract::Extension(Arc::new(crate::HandlerRuntime::disabled())),
axum::extract::Query(DeployFunctionQuery::default()),
Path("greeter".to_string()),
Json(FunctionUpsert {
component: v2.clone(),
config: Default::default(),
lifecycle: Lifecycle::Independent,
}),
)
.await,
)
.await;
assert_eq!(body["active"], v2);
assert_eq!(body["versions"].as_array().unwrap().len(), 2);
let (st, body) = body_json(
rollback_function(
State(deploy.clone()),
axum::extract::Extension(crate::ProjectContext::default()),
Path("greeter".to_string()),
Json(RollbackBody { to: v1.clone() }),
)
.await,
)
.await;
assert_eq!(st, StatusCode::OK);
assert_eq!(body["active"], v1);
let resp = rollback_function(
State(deploy.clone()),
axum::extract::Extension(crate::ProjectContext::default()),
Path("greeter".to_string()),
Json(RollbackBody { to: "c".repeat(64) }),
)
.await;
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
let (st, body) = body_json(
alias_function(
State(deploy.clone()),
axum::extract::Extension(crate::ProjectContext::default()),
Path(("greeter".to_string(), "prod".to_string())),
Json(AliasBody {
version: v2.clone(),
}),
)
.await,
)
.await;
assert_eq!(st, StatusCode::OK);
assert_eq!(body["aliases"]["prod"], v2);
let (st, _) = body_json(
remove_function(
State(deploy.clone()),
axum::extract::Extension(crate::ProjectContext::default()),
Path("greeter".to_string()),
)
.await,
)
.await;
assert_eq!(st, StatusCode::NO_CONTENT);
assert!(deploy
.get_function(ProjectRef::DEFAULT, "greeter")
.await
.unwrap()
.is_none());
let empty = DeployStore::new(
Arc::new(FakeStorage { present: false }),
Arc::new(MemoryKv::new()),
);
let resp = deploy_function(
State(empty),
axum::extract::Extension(crate::ProjectContext::default()),
axum::extract::Extension(Arc::new(crate::HandlerRuntime::disabled())),
axum::extract::Query(DeployFunctionQuery::default()),
Path("orphan".to_string()),
Json(FunctionUpsert {
component: v1.clone(),
config: Default::default(),
lifecycle: Lifecycle::default(),
}),
)
.await;
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
}
struct StubControl {
admits: std::sync::Mutex<Vec<(String, String)>>,
respond: StubJoin,
}
#[derive(Clone, Copy)]
enum StubJoin {
Admit,
Spent,
Invalid,
Revoked,
}
#[async_trait::async_trait]
impl MeshControl for StubControl {
async fn admit(
&self,
mesh_pubkey_hex: &str,
jti: &str,
_proof: &[u8],
_proof_iat: u64,
_now: u64,
_advertise_addr: Option<&str>,
) -> Result<JoinOutcome, String> {
self.admits
.lock()
.unwrap()
.push((mesh_pubkey_hex.to_string(), jti.to_string()));
Ok(match self.respond {
StubJoin::Admit => JoinOutcome::Admitted {
members: vec!["signed-member".to_string()],
addrs: std::collections::BTreeMap::from([(7u64, "https://x:7000".to_string())]),
},
StubJoin::Spent => JoinOutcome::TokenSpent,
StubJoin::Invalid => JoinOutcome::ProofInvalid,
StubJoin::Revoked => JoinOutcome::Revoked,
})
}
async fn rotate_key(&self) -> Result<String, String> {
Ok("cafe".to_string())
}
async fn revoke(&self, _node: u64) -> Result<(), String> {
Ok(())
}
async fn members(&self) -> Result<Vec<MeshMember>, String> {
Ok(Vec::new())
}
async fn promote(&self, _node: u64) -> Result<(), String> {
Ok(())
}
}
#[tokio::test]
async fn cluster_join_dispatches_and_maps_outcomes() {
let keys: Arc<dyn Signer> = Arc::new(LocalSigner::generate(TokenAlg::Es256));
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let auth = Auth::with_key(keys.public_key(), kv);
let token = cose::mint_join(600, now_unix(), &*keys).await.unwrap();
let req = |proof: &str| JoinRequest {
token: token.clone(),
mesh_pubkey: "302a300506032b6570032100feed".into(),
possession_proof: proof.to_string(),
proof_iat: now_unix(),
advertise_addr: Some("https://joiner:7000".into()),
};
let admitter = Arc::new(StubControl {
admits: std::sync::Mutex::new(Vec::new()),
respond: StubJoin::Admit,
});
let resp = cluster_join(
Extension(auth.clone()),
Extension(MeshControlHandle(Some(admitter.clone()))),
Json(req("aa01")),
)
.await;
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(admitter.admits.lock().unwrap().len(), 1);
let spent = Arc::new(StubControl {
admits: std::sync::Mutex::new(Vec::new()),
respond: StubJoin::Spent,
});
assert_eq!(
cluster_join(
Extension(auth.clone()),
Extension(MeshControlHandle(Some(spent))),
Json(req("aa01")),
)
.await
.status(),
StatusCode::CONFLICT
);
let invalid = Arc::new(StubControl {
admits: std::sync::Mutex::new(Vec::new()),
respond: StubJoin::Invalid,
});
assert_eq!(
cluster_join(
Extension(auth.clone()),
Extension(MeshControlHandle(Some(invalid))),
Json(req("aa01")),
)
.await
.status(),
StatusCode::FORBIDDEN
);
let revoked = Arc::new(StubControl {
admits: std::sync::Mutex::new(Vec::new()),
respond: StubJoin::Revoked,
});
assert_eq!(
cluster_join(
Extension(auth.clone()),
Extension(MeshControlHandle(Some(revoked))),
Json(req("aa01")),
)
.await
.status(),
StatusCode::FORBIDDEN
);
let ok = Arc::new(StubControl {
admits: std::sync::Mutex::new(Vec::new()),
respond: StubJoin::Admit,
});
assert_eq!(
cluster_join(
Extension(auth.clone()),
Extension(MeshControlHandle(Some(ok))),
Json(req("not-hex")),
)
.await
.status(),
StatusCode::BAD_REQUEST
);
let none = cluster_join(
Extension(auth),
Extension(MeshControlHandle(None)),
Json(req("aa01")),
)
.await;
assert_eq!(none.status(), StatusCode::NOT_IMPLEMENTED);
}
#[tokio::test]
async fn bootstrap_mints_the_first_token_once() {
use axum::http::{header::AUTHORIZATION, HeaderMap, HeaderValue};
let keys: Arc<dyn Signer> = Arc::new(LocalSigner::generate(TokenAlg::Es256));
let public = keys.public_key();
let deploy = DeployStore::new(
Arc::new(MemStorage::default()),
Arc::new(MemoryKv::new()) as Arc<dyn KvStore>,
);
let secret = "s3cr3t-bootstrap-value";
let gate = BootstrapGate::new(Some(secret));
let issuer = Issuer(Some(keys.clone()));
let bearer = |s: &str| {
let mut h = HeaderMap::new();
h.insert(
AUTHORIZATION,
HeaderValue::from_str(&format!("Bearer {s}")).unwrap(),
);
h
};
let req = || BootstrapRequest {
roles: vec!["admin".to_string()],
ttl_secs: None,
};
let bad = bootstrap_token(
State(deploy.clone()),
Extension(issuer.clone()),
Extension(gate.clone()),
bearer("wrong"),
Json(req()),
)
.await;
assert_eq!(bad.status(), StatusCode::UNAUTHORIZED);
let ok = bootstrap_token(
State(deploy.clone()),
Extension(issuer.clone()),
Extension(gate.clone()),
bearer(secret),
Json(req()),
)
.await;
assert_eq!(ok.status(), StatusCode::CREATED);
let body = axum::body::to_bytes(ok.into_body(), usize::MAX)
.await
.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
let token = json["token"].as_str().unwrap();
let id = json["id"].as_str().unwrap();
let verified = cose::verify(token, &public, now_unix()).unwrap();
assert!(verified.roles.iter().any(|r| r.name == "admin"));
assert!(deploy
.list_token_meta()
.await
.unwrap()
.iter()
.any(|m| m.revocation_id == id));
let reuse = bootstrap_token(
State(deploy.clone()),
Extension(issuer.clone()),
Extension(gate),
bearer(secret),
Json(req()),
)
.await;
assert_eq!(reuse.status(), StatusCode::CONFLICT);
let disabled = bootstrap_token(
State(deploy),
Extension(issuer),
Extension(BootstrapGate(None)),
bearer(secret),
Json(req()),
)
.await;
assert_eq!(disabled.status(), StatusCode::NOT_IMPLEMENTED);
}
#[tokio::test]
async fn cluster_rotate_key_returns_the_new_pubkey_or_501() {
let control = Arc::new(StubControl {
admits: std::sync::Mutex::new(Vec::new()),
respond: StubJoin::Admit,
});
let resp = cluster_rotate_key(Extension(MeshControlHandle(Some(control)))).await;
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let parsed: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(parsed["pubkey"].as_str(), Some("cafe"));
let none = cluster_rotate_key(Extension(MeshControlHandle(None))).await;
assert_eq!(none.status(), StatusCode::NOT_IMPLEMENTED);
}
#[test]
fn gateway_addr_gate_refuses_metadata_and_private_per_posture() {
use boatramp_core::security::SecurityProfile;
let strict = SecurityProfile::MultiTenant.preset();
let loose = SecurityProfile::SingleTenant.preset();
let public: IpAddr = "93.184.216.34".parse().unwrap(); let private: IpAddr = "10.1.2.3".parse().unwrap();
let loopback: IpAddr = "127.0.0.1".parse().unwrap();
let metadata: IpAddr = IpAddr::V4(CLOUD_METADATA_IPV4);
assert!(gateway_addr_allowed(public, &strict));
assert!(!gateway_addr_allowed(private, &strict));
assert!(!gateway_addr_allowed(loopback, &strict));
assert!(!gateway_addr_allowed(metadata, &strict));
assert!(gateway_addr_allowed(public, &loose));
assert!(gateway_addr_allowed(private, &loose));
assert!(gateway_addr_allowed(loopback, &loose));
assert!(!gateway_addr_allowed(metadata, &loose));
}
#[tokio::test]
async fn resolve_env_merges_static_and_host_secrets() {
use boatramp_core::config::HandlersSiteConfig;
std::env::set_var("BOATRAMP_TEST_RESOLVE_SECRET", "topsecret");
let deploy_env = std::collections::BTreeMap::from([
("GREETING".to_string(), "hi".to_string()),
("OVERRIDE_ME".to_string(), "static".to_string()),
]);
let site_handlers = HandlersSiteConfig {
enabled: true,
secrets: std::collections::BTreeMap::from([
(
"SECRET_TOKEN".to_string(),
"BOATRAMP_TEST_RESOLVE_SECRET".to_string(),
),
(
"OVERRIDE_ME".to_string(),
"BOATRAMP_TEST_RESOLVE_SECRET".to_string(),
),
(
"MISSING".to_string(),
"BOATRAMP_TEST_NOT_SET_VAR".to_string(),
),
]),
..Default::default()
};
let env = resolve_env(
"blog",
boatramp_core::project::ProjectRef::DEFAULT,
&deploy_env,
&site_handlers,
true,
None,
)
.await
.expect("resolves");
assert!(env.contains(&("GREETING".to_string(), "hi".to_string())));
assert!(env.contains(&("SECRET_TOKEN".to_string(), "topsecret".to_string())));
assert!(env.contains(&("OVERRIDE_ME".to_string(), "topsecret".to_string())));
assert!(!env.iter().any(|(k, _)| k == "MISSING"));
std::env::remove_var("BOATRAMP_TEST_RESOLVE_SECRET");
}
#[tokio::test]
async fn multi_tenant_posture_refuses_a_host_env_handler_secret() {
use boatramp_core::config::HandlersSiteConfig;
std::env::set_var("BOATRAMP_TEST_OTHER_TENANT_SECRET", "leak-me");
let deploy_env = std::collections::BTreeMap::new();
let bare = HandlersSiteConfig {
enabled: true,
secrets: std::collections::BTreeMap::from([(
"STOLEN".to_string(),
"BOATRAMP_TEST_OTHER_TENANT_SECRET".to_string(),
)]),
..Default::default()
};
let err = resolve_env(
"evil",
boatramp_core::project::ProjectRef::DEFAULT,
&deploy_env,
&bare,
false,
None,
)
.await
.expect_err("multi-tenant must refuse a bare host-env ref");
assert!(
err.contains("STOLEN"),
"error names the offending guest var: {err}"
);
assert!(
err.contains("multi-tenant"),
"error steers the tenant: {err}"
);
assert!(
!err.contains("leak-me"),
"the host value must never appear (never read): {err}"
);
let explicit = HandlersSiteConfig {
enabled: true,
secrets: std::collections::BTreeMap::from([(
"STOLEN".to_string(),
"env:BOATRAMP_TEST_OTHER_TENANT_SECRET".to_string(),
)]),
..Default::default()
};
assert!(resolve_env(
"evil",
boatramp_core::project::ProjectRef::DEFAULT,
&deploy_env,
&explicit,
false,
None,
)
.await
.is_err());
let reserved = HandlersSiteConfig {
enabled: true,
secrets: std::collections::BTreeMap::from([(
"TOKEN".to_string(),
"vault:kv/data/app#token".to_string(),
)]),
..Default::default()
};
let err = resolve_env(
"evil",
boatramp_core::project::ProjectRef::DEFAULT,
&deploy_env,
&reserved,
true,
None,
)
.await
.expect_err("a reserved scheme is not yet supported, even under single-tenant");
assert!(err.contains("not yet supported"), "{err}");
let arbitrary = HandlersSiteConfig {
enabled: true,
secrets: std::collections::BTreeMap::from([(
"KEY".to_string(),
"aws:sm/prod/apikey".to_string(),
)]),
..Default::default()
};
let err = resolve_env(
"evil",
boatramp_core::project::ProjectRef::DEFAULT,
&deploy_env,
&arbitrary,
true,
None,
)
.await
.expect_err("any unknown scheme is reserved, even under single-tenant");
assert!(
err.contains("not yet supported") && err.contains("aws"),
"provider-neutral reservation names the scheme: {err}"
);
std::env::remove_var("BOATRAMP_TEST_OTHER_TENANT_SECRET");
}
#[tokio::test]
async fn function_resolve_secret_env_reads_host_and_matches_handler_semantics() {
std::env::set_var("BOATRAMP_TEST_FN_SECRET", "fnsecret");
let static_env = std::collections::BTreeMap::from([
("STAGE".to_string(), "prod".to_string()),
("OVERRIDE_ME".to_string(), "static".to_string()),
]);
let secrets = std::collections::BTreeMap::from([
("DB_URL".to_string(), "BOATRAMP_TEST_FN_SECRET".to_string()),
(
"OVERRIDE_ME".to_string(),
"BOATRAMP_TEST_FN_SECRET".to_string(),
),
(
"MISSING".to_string(),
"BOATRAMP_TEST_FN_NOT_SET".to_string(),
),
]);
let env = resolve_secret_env(
"fn/api",
boatramp_core::project::ProjectRef::DEFAULT,
&static_env,
&secrets,
true,
None,
)
.await
.expect("resolves");
assert!(env.contains(&("STAGE".to_string(), "prod".to_string())));
assert!(env.contains(&("DB_URL".to_string(), "fnsecret".to_string())));
assert!(env.contains(&("OVERRIDE_ME".to_string(), "fnsecret".to_string())));
assert!(!env.iter().any(|(k, _)| k == "MISSING"));
std::env::remove_var("BOATRAMP_TEST_FN_SECRET");
}
#[tokio::test]
async fn multi_tenant_posture_refuses_a_host_env_function_secret() {
std::env::set_var("BOATRAMP_TEST_FN_LEAK", "leak-me");
let static_env = std::collections::BTreeMap::new();
let secrets = std::collections::BTreeMap::from([(
"DB_URL".to_string(),
"BOATRAMP_TEST_FN_LEAK".to_string(),
)]);
let err = resolve_secret_env(
"fn/api",
boatramp_core::project::ProjectRef::DEFAULT,
&static_env,
&secrets,
false,
None,
)
.await
.expect_err("multi-tenant must refuse a function host-env ref");
assert!(
err.contains("DB_URL"),
"error names the offending guest var: {err}"
);
assert!(
!err.contains("leak-me"),
"host value must never appear: {err}"
);
let env = resolve_secret_env(
"fn/api",
boatramp_core::project::ProjectRef::DEFAULT,
&static_env,
&secrets,
true,
None,
)
.await
.expect("resolves");
assert!(env.contains(&("DB_URL".to_string(), "leak-me".to_string())));
std::env::remove_var("BOATRAMP_TEST_FN_LEAK");
}
#[tokio::test]
async fn boatramp_scheme_resolves_from_the_project_scoped_store() {
use boatramp_core::project::ProjectRef;
use boatramp_core::secret_store::SecretStore;
use std::sync::Arc;
struct XorEnvelope;
#[async_trait::async_trait]
impl boatramp_core::envelope::KeyEnvelope for XorEnvelope {
async fn wrap(
&self,
p: &[u8],
) -> Result<Vec<u8>, boatramp_core::envelope::EnvelopeError> {
Ok(p.iter().map(|b| b ^ 0x5a).collect())
}
async fn unwrap(
&self,
c: &[u8],
) -> Result<Vec<u8>, boatramp_core::envelope::EnvelopeError> {
Ok(c.iter().map(|b| b ^ 0x5a).collect())
}
}
let store = SecretStore::new(
Arc::new(boatramp_core::kv::MemoryKv::new()),
Arc::new(XorEnvelope),
);
store
.set(ProjectRef::new("acme"), "api-key", b"s3cr3t")
.await
.unwrap();
let static_env = std::collections::BTreeMap::new();
let secrets = std::collections::BTreeMap::from([
("API_KEY".to_string(), "boatramp:api-key".to_string()),
("MISSING".to_string(), "boatramp:not-set".to_string()),
]);
let env = resolve_secret_env(
"site",
ProjectRef::new("acme"),
&static_env,
&secrets,
false,
Some(&store),
)
.await
.expect("boatramp refs resolve without the host-env gate");
assert!(env.contains(&("API_KEY".to_string(), "s3cr3t".to_string())));
assert!(!env.iter().any(|(k, _)| k == "MISSING"));
let other_secrets = std::collections::BTreeMap::from([(
"API_KEY".to_string(),
"boatramp:api-key".to_string(),
)]);
let other = resolve_secret_env(
"site",
ProjectRef::new("globex"),
&static_env,
&other_secrets,
false,
Some(&store),
)
.await
.expect("resolves (a foreign project's secret is simply absent → skipped)");
assert!(
!other.iter().any(|(k, _)| k == "API_KEY"),
"a tenant must not read another project's secret"
);
let err = resolve_secret_env(
"site",
ProjectRef::new("acme"),
&static_env,
&other_secrets,
false,
None,
)
.await
.expect_err("no store configured must fail closed");
assert!(err.contains("no internal secret store"), "{err}");
}
fn req() -> Request {
Request::builder()
.uri("/")
.header(header::HOST, "example.com")
.body(Body::empty())
.unwrap()
}
#[test]
fn forwarded_headers_set_standard_triple() {
let mut request = req();
set_forwarded_headers(&mut request, "203.0.113.7".parse().unwrap());
let h = request.headers();
assert_eq!(h.get("x-forwarded-for").unwrap(), "203.0.113.7");
assert_eq!(h.get("x-forwarded-host").unwrap(), "example.com");
assert_eq!(h.get("x-forwarded-proto").unwrap(), "http");
}
#[test]
fn forwarded_for_overwrites_spoofed_value() {
let mut request = Request::builder()
.uri("/")
.header(header::HOST, "example.com")
.header("x-forwarded-for", "10.0.0.1, 1.2.3.4")
.body(Body::empty())
.unwrap();
set_forwarded_headers(&mut request, "203.0.113.7".parse().unwrap());
let values: Vec<_> = request
.headers()
.get_all("x-forwarded-for")
.iter()
.collect();
assert_eq!(values.len(), 1);
assert_eq!(values[0], "203.0.113.7");
}
#[test]
fn forwarded_proto_preserves_upstream_tls() {
let mut request = Request::builder()
.uri("/")
.header(header::HOST, "example.com")
.header("x-forwarded-proto", "https")
.body(Body::empty())
.unwrap();
set_forwarded_headers(&mut request, "203.0.113.7".parse().unwrap());
assert_eq!(request.headers().get("x-forwarded-proto").unwrap(), "https");
}
#[test]
fn forwarded_host_absent_when_no_host_header() {
let mut request = Request::builder().uri("/").body(Body::empty()).unwrap();
set_forwarded_headers(&mut request, "203.0.113.7".parse().unwrap());
assert!(request.headers().get("x-forwarded-host").is_none());
assert_eq!(
request.headers().get("x-forwarded-for").unwrap(),
"203.0.113.7"
);
}
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_core::messaging::{LogMessaging, Messaging};
use boatramp_core::{ByteStream, GetObject, ObjectMeta, PutMeta, StorageError};
const EVENT_CONSUMER: &[u8] =
include_bytes!("../../boatramp-handlers/tests/fixtures/event-consumer.wasm");
#[derive(Default)]
struct MemStorage {
objects: std::sync::Mutex<std::collections::HashMap<String, Vec<u8>>>,
}
#[async_trait::async_trait]
impl boatramp_core::Storage for MemStorage {
async fn get(&self, key: &str) -> Result<GetObject, StorageError> {
let bytes = self
.objects
.lock()
.unwrap()
.get(key)
.cloned()
.ok_or_else(|| StorageError::NotFound(key.to_string()))?;
let body: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from(bytes)) }).boxed();
Ok(GetObject {
meta: ObjectMeta {
key: key.to_string(),
..Default::default()
},
body,
})
}
async fn get_range(
&self,
key: &str,
_: u64,
_: Option<u64>,
) -> Result<GetObject, StorageError> {
self.get(key).await
}
async fn put(
&self,
key: &str,
mut body: ByteStream,
_: PutMeta,
) -> Result<ObjectMeta, StorageError> {
use futures::StreamExt;
let mut buf = Vec::new();
while let Some(chunk) = body.next().await {
buf.extend_from_slice(&chunk?);
}
self.objects.lock().unwrap().insert(key.to_string(), buf);
Ok(ObjectMeta {
key: key.to_string(),
..Default::default()
})
}
async fn head(&self, key: &str) -> Result<ObjectMeta, StorageError> {
self.objects
.lock()
.unwrap()
.get(key)
.map(|_| ObjectMeta {
key: key.to_string(),
..Default::default()
})
.ok_or_else(|| StorageError::NotFound(key.to_string()))
}
async fn delete(&self, key: &str) -> Result<(), StorageError> {
self.objects.lock().unwrap().remove(key);
Ok(())
}
async fn list(&self, _: &str) -> Result<Vec<ObjectMeta>, StorageError> {
Ok(Vec::new())
}
}
fn observed_state_in(
project: &str,
workload: &str,
host: &str,
healthy: bool,
phase: boatramp_core::compute::ReplicaPhase,
) -> boatramp_core::compute::ObservedInstance {
use boatramp_core::compute::{Endpoint, InstanceHandle, ReplicaPhase, Scheme, Snapshot};
boatramp_core::compute::ObservedInstance {
handle: InstanceHandle {
project: project.into(),
workload: workload.into(),
replica: 0,
backend_ref: "ref-0".into(),
},
node: 1,
backend: "vmm".into(),
endpoint: Endpoint {
scheme: Scheme::Http,
host: host.into(),
port: 80,
},
region: None,
healthy,
started_at: None,
phase,
snapshot: matches!(phase, ReplicaPhase::Zero).then(|| Snapshot {
project: project.into(),
workload: workload.into(),
replica: 0,
data_ref: "snap-0".into(),
}),
}
}
fn observed_state(
workload: &str,
healthy: bool,
phase: boatramp_core::compute::ReplicaPhase,
) -> boatramp_core::compute::ObservedInstance {
observed_state_in("default", workload, "10.0.0.2", healthy, phase)
}
#[tokio::test]
async fn has_parked_replica_detects_a_zeroed_replica() {
use boatramp_core::compute::ReplicaPhase;
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage, kv);
assert!(!has_parked_replica(&deploy, "default", "w").await);
deploy
.set_replica_state(
ProjectRef::DEFAULT,
&observed_state("w", true, ReplicaPhase::Running),
)
.await
.unwrap();
assert!(!has_parked_replica(&deploy, "default", "w").await);
deploy
.set_replica_state(
ProjectRef::DEFAULT,
&observed_state("w", false, ReplicaPhase::Zero),
)
.await
.unwrap();
assert!(has_parked_replica(&deploy, "default", "w").await);
}
#[tokio::test]
async fn await_warm_returns_immediately_when_healthy_and_times_out_otherwise() {
use boatramp_core::compute::ReplicaPhase;
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage, kv);
let empty = await_warm(
&deploy,
"default",
"w",
std::time::Duration::from_millis(150),
)
.await;
assert!(empty.is_empty());
deploy
.set_replica_state(
ProjectRef::DEFAULT,
&observed_state("w", true, ReplicaPhase::Running),
)
.await
.unwrap();
let warm = await_warm(&deploy, "default", "w", std::time::Duration::from_secs(5)).await;
assert_eq!(warm, vec!["http://10.0.0.2:80".to_string()]);
}
#[tokio::test]
async fn compute_endpoints_are_project_scoped() {
use boatramp_core::compute::ReplicaPhase;
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage, kv);
deploy
.set_replica_state(
ProjectRef::new("acme"),
&observed_state_in("acme", "web", "10.0.0.5", true, ReplicaPhase::Running),
)
.await
.unwrap();
deploy
.set_replica_state(
ProjectRef::DEFAULT,
&observed_state_in("default", "web", "10.0.0.9", true, ReplicaPhase::Running),
)
.await
.unwrap();
assert_eq!(
compute_endpoints(&deploy, "acme", "web").await,
vec!["http://10.0.0.5:80".to_string()],
"acme's web resolves against acme, not default"
);
assert_eq!(
compute_endpoints(&deploy, "default", "web").await,
vec!["http://10.0.0.9:80".to_string()]
);
assert!(compute_endpoints(&deploy, "beta", "web").await.is_empty());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dispatcher_delivers_at_least_once_then_dead_letters() {
use boatramp_handlers::{Bindings, HandlerEngine, Limits};
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let mq = LogMessaging::new(storage, kv.clone());
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let hash = boatramp_core::deploy::sha256_hex(EVENT_CONSUMER);
let bindings = Bindings::new("blog").with_keyvalue("blog", kv.clone());
let topic = "blog/orders/created";
for _ in 0..3 {
mq.publish(topic, b"ok").await.unwrap();
}
loop {
let acked = dispatch_consumer_batch(
&engine,
&mq,
&metrics::Metrics::default(),
"blog",
topic,
"blog/",
"",
boatramp_core::messaging::StartPosition::Latest,
&hash,
EVENT_CONSUMER,
&bindings,
Limits::default(),
Duration::from_secs(30),
5,
10,
)
.await;
if acked == 0 {
break;
}
}
assert_eq!(
kv.get("hkv/blog/delivered/orders/created").await.unwrap(),
Some(b"3".to_vec())
);
mq.publish(topic, b"fail").await.unwrap();
for _ in 0..5 {
dispatch_consumer_batch(
&engine,
&mq,
&metrics::Metrics::default(),
"blog",
topic,
"blog/",
"",
boatramp_core::messaging::StartPosition::Latest,
&hash,
EVENT_CONSUMER,
&bindings,
Limits::default(),
Duration::ZERO,
2,
10,
)
.await;
}
assert_eq!(mq.dead_letter_count(topic).await.unwrap(), 1);
assert_eq!(
kv.get("hkv/blog/delivered/orders/created").await.unwrap(),
Some(b"3".to_vec())
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn consumer_groups_fan_out_through_the_dispatcher() {
use boatramp_handlers::{Bindings, HandlerEngine, Limits};
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let mq = LogMessaging::new(storage, kv.clone());
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let hash = boatramp_core::deploy::sha256_hex(EVENT_CONSUMER);
let bindings = Bindings::new("blog").with_keyvalue("blog", kv.clone());
let topic = "blog/orders/created";
let start = boatramp_core::messaging::StartPosition::Latest;
for g in ["billing", "audit"] {
let n = dispatch_consumer_batch(
&engine,
&mq,
&metrics::Metrics::default(),
"blog",
topic,
"blog/",
g,
start,
&hash,
EVENT_CONSUMER,
&bindings,
Limits::default(),
Duration::from_secs(30),
5,
10,
)
.await;
assert_eq!(n, 0, "no events yet for group {g}");
}
mq.publish(topic, b"ok").await.unwrap();
for g in ["billing", "audit"] {
let n = dispatch_consumer_batch(
&engine,
&mq,
&metrics::Metrics::default(),
"blog",
topic,
"blog/",
g,
start,
&hash,
EVENT_CONSUMER,
&bindings,
Limits::default(),
Duration::from_secs(30),
5,
10,
)
.await;
assert_eq!(n, 1, "group {g} should receive the message");
}
assert_eq!(
kv.get("hkv/blog/delivered/orders/created").await.unwrap(),
Some(b"2".to_vec())
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn scheduler_runs_current_consumers_not_previews() {
use boatramp_core::config::{ConsumerConfig, DeployConfig, HandlersSiteConfig, SiteConfig};
use boatramp_core::deploy::{DeployStore, FileEntry, Manifest};
use boatramp_handlers::{HandlerEngine, Limits};
use futures::StreamExt;
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let messaging: Arc<dyn Messaging> =
Arc::new(LogMessaging::new(storage.clone(), kv.clone()));
let hash = boatramp_core::deploy::sha256_hex(EVENT_CONSUMER);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(EVENT_CONSUMER)) })
.boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let mut files = std::collections::BTreeMap::new();
files.insert(
"consumer.wasm".to_string(),
FileEntry {
hash: hash.clone(),
size: EVENT_CONSUMER.len() as u64,
content_type: None,
variants: std::collections::BTreeMap::new(),
},
);
let manifest = Manifest {
files,
config: DeployConfig {
consumers: vec![ConsumerConfig {
tenancy: None,
token_claims: None,
topic: "orders/created".into(),
component: "consumer.wasm".into(),
imports: vec!["wasi:keyvalue".into()],
group: String::new(),
start: Default::default(),
}],
..Default::default()
},
..Default::default()
};
let id = deploy.put_manifest(&manifest).await.unwrap();
deploy
.activate(ProjectRef::DEFAULT, "blog", &id)
.await
.unwrap();
deploy
.set_site_config(
ProjectRef::DEFAULT,
"blog",
&SiteConfig {
handlers: Some(HandlersSiteConfig {
enabled: true,
allow_imports: vec!["wasi:keyvalue".into()],
..Default::default()
}),
..Default::default()
},
)
.await
.unwrap();
messaging
.publish("blog/orders/created", b"live")
.await
.unwrap();
messaging
.publish("blog/_preview/abc/orders/created", b"preview")
.await
.unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, None, Some(messaging));
let inner = rt.inner.clone().unwrap();
let mut cache = std::collections::HashMap::new();
let mut crons = std::collections::HashMap::new();
let mut sweep = std::collections::HashMap::new();
let now = CronNow {
minute: 0,
hour: 0,
dom: 1,
month: 1,
dow: 0,
minute_stamp: 0,
};
for _ in 0..3 {
run_scheduler_tick(&inner, &deploy, &mut cache, &mut crons, &mut sweep, now)
.await
.unwrap();
}
assert_eq!(
kv.get("hkv/blog/delivered/orders/created").await.unwrap(),
Some(b"1".to_vec())
);
assert_eq!(
kv.get("hkv/blog/_preview/abc/delivered/orders/created")
.await
.unwrap(),
None
);
}
const KV_COUNTER: &[u8] =
include_bytes!("../../boatramp-handlers/tests/fixtures/kv-counter.wasm");
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn function_invoker_runs_target_buffers_and_meters() {
use boatramp_core::deploy::DeployStore;
use boatramp_core::function::{Function, FunctionVersion, Lifecycle, Owner};
use boatramp_handlers::{HandlerEngine, InvokeError, InvokeRequest, Invoker, Limits};
use futures::StreamExt;
const HTTP_200: &[u8] =
include_bytes!("../../boatramp-handlers/tests/fixtures/http-200.wasm");
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let hash = boatramp_core::deploy::sha256_hex(HTTP_200);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(HTTP_200)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let function = Function {
name: "target".into(),
owner: Owner::Project("default".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: Default::default(),
};
deploy
.put_function(ProjectRef::DEFAULT, &function)
.await
.unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv, storage, None, None);
rt.set_invoker(deploy.clone());
let invoker = rt.inner.as_ref().unwrap().invoker.get().unwrap().clone();
let request = || InvokeRequest {
method: "GET".into(),
path: "/".into(),
headers: vec![],
body: vec![],
};
let response = invoker.invoke("target", request(), 1).await.unwrap();
assert_eq!(response.status, 200);
let metering = deploy
.get_metering(ProjectRef::DEFAULT, "target")
.await
.unwrap()
.unwrap();
assert_eq!(metering.invocations, 1);
let err = invoker.invoke("ghost", request(), 1).await.unwrap_err();
assert!(matches!(err, InvokeError::NotFound));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn federation_runner_enforces_the_safelist_before_planning() {
use boatramp_core::deploy::DeployStore;
use boatramp_core::project::ProjectRef;
use boatramp_handlers::{GraphqlRequest, HandlerEngine, Limits, SupergraphRunError};
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, None, None);
rt.set_invoker(deploy.clone());
let runner = rt
.inner
.as_ref()
.unwrap()
.federation_runner
.get()
.unwrap()
.scoped(ProjectRef::new("default"), Vec::new());
let req = |query: &str| GraphqlRequest {
query: Some(query.to_string()),
persisted_hash: None,
variables: "{}".to_string(),
operation_name: None,
authorization: None,
};
assert!(matches!(
runner.run(req("{ me { id } }"), 1).await,
Err(SupergraphRunError::NotSafelisted)
));
let query = "{ me { id } }";
let hash = crate::graphql_apq::sha256_hex(query);
kv.put(&format!("hapq/default/{hash}"), query.as_bytes().to_vec())
.await
.unwrap();
assert!(matches!(
runner.run(req(query), 1).await,
Err(SupergraphRunError::PlanFailed(_))
));
let persisted = GraphqlRequest {
query: None,
persisted_hash: Some("deadbeef".to_string()),
variables: "{}".to_string(),
operation_name: None,
authorization: None,
};
assert!(matches!(
runner.run(persisted, 1).await,
Err(SupergraphRunError::NotSafelisted)
));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "run on the host toolchain (real libsql static-musl segfault); wired in the CI graphql-propagation gate"]
async fn graphql_run_propagates_caller_principal_to_a_scoped_subfetch() {
use boatramp_core::deploy::{sha256_hex, DeployStore};
use boatramp_core::function::{
Function, FunctionConfig, FunctionVersion, Lifecycle, Owner,
};
use boatramp_core::project::ProjectRef;
use boatramp_core::sql::{SqlBackends, SqlValue};
use boatramp_core::tenancy::{AccessMode, ScopeAxis, Tenancy, TenantSource};
use boatramp_handlers::{GraphqlRequest, HandlerEngine, Limits, ScopeFact};
const PROBE: &[u8] = include_bytes!("../tests/fixtures/graphql-scope-probe.wasm");
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let sql_dir = std::env::temp_dir().join(format!("br-gqlprop-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&sql_dir);
let backends = boatramp_storage::LibsqlSqlBackends::local(&sql_dir);
let db = backends
.database("default", "fn/scopeprobe", "")
.await
.unwrap();
{
let mut tx = db.begin().await.unwrap();
tx.execute(
"CREATE TABLE items (id TEXT PRIMARY KEY, tenant_id TEXT)",
&[],
)
.await
.unwrap();
for (id, tenant) in [("a1", "tenant_A"), ("b1", "tenant_B"), ("b2", "tenant_B")] {
tx.execute(
"INSERT INTO items (id, tenant_id) VALUES (?1, ?2)",
&[SqlValue::Text(id.into()), SqlValue::Text(tenant.into())],
)
.await
.unwrap();
}
tx.commit().await.unwrap();
}
let sql: Arc<dyn SqlBackends> = Arc::new(backends);
let hash = sha256_hex(PROBE);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(PROBE)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let function = Function {
name: "scopeprobe".into(),
owner: Owner::Project("default".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: FunctionConfig {
imports: vec!["sql".into()],
tenancy: Some(Tenancy::Scoped {
column: "tenant_id".into(),
sources: vec![TenantSource::None],
read: AccessMode::Own,
write: AccessMode::None,
}),
..Default::default()
},
};
deploy
.put_function(ProjectRef::DEFAULT, &function)
.await
.unwrap();
crate::graphql_registry::publish(
kv.as_ref(),
"default",
"scopeprobe",
"type Query { items: [Item!]! }\ntype Item @key(fields: \"id\") { id: ID! }",
)
.await
.unwrap();
let query = "{ items { id } }";
let op_hash = crate::graphql_apq::sha256_hex(query);
kv.put(
&format!("hapq/default/{op_hash}"),
query.as_bytes().to_vec(),
)
.await
.unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, Some(sql), None);
rt.set_invoker(deploy.clone());
let fed = rt
.inner
.as_ref()
.unwrap()
.federation_runner
.get()
.unwrap()
.clone();
let req = || GraphqlRequest {
query: Some(query.to_string()),
persisted_hash: None,
variables: "{}".to_string(),
operation_name: None,
authorization: None,
};
let runner_b = fed.scoped(
ProjectRef::new("default"),
vec![ScopeFact {
axis: ScopeAxis::Tenant,
value: SqlValue::Text("tenant_B".into()),
}],
);
let body_b = String::from_utf8_lossy(&runner_b.run(req(), 0).await.unwrap()).into_owned();
assert!(
body_b.contains("\"b1\"") && body_b.contains("\"b2\"") && !body_b.contains("\"a1\""),
"graphql::run propagated principal B → the subgraph read ONLY tenant B's rows: {body_b}"
);
let runner_empty = fed.scoped(ProjectRef::new("default"), Vec::new());
let body_none =
String::from_utf8_lossy(&runner_empty.run(req(), 0).await.unwrap()).into_owned();
assert!(
!body_none.contains("\"a1\"")
&& !body_none.contains("\"b1\"")
&& !body_none.contains("\"b2\""),
"with NO propagated principal the subgraph fails closed — no rows leak: {body_none}"
);
let _ = std::fs::remove_dir_all(&sql_dir);
println!(
"GRAPHQL-RUN PRINCIPAL PROPAGATION OK: graphql::run carried the caller's resolved principal \
(tenant B) to a federated subgraph sub-fetch, which resolved its own-tenancy from the \
inherited principal and returned ONLY tenant B's rows over a real libsql engine; with no \
principal the same sub-fetch failed closed (no rows) — the async lane can drive a \
tenant-scoped supergraph read/write with no request bearer, symmetric to emit::invoke"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "run on the host toolchain (real libsql static-musl segfault); wired in the CI gateway-target gate"]
async fn gateway_forces_target_confinement_onto_a_wasm_subgraph() {
use boatramp_core::deploy::{sha256_hex, DeployStore};
use boatramp_core::function::{
Function, FunctionConfig, FunctionVersion, Lifecycle, Owner,
};
use boatramp_core::project::ProjectRef;
use boatramp_core::sql::{SqlBackends, SqlValue};
use boatramp_core::tenancy::{
AccessMode, PublicPredicate, PublicSubset, PublicTerm, TableScope, Tenancy,
TenancySchema, TenantSource,
};
use boatramp_handlers::{HandlerEngine, Limits};
const PROBE: &[u8] = include_bytes!("../tests/fixtures/graphql-scope-probe.wasm");
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let sql_dir = std::env::temp_dir().join(format!("br-gwtarget-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&sql_dir);
let backends = boatramp_storage::LibsqlSqlBackends::local(&sql_dir);
let db = backends
.database("default", "fn/scopeprobe", "")
.await
.unwrap();
{
let mut tx = db.begin().await.unwrap();
tx.execute(
"CREATE TABLE items (id TEXT PRIMARY KEY, tenant_id TEXT, published INTEGER)",
&[],
)
.await
.unwrap();
for (id, tenant, published) in [
("a_pub", "tenant_A", 1),
("b_pub", "tenant_B", 1),
("b_priv", "tenant_B", 0),
] {
tx.execute(
"INSERT INTO items (id, tenant_id, published) VALUES (?1, ?2, ?3)",
&[
SqlValue::Text(id.into()),
SqlValue::Text(tenant.into()),
SqlValue::Integer(published),
],
)
.await
.unwrap();
}
tx.commit().await.unwrap();
}
let sql: Arc<dyn SqlBackends> = Arc::new(backends);
let hash = sha256_hex(PROBE);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(PROBE)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let function = Function {
name: "scopeprobe".into(),
owner: Owner::Project("default".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: FunctionConfig {
imports: vec!["sql".into()],
tenancy: Some(Tenancy::Scoped {
column: "tenant_id".into(),
sources: vec![TenantSource::None],
read: AccessMode::Own,
write: AccessMode::None,
}),
..Default::default()
},
};
deploy
.put_function(ProjectRef::DEFAULT, &function)
.await
.unwrap();
let sdl = "type Query { items: [Item!]! @tenant(scope: target, via: [domain], public: \"items\") }\n\
type Item @key(fields: \"id\") { id: ID! }";
let sg = crate::graphql_federation::compose(&[("scopeprobe".into(), sdl.into())]).unwrap();
let plan = crate::graphql_plan::plan("{ items { id } }", &sg).unwrap();
let mut schema = TenancySchema {
default_tenant_key: "tenant_id".into(),
tables: std::collections::BTreeMap::from([("items".into(), TableScope::Tenant)]),
..Default::default()
};
schema.public_subsets.insert(
"items".into(),
PublicSubset {
predicate: PublicPredicate {
terms: vec![PublicTerm::Cmp {
column: "published".into(),
op: boatramp_core::tenancy::PublicCmp::Eq,
value: boatramp_core::tenancy::PublicLiteral::Int(1),
}],
},
world_public: true,
listable: true,
},
);
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, Some(sql), None);
rt.set_invoker(deploy.clone());
let inner = rt.inner.as_ref().unwrap();
let invoker = inner.invoker.get().unwrap().clone();
let make_router = |domain: Option<&str>| {
crate::graphql_gateway::BackendRouter::new(
invoker.scoped(ProjectRef::new("default"), Vec::new()),
"default".to_string(),
inner.sql.clone(),
std::collections::BTreeMap::new(),
None,
)
.with_target_inputs(Some(crate::graphql_gateway::TargetInputs {
schema: Arc::new(schema.clone()),
domain_context: domain.map(str::to_string),
target_handle: None,
capability_anchor: None,
}))
};
let router_b = make_router(Some("tenant_B"));
let out_b = crate::graphql_gateway::execute(&plan, &router_b, &serde_json::json!({})).await;
let body_b = out_b.to_string();
assert!(
body_b.contains("\"b_pub\"")
&& !body_b.contains("\"b_priv\"")
&& !body_b.contains("\"a_pub\""),
"gateway forced target B onto the wasm subgraph → ONLY B's PUBLIC row (b_pub), never B's \
private row nor A's: {body_b}"
);
let router_none = make_router(None);
let out_none =
crate::graphql_gateway::execute(&plan, &router_none, &serde_json::json!({})).await;
let body_none = out_none.to_string();
assert!(
!body_none.contains("\"a_pub\"")
&& !body_none.contains("\"b_pub\"")
&& !body_none.contains("\"b_priv\""),
"with no resolved target the wasm-subgraph target fetch fails closed — no rows: {body_none}"
);
let _ = std::fs::remove_dir_all(&sql_dir);
println!(
"GATEWAY WASM-TARGET OK: the /graphql gateway resolved target tenant B from the routed \
domain and FORCED a HostTenancy::target binding onto the wasm subgraph invocation \
(invoke_target), confining the guest's own SQL to tenant=B AND published=1 over a real \
libsql engine — returned ONLY B's public row (b_pub), never B's private row (b_priv) nor \
tenant A's (a_pub); with no resolved target the fetch failed closed"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "run on the host toolchain (real libsql static-musl segfault); wired in the CI gateway-own-propagation gate"]
async fn gateway_propagates_caller_own_principal_to_a_wasm_subgraph() {
use boatramp_core::deploy::{sha256_hex, DeployStore};
use boatramp_core::function::{
Function, FunctionConfig, FunctionVersion, Lifecycle, Owner,
};
use boatramp_core::project::ProjectRef;
use boatramp_core::sql::{SqlBackends, SqlValue};
use boatramp_core::tenancy::{AccessMode, ScopeAxis, Tenancy, TenantSource};
use boatramp_handlers::{HandlerEngine, Limits, ScopeFact};
const PROBE: &[u8] = include_bytes!("../tests/fixtures/graphql-scope-probe.wasm");
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let sql_dir = std::env::temp_dir().join(format!("br-gwown-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&sql_dir);
let backends = boatramp_storage::LibsqlSqlBackends::local(&sql_dir);
let db = backends
.database("default", "fn/scopeprobe", "")
.await
.unwrap();
{
let mut tx = db.begin().await.unwrap();
tx.execute(
"CREATE TABLE items (id TEXT PRIMARY KEY, tenant_id TEXT)",
&[],
)
.await
.unwrap();
for (id, tenant) in [("a1", "tenant_A"), ("b1", "tenant_B"), ("b2", "tenant_B")] {
tx.execute(
"INSERT INTO items (id, tenant_id) VALUES (?1, ?2)",
&[SqlValue::Text(id.into()), SqlValue::Text(tenant.into())],
)
.await
.unwrap();
}
tx.commit().await.unwrap();
}
let sql: Arc<dyn SqlBackends> = Arc::new(backends);
let hash = sha256_hex(PROBE);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(PROBE)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let function = Function {
name: "scopeprobe".into(),
owner: Owner::Project("default".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: FunctionConfig {
imports: vec!["sql".into()],
tenancy: Some(Tenancy::Scoped {
column: "tenant_id".into(),
sources: vec![TenantSource::None],
read: AccessMode::Own,
write: AccessMode::None,
}),
..Default::default()
},
};
deploy
.put_function(ProjectRef::DEFAULT, &function)
.await
.unwrap();
let sdl = "type Query { items: [Item!]! }\ntype Item @key(fields: \"id\") { id: ID! }";
let sg = crate::graphql_federation::compose(&[("scopeprobe".into(), sdl.into())]).unwrap();
let plan = crate::graphql_plan::plan("{ items { id } }", &sg).unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, Some(sql), None);
rt.set_invoker(deploy.clone());
let inner = rt.inner.as_ref().unwrap();
let invoker = inner.invoker.get().unwrap().clone();
let make_router = |facts: Vec<ScopeFact>| {
crate::graphql_gateway::BackendRouter::new(
invoker.scoped(ProjectRef::new("default"), facts),
"default".to_string(),
inner.sql.clone(),
std::collections::BTreeMap::new(),
None,
)
};
let router_b = make_router(vec![ScopeFact {
axis: ScopeAxis::Tenant,
value: SqlValue::Text("tenant_B".into()),
}]);
let body_b = crate::graphql_gateway::execute(&plan, &router_b, &serde_json::json!({}))
.await
.to_string();
assert!(
body_b.contains("\"b1\"") && body_b.contains("\"b2\"") && !body_b.contains("\"a1\""),
"gateway propagated principal B → the wasm subgraph read ONLY tenant B's rows: {body_b}"
);
let router_anon = make_router(Vec::new());
let body_anon =
crate::graphql_gateway::execute(&plan, &router_anon, &serde_json::json!({}))
.await
.to_string();
assert!(
!body_anon.contains("\"a1\"")
&& !body_anon.contains("\"b1\"")
&& !body_anon.contains("\"b2\""),
"with an empty principal the wasm-subgraph own fetch fails closed — no rows: {body_anon}"
);
let _ = std::fs::remove_dir_all(&sql_dir);
println!(
"GATEWAY OWN-PROPAGATION OK: the /graphql gateway propagated the caller's resolved OWN \
principal (tenant B) to a wasm subgraph fetch, which scoped its own read to the \
inherited principal and returned ONLY tenant B's rows over a real libsql engine; with \
an empty principal the same own fetch failed closed (no rows) — anon is never widened"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "run on the host toolchain (real libsql static-musl segfault); wired in the CI gateway-target gate"]
async fn gateway_target_or_null_includes_the_shared_base() {
use boatramp_core::deploy::{sha256_hex, DeployStore};
use boatramp_core::function::{
Function, FunctionConfig, FunctionVersion, Lifecycle, Owner,
};
use boatramp_core::project::ProjectRef;
use boatramp_core::sql::{SqlBackends, SqlValue};
use boatramp_core::tenancy::{
AccessMode, PublicPredicate, PublicSubset, PublicTerm, TableScope, Tenancy,
TenancySchema, TenantSource,
};
use boatramp_handlers::{HandlerEngine, Limits};
const PROBE: &[u8] = include_bytes!("../tests/fixtures/graphql-scope-probe.wasm");
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let sql_dir = std::env::temp_dir().join(format!("br-gwton-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&sql_dir);
let backends = boatramp_storage::LibsqlSqlBackends::local(&sql_dir);
let db = backends
.database("default", "fn/scopeprobe", "")
.await
.unwrap();
{
let mut tx = db.begin().await.unwrap();
tx.execute(
"CREATE TABLE items (id TEXT PRIMARY KEY, tenant_id TEXT, published INTEGER)",
&[],
)
.await
.unwrap();
for (id, tenant, published) in [
("base_pub", None, 1),
("base_priv", None, 0),
("b_pub", Some("tenant_B"), 1),
("b_priv", Some("tenant_B"), 0),
("a_pub", Some("tenant_A"), 1),
] {
tx.execute(
"INSERT INTO items (id, tenant_id, published) VALUES (?1, ?2, ?3)",
&[
SqlValue::Text(id.into()),
tenant
.map(|t| SqlValue::Text(t.into()))
.unwrap_or(SqlValue::Null),
SqlValue::Integer(published),
],
)
.await
.unwrap();
}
tx.commit().await.unwrap();
}
let sql: Arc<dyn SqlBackends> = Arc::new(backends);
let hash = sha256_hex(PROBE);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(PROBE)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let function = Function {
name: "scopeprobe".into(),
owner: Owner::Project("default".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: FunctionConfig {
imports: vec!["sql".into()],
tenancy: Some(Tenancy::Scoped {
column: "tenant_id".into(),
sources: vec![TenantSource::None],
read: AccessMode::Own,
write: AccessMode::None,
}),
..Default::default()
},
};
deploy
.put_function(ProjectRef::DEFAULT, &function)
.await
.unwrap();
let sdl = "type Query { items: [Item!]! @tenant(scope: target_or_null, via: [domain], public: \"items\") }\n\
type Item @key(fields: \"id\") { id: ID! }";
let sg = crate::graphql_federation::compose(&[("scopeprobe".into(), sdl.into())]).unwrap();
let plan = crate::graphql_plan::plan("{ items { id } }", &sg).unwrap();
let mut schema = TenancySchema {
default_tenant_key: "tenant_id".into(),
tables: std::collections::BTreeMap::from([("items".into(), TableScope::Tenant)]),
..Default::default()
};
schema.public_subsets.insert(
"items".into(),
PublicSubset {
predicate: PublicPredicate {
terms: vec![PublicTerm::Cmp {
column: "published".into(),
op: boatramp_core::tenancy::PublicCmp::Eq,
value: boatramp_core::tenancy::PublicLiteral::Int(1),
}],
},
world_public: true,
listable: true,
},
);
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, Some(sql), None);
rt.set_invoker(deploy.clone());
let inner = rt.inner.as_ref().unwrap();
let router = crate::graphql_gateway::BackendRouter::new(
inner
.invoker
.get()
.unwrap()
.clone()
.scoped(ProjectRef::new("default"), Vec::new()),
"default".to_string(),
inner.sql.clone(),
std::collections::BTreeMap::new(),
None,
)
.with_target_inputs(Some(crate::graphql_gateway::TargetInputs {
schema: Arc::new(schema),
domain_context: Some("tenant_B".into()),
target_handle: None,
capability_anchor: None,
}));
let out = crate::graphql_gateway::execute(&plan, &router, &serde_json::json!({})).await;
let body = out.to_string();
assert!(
body.contains("\"base_pub\"") && body.contains("\"b_pub\""),
"target_or_null returned B's public row AND the shared base floor: {body}"
);
assert!(
!body.contains("\"base_priv\"")
&& !body.contains("\"b_priv\"")
&& !body.contains("\"a_pub\""),
"the public subset confines BOTH B and base (no base_priv, no b_priv), and no other \
tenant's rows (no a_pub): {body}"
);
let _ = std::fs::remove_dir_all(&sql_dir);
println!(
"GATEWAY TARGET-OR-NULL OK: scope:target_or_null on a wasm subgraph read tenant B's \
public row (b_pub) PLUS the shared NULL-tenant base floor (base_pub), each confined to \
published=1 — never B's private row (b_priv), never a non-public base row (base_priv), \
never another tenant's row (a_pub); the base-only funnel keeps its inheritance floor"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn scheduler_drains_a_non_default_projects_invocation_in_its_own_tenant() {
use crate::scheduler::{run_scheduler_tick, CronNow};
use boatramp_core::deploy::DeployStore;
use boatramp_core::function::{
Function, FunctionVersion, Invocation, InvocationStatus, InvokeMode, Lifecycle, Owner,
};
use boatramp_handlers::{HandlerEngine, Limits};
use futures::StreamExt;
const HTTP_200: &[u8] =
include_bytes!("../../boatramp-handlers/tests/fixtures/http-200.wasm");
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let hash = boatramp_core::deploy::sha256_hex(HTTP_200);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(HTTP_200)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let acme = ProjectRef::new("acme");
let function = Function {
name: "worker".into(),
owner: Owner::Project("acme".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: Default::default(),
};
deploy.put_function(acme, &function).await.unwrap();
let inv = Invocation {
id: "inv1".into(),
function: "worker".into(),
version: "v1".into(),
mode: InvokeMode::Async,
status: InvocationStatus::Queued,
idempotency_key: None,
attempts: 0,
lease_expires: None,
request_b64: None,
request_content_type: None,
result: None,
created: 0,
updated: 0,
};
deploy.put_invocation(acme, &inv).await.unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage.clone(), None, None);
let inner = rt.inner.as_ref().unwrap();
let mut wasm_cache = std::collections::HashMap::new();
let mut cron_state = std::collections::HashMap::new();
let mut sweep = std::collections::HashMap::new();
let now = CronNow {
minute: 0,
hour: 0,
dom: 1,
month: 1,
dow: 0,
minute_stamp: 0,
};
run_scheduler_tick(
inner,
&deploy,
&mut wasm_cache,
&mut cron_state,
&mut sweep,
now,
)
.await
.unwrap();
let settled = poll_invocation_settled(&deploy, acme, "worker", "inv1").await;
assert_eq!(settled.status, InvocationStatus::Succeeded);
let metering = deploy.get_metering(acme, "worker").await.unwrap().unwrap();
assert_eq!(metering.invocations, 1);
assert!(deploy
.get_invocation(ProjectRef::DEFAULT, "worker", "inv1")
.await
.unwrap()
.is_none());
assert!(deploy
.get_metering(ProjectRef::DEFAULT, "worker")
.await
.unwrap()
.is_none());
}
#[cfg(feature = "handlers")]
async fn poll_invocation_settled(
deploy: &boatramp_core::deploy::DeployStore,
project: ProjectRef<'_>,
function: &str,
id: &str,
) -> boatramp_core::function::Invocation {
use boatramp_core::function::InvocationStatus;
for _ in 0..200 {
if let Some(inv) = deploy.get_invocation(project, function, id).await.unwrap() {
if matches!(
inv.status,
InvocationStatus::Succeeded | InvocationStatus::Failed
) {
return inv;
}
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
panic!("invocation {function}/{id} never settled");
}
#[cfg(feature = "handlers")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drain_reclaims_an_expired_lease_and_skips_a_live_one() {
use crate::scheduler::{run_scheduler_tick, CronNow};
use boatramp_core::deploy::DeployStore;
use boatramp_core::function::{
Function, FunctionVersion, Invocation, InvocationStatus, InvokeMode, Lifecycle, Owner,
};
use boatramp_handlers::{HandlerEngine, Limits};
use futures::StreamExt;
const HTTP_200: &[u8] =
include_bytes!("../../boatramp-handlers/tests/fixtures/http-200.wasm");
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let hash = boatramp_core::deploy::sha256_hex(HTTP_200);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(HTTP_200)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let function = Function {
name: "worker".into(),
owner: Owner::Project("default".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: Default::default(),
};
deploy
.put_function(ProjectRef::DEFAULT, &function)
.await
.unwrap();
let base = Invocation {
id: String::new(),
function: "worker".into(),
version: "v1".into(),
mode: InvokeMode::Async,
status: InvocationStatus::Running,
idempotency_key: None,
attempts: 1,
lease_expires: None,
request_b64: None,
request_content_type: None,
result: None,
created: 0,
updated: 0,
};
let orphan = Invocation {
id: "orphan".into(),
lease_expires: Some(1),
..base.clone()
};
deploy
.put_invocation(ProjectRef::DEFAULT, &orphan)
.await
.unwrap();
let live = Invocation {
id: "live".into(),
lease_expires: Some(u64::MAX),
..base.clone()
};
deploy
.put_invocation(ProjectRef::DEFAULT, &live)
.await
.unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage.clone(), None, None);
let inner = rt.inner.as_ref().unwrap();
let now = CronNow {
minute: 0,
hour: 0,
dom: 1,
month: 1,
dow: 0,
minute_stamp: 0,
};
let mut wasm_cache = std::collections::HashMap::new();
let mut cron_state = std::collections::HashMap::new();
let mut sweep = std::collections::HashMap::new();
run_scheduler_tick(
inner,
&deploy,
&mut wasm_cache,
&mut cron_state,
&mut sweep,
now,
)
.await
.unwrap();
let settled =
poll_invocation_settled(&deploy, ProjectRef::DEFAULT, "worker", "orphan").await;
assert_eq!(settled.status, InvocationStatus::Succeeded);
assert_eq!(settled.attempts, 2, "a reclaim counts as another attempt");
assert_eq!(
settled.lease_expires, None,
"a settled invocation drops its lease"
);
let live_after = deploy
.get_invocation(ProjectRef::DEFAULT, "worker", "live")
.await
.unwrap()
.unwrap();
assert_eq!(live_after.status, InvocationStatus::Running);
assert_eq!(live_after.attempts, 1, "a live lease is never reclaimed");
assert_eq!(live_after.lease_expires, Some(u64::MAX));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guest_kv_is_isolated_between_same_named_functions_in_two_projects() {
use boatramp_core::deploy::DeployStore;
use boatramp_core::function::{
Function, FunctionConfig, FunctionVersion, Lifecycle, Owner,
};
use boatramp_handlers::{HandlerEngine, Limits};
use futures::StreamExt;
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let hash = boatramp_core::deploy::sha256_hex(KV_COUNTER);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(KV_COUNTER)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let store = Function {
name: "store".into(),
owner: Owner::Project("default".into()),
versions: vec![FunctionVersion {
id: "v1".into(),
component: hash.clone(),
created: 0,
lifecycle: Lifecycle::Independent,
}],
active: "v1".into(),
aliases: Default::default(),
config: FunctionConfig {
imports: vec!["wasi:keyvalue".into()],
..Default::default()
},
};
let acme = ProjectRef::new("acme");
let globex = ProjectRef::new("globex");
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, None, None);
let inner = rt.inner.as_ref().unwrap();
let request = || {
axum::http::Request::builder()
.method("GET")
.uri("/")
.body(axum::body::Body::empty())
.unwrap()
};
let component = store.resolve(&store.active).unwrap().to_owned();
for project in [acme, globex, ProjectRef::DEFAULT] {
let (response, _) = execute_function(
inner,
&deploy,
project,
&store,
&component,
request(),
0,
boatramp_handlers::Lane::Sync,
crate::function_runtime::FnTenant::Request,
)
.await;
assert!(response.status().is_success(), "invocation should succeed");
}
assert_eq!(
kv.get("hkv/acme/fn/store/hits").await.unwrap(),
Some(b"1".to_vec()),
"acme's write must be tenant-qualified"
);
assert_eq!(
kv.get("hkv/globex/fn/store/hits").await.unwrap(),
Some(b"1".to_vec()),
"globex's write must be tenant-qualified"
);
assert_eq!(
kv.get("hkv/fn/store/hits").await.unwrap(),
Some(b"1".to_vec()),
"the default project must keep the byte-identical pre-project key"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn scheduler_fires_crons_with_dedup_and_overlap_skip() {
use boatramp_core::config::{
CronConfig, DeployConfig, HandlerConfig, HandlersSiteConfig, Overlap, SiteConfig,
};
use boatramp_core::deploy::{DeployStore, FileEntry, Manifest};
use boatramp_handlers::{HandlerEngine, Limits};
use futures::StreamExt;
use std::sync::atomic::Ordering;
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let hash = boatramp_core::deploy::sha256_hex(KV_COUNTER);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(KV_COUNTER)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let mut files = std::collections::BTreeMap::new();
files.insert(
"counter.wasm".to_string(),
FileEntry {
hash: hash.clone(),
size: KV_COUNTER.len() as u64,
content_type: None,
variants: std::collections::BTreeMap::new(),
},
);
let manifest = Manifest {
files,
config: DeployConfig {
handlers: vec![HandlerConfig {
tenancy: None,
token_claims: None,
route: "/".into(),
methods: Vec::new(),
component: "counter.wasm".into(),
imports: vec!["wasi:keyvalue".into()],
streaming: false,
limits: None,
env: std::collections::BTreeMap::new(),
invoke_targets: Vec::new(),
}],
crons: vec![CronConfig {
schedule: "* * * * *".into(),
route: "/".into(),
overlap: Overlap::Skip,
}],
..Default::default()
},
..Default::default()
};
let id = deploy.put_manifest(&manifest).await.unwrap();
deploy
.activate(ProjectRef::DEFAULT, "blog", &id)
.await
.unwrap();
deploy
.set_site_config(
ProjectRef::DEFAULT,
"blog",
&SiteConfig {
handlers: Some(HandlersSiteConfig {
enabled: true,
allow_imports: vec!["wasi:keyvalue".into()],
..Default::default()
}),
..Default::default()
},
)
.await
.unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, None, None);
let inner = rt.inner.clone().unwrap();
let mut wasm = std::collections::HashMap::new();
let mut crons = std::collections::HashMap::new();
let mut sweep = std::collections::HashMap::new();
let at = |stamp| CronNow {
minute: 0,
hour: 0,
dom: 1,
month: 1,
dow: 0,
minute_stamp: stamp,
};
let (_, handles) =
run_scheduler_tick(&inner, &deploy, &mut wasm, &mut crons, &mut sweep, at(100))
.await
.unwrap();
for h in handles {
h.await.unwrap();
}
assert_eq!(kv.get("hkv/blog/hits").await.unwrap(), Some(b"1".to_vec()));
let (_, handles) =
run_scheduler_tick(&inner, &deploy, &mut wasm, &mut crons, &mut sweep, at(100))
.await
.unwrap();
assert!(handles.is_empty());
assert_eq!(kv.get("hkv/blog/hits").await.unwrap(), Some(b"1".to_vec()));
let (_, handles) =
run_scheduler_tick(&inner, &deploy, &mut wasm, &mut crons, &mut sweep, at(101))
.await
.unwrap();
for h in handles {
h.await.unwrap();
}
assert_eq!(kv.get("hkv/blog/hits").await.unwrap(), Some(b"2".to_vec()));
crons
.get("default|blog|cron|0")
.unwrap()
.running
.store(true, Ordering::Release);
let (_, handles) =
run_scheduler_tick(&inner, &deploy, &mut wasm, &mut crons, &mut sweep, at(102))
.await
.unwrap();
assert!(handles.is_empty());
assert_eq!(kv.get("hkv/blog/hits").await.unwrap(), Some(b"2".to_vec()));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cron_leader_gate_suppresses_crons_off_leader() {
use boatramp_core::config::{
CronConfig, DeployConfig, HandlerConfig, HandlersSiteConfig, Overlap, SiteConfig,
};
use boatramp_core::deploy::{DeployStore, FileEntry, Manifest};
use boatramp_handlers::{HandlerEngine, Limits};
use futures::StreamExt;
let storage = Arc::new(MemStorage::default());
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let deploy = DeployStore::new(storage.clone(), kv.clone());
let hash = boatramp_core::deploy::sha256_hex(KV_COUNTER);
let stream: ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from_static(KV_COUNTER)) }).boxed();
deploy.put_blob(&hash, stream).await.unwrap();
let mut files = std::collections::BTreeMap::new();
files.insert(
"counter.wasm".to_string(),
FileEntry {
hash: hash.clone(),
size: KV_COUNTER.len() as u64,
content_type: None,
variants: std::collections::BTreeMap::new(),
},
);
let manifest = Manifest {
files,
config: DeployConfig {
handlers: vec![HandlerConfig {
tenancy: None,
token_claims: None,
route: "/".into(),
methods: Vec::new(),
component: "counter.wasm".into(),
imports: vec!["wasi:keyvalue".into()],
streaming: false,
limits: None,
env: std::collections::BTreeMap::new(),
invoke_targets: Vec::new(),
}],
crons: vec![CronConfig {
schedule: "* * * * *".into(),
route: "/".into(),
overlap: Overlap::Skip,
}],
..Default::default()
},
..Default::default()
};
let id = deploy.put_manifest(&manifest).await.unwrap();
deploy
.activate(ProjectRef::DEFAULT, "blog", &id)
.await
.unwrap();
deploy
.set_site_config(
ProjectRef::DEFAULT,
"blog",
&SiteConfig {
handlers: Some(HandlersSiteConfig {
enabled: true,
allow_imports: vec!["wasi:keyvalue".into()],
..Default::default()
}),
..Default::default()
},
)
.await
.unwrap();
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv.clone(), storage, None, None);
rt.set_cron_leader_gate(Arc::new(|| false));
let inner = rt.inner.clone().unwrap();
let mut wasm = std::collections::HashMap::new();
let mut crons = std::collections::HashMap::new();
let mut sweep = std::collections::HashMap::new();
let now = CronNow {
minute: 0,
hour: 0,
dom: 1,
month: 1,
dow: 0,
minute_stamp: 100,
};
let (_, handles) =
run_scheduler_tick(&inner, &deploy, &mut wasm, &mut crons, &mut sweep, now)
.await
.unwrap();
assert!(handles.is_empty(), "a non-leader must not fire crons");
assert_eq!(kv.get("hkv/blog/hits").await.unwrap(), None);
}
#[tokio::test]
async fn project_tenancy_knobs_override_wins_else_node_base() {
use boatramp_core::security::ResolvedProjectTenancy;
use boatramp_handlers::{HandlerEngine, Limits};
let kv: Arc<dyn boatramp_core::kv::KvStore> = Arc::new(boatramp_core::kv::MemoryKv::new());
let storage: Arc<dyn boatramp_core::Storage> = Arc::new(MemStorage::default());
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv, storage, None, None);
rt.set_tenancy_posture(true, false);
let mut overrides = std::collections::BTreeMap::new();
overrides.insert(
"preview".to_string(),
ResolvedProjectTenancy {
require_tenancy_declaration: true,
allow_cross_tenant_db: true,
capability_max_ttl_secs: Some(1800),
},
);
rt.set_project_tenancy_overrides(overrides);
let inner = rt.inner.as_ref().unwrap();
let p = inner.project_tenancy_knobs("preview");
assert!(p.require_tenancy_declaration);
assert!(p.allow_cross_tenant_db);
assert_eq!(p.capability_max_ttl_secs, Some(1800));
let b = inner.project_tenancy_knobs("prod");
assert!(b.require_tenancy_declaration);
assert!(!b.allow_cross_tenant_db);
assert_eq!(b.capability_max_ttl_secs, None);
}
#[tokio::test]
async fn build_bindings_dispatches_named_sql_databases_with_least_privilege() {
use boatramp_core::config::HandlersSiteConfig;
use boatramp_core::project::ProjectRef;
use boatramp_handlers::{HandlerEngine, Limits};
let kv: Arc<dyn boatramp_core::kv::KvStore> = Arc::new(boatramp_core::kv::MemoryKv::new());
let storage: Arc<dyn boatramp_core::Storage> = Arc::new(MemStorage::default());
let sql_dir =
std::env::temp_dir().join(format!("boatramp-named-sql-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&sql_dir);
let sql: Arc<dyn boatramp_core::sql::SqlBackends> =
Arc::new(boatramp_storage::LibsqlSqlBackends::local(&sql_dir));
let engine = HandlerEngine::new(Limits::default(), 16).unwrap();
let rt = HandlerRuntime::new(engine, kv, storage, Some(sql), None);
let inner = rt.inner.as_ref().unwrap();
let site = HandlersSiteConfig {
enabled: true,
allow_imports: vec!["sql".into(), "sql:product".into(), "sql:privileged".into()],
tenancy: Some(boatramp_core::tenancy::Tenancy::Disabled),
..Default::default()
};
let env = std::collections::BTreeMap::new();
let build = |imports: &[&str]| {
let imports: Vec<String> = imports.iter().copied().map(String::from).collect();
let site = &site;
let env = &env;
async move {
crate::handler_dispatch::build_bindings(
inner,
ProjectRef::new("default"),
"shop",
"shop",
None,
&imports,
site,
env,
&[],
0,
None,
None,
None,
None,
None,
None,
None,
)
.await
.expect("no secrets → resolves")
.sql_database_names()
}
};
assert_eq!(build(&["sql", "sql:product"]).await, vec!["", "product"]);
assert_eq!(
build(&["sql", "sql:*"]).await,
vec!["", "privileged", "product"]
);
assert!(build(&["sql:secret"]).await.is_empty());
assert_eq!(build(&["sql:product"]).await, vec!["product"]);
let _ = std::fs::remove_dir_all(&sql_dir);
}
}