1use std::collections::BTreeSet;
10use std::collections::HashMap;
11use std::collections::HashSet;
12use std::sync::Arc;
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::time::Duration;
15
16use axum::Router;
17use axum::extract::{DefaultBodyLimit, State};
18use axum::http::{HeaderMap, Method, StatusCode, header};
19use axum::middleware::{Next, from_fn_with_state};
20use axum::response::{IntoResponse, Response};
21use axum::routing::{get, post};
22use bytes::Bytes;
23use serde_json::json;
24use tracing::error;
25
26use crate::admin::require_admin;
27use crate::config::Config;
28use crate::config::ConfigError;
29use crate::events::EventBus;
30use crate::http::{
31 Json, MAX_BODY_BYTES, ScopeState, cors_layer, rate_limited, scope_layer,
32 security_headers_layer, token_response_layer,
33};
34use crate::module::{HARNESS_API, Module, ModuleContext, harness_api_mismatch};
35use crate::ports::Dispatcher;
36use crate::ports::{
37 ActorHandler, ActorHandlers, Clock, Database, Mailer, Port, Ports, RateLimiter, Statement,
38 SystemClock, validate_actor_kind, warn_undeclared_ports,
39};
40use crate::problem::Problem;
41use crate::problems::SLUGS;
42use crate::rate_limit::{RateLimit, RateLimitFailure};
43use crate::route_policy::deployed_env;
44use crate::scheduled::ScheduledBudget;
45use crate::scope::Scope;
46use crate::sidecar::{
47 GatewayGuard, SIDECAR_REQUIRE_GATEWAY, SidecarMount, X_HARNESS_GATEWAY, gateway_guard,
48 gateway_signer, mint_gateway, truthy,
49};
50use crate::signer::HmacSigner;
51use crate::stream::{RequestStream, StreamRoute};
52use crate::surface::{
53 MAX_SIDECAR_SURFACE_BYTES, RenderedSurface, SurfaceDocument, SurfaceSource, UiContext, UiMount,
54 sanitize_sidecar_document, strip_unguarded_captcha_actions,
55};
56use crate::template::{Template, TemplateRegistry};
57use crate::venture::{Venture, VentureEnv};
58
59fn check_tables(
68 module: &dyn Module,
69 tables: &mut HashMap<&'static str, &'static str>,
70 errors: &mut ConfigError,
71) {
72 let name = module.name();
73 for table in module.tables() {
74 match tables.get(table) {
75 Some(owner) => errors.push(format!(
76 "duplicate table `{table}` claimed by modules `{owner}` and `{name}`"
77 )),
78 None => {
79 tables.insert(table, name);
80 }
81 }
82 }
83
84 let owns = module.tables();
88 let mut declared: Vec<&'static str> = Vec::new();
89 for set in module.personal_data() {
90 for problem in set.validate(name, owns) {
91 errors.push(problem);
92 }
93 if declared.contains(&set.table) {
94 errors.push(format!(
95 "module `{name}` declares table `{}` twice in personal_data()",
96 set.table
97 ));
98 }
99 declared.push(set.table);
100 }
101}
102
103fn collect_actors(
109 handlers: &[(String, Arc<dyn ActorHandler>)],
110 modules: &[Arc<dyn Module>],
111 errors: &mut ConfigError,
112) -> ActorHandlers {
113 let mut registered: HashSet<&str> = HashSet::new();
114 let mut registry = ActorHandlers::new();
115 for (kind, handler) in handlers {
116 if let Err(err) = validate_actor_kind(kind) {
117 errors.push(format!(
118 "actor handler registered for invalid kind `{kind}`: {err}"
119 ));
120 }
121 if !registered.insert(kind.as_str()) {
122 errors.push(format!(
123 "duplicate actor handler registered for kind `{kind}`"
124 ));
125 }
126 registry = registry.with(kind, Arc::clone(handler));
127 }
128
129 let mut declared: HashMap<&'static str, &'static str> = HashMap::new();
130 for module in modules {
131 let name = module.name();
132 let kinds = module.actor_kinds();
133 if !kinds.is_empty()
134 && !module.requires().contains(&Port::Actor)
135 && !module.optional().contains(&Port::Actor)
136 {
137 errors.push(format!(
138 "module `{name}` declares actor kinds but lists port {} in neither requires() nor optional()",
139 Port::Actor.name(),
140 ));
141 }
142 for kind in kinds {
143 if let Err(err) = validate_actor_kind(kind) {
144 errors.push(format!(
145 "module `{name}` declares invalid actor kind `{kind}`: {err}"
146 ));
147 }
148 match declared.get(kind) {
149 Some(owner) => errors.push(format!(
150 "duplicate actor kind `{kind}` declared by modules `{owner}` and `{name}`"
151 )),
152 None => {
153 declared.insert(kind, name);
154 }
155 }
156 if !registered.contains(kind) {
157 errors.push(format!(
158 "module `{name}` declares actor kind `{kind}` with no registered handler"
159 ));
160 }
161 }
162 }
163
164 registry
165}
166
167pub trait Runtime: Send + Sync + 'static {
172 fn provides(&self) -> Vec<Port>;
173 fn effectively_configured(&self, port: Port) -> bool {
179 self.provides().contains(&port)
180 }
181}
182
183pub struct Harness {
185 venture: Arc<Venture>,
186 modules: Vec<Arc<dyn Module>>,
187 templates: Arc<TemplateRegistry>,
188 events: EventBus,
189 runtime: Option<Arc<dyn Runtime>>,
190 well_known: Option<Router>,
193 surface: Arc<SurfaceVariants>,
196 actor_handlers: ActorHandlers,
199 ui: Option<Arc<dyn UiMount>>,
201 personal_data: Arc<crate::PersonalDataCatalog>,
204 unprotected_acceptance_recorded: AtomicBool,
214 unlimited_acceptance_recorded: AtomicBool,
215}
216
217struct SurfaceVariants {
218 document: Arc<SurfaceDocument>,
219 full: RenderedSurface,
220 public: RenderedSurface,
221}
222
223impl SurfaceVariants {
224 fn compose(
225 venture: &Venture,
226 modules: &[Arc<dyn Module>],
227 ui: Option<&Arc<dyn UiMount>>,
228 ) -> Self {
229 let mut document = SurfaceDocument::compose(venture, modules);
230 document.ui = ui.and_then(|ui| ui.describe());
231 Self {
232 full: RenderedSurface::render(&document),
233 public: RenderedSurface::render(&document.public()),
234 document: Arc::new(document),
235 }
236 }
237}
238
239impl std::fmt::Debug for Harness {
240 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
241 let modules: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
242 f.debug_struct("Harness")
243 .field("venture", &self.venture.name)
244 .field("modules", &modules)
245 .finish_non_exhaustive()
246 }
247}
248
249impl Harness {
250 #[must_use]
254 pub fn clone_arc(self: &std::sync::Arc<Self>) -> std::sync::Arc<Self> {
255 std::sync::Arc::clone(self)
256 }
257 pub fn builder() -> HarnessBuilder {
258 HarnessBuilder::default()
259 }
260
261 pub fn venture(&self) -> &Arc<Venture> {
262 &self.venture
263 }
264
265 pub fn modules(&self) -> &[Arc<dyn Module>] {
266 &self.modules
267 }
268
269 pub fn actor_handlers(&self) -> &ActorHandlers {
272 &self.actor_handlers
273 }
274
275 pub fn max_body_bytes(&self, path: &str, cfg: &dyn Config) -> usize {
298 let Some((name, _)) = module_and_segments(path) else {
299 return MAX_BODY_BYTES;
300 };
301 self.modules
302 .iter()
303 .find(|module| module.name() == name)
304 .map_or(MAX_BODY_BYTES, |module| {
305 MAX_BODY_BYTES.max(module.max_body_bytes(cfg))
306 })
307 }
308
309 pub fn streaming_route(&self, path: &str, method: &axum::http::Method) -> Option<usize> {
327 let (name, segments) = exact_module_and_segments(path)?;
328 let module = self.modules.iter().find(|module| module.name() == name)?;
329 match_streaming(module.streaming_routes(), method, &segments)
330 }
331
332 pub fn templates(&self) -> &Arc<TemplateRegistry> {
333 &self.templates
334 }
335
336 pub fn events(&self) -> &EventBus {
337 &self.events
338 }
339
340 pub fn module_context(&self, module: &dyn Module, ports: &Ports) -> ModuleContext {
345 let venture = self.venture_as_deployed(ports.config.as_ref());
360 ModuleContext {
361 config: Arc::clone(&ports.config),
362 ports: ports.view_for(module),
363 events: self.events_for(ports),
364 templates: Arc::clone(&self.templates),
365 venture,
366 unprotected_writes_accepted: crate::route_policy::unprotected_writes_override(
367 ports.config.as_ref(),
368 )
369 .is_some(),
370 personal_data: Arc::clone(&self.personal_data),
371 ui_mounted: self.ui.is_some(),
372 scheduled: Arc::new(ScheduledBudget::unbounded()),
376 }
377 }
378
379 fn gateway_guard_state(ports: &Ports, gateway: Option<&Arc<HmacSigner>>) -> GatewayGuard {
389 GatewayGuard {
390 require: truthy(ports.config.as_ref(), SIDECAR_REQUIRE_GATEWAY),
391 signer: gateway.map(Arc::clone),
392 admin_token: ports
393 .config
394 .get("ADMIN_TOKEN")
395 .filter(|token| !token.is_empty()),
396 }
397 }
398
399 fn merged_surface(
402 &self,
403 mounts: Vec<SidecarMount>,
404 ports: &Ports,
405 gateway: Option<Arc<HmacSigner>>,
406 env: VentureEnv,
407 ) -> Arc<dyn SurfaceSource> {
408 Arc::new(MergedSurface {
409 base: Arc::clone(&self.surface),
410 mounts,
411 dispatcher: ports.dispatcher.clone(),
412 gateway,
413 env,
414 captcha_configured: ports.captcha.is_some(),
415 })
416 }
417
418 fn venture_as_deployed(&self, config: &dyn Config) -> Arc<Venture> {
424 let env = deployed_env(self.venture.env, config);
425 if env == self.venture.env {
426 return Arc::clone(&self.venture);
427 }
428 let mut deployed = (*self.venture).clone();
429 deployed.env = env;
430 Arc::new(deployed)
431 }
432
433 fn nest_modules(&self, ports: &Ports) -> Router {
439 let mut api = Router::new();
440 for module in &self.modules {
441 let ctx = self.module_context(module.as_ref(), ports);
442 api = api.nest(&format!("/v1/{}", module.name()), module.router(ctx));
443 }
444 api
445 }
446
447 fn production_readiness_now(
463 &self,
464 env: VentureEnv,
465 config: &dyn Config,
466 rate_limiter_ready: bool,
467 signer_ready: bool,
468 ) -> Vec<String> {
469 if let Some(note) = crate::route_policy::env_disagreement(self.venture.env, env) {
470 tracing::warn!("{note}");
471 crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, ¬e);
472 }
473 let unprotected = crate::route_policy::unprotected_writes_override(config);
477 let unlimited = crate::route_policy::unlimited_public_routes_override(config);
478 let guards = crate::route_policy::WriteGuards::collect(&self.modules);
479 let mut problems = crate::route_policy::production_readiness(
480 env,
481 &guards,
482 self.runtime.as_ref(),
483 rate_limiter_ready,
484 signer_ready,
485 unprotected.as_deref(),
486 unlimited.as_deref(),
487 );
488 let hmac_problems = crate::route_policy::webhook_secret_readiness(env, &guards, config);
493 if problems.is_empty() && hmac_problems.is_empty() {
494 let accepted = crate::route_policy::production_readiness(
503 env,
504 &guards,
505 self.runtime.as_ref(),
506 rate_limiter_ready,
507 signer_ready,
508 None,
509 None,
510 );
511 self.record_acceptances_once(
512 &guards,
513 rate_limiter_ready,
514 unprotected.as_deref(),
515 unlimited.as_deref(),
516 &accepted,
517 );
518 return problems;
519 }
520 for problem in &problems {
528 let detail = format!(
529 "refusing guarded routes: {problem} — set {} to a reason to accept this \
530 explicitly while the port is wired",
531 crate::route_policy::ALLOW_UNPROTECTED_WRITES
532 );
533 error!("{detail}");
534 crate::logging::forward_control_event(crate::logging::ControlLevel::Error, &detail);
535 }
536 problems.extend(hmac_problems);
537 problems
538 }
539
540 fn record_acceptances_once(
546 &self,
547 guards: &crate::route_policy::WriteGuards,
548 rate_limiter_ready: bool,
549 unprotected: Option<&str>,
550 unlimited: Option<&str>,
551 accepted: &[String],
552 ) {
553 if guards.needs_captcha()
554 && !crate::route_policy::captcha_effective(self.runtime.as_ref())
555 && let Some(reason) = unprotected
556 && !self
557 .unprotected_acceptance_recorded
558 .swap(true, Ordering::Relaxed)
559 {
560 let detail = format!(
561 "serving guarded routes unprotected on an operator's recorded acceptance \
562 (reason: {reason}; problems: {})",
563 accepted.join("; ")
564 );
565 tracing::warn!(
566 control = "production-readiness",
567 acceptance = crate::route_policy::ALLOW_UNPROTECTED_WRITES,
568 reason,
569 "serving captcha-guarded routes unprotected on an operator's recorded acceptance"
570 );
571 crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, &detail);
575 }
576 if guards.needs_rate_limiter()
577 && !rate_limiter_ready
578 && let Some(reason) = unlimited
579 && !self
580 .unlimited_acceptance_recorded
581 .swap(true, Ordering::Relaxed)
582 {
583 let detail = format!(
584 "serving public writes and admin routes without a rate limiter on an \
585 operator's recorded acceptance (reason: {reason}; problems: {})",
586 accepted.join("; ")
587 );
588 tracing::warn!(
589 control = "production-readiness",
590 acceptance = crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES,
591 reason,
592 "serving public writes and admin routes without a rate limiter on an \
593 operator's recorded acceptance"
594 );
595 crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, &detail);
597 }
598 }
599
600 fn nest_sidecars(
603 mut api: Router,
604 mounted: &[SidecarMount],
605 ports: &Ports,
606 gateway: Option<&Arc<HmacSigner>>,
607 ) -> Router {
608 for mount in mounted {
609 api = api.nest(
610 &format!("/v1/{}", mount.name),
611 crate::sidecar::router(
612 mount.clone(),
613 ports.dispatcher.clone(),
614 Arc::clone(&ports.config),
615 ports.rate_limiter.clone(),
616 gateway.map(Arc::clone),
617 ),
618 );
619 }
620 api
621 }
622
623 fn sidecar_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
624 let mounts = match crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()) {
625 Ok(mounts) => mounts,
626 Err(errors) => {
627 for error in errors {
628 let detail = format!("ignoring the sidecar mount table: {error}");
629 tracing::error!(error, "ignoring the sidecar mount table");
630 crate::logging::forward_control_event(
633 crate::logging::ControlLevel::Error,
634 &detail,
635 );
636 }
637 crate::sidecar::SidecarMounts::default()
638 }
639 };
640 let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
641 for collision in mounts.collisions(&module_names) {
642 let detail = format!("ignoring the colliding sidecar mount: {collision}");
643 tracing::error!(error = collision, "ignoring the colliding sidecar mount");
644 crate::logging::forward_control_event(crate::logging::ControlLevel::Error, &detail);
645 }
646 mounts
647 .iter()
648 .filter(|m| !module_names.contains(&m.name.as_str()))
649 .cloned()
650 .collect()
651 }
652
653 fn resolved_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
657 let mounts =
658 crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()).unwrap_or_default();
659 let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
660 mounts
661 .iter()
662 .filter(|m| !module_names.contains(&m.name.as_str()))
663 .cloned()
664 .collect()
665 }
666
667 fn events_for(&self, ports: &Ports) -> EventBus {
675 let mounts = self.resolved_mounts(ports);
676 if mounts.is_empty() {
677 return self.events.clone();
678 }
679 let Some(dispatcher) = ports.dispatcher.clone() else {
680 return self.events.clone();
681 };
682 let gateway = gateway_signer(
683 ports.config.as_ref(),
684 ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
685 );
686 self.events
687 .clone()
688 .forwarding_to(crate::sidecar::EventForwarder::new(
689 mounts, dispatcher, gateway,
690 ))
691 }
692
693 fn ui_router(
698 &self,
699 api: &Router,
700 ports: &Ports,
701 surface: &Arc<dyn SurfaceSource>,
702 ) -> Option<Router> {
703 self.ui.as_ref().map(|ui| {
704 ui.router(UiContext {
705 surface: Arc::clone(surface),
706 api: api.clone(),
707 config: Arc::clone(&ports.config),
708 venture: Arc::clone(&self.venture),
709 captcha_configured: ports.captcha.is_some(),
710 signer: ports.signer.clone(),
711 rate_limiter: ports.rate_limiter.clone(),
712 })
713 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
714 })
715 }
716
717 fn root_router(
723 &self,
724 api: Router,
725 ui: Option<Router>,
726 gateway: Option<Arc<HmacSigner>>,
727 health_state: HealthState,
728 ready_state: ReadyState,
729 surface_state: SurfaceState,
730 ) -> Router {
731 let root = Router::new()
732 .route("/__health", get(health_handler))
733 .with_state(health_state)
734 .route("/__ready", get(ready_handler))
735 .with_state(ready_state)
736 .route("/__surface", get(surface_handler))
737 .with_state(surface_state)
738 .merge(self.inbound_events_route(gateway).unwrap_or_default())
741 .merge(api);
742 let root = match &self.well_known {
743 Some(well_known) => root.nest(
744 "/.well-known",
745 well_known
746 .clone()
747 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
748 ),
749 None => root,
750 };
751 match ui {
752 Some(ui) => root.nest("/ui", ui),
753 None => root,
754 }
755 }
756
757 fn inbound_events_route(&self, gateway: Option<Arc<HmacSigner>>) -> Option<Router> {
771 gateway.as_ref()?;
772 Some(
773 Router::new()
774 .route(
775 "/__events",
776 post(crate::sidecar::events_inbound)
777 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
778 )
779 .with_state(Arc::new(crate::sidecar::InboundEvents {
780 bus: self.events.clone(),
781 gateway,
782 })),
783 )
784 }
785
786 pub fn runtime(&self) -> Option<&Arc<dyn Runtime>> {
788 self.runtime.as_ref()
789 }
790
791 #[must_use]
795 pub fn surface(&self) -> &SurfaceDocument {
796 &self.surface.document
797 }
798
799 fn health_state(&self, ports: &Ports, sidecars: Vec<SidecarMount>) -> HealthState {
813 HealthState {
814 venture: self.venture_as_deployed(ports.config.as_ref()),
819 modules: self.modules.clone(),
820 harness_build: ports.config.get("HARNESS_BUILD").filter(|b| !b.is_empty()),
821 mailer: ports.mailer.clone(),
822 captcha_configured: ports.captcha.is_some(),
823 sidecars,
824 dispatcher: ports.dispatcher.clone(),
825 clock: ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
826 probe_cache: new_probe_cache(),
827 }
828 }
829
830 pub fn router(&self, ports: Ports) -> Router {
831 let env = deployed_env(self.venture.env, ports.config.as_ref());
836 let (limiter, signer) = (ports.rate_limiter.is_some(), ports.signer.is_some());
837 let readiness = self.production_readiness_now(env, ports.config.as_ref(), limiter, signer);
838 let problem_type_base: Arc<str> = self.venture.problem_type_base().into();
842 let mut api = self.nest_modules(&ports);
843 let gateway = gateway_signer(
846 ports.config.as_ref(),
847 ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
848 );
849 let mounted = self.sidecar_mounts(&ports);
850 api = Self::nest_sidecars(api, &mounted, &ports, gateway.as_ref());
851 let gateway_state = Self::gateway_guard_state(&ports, gateway.as_ref());
852 let mounts_for_health = mounted.clone();
853 let tenant_layer = Arc::new(TenantLayer {
855 routing: ports.tenants.clone(),
856 fallback: ports.db.clone(),
857 });
858 let streaming = StreamingState::build(&self.modules);
880 let api = api.layer(from_fn_with_state(streaming, streaming_request_layer));
881 let api = with_abuse_floor(api, ports.rate_limiter.clone(), tenant_layer);
882
883 let surface_source = self.merged_surface(mounted, &ports, gateway.clone(), env);
884
885 let ui = self.ui_router(&api, &ports, &surface_source);
886
887 let health_state = self.health_state(&ports, mounts_for_health);
888 let missing_ports = missing_module_ports(&self.modules, &ports);
892 let Ports {
893 config,
894 db,
895 auth: _,
896 mailer: _,
897 captcha: _,
898 rate_limiter: _,
899 signer: _,
900 kv: _,
901 blob: _,
902 push: _,
903 payments: _,
904 tracker: _,
905 realtime: _,
906 text_model: _,
907 classifier: _,
908 vector_index: _,
909 embedder: _,
910 custom_hostnames: _,
911 http: _,
912 clock,
913 id_gen,
914 defer,
915 actors: _,
916 dispatcher: _,
917 tenants: _,
918 } = ports;
919
920 let scope_state = ScopeState {
921 defer: defer.unwrap_or_else(|| Arc::new(crate::ports::NoopDefer)),
922 id_gen: id_gen.unwrap_or_else(|| Arc::new(crate::ports::UlidIdGen)),
923 module: (self.modules.len() == 1).then(|| self.modules[0].name()),
926 };
927 let ready_state = ReadyState {
928 db,
929 clock: clock.unwrap_or_else(|| Arc::new(SystemClock)),
930 missing_ports,
931 };
932
933 let surface_state = SurfaceState {
934 config,
935 source: surface_source,
936 };
937
938 let root = self.root_router(api, ui, gateway, health_state, ready_state, surface_state);
939
940 root.layer(from_fn_with_state(
949 Arc::new(readiness),
950 production_readiness_guard,
951 ))
952 .layer(from_fn_with_state(gateway_state, gateway_guard))
953 .layer(from_fn_with_state(scope_state, scope_layer))
954 .layer(axum::middleware::from_fn(token_response_layer))
955 .layer(cors_layer(&self.venture.cors_origins))
956 .layer(from_fn_with_state(problem_type_base, problem_type_layer))
965 }
966}
967
968fn module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
976 let path = match path.split_once(['?', '#']) {
977 Some((path, _)) => path,
978 None => path,
979 };
980 let mut segments = path.split('/').filter(|segment| !segment.is_empty());
981 if segments.next() != Some("v1") {
982 return None;
983 }
984 let name = segments.next()?;
985 Some((name, segments.collect()))
986}
987
988fn exact_module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
995 let path = match path.split_once(['?', '#']) {
996 Some((path, _)) => path,
997 None => path,
998 };
999 let mut segments = path.split('/');
1000 if segments.next() != Some("") || segments.next() != Some("v1") {
1002 return None;
1003 }
1004 let name = segments.next()?;
1005 if name.is_empty() {
1006 return None;
1007 }
1008 Some((name, segments.collect()))
1009}
1010
1011fn match_streaming(routes: &[StreamRoute], method: &Method, segments: &[&str]) -> Option<usize> {
1014 routes.iter().find_map(|route| {
1015 (route.method == *method && pattern_matches(route.path, segments))
1016 .then_some(route.max_bytes)
1017 })
1018}
1019
1020fn pattern_matches(pattern: &str, segments: &[&str]) -> bool {
1024 let mut pattern_segments = pattern.split('/').filter(|segment| !segment.is_empty());
1025 let mut rest = segments.iter();
1026 loop {
1027 match pattern_segments.next() {
1028 Some("{*rest}") => {
1029 return pattern_segments.next().is_none() && rest.next().is_some();
1030 }
1031 Some(param) if param.starts_with('{') && param.ends_with('}') => {
1032 if rest.next().is_none() {
1033 return false;
1034 }
1035 }
1036 Some(literal) => {
1037 if rest.next() != Some(&literal) {
1038 return false;
1039 }
1040 }
1041 None => return rest.next().is_none(),
1042 }
1043 }
1044}
1045
1046#[derive(Clone)]
1049struct StreamingState {
1050 modules: Arc<Vec<(&'static str, &'static [StreamRoute])>>,
1051}
1052
1053impl StreamingState {
1054 fn build(modules: &[Arc<dyn Module>]) -> Self {
1055 let routes = modules
1058 .iter()
1059 .filter(|module| !module.streaming_routes().is_empty())
1060 .map(|module| (module.name(), module.streaming_routes()))
1061 .collect();
1062 Self {
1063 modules: Arc::new(routes),
1064 }
1065 }
1066
1067 fn max_bytes(&self, path: &str, method: &Method) -> Option<usize> {
1068 let (name, segments) = exact_module_and_segments(path)?;
1069 let (_, routes) = self.modules.iter().find(|(module, _)| *module == name)?;
1070 match_streaming(routes, method, &segments)
1071 }
1072}
1073
1074async fn streaming_request_layer(
1086 State(state): State<StreamingState>,
1087 mut request: axum::extract::Request,
1088 next: Next,
1089) -> Response {
1090 if state.modules.is_empty() {
1091 return next.run(request).await;
1092 }
1093 let Some(max_bytes) = state.max_bytes(request.uri().path(), request.method()) else {
1094 return next.run(request).await;
1095 };
1096 if request.extensions().get::<RequestStream>().is_some() {
1097 return next.run(request).await;
1098 }
1099 if let Some(declared) = crate::ports::declared_content_length(request.headers())
1102 && declared > max_bytes
1103 {
1104 let problem = Problem::request_too_large();
1105 let problem = match request.extensions().get::<Scope>() {
1106 Some(scope) => problem.instance(&scope.request_id),
1107 None => problem,
1108 };
1109 return problem.into_response();
1110 }
1111 let body = std::mem::replace(request.body_mut(), axum::body::Body::empty());
1112 request
1113 .extensions_mut()
1114 .insert(RequestStream::new(body.into_data_stream(), max_bytes));
1115 next.run(request).await
1116}
1117
1118fn with_abuse_floor(
1123 api: Router,
1124 rate_limiter: Option<Arc<dyn RateLimiter>>,
1125 tenant_layer: Arc<TenantLayer>,
1126) -> Router {
1127 api.layer(from_fn_with_state(rate_limiter, admin_rate_limit_layer))
1128 .layer(axum::middleware::from_fn_with_state(
1129 tenant_layer,
1130 resolve_tenant_layer,
1131 ))
1132 .layer(axum::middleware::from_fn(security_headers_layer))
1133 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
1134}
1135
1136async fn problem_type_layer(
1141 State(base): State<Arc<str>>,
1142 request: axum::extract::Request,
1143 next: Next,
1144) -> Response {
1145 let mut response = next.run(request).await;
1146 let Some(problem) = response.extensions().get::<Problem>().cloned() else {
1147 return response;
1148 };
1149 let body = problem.body(&base).to_string();
1150 *response.body_mut() = axum::body::Body::from(body);
1151 response.headers_mut().remove(header::CONTENT_LENGTH);
1152 response.extensions_mut().remove::<Problem>();
1156 response
1157}
1158
1159type ProbeCache = Option<(i64, Vec<serde_json::Value>)>;
1162
1163fn new_probe_cache() -> Arc<futures_util::lock::Mutex<ProbeCache>> {
1166 Arc::new(futures_util::lock::Mutex::new(None))
1167}
1168
1169#[derive(Clone)]
1170struct HealthState {
1171 venture: Arc<Venture>,
1172 modules: Vec<Arc<dyn Module>>,
1173 harness_build: Option<String>,
1175 mailer: Option<Arc<dyn Mailer>>,
1178 captcha_configured: bool,
1179 sidecars: Vec<SidecarMount>,
1180 dispatcher: Option<Arc<dyn Dispatcher>>,
1181 clock: Arc<dyn Clock>,
1182 probe_cache: Arc<futures_util::lock::Mutex<ProbeCache>>,
1193}
1194
1195const SIDECAR_PROBE_TTL_SECS: i64 = 30;
1198
1199const SIDECAR_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
1202
1203async fn health_handler(State(state): State<HealthState>) -> impl IntoResponse {
1204 let modules: Vec<serde_json::Value> = state
1205 .modules
1206 .iter()
1207 .map(|module| {
1208 json!({
1209 "name": module.name(),
1210 "version": module.version(),
1211 "requires": module.requires().iter().map(Port::name).collect::<Vec<_>>(),
1212 "optional": module.optional().iter().map(Port::name).collect::<Vec<_>>(),
1213 "tables": module.tables(),
1214 "emits": module.emits(),
1215 })
1216 })
1217 .collect();
1218 let sidecars = probe_sidecars(&state).await;
1219 let mailers = state
1223 .mailer
1224 .as_ref()
1225 .map(|mailer| mailer.providers())
1226 .unwrap_or_default();
1227 Json(json!({
1228 "venture": state.venture.name,
1229 "env": state.venture.env.as_str(),
1230 "harness_api": HARNESS_API,
1231 "harness_build": state.harness_build,
1232 "mailer": if state.mailer.is_some() { "configured" } else { "not_configured" },
1238 "captcha": if state.captcha_configured { "configured" } else { "absent" },
1239 "mailers": mailers,
1240 "modules": modules,
1241 "sidecars": sidecars,
1242 }))
1243}
1244
1245async fn probe_sidecars(state: &HealthState) -> Vec<serde_json::Value> {
1257 if state.sidecars.is_empty() {
1258 return Vec::new();
1259 }
1260 let mut cache = state.probe_cache.lock().await;
1261 let now_ms =
1262 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1263 if let Some((at_ms, entries)) = cache.as_ref()
1264 && now_ms - at_ms < SIDECAR_PROBE_TTL_SECS * 1000
1265 {
1266 return entries.clone();
1267 }
1268
1269 let mut entries = Vec::new();
1270 for mount in &state.sidecars {
1271 entries.push(probe_one_sidecar(state, mount).await);
1272 }
1273 let stamped_ms =
1274 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1275 *cache = Some((stamped_ms, entries.clone()));
1276 entries
1277}
1278
1279async fn probe_one_sidecar(state: &HealthState, mount: &SidecarMount) -> serde_json::Value {
1280 let mut missing = json!({
1281 "name": mount.name,
1282 "binding": mount.binding,
1283 "probe": "unreachable",
1284 "contract": serde_json::Value::Null,
1285 "module": serde_json::Value::Null,
1286 "version": serde_json::Value::Null,
1287 });
1288 let Some(dispatcher) = state.dispatcher.as_ref() else {
1289 return missing;
1290 };
1291 let request = http::Request::builder()
1292 .method(http::Method::GET)
1293 .uri("/__health")
1294 .body(Bytes::new());
1295 let Ok(request) = request else {
1296 return missing;
1297 };
1298 let dispatcher = Arc::clone(dispatcher);
1303 let binding = mount.binding.clone();
1304 let probe = async move { dispatcher.dispatch(&binding, request).await };
1305 let Some(Ok(response)) =
1306 crate::ports::timeout(&*state.clock, probe, SIDECAR_PROBE_TIMEOUT).await
1307 else {
1308 return missing;
1309 };
1310 let contract = response
1311 .headers()
1312 .get(crate::sidecar::X_HARNESS_API)
1313 .and_then(|v| v.to_str().ok())
1314 .and_then(|v| v.parse::<u32>().ok());
1315 let body = serde_json::from_slice::<serde_json::Value>(response.body()).ok();
1316 let module = response
1317 .headers()
1318 .get(crate::sidecar::X_HARNESS_MODULE)
1319 .and_then(|v| v.to_str().ok())
1320 .map(str::to_owned)
1321 .or_else(|| first_module_string(body.as_ref(), "name"));
1322 let version = first_module_string(body.as_ref(), "version");
1323 let tables = declared_tables(body.as_ref());
1324 let contract = contract.or_else(|| {
1325 body.as_ref()
1326 .and_then(|b| b.get("harness_api"))
1327 .and_then(serde_json::Value::as_u64)
1328 .and_then(|api| u32::try_from(api).ok())
1329 });
1330 missing["contract"] = contract.map_or(serde_json::Value::Null, |api| json!(api));
1331 missing["module"] = module.map_or(serde_json::Value::Null, |name| json!(name));
1332 missing["version"] = version.map_or(serde_json::Value::Null, |version| json!(version));
1333 missing["tables"] = json!(tables);
1334 match contract {
1335 Some(api) if api != HARNESS_API => missing["probe"] = json!("mismatch"),
1336 Some(_) => missing["probe"] = json!("ok"),
1337 None => missing["probe"] = json!("unreachable"),
1338 }
1339 let clashes = collisions_with(&tables, state, &mount.name);
1351 if !clashes.is_empty() {
1352 let detail = format!(
1353 "sidecar `{}` claims table(s) {} already claimed in this deployment",
1354 mount.name,
1355 clashes.join(", ")
1356 );
1357 tracing::error!(sidecar = %mount.name, tables = %clashes.join(","), "{detail}");
1358 crate::logging::forward_internal_error(&detail);
1359 missing["probe"] = json!("table-collision");
1360 missing["table_collisions"] = json!(clashes);
1361 }
1362 missing
1363}
1364
1365fn declared_tables(body: Option<&serde_json::Value>) -> Vec<String> {
1369 let Some(modules) = body
1370 .and_then(|b| b.get("modules"))
1371 .and_then(|m| m.as_array())
1372 else {
1373 return Vec::new();
1374 };
1375 let mut tables: Vec<String> = modules
1376 .iter()
1377 .filter_map(|module| module.get("tables")?.as_array())
1378 .flatten()
1379 .filter_map(|table| table.as_str().map(str::to_owned))
1380 .collect();
1381 tables.sort_unstable();
1382 tables.dedup();
1383 tables
1384}
1385
1386fn collisions_with(claimed: &[String], state: &HealthState, mount: &str) -> Vec<String> {
1389 let mut clashes: Vec<String> = Vec::new();
1390 for module in &state.modules {
1391 if module.name() == mount {
1392 continue;
1393 }
1394 for table in module.tables() {
1395 if claimed.iter().any(|claimed| claimed == table)
1396 && !clashes.iter().any(|seen| seen == table)
1397 {
1398 clashes.push(format!("`{table}` (also module `{}`)", module.name()));
1399 }
1400 }
1401 }
1402 clashes
1403}
1404
1405fn first_module_string(body: Option<&serde_json::Value>, key: &str) -> Option<String> {
1408 body?
1409 .get("modules")?
1410 .get(0)?
1411 .get(key)?
1412 .as_str()
1413 .map(str::to_owned)
1414}
1415
1416struct TenantLayer {
1419 routing: Option<Arc<dyn crate::tenant::TenantRouting>>,
1420 fallback: Option<Arc<dyn Database>>,
1426}
1427
1428async fn resolve_tenant_layer(
1435 State(layer): State<Arc<TenantLayer>>,
1436 scope: Scope,
1437 mut request: axum::extract::Request,
1438 next: Next,
1439) -> Response {
1440 let host = request
1441 .headers()
1442 .get(header::HOST)
1443 .and_then(|value| value.to_str().ok())
1444 .unwrap_or_default()
1445 .to_owned();
1446
1447 let resolution = match layer.routing.as_ref() {
1452 Some(routing) => routing.resolve(&host),
1453 None => <crate::tenant::ImplicitTenant as crate::tenant::ResolveTenant>::resolve(
1454 &crate::tenant::ImplicitTenant,
1455 &host,
1456 ),
1457 };
1458
1459 let tenant = match resolution.admit() {
1460 Ok(tenant) => tenant,
1461 Err(problem) => {
1462 return Problem::new(problem)
1463 .instance(&scope.request_id)
1464 .into_response();
1465 }
1466 };
1467
1468 let db = match layer.routing.as_ref() {
1469 Some(routing) => match routing.database(&tenant).await {
1470 Ok(db) => Some(db),
1471 Err(err) => {
1472 tracing::error!(tenant = %tenant.id(), "{err}");
1475 crate::logging::forward_internal_error(&err.to_string());
1476 return Problem::new(&SLUGS.tenant_degraded)
1477 .instance(&scope.request_id)
1478 .into_response();
1479 }
1480 },
1481 None => layer.fallback.clone(),
1482 };
1483
1484 if let Some(db) = db {
1485 let tenancy = if layer.routing.is_some() {
1490 crate::tenant::Tenancy::FromRegistry
1491 } else {
1492 crate::tenant::Tenancy::Sole
1493 };
1494 request
1495 .extensions_mut()
1496 .insert(crate::tenant_conn::ResolvedTenant {
1497 tenant,
1498 tenancy,
1499 db,
1500 });
1501 }
1502 next.run(request).await
1503}
1504
1505#[derive(Clone)]
1506struct ReadyState {
1507 db: Option<Arc<dyn Database>>,
1508 clock: Arc<dyn Clock>,
1509 missing_ports: Vec<Port>,
1513}
1514
1515fn missing_module_ports(modules: &[Arc<dyn Module>], ports: &Ports) -> Vec<Port> {
1520 const PROBED: &[Port] = &[Port::VectorIndex, Port::Embedder];
1521 PROBED
1522 .iter()
1523 .copied()
1524 .filter(|port| {
1525 modules
1526 .iter()
1527 .any(|module| module.requires().contains(port))
1528 && !ports.has(*port)
1529 })
1530 .collect()
1531}
1532
1533async fn ready_handler(State(state): State<ReadyState>) -> impl IntoResponse {
1537 if !state.missing_ports.is_empty() {
1538 let detail = state
1539 .missing_ports
1540 .iter()
1541 .map(|port| format!("{} port is not configured", port.name()))
1542 .collect::<Vec<_>>()
1543 .join("; ");
1544 return Problem::not_ready(detail).into_response();
1545 }
1546 let Some(db) = state.db else {
1547 return Problem::not_ready("database port is not configured").into_response();
1548 };
1549 let stmt = Statement::new("SELECT 1");
1550 let query = async move { db.query(&stmt).await };
1551 match crate::ports::timeout(&*state.clock, query, Duration::from_secs(2)).await {
1552 Some(Ok(_rows)) => Json(json!({ "ok": true })).into_response(),
1553 Some(Err(err)) => {
1554 error!(error = %err, "readiness probe query failed");
1555 crate::logging::forward_internal_error(&format!("readiness probe query failed: {err}"));
1556 Problem::not_ready("database query failed").into_response()
1557 }
1558 None => Problem::not_ready("database did not answer within 2 s").into_response(),
1559 }
1560}
1561
1562#[derive(Clone)]
1563struct SurfaceState {
1564 config: Arc<dyn Config>,
1565 source: Arc<dyn SurfaceSource>,
1566}
1567
1568struct MergedSurface {
1575 base: Arc<SurfaceVariants>,
1576 mounts: Vec<SidecarMount>,
1577 dispatcher: Option<Arc<dyn Dispatcher>>,
1578 gateway: Option<Arc<HmacSigner>>,
1581 env: VentureEnv,
1582 captcha_configured: bool,
1583}
1584
1585impl MergedSurface {
1586 async fn sidecar_modules(&self) -> Vec<crate::surface::ModuleSurface> {
1588 let mut extra = Vec::new();
1589 let Some(dispatcher) = &self.dispatcher else {
1590 return extra;
1591 };
1592 for mount in &self.mounts {
1593 if !dispatcher.has(&mount.binding) {
1594 continue;
1595 }
1596 let mut builder = axum::http::Request::builder()
1597 .method(axum::http::Method::GET)
1598 .uri("/__surface")
1599 .header(header::ACCEPT, "application/json");
1600 if let Some(signer) = &self.gateway {
1601 builder =
1603 builder.header(X_HARNESS_GATEWAY, mint_gateway(signer, &mount.name, false));
1604 }
1605 let request = builder
1606 .body(bytes::Bytes::new())
1607 .expect("static request builds");
1608 let answer = match dispatcher.dispatch(&mount.binding, request).await {
1609 Ok(response) if response.status().is_success() => response,
1610 Ok(response) => {
1611 tracing::warn!(module = mount.name, status = %response.status(), "sidecar surface not available");
1612 continue;
1613 }
1614 Err(err) => {
1615 tracing::warn!(module = mount.name, error = %err, "sidecar surface fetch failed");
1616 continue;
1617 }
1618 };
1619 if answer.body().len() > MAX_SIDECAR_SURFACE_BYTES {
1622 tracing::warn!(
1623 module = mount.name,
1624 "sidecar surface exceeds {MAX_SIDECAR_SURFACE_BYTES} bytes and was not merged"
1625 );
1626 continue;
1627 }
1628 match serde_json::from_slice::<SurfaceDocument>(answer.body()) {
1629 Ok(document) => match sanitize_sidecar_document(&document, &mount.name) {
1630 Ok(entries) => extra.extend(entries.into_iter().filter_map(|mut entry| {
1631 if self.env == VentureEnv::Production
1635 && !self.captcha_configured
1636 && strip_unguarded_captcha_actions(&mut entry.surface) > 0
1637 {
1638 tracing::warn!(
1639 module = entry.name,
1640 "merged sidecar actions requiring a captcha were dropped: no Captcha port here"
1641 );
1642 }
1643 (!entry.surface.is_empty()).then_some(entry)
1644 })),
1645 Err(problems) => {
1646 for problem in problems {
1647 tracing::warn!(module = mount.name, error = problem, "sidecar surface rejected at merge");
1648 }
1649 }
1650 },
1651 Err(err) => {
1652 tracing::warn!(module = mount.name, error = %err, "sidecar surface is not a surface document");
1653 }
1654 }
1655 }
1656 extra
1657 }
1658}
1659
1660#[async_trait::async_trait]
1661impl SurfaceSource for MergedSurface {
1662 async fn current(&self) -> Arc<SurfaceDocument> {
1663 if self.mounts.is_empty() {
1664 return Arc::clone(&self.base.document);
1665 }
1666 let mut document = (*self.base.document).clone();
1667 document.modules.extend(self.sidecar_modules().await);
1668 Arc::new(document)
1669 }
1670
1671 fn built(&self) -> Arc<SurfaceDocument> {
1672 Arc::clone(&self.base.document)
1673 }
1674
1675 fn rendered(&self, admin: bool) -> Option<&RenderedSurface> {
1676 Some(if admin {
1677 &self.base.full
1678 } else {
1679 &self.base.public
1680 })
1681 }
1682}
1683
1684async fn surface_handler(
1700 State(state): State<SurfaceState>,
1701 headers: HeaderMap,
1702) -> impl IntoResponse {
1703 let admin = require_admin(&*state.config, &headers).is_ok();
1704 let current = state.source.current().await;
1708 let built = state.source.built();
1709 let prerendered = Arc::ptr_eq(¤t, &built).then(|| state.source.rendered(admin));
1710 let fresh;
1711 let rendered: &RenderedSurface = if let Some(rendered) = prerendered.flatten() {
1712 rendered
1713 } else {
1714 fresh = if admin {
1715 RenderedSurface::render(¤t)
1716 } else {
1717 RenderedSurface::render(¤t.public())
1718 };
1719 &fresh
1720 };
1721 let matches = headers
1722 .get(header::IF_NONE_MATCH)
1723 .and_then(|value| value.to_str().ok())
1724 .is_some_and(|value| {
1725 value
1726 .split(',')
1727 .map(str::trim)
1728 .any(|tag| tag == "*" || tag == rendered.etag)
1729 });
1730 let mut response = if matches {
1731 StatusCode::NOT_MODIFIED.into_response()
1732 } else {
1733 (
1734 [(header::CONTENT_TYPE, "application/json")],
1735 rendered.json.clone(),
1736 )
1737 .into_response()
1738 };
1739 let response_headers = response.headers_mut();
1740 response_headers.insert(
1741 header::ETAG,
1742 header::HeaderValue::from_str(&rendered.etag).expect("hex etag is a valid header"),
1743 );
1744 response_headers.insert(
1745 header::CACHE_CONTROL,
1746 if admin {
1747 header::HeaderValue::from_static("private, no-store")
1748 } else {
1749 header::HeaderValue::from_static("no-cache")
1750 },
1751 );
1752 response_headers.insert(
1753 header::VARY,
1754 header::HeaderValue::from_static("Authorization"),
1755 );
1756 response
1757}
1758
1759#[derive(Default)]
1762pub struct HarnessBuilder {
1763 venture: Option<Venture>,
1764 modules: Vec<Arc<dyn Module>>,
1765 provides: Vec<Port>,
1766 runtime: Option<Arc<dyn Runtime>>,
1767 module_templates: Vec<(String, Box<dyn Template>)>,
1768 overrides: Vec<(String, Box<dyn Template>)>,
1769 ui: Option<Arc<dyn UiMount>>,
1770 actor_handlers: Vec<(String, Arc<dyn ActorHandler>)>,
1771}
1772
1773impl HarnessBuilder {
1774 #[must_use]
1775 pub fn venture(mut self, venture: Venture) -> Self {
1776 self.venture = Some(venture);
1777 self
1778 }
1779
1780 #[must_use]
1783 pub fn module(mut self, module: impl Module) -> Self {
1784 self.modules.push(Arc::new(module));
1785 self
1786 }
1787
1788 #[must_use]
1791 pub fn module_arc(mut self, module: Arc<dyn Module>) -> Self {
1792 self.modules.push(module);
1793 self
1794 }
1795
1796 #[must_use]
1800 pub fn runtime(mut self, runtime: impl Runtime) -> Self {
1801 let runtime: Arc<dyn Runtime> = Arc::new(runtime);
1802 self.provides = runtime.provides();
1803 self.runtime = Some(runtime);
1804 self
1805 }
1806
1807 #[must_use]
1810 pub fn templates(
1811 mut self,
1812 templates: impl IntoIterator<Item = (String, Box<dyn Template>)>,
1813 ) -> Self {
1814 self.module_templates.extend(templates);
1815 self
1816 }
1817
1818 #[must_use]
1821 pub fn ui(mut self, ui: impl UiMount) -> Self {
1822 self.ui = Some(Arc::new(ui));
1823 self
1824 }
1825
1826 #[must_use]
1829 pub fn template(mut self, id: impl Into<String>, template: Box<dyn Template>) -> Self {
1830 self.overrides.push((id.into(), template));
1831 self
1832 }
1833
1834 #[must_use]
1839 pub fn actor(mut self, kind: &str, handler: Arc<dyn ActorHandler>) -> Self {
1840 self.actor_handlers.push((kind.to_owned(), handler));
1841 self
1842 }
1843
1844 pub fn build(self) -> Result<Harness, ConfigError> {
1855 let mut errors = ConfigError::default();
1856
1857 let venture = if let Some(venture) = self.venture {
1858 venture.validate(&mut errors);
1859 venture
1860 } else {
1861 errors.push("missing venture: call .venture(Venture::new(..)) before .build()");
1862 Venture::new("invalid", "invalid.invalid")
1863 };
1864
1865 let mut names: HashMap<&'static str, usize> = HashMap::new();
1866 let mut tables: HashMap<&'static str, &'static str> = HashMap::new();
1867
1868 for module in &self.modules {
1869 if module.harness_api() != HARNESS_API {
1870 errors.push(harness_api_mismatch(module.as_ref()));
1871 }
1872
1873 let name = module.name();
1874 if name.is_empty() || !is_module_name(name) {
1875 errors.push(format!(
1876 "module name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
1877 ));
1878 }
1879 if names.insert(name, 1).is_some() {
1880 errors.push(format!("duplicate module name `{name}`"));
1881 }
1882
1883 for port in module.requires().iter().chain(module.optional()) {
1884 if !Port::ALL.contains(port) {
1885 errors.push(format!(
1886 "module `{name}` declares unknown port {}",
1887 port.name()
1888 ));
1889 }
1890 }
1891 for port in module.requires() {
1892 if module.optional().contains(port) {
1893 errors.push(format!(
1894 "module `{name}` lists port {} in both requires() and optional()",
1895 port.name()
1896 ));
1897 }
1898 }
1899
1900 module.surface().validate(name, &mut errors);
1901
1902 check_tables(module.as_ref(), &mut tables, &mut errors);
1903 }
1904
1905 let actor_handlers = collect_actors(&self.actor_handlers, &self.modules, &mut errors);
1906
1907 let well_known = collect_well_known(&self.modules, &mut errors);
1908
1909 for module in &self.modules {
1910 for port in module.requires() {
1911 if !self.provides.contains(port) {
1912 errors.push(format!(
1913 "module `{}` requires port {} which the runtime does not provide",
1914 module.name(),
1915 port.name()
1916 ));
1917 }
1918 }
1919 warn_undeclared_ports(module.as_ref(), &self.provides);
1920 }
1921
1922 let ordered = resolve_dependency_order(&self.modules, &mut errors);
1923
1924 append_production_readiness(&venture, &self.modules, self.runtime.as_ref(), &mut errors);
1925
1926 check_template_ids(
1927 self.overrides.iter().chain(self.module_templates.iter()),
1928 &names,
1929 &mut errors,
1930 );
1931
1932 let surface = Arc::new(SurfaceVariants::compose(
1933 &venture,
1934 &self.modules,
1935 self.ui.as_ref(),
1936 ));
1937 if let Some(ui) = &self.ui {
1938 ui.validate(&surface.document, &mut errors);
1939 }
1940
1941 errors.into_result()?;
1942
1943 let mut registry = TemplateRegistry::new();
1944 registry.register_all(self.module_templates);
1945 registry.register_all(self.overrides);
1946
1947 let mut events = EventBus::new();
1948 for module in &self.modules {
1949 for (name, handler) in module.events() {
1950 events = events.on(name, handler);
1951 }
1952 }
1953
1954 Ok(Harness {
1955 venture: Arc::new(venture),
1956 unprotected_acceptance_recorded: AtomicBool::new(false),
1957 unlimited_acceptance_recorded: AtomicBool::new(false),
1958 personal_data: Arc::new(crate::PersonalDataCatalog::compose(
1964 ordered.iter().map(|m| (m.name(), m.personal_data())),
1965 )),
1966 modules: ordered,
1967 templates: Arc::new(registry),
1968 events,
1969 runtime: self.runtime,
1970 well_known,
1971 surface,
1972 ui: self.ui,
1973 actor_handlers,
1974 })
1975 }
1976}
1977
1978fn resolve_dependency_order(
1991 modules: &[Arc<dyn Module>],
1992 errors: &mut ConfigError,
1993) -> Vec<Arc<dyn Module>> {
1994 let names: BTreeSet<&str> = modules.iter().map(|module| module.name()).collect();
1995 let mut unknown = false;
1996 for module in modules {
1997 for needed in module.depends_on() {
1998 if !names.contains(needed) {
1999 unknown = true;
2000 errors.push(format!(
2001 "module `{}` depends on `{needed}`, which this venture does not compose: \
2002 add the module or drop the dependency",
2003 module.name()
2004 ));
2005 }
2006 }
2007 }
2008 if unknown {
2009 return modules.to_vec();
2010 }
2011
2012 let mut remaining: Vec<Option<Arc<dyn Module>>> = modules
2016 .iter()
2017 .map(|module| Some(Arc::clone(module)))
2018 .collect();
2019 let mut placed: BTreeSet<&str> = BTreeSet::new();
2020 let mut ordered: Vec<Arc<dyn Module>> = Vec::with_capacity(modules.len());
2021
2022 while ordered.len() < modules.len() {
2023 let mut progressed = false;
2024 for slot in &mut remaining {
2025 let ready = slot.as_ref().is_some_and(|module| {
2026 module
2027 .depends_on()
2028 .iter()
2029 .all(|needed| placed.contains(needed))
2030 });
2031 if ready {
2032 let module = slot.take().expect("checked above");
2033 placed.insert(module.name());
2034 ordered.push(module);
2035 progressed = true;
2036 }
2037 }
2038 if !progressed {
2039 let stuck: Vec<&str> = remaining
2041 .iter()
2042 .filter_map(|slot| slot.as_ref().map(|module| module.name()))
2043 .collect();
2044 errors.push(format!(
2045 "modules [{}] depend on each other in a cycle: migrations cannot be ordered so \
2046 that every module follows the ones it sits on top of",
2047 stuck.join(", ")
2048 ));
2049 return modules.to_vec();
2050 }
2051 }
2052 ordered
2053}
2054
2055fn check_template_ids<'a>(
2056 ids: impl Iterator<Item = &'a (String, Box<dyn Template>)>,
2057 names: &HashMap<&'static str, usize>,
2058 errors: &mut ConfigError,
2059) {
2060 for (id, _) in ids {
2061 let Some(module_name) = id.split('/').next() else {
2062 continue;
2063 };
2064 if !names.contains_key(module_name) {
2065 errors.push(format!(
2066 "template `{id}` names module `{module_name}` which is not registered"
2067 ));
2068 }
2069 }
2070}
2071
2072fn is_module_name(name: &str) -> bool {
2073 name.split('-').all(|part| {
2074 !part.is_empty()
2075 && part
2076 .chars()
2077 .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
2078 })
2079}
2080
2081async fn production_readiness_guard(
2091 State(problems): State<Arc<Vec<String>>>,
2092 request: axum::extract::Request,
2093 next: Next,
2094) -> Response {
2095 if problems.is_empty() || !request.uri().path().starts_with("/v1/") {
2096 return next.run(request).await;
2097 }
2098 let problem = Problem::new(&SLUGS.not_production_ready).with_detail(problems.join("; "));
2099 match request.extensions().get::<Scope>() {
2100 Some(scope) => problem.instance(&scope.request_id.clone()),
2101 None => problem,
2102 }
2103 .into_response()
2104}
2105
2106async fn admin_rate_limit_layer(
2115 State(limiter): State<Option<Arc<dyn RateLimiter>>>,
2116 request: axum::extract::Request,
2117 next: Next,
2118) -> Response {
2119 if !crate::sidecar::is_admin_path(request.uri().path()) {
2120 return next.run(request).await;
2121 }
2122 let keys: Vec<String> = crate::rate_limit::rate_limit_keys(
2123 crate::rate_limit::client_ip(request.headers()).as_deref(),
2124 None,
2125 )
2126 .into_iter()
2127 .map(|key| format!("admin:{key}"))
2128 .collect();
2129 if let RateLimit::Denied { decision } =
2130 crate::rate_limit::check_rate_limit(limiter.as_ref(), &keys, RateLimitFailure::FailClosed)
2131 .await
2132 {
2133 return rate_limited(&decision);
2134 }
2135 next.run(request).await
2136}
2137
2138fn append_production_readiness(
2150 venture: &Venture,
2151 modules: &[Arc<dyn Module>],
2152 runtime: Option<&Arc<dyn Runtime>>,
2153 errors: &mut ConfigError,
2154) {
2155 for error in crate::route_policy::production_readiness(
2156 venture.env,
2157 &crate::route_policy::WriteGuards::collect(modules),
2158 runtime,
2159 crate::route_policy::rate_limiter_effective(runtime),
2165 crate::route_policy::signer_effective(runtime),
2166 None,
2167 None,
2168 ) {
2169 errors.push(error);
2170 }
2171}
2172
2173fn collect_well_known(modules: &[Arc<dyn Module>], errors: &mut ConfigError) -> Option<Router> {
2174 let mut providers: Vec<&'static str> = Vec::new();
2175 let mut well_known = None;
2176 for module in modules {
2177 if let Some(router) = module.well_known() {
2178 providers.push(module.name());
2179 well_known = Some(router);
2180 }
2181 }
2182 if providers.len() > 1 {
2183 let listed = providers
2184 .iter()
2185 .map(|name| format!("`{name}`"))
2186 .collect::<Vec<_>>()
2187 .join(", ");
2188 errors.push(format!(
2189 "modules {listed} all provide a well-known router; at most one module may occupy /.well-known"
2190 ));
2191 }
2192 well_known
2193}
2194
2195#[cfg(test)]
2196mod acceptance_recording {
2197 use super::*;
2203 use crate::Migrations;
2204 use crate::config::MapConfig;
2205
2206 struct PubWriter;
2209
2210 impl Module for PubWriter {
2211 fn name(&self) -> &'static str {
2212 "writer"
2213 }
2214 fn version(&self) -> &'static str {
2215 "0.0.0-test"
2216 }
2217 fn requires(&self) -> &'static [Port] {
2218 &[]
2219 }
2220 fn public_writes(&self) -> bool {
2221 true
2222 }
2223 fn migrations(&self) -> Migrations {
2224 Migrations::default()
2225 }
2226 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2227 Ok(())
2228 }
2229 fn router(&self, _: ModuleContext) -> Router {
2230 Router::new()
2231 }
2232 }
2233
2234 struct AllPorts;
2237
2238 impl Runtime for AllPorts {
2239 fn provides(&self) -> Vec<Port> {
2240 Port::ALL.to_vec()
2241 }
2242 }
2243
2244 fn harness() -> Harness {
2245 Harness::builder()
2246 .venture(
2247 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2248 )
2249 .module(PubWriter)
2250 .runtime(AllPorts)
2251 .build()
2252 .expect("a development venture builds")
2253 }
2254
2255 fn deployed(unlimited: Option<&str>) -> MapConfig {
2256 let pairs = [("ENV", "production")].into_iter().chain(
2257 unlimited.map(|reason| (crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES, reason)),
2258 );
2259 MapConfig::from_pairs(pairs)
2260 }
2261
2262 #[test]
2263 fn the_waiver_is_recorded_once_and_only_when_it_waived() {
2264 let harness = harness();
2265 let config = deployed(Some("issue #437: limiter binding pending"));
2266 assert!(
2267 harness
2268 .production_readiness_now(VentureEnv::Production, &config, false, true)
2269 .is_empty(),
2270 "the recorded acceptance serves"
2271 );
2272 assert!(
2273 harness
2274 .unlimited_acceptance_recorded
2275 .load(Ordering::Relaxed)
2276 );
2277
2278 assert!(
2281 harness
2282 .production_readiness_now(VentureEnv::Production, &config, false, true)
2283 .is_empty()
2284 );
2285 }
2286
2287 #[test]
2288 fn nothing_is_recorded_when_the_port_was_fine_all_along() {
2289 let harness = harness();
2290 let config = deployed(Some("unused: the limiter resolved"));
2291 assert!(
2292 harness
2293 .production_readiness_now(VentureEnv::Production, &config, true, true)
2294 .is_empty()
2295 );
2296 assert!(
2297 !harness
2298 .unlimited_acceptance_recorded
2299 .load(Ordering::Relaxed),
2300 "a waiver that removed no refusal is not a recorded acceptance"
2301 );
2302 }
2303
2304 #[test]
2305 fn a_refusing_deployment_records_no_acceptance_at_all() {
2306 let harness = harness();
2307 let config = MapConfig::from_pairs([
2311 ("ENV", "production"),
2312 (
2313 crate::route_policy::ALLOW_UNPROTECTED_WRITES,
2314 "captcha pending",
2315 ),
2316 ]);
2317 assert_eq!(
2318 harness
2319 .production_readiness_now(VentureEnv::Production, &config, false, true)
2320 .len(),
2321 1
2322 );
2323 assert!(
2324 !harness
2325 .unprotected_acceptance_recorded
2326 .load(Ordering::Relaxed)
2327 );
2328 assert!(
2329 !harness
2330 .unlimited_acceptance_recorded
2331 .load(Ordering::Relaxed)
2332 );
2333 }
2334
2335 struct HmacWriter;
2339
2340 impl Module for HmacWriter {
2341 fn name(&self) -> &'static str {
2342 "webhooks"
2343 }
2344 fn version(&self) -> &'static str {
2345 "0.0.0-test"
2346 }
2347 fn requires(&self) -> &'static [Port] {
2348 &[]
2349 }
2350 fn public_writes(&self) -> bool {
2351 true
2352 }
2353 fn public_write_policy(&self) -> crate::route_policy::RoutePolicy {
2354 crate::route_policy::RoutePolicy::Signature
2355 }
2356 fn signature_verification(&self) -> crate::route_policy::SignatureVerification {
2357 crate::route_policy::SignatureVerification::Hmac {
2358 secret: "WEBHOOK_SECRET",
2359 }
2360 }
2361 fn migrations(&self) -> Migrations {
2362 Migrations::default()
2363 }
2364 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2365 Ok(())
2366 }
2367 fn router(&self, _: ModuleContext) -> Router {
2368 Router::new()
2369 }
2370 }
2371
2372 #[test]
2373 fn a_non_payments_signature_venture_boots_on_its_webhook_secret_alone() {
2374 let harness = Harness::builder()
2375 .venture(
2376 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2377 )
2378 .module(HmacWriter)
2379 .runtime(AllPorts)
2380 .build()
2381 .expect("a development venture builds");
2382 let deployed = MapConfig::from_pairs([("ENV", "production")]);
2383 let problems =
2384 harness.production_readiness_now(VentureEnv::Production, &deployed, true, true);
2385 assert_eq!(problems.len(), 1, "{problems:?}");
2386 assert!(
2387 problems[0].contains("WEBHOOKS_WEBHOOK_SECRET"),
2388 "{}",
2389 problems[0]
2390 );
2391 assert!(
2392 !problems[0].contains("Payments"),
2393 "a module verifying through the HMAC scheme is not the Payments leg's business: {}",
2394 problems[0]
2395 );
2396
2397 let ready = MapConfig::from_pairs([
2399 ("ENV", "production"),
2400 ("WEBHOOKS_WEBHOOK_SECRET", "whsec-test-dummy"),
2401 ]);
2402 assert!(
2403 harness
2404 .production_readiness_now(VentureEnv::Production, &ready, true, true)
2405 .is_empty()
2406 );
2407 }
2408}