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, Port, Ports, RateLimiter, Statement, SystemClock,
38 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_configured: ports.mailer.is_some(),
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_configured: bool,
1176 captcha_configured: bool,
1177 sidecars: Vec<SidecarMount>,
1178 dispatcher: Option<Arc<dyn Dispatcher>>,
1179 clock: Arc<dyn Clock>,
1180 probe_cache: Arc<futures_util::lock::Mutex<ProbeCache>>,
1191}
1192
1193const SIDECAR_PROBE_TTL_SECS: i64 = 30;
1196
1197const SIDECAR_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
1200
1201async fn health_handler(State(state): State<HealthState>) -> impl IntoResponse {
1202 let modules: Vec<serde_json::Value> = state
1203 .modules
1204 .iter()
1205 .map(|module| {
1206 json!({
1207 "name": module.name(),
1208 "version": module.version(),
1209 "requires": module.requires().iter().map(Port::name).collect::<Vec<_>>(),
1210 "optional": module.optional().iter().map(Port::name).collect::<Vec<_>>(),
1211 "tables": module.tables(),
1212 "emits": module.emits(),
1213 })
1214 })
1215 .collect();
1216 let sidecars = probe_sidecars(&state).await;
1217 Json(json!({
1218 "venture": state.venture.name,
1219 "env": state.venture.env.as_str(),
1220 "harness_api": HARNESS_API,
1221 "harness_build": state.harness_build,
1222 "mailer": if state.mailer_configured { "configured" } else { "not_configured" },
1225 "captcha": if state.captcha_configured { "configured" } else { "absent" },
1226 "modules": modules,
1227 "sidecars": sidecars,
1228 }))
1229}
1230
1231async fn probe_sidecars(state: &HealthState) -> Vec<serde_json::Value> {
1243 if state.sidecars.is_empty() {
1244 return Vec::new();
1245 }
1246 let mut cache = state.probe_cache.lock().await;
1247 let now_ms =
1248 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1249 if let Some((at_ms, entries)) = cache.as_ref()
1250 && now_ms - at_ms < SIDECAR_PROBE_TTL_SECS * 1000
1251 {
1252 return entries.clone();
1253 }
1254
1255 let mut entries = Vec::new();
1256 for mount in &state.sidecars {
1257 entries.push(probe_one_sidecar(state, mount).await);
1258 }
1259 let stamped_ms =
1260 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1261 *cache = Some((stamped_ms, entries.clone()));
1262 entries
1263}
1264
1265async fn probe_one_sidecar(state: &HealthState, mount: &SidecarMount) -> serde_json::Value {
1266 let mut missing = json!({
1267 "name": mount.name,
1268 "binding": mount.binding,
1269 "probe": "unreachable",
1270 "contract": serde_json::Value::Null,
1271 "module": serde_json::Value::Null,
1272 "version": serde_json::Value::Null,
1273 });
1274 let Some(dispatcher) = state.dispatcher.as_ref() else {
1275 return missing;
1276 };
1277 let request = http::Request::builder()
1278 .method(http::Method::GET)
1279 .uri("/__health")
1280 .body(Bytes::new());
1281 let Ok(request) = request else {
1282 return missing;
1283 };
1284 let dispatcher = Arc::clone(dispatcher);
1289 let binding = mount.binding.clone();
1290 let probe = async move { dispatcher.dispatch(&binding, request).await };
1291 let Some(Ok(response)) =
1292 crate::ports::timeout(&*state.clock, probe, SIDECAR_PROBE_TIMEOUT).await
1293 else {
1294 return missing;
1295 };
1296 let contract = response
1297 .headers()
1298 .get(crate::sidecar::X_HARNESS_API)
1299 .and_then(|v| v.to_str().ok())
1300 .and_then(|v| v.parse::<u32>().ok());
1301 let body = serde_json::from_slice::<serde_json::Value>(response.body()).ok();
1302 let module = response
1303 .headers()
1304 .get(crate::sidecar::X_HARNESS_MODULE)
1305 .and_then(|v| v.to_str().ok())
1306 .map(str::to_owned)
1307 .or_else(|| first_module_string(body.as_ref(), "name"));
1308 let version = first_module_string(body.as_ref(), "version");
1309 let tables = declared_tables(body.as_ref());
1310 let contract = contract.or_else(|| {
1311 body.as_ref()
1312 .and_then(|b| b.get("harness_api"))
1313 .and_then(serde_json::Value::as_u64)
1314 .and_then(|api| u32::try_from(api).ok())
1315 });
1316 missing["contract"] = contract.map_or(serde_json::Value::Null, |api| json!(api));
1317 missing["module"] = module.map_or(serde_json::Value::Null, |name| json!(name));
1318 missing["version"] = version.map_or(serde_json::Value::Null, |version| json!(version));
1319 missing["tables"] = json!(tables);
1320 match contract {
1321 Some(api) if api != HARNESS_API => missing["probe"] = json!("mismatch"),
1322 Some(_) => missing["probe"] = json!("ok"),
1323 None => missing["probe"] = json!("unreachable"),
1324 }
1325 let clashes = collisions_with(&tables, state, &mount.name);
1337 if !clashes.is_empty() {
1338 let detail = format!(
1339 "sidecar `{}` claims table(s) {} already claimed in this deployment",
1340 mount.name,
1341 clashes.join(", ")
1342 );
1343 tracing::error!(sidecar = %mount.name, tables = %clashes.join(","), "{detail}");
1344 crate::logging::forward_internal_error(&detail);
1345 missing["probe"] = json!("table-collision");
1346 missing["table_collisions"] = json!(clashes);
1347 }
1348 missing
1349}
1350
1351fn declared_tables(body: Option<&serde_json::Value>) -> Vec<String> {
1355 let Some(modules) = body
1356 .and_then(|b| b.get("modules"))
1357 .and_then(|m| m.as_array())
1358 else {
1359 return Vec::new();
1360 };
1361 let mut tables: Vec<String> = modules
1362 .iter()
1363 .filter_map(|module| module.get("tables")?.as_array())
1364 .flatten()
1365 .filter_map(|table| table.as_str().map(str::to_owned))
1366 .collect();
1367 tables.sort_unstable();
1368 tables.dedup();
1369 tables
1370}
1371
1372fn collisions_with(claimed: &[String], state: &HealthState, mount: &str) -> Vec<String> {
1375 let mut clashes: Vec<String> = Vec::new();
1376 for module in &state.modules {
1377 if module.name() == mount {
1378 continue;
1379 }
1380 for table in module.tables() {
1381 if claimed.iter().any(|claimed| claimed == table)
1382 && !clashes.iter().any(|seen| seen == table)
1383 {
1384 clashes.push(format!("`{table}` (also module `{}`)", module.name()));
1385 }
1386 }
1387 }
1388 clashes
1389}
1390
1391fn first_module_string(body: Option<&serde_json::Value>, key: &str) -> Option<String> {
1394 body?
1395 .get("modules")?
1396 .get(0)?
1397 .get(key)?
1398 .as_str()
1399 .map(str::to_owned)
1400}
1401
1402struct TenantLayer {
1405 routing: Option<Arc<dyn crate::tenant::TenantRouting>>,
1406 fallback: Option<Arc<dyn Database>>,
1412}
1413
1414async fn resolve_tenant_layer(
1421 State(layer): State<Arc<TenantLayer>>,
1422 scope: Scope,
1423 mut request: axum::extract::Request,
1424 next: Next,
1425) -> Response {
1426 let host = request
1427 .headers()
1428 .get(header::HOST)
1429 .and_then(|value| value.to_str().ok())
1430 .unwrap_or_default()
1431 .to_owned();
1432
1433 let resolution = match layer.routing.as_ref() {
1438 Some(routing) => routing.resolve(&host),
1439 None => <crate::tenant::ImplicitTenant as crate::tenant::ResolveTenant>::resolve(
1440 &crate::tenant::ImplicitTenant,
1441 &host,
1442 ),
1443 };
1444
1445 let tenant = match resolution.admit() {
1446 Ok(tenant) => tenant,
1447 Err(problem) => {
1448 return Problem::new(problem)
1449 .instance(&scope.request_id)
1450 .into_response();
1451 }
1452 };
1453
1454 let db = match layer.routing.as_ref() {
1455 Some(routing) => match routing.database(&tenant).await {
1456 Ok(db) => Some(db),
1457 Err(err) => {
1458 tracing::error!(tenant = %tenant.id(), "{err}");
1461 crate::logging::forward_internal_error(&err.to_string());
1462 return Problem::new(&SLUGS.tenant_degraded)
1463 .instance(&scope.request_id)
1464 .into_response();
1465 }
1466 },
1467 None => layer.fallback.clone(),
1468 };
1469
1470 if let Some(db) = db {
1471 let tenancy = if layer.routing.is_some() {
1476 crate::tenant::Tenancy::FromRegistry
1477 } else {
1478 crate::tenant::Tenancy::Sole
1479 };
1480 request
1481 .extensions_mut()
1482 .insert(crate::tenant_conn::ResolvedTenant {
1483 tenant,
1484 tenancy,
1485 db,
1486 });
1487 }
1488 next.run(request).await
1489}
1490
1491#[derive(Clone)]
1492struct ReadyState {
1493 db: Option<Arc<dyn Database>>,
1494 clock: Arc<dyn Clock>,
1495 missing_ports: Vec<Port>,
1499}
1500
1501fn missing_module_ports(modules: &[Arc<dyn Module>], ports: &Ports) -> Vec<Port> {
1506 const PROBED: &[Port] = &[Port::VectorIndex, Port::Embedder];
1507 PROBED
1508 .iter()
1509 .copied()
1510 .filter(|port| {
1511 modules
1512 .iter()
1513 .any(|module| module.requires().contains(port))
1514 && !ports.has(*port)
1515 })
1516 .collect()
1517}
1518
1519async fn ready_handler(State(state): State<ReadyState>) -> impl IntoResponse {
1523 if !state.missing_ports.is_empty() {
1524 let detail = state
1525 .missing_ports
1526 .iter()
1527 .map(|port| format!("{} port is not configured", port.name()))
1528 .collect::<Vec<_>>()
1529 .join("; ");
1530 return Problem::not_ready(detail).into_response();
1531 }
1532 let Some(db) = state.db else {
1533 return Problem::not_ready("database port is not configured").into_response();
1534 };
1535 let stmt = Statement::new("SELECT 1");
1536 let query = async move { db.query(&stmt).await };
1537 match crate::ports::timeout(&*state.clock, query, Duration::from_secs(2)).await {
1538 Some(Ok(_rows)) => Json(json!({ "ok": true })).into_response(),
1539 Some(Err(err)) => {
1540 error!(error = %err, "readiness probe query failed");
1541 crate::logging::forward_internal_error(&format!("readiness probe query failed: {err}"));
1542 Problem::not_ready("database query failed").into_response()
1543 }
1544 None => Problem::not_ready("database did not answer within 2 s").into_response(),
1545 }
1546}
1547
1548#[derive(Clone)]
1549struct SurfaceState {
1550 config: Arc<dyn Config>,
1551 source: Arc<dyn SurfaceSource>,
1552}
1553
1554struct MergedSurface {
1561 base: Arc<SurfaceVariants>,
1562 mounts: Vec<SidecarMount>,
1563 dispatcher: Option<Arc<dyn Dispatcher>>,
1564 gateway: Option<Arc<HmacSigner>>,
1567 env: VentureEnv,
1568 captcha_configured: bool,
1569}
1570
1571impl MergedSurface {
1572 async fn sidecar_modules(&self) -> Vec<crate::surface::ModuleSurface> {
1574 let mut extra = Vec::new();
1575 let Some(dispatcher) = &self.dispatcher else {
1576 return extra;
1577 };
1578 for mount in &self.mounts {
1579 if !dispatcher.has(&mount.binding) {
1580 continue;
1581 }
1582 let mut builder = axum::http::Request::builder()
1583 .method(axum::http::Method::GET)
1584 .uri("/__surface")
1585 .header(header::ACCEPT, "application/json");
1586 if let Some(signer) = &self.gateway {
1587 builder =
1589 builder.header(X_HARNESS_GATEWAY, mint_gateway(signer, &mount.name, false));
1590 }
1591 let request = builder
1592 .body(bytes::Bytes::new())
1593 .expect("static request builds");
1594 let answer = match dispatcher.dispatch(&mount.binding, request).await {
1595 Ok(response) if response.status().is_success() => response,
1596 Ok(response) => {
1597 tracing::warn!(module = mount.name, status = %response.status(), "sidecar surface not available");
1598 continue;
1599 }
1600 Err(err) => {
1601 tracing::warn!(module = mount.name, error = %err, "sidecar surface fetch failed");
1602 continue;
1603 }
1604 };
1605 if answer.body().len() > MAX_SIDECAR_SURFACE_BYTES {
1608 tracing::warn!(
1609 module = mount.name,
1610 "sidecar surface exceeds {MAX_SIDECAR_SURFACE_BYTES} bytes and was not merged"
1611 );
1612 continue;
1613 }
1614 match serde_json::from_slice::<SurfaceDocument>(answer.body()) {
1615 Ok(document) => match sanitize_sidecar_document(&document, &mount.name) {
1616 Ok(entries) => extra.extend(entries.into_iter().filter_map(|mut entry| {
1617 if self.env == VentureEnv::Production
1621 && !self.captcha_configured
1622 && strip_unguarded_captcha_actions(&mut entry.surface) > 0
1623 {
1624 tracing::warn!(
1625 module = entry.name,
1626 "merged sidecar actions requiring a captcha were dropped: no Captcha port here"
1627 );
1628 }
1629 (!entry.surface.is_empty()).then_some(entry)
1630 })),
1631 Err(problems) => {
1632 for problem in problems {
1633 tracing::warn!(module = mount.name, error = problem, "sidecar surface rejected at merge");
1634 }
1635 }
1636 },
1637 Err(err) => {
1638 tracing::warn!(module = mount.name, error = %err, "sidecar surface is not a surface document");
1639 }
1640 }
1641 }
1642 extra
1643 }
1644}
1645
1646#[async_trait::async_trait]
1647impl SurfaceSource for MergedSurface {
1648 async fn current(&self) -> Arc<SurfaceDocument> {
1649 if self.mounts.is_empty() {
1650 return Arc::clone(&self.base.document);
1651 }
1652 let mut document = (*self.base.document).clone();
1653 document.modules.extend(self.sidecar_modules().await);
1654 Arc::new(document)
1655 }
1656
1657 fn built(&self) -> Arc<SurfaceDocument> {
1658 Arc::clone(&self.base.document)
1659 }
1660
1661 fn rendered(&self, admin: bool) -> Option<&RenderedSurface> {
1662 Some(if admin {
1663 &self.base.full
1664 } else {
1665 &self.base.public
1666 })
1667 }
1668}
1669
1670async fn surface_handler(
1686 State(state): State<SurfaceState>,
1687 headers: HeaderMap,
1688) -> impl IntoResponse {
1689 let admin = require_admin(&*state.config, &headers).is_ok();
1690 let current = state.source.current().await;
1694 let built = state.source.built();
1695 let prerendered = Arc::ptr_eq(¤t, &built).then(|| state.source.rendered(admin));
1696 let fresh;
1697 let rendered: &RenderedSurface = if let Some(rendered) = prerendered.flatten() {
1698 rendered
1699 } else {
1700 fresh = if admin {
1701 RenderedSurface::render(¤t)
1702 } else {
1703 RenderedSurface::render(¤t.public())
1704 };
1705 &fresh
1706 };
1707 let matches = headers
1708 .get(header::IF_NONE_MATCH)
1709 .and_then(|value| value.to_str().ok())
1710 .is_some_and(|value| {
1711 value
1712 .split(',')
1713 .map(str::trim)
1714 .any(|tag| tag == "*" || tag == rendered.etag)
1715 });
1716 let mut response = if matches {
1717 StatusCode::NOT_MODIFIED.into_response()
1718 } else {
1719 (
1720 [(header::CONTENT_TYPE, "application/json")],
1721 rendered.json.clone(),
1722 )
1723 .into_response()
1724 };
1725 let response_headers = response.headers_mut();
1726 response_headers.insert(
1727 header::ETAG,
1728 header::HeaderValue::from_str(&rendered.etag).expect("hex etag is a valid header"),
1729 );
1730 response_headers.insert(
1731 header::CACHE_CONTROL,
1732 if admin {
1733 header::HeaderValue::from_static("private, no-store")
1734 } else {
1735 header::HeaderValue::from_static("no-cache")
1736 },
1737 );
1738 response_headers.insert(
1739 header::VARY,
1740 header::HeaderValue::from_static("Authorization"),
1741 );
1742 response
1743}
1744
1745#[derive(Default)]
1748pub struct HarnessBuilder {
1749 venture: Option<Venture>,
1750 modules: Vec<Arc<dyn Module>>,
1751 provides: Vec<Port>,
1752 runtime: Option<Arc<dyn Runtime>>,
1753 module_templates: Vec<(String, Box<dyn Template>)>,
1754 overrides: Vec<(String, Box<dyn Template>)>,
1755 ui: Option<Arc<dyn UiMount>>,
1756 actor_handlers: Vec<(String, Arc<dyn ActorHandler>)>,
1757}
1758
1759impl HarnessBuilder {
1760 #[must_use]
1761 pub fn venture(mut self, venture: Venture) -> Self {
1762 self.venture = Some(venture);
1763 self
1764 }
1765
1766 #[must_use]
1769 pub fn module(mut self, module: impl Module) -> Self {
1770 self.modules.push(Arc::new(module));
1771 self
1772 }
1773
1774 #[must_use]
1777 pub fn module_arc(mut self, module: Arc<dyn Module>) -> Self {
1778 self.modules.push(module);
1779 self
1780 }
1781
1782 #[must_use]
1786 pub fn runtime(mut self, runtime: impl Runtime) -> Self {
1787 let runtime: Arc<dyn Runtime> = Arc::new(runtime);
1788 self.provides = runtime.provides();
1789 self.runtime = Some(runtime);
1790 self
1791 }
1792
1793 #[must_use]
1796 pub fn templates(
1797 mut self,
1798 templates: impl IntoIterator<Item = (String, Box<dyn Template>)>,
1799 ) -> Self {
1800 self.module_templates.extend(templates);
1801 self
1802 }
1803
1804 #[must_use]
1807 pub fn ui(mut self, ui: impl UiMount) -> Self {
1808 self.ui = Some(Arc::new(ui));
1809 self
1810 }
1811
1812 #[must_use]
1815 pub fn template(mut self, id: impl Into<String>, template: Box<dyn Template>) -> Self {
1816 self.overrides.push((id.into(), template));
1817 self
1818 }
1819
1820 #[must_use]
1825 pub fn actor(mut self, kind: &str, handler: Arc<dyn ActorHandler>) -> Self {
1826 self.actor_handlers.push((kind.to_owned(), handler));
1827 self
1828 }
1829
1830 pub fn build(self) -> Result<Harness, ConfigError> {
1841 let mut errors = ConfigError::default();
1842
1843 let venture = if let Some(venture) = self.venture {
1844 venture.validate(&mut errors);
1845 venture
1846 } else {
1847 errors.push("missing venture: call .venture(Venture::new(..)) before .build()");
1848 Venture::new("invalid", "invalid.invalid")
1849 };
1850
1851 let mut names: HashMap<&'static str, usize> = HashMap::new();
1852 let mut tables: HashMap<&'static str, &'static str> = HashMap::new();
1853
1854 for module in &self.modules {
1855 if module.harness_api() != HARNESS_API {
1856 errors.push(harness_api_mismatch(module.as_ref()));
1857 }
1858
1859 let name = module.name();
1860 if name.is_empty() || !is_module_name(name) {
1861 errors.push(format!(
1862 "module name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
1863 ));
1864 }
1865 if names.insert(name, 1).is_some() {
1866 errors.push(format!("duplicate module name `{name}`"));
1867 }
1868
1869 for port in module.requires().iter().chain(module.optional()) {
1870 if !Port::ALL.contains(port) {
1871 errors.push(format!(
1872 "module `{name}` declares unknown port {}",
1873 port.name()
1874 ));
1875 }
1876 }
1877 for port in module.requires() {
1878 if module.optional().contains(port) {
1879 errors.push(format!(
1880 "module `{name}` lists port {} in both requires() and optional()",
1881 port.name()
1882 ));
1883 }
1884 }
1885
1886 module.surface().validate(name, &mut errors);
1887
1888 check_tables(module.as_ref(), &mut tables, &mut errors);
1889 }
1890
1891 let actor_handlers = collect_actors(&self.actor_handlers, &self.modules, &mut errors);
1892
1893 let well_known = collect_well_known(&self.modules, &mut errors);
1894
1895 for module in &self.modules {
1896 for port in module.requires() {
1897 if !self.provides.contains(port) {
1898 errors.push(format!(
1899 "module `{}` requires port {} which the runtime does not provide",
1900 module.name(),
1901 port.name()
1902 ));
1903 }
1904 }
1905 warn_undeclared_ports(module.as_ref(), &self.provides);
1906 }
1907
1908 let ordered = resolve_dependency_order(&self.modules, &mut errors);
1909
1910 append_production_readiness(&venture, &self.modules, self.runtime.as_ref(), &mut errors);
1911
1912 check_template_ids(
1913 self.overrides.iter().chain(self.module_templates.iter()),
1914 &names,
1915 &mut errors,
1916 );
1917
1918 let surface = Arc::new(SurfaceVariants::compose(
1919 &venture,
1920 &self.modules,
1921 self.ui.as_ref(),
1922 ));
1923 if let Some(ui) = &self.ui {
1924 ui.validate(&surface.document, &mut errors);
1925 }
1926
1927 errors.into_result()?;
1928
1929 let mut registry = TemplateRegistry::new();
1930 registry.register_all(self.module_templates);
1931 registry.register_all(self.overrides);
1932
1933 let mut events = EventBus::new();
1934 for module in &self.modules {
1935 for (name, handler) in module.events() {
1936 events = events.on(name, handler);
1937 }
1938 }
1939
1940 Ok(Harness {
1941 venture: Arc::new(venture),
1942 unprotected_acceptance_recorded: AtomicBool::new(false),
1943 unlimited_acceptance_recorded: AtomicBool::new(false),
1944 personal_data: Arc::new(crate::PersonalDataCatalog::compose(
1950 ordered.iter().map(|m| (m.name(), m.personal_data())),
1951 )),
1952 modules: ordered,
1953 templates: Arc::new(registry),
1954 events,
1955 runtime: self.runtime,
1956 well_known,
1957 surface,
1958 ui: self.ui,
1959 actor_handlers,
1960 })
1961 }
1962}
1963
1964fn resolve_dependency_order(
1977 modules: &[Arc<dyn Module>],
1978 errors: &mut ConfigError,
1979) -> Vec<Arc<dyn Module>> {
1980 let names: BTreeSet<&str> = modules.iter().map(|module| module.name()).collect();
1981 let mut unknown = false;
1982 for module in modules {
1983 for needed in module.depends_on() {
1984 if !names.contains(needed) {
1985 unknown = true;
1986 errors.push(format!(
1987 "module `{}` depends on `{needed}`, which this venture does not compose: \
1988 add the module or drop the dependency",
1989 module.name()
1990 ));
1991 }
1992 }
1993 }
1994 if unknown {
1995 return modules.to_vec();
1996 }
1997
1998 let mut remaining: Vec<Option<Arc<dyn Module>>> = modules
2002 .iter()
2003 .map(|module| Some(Arc::clone(module)))
2004 .collect();
2005 let mut placed: BTreeSet<&str> = BTreeSet::new();
2006 let mut ordered: Vec<Arc<dyn Module>> = Vec::with_capacity(modules.len());
2007
2008 while ordered.len() < modules.len() {
2009 let mut progressed = false;
2010 for slot in &mut remaining {
2011 let ready = slot.as_ref().is_some_and(|module| {
2012 module
2013 .depends_on()
2014 .iter()
2015 .all(|needed| placed.contains(needed))
2016 });
2017 if ready {
2018 let module = slot.take().expect("checked above");
2019 placed.insert(module.name());
2020 ordered.push(module);
2021 progressed = true;
2022 }
2023 }
2024 if !progressed {
2025 let stuck: Vec<&str> = remaining
2027 .iter()
2028 .filter_map(|slot| slot.as_ref().map(|module| module.name()))
2029 .collect();
2030 errors.push(format!(
2031 "modules [{}] depend on each other in a cycle: migrations cannot be ordered so \
2032 that every module follows the ones it sits on top of",
2033 stuck.join(", ")
2034 ));
2035 return modules.to_vec();
2036 }
2037 }
2038 ordered
2039}
2040
2041fn check_template_ids<'a>(
2042 ids: impl Iterator<Item = &'a (String, Box<dyn Template>)>,
2043 names: &HashMap<&'static str, usize>,
2044 errors: &mut ConfigError,
2045) {
2046 for (id, _) in ids {
2047 let Some(module_name) = id.split('/').next() else {
2048 continue;
2049 };
2050 if !names.contains_key(module_name) {
2051 errors.push(format!(
2052 "template `{id}` names module `{module_name}` which is not registered"
2053 ));
2054 }
2055 }
2056}
2057
2058fn is_module_name(name: &str) -> bool {
2059 name.split('-').all(|part| {
2060 !part.is_empty()
2061 && part
2062 .chars()
2063 .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
2064 })
2065}
2066
2067async fn production_readiness_guard(
2077 State(problems): State<Arc<Vec<String>>>,
2078 request: axum::extract::Request,
2079 next: Next,
2080) -> Response {
2081 if problems.is_empty() || !request.uri().path().starts_with("/v1/") {
2082 return next.run(request).await;
2083 }
2084 let problem = Problem::new(&SLUGS.not_production_ready).with_detail(problems.join("; "));
2085 match request.extensions().get::<Scope>() {
2086 Some(scope) => problem.instance(&scope.request_id.clone()),
2087 None => problem,
2088 }
2089 .into_response()
2090}
2091
2092async fn admin_rate_limit_layer(
2101 State(limiter): State<Option<Arc<dyn RateLimiter>>>,
2102 request: axum::extract::Request,
2103 next: Next,
2104) -> Response {
2105 if !crate::sidecar::is_admin_path(request.uri().path()) {
2106 return next.run(request).await;
2107 }
2108 let keys: Vec<String> = crate::rate_limit::rate_limit_keys(
2109 crate::rate_limit::client_ip(request.headers()).as_deref(),
2110 None,
2111 )
2112 .into_iter()
2113 .map(|key| format!("admin:{key}"))
2114 .collect();
2115 if let RateLimit::Denied { decision } =
2116 crate::rate_limit::check_rate_limit(limiter.as_ref(), &keys, RateLimitFailure::FailClosed)
2117 .await
2118 {
2119 return rate_limited(&decision);
2120 }
2121 next.run(request).await
2122}
2123
2124fn append_production_readiness(
2136 venture: &Venture,
2137 modules: &[Arc<dyn Module>],
2138 runtime: Option<&Arc<dyn Runtime>>,
2139 errors: &mut ConfigError,
2140) {
2141 for error in crate::route_policy::production_readiness(
2142 venture.env,
2143 &crate::route_policy::WriteGuards::collect(modules),
2144 runtime,
2145 crate::route_policy::rate_limiter_effective(runtime),
2151 crate::route_policy::signer_effective(runtime),
2152 None,
2153 None,
2154 ) {
2155 errors.push(error);
2156 }
2157}
2158
2159fn collect_well_known(modules: &[Arc<dyn Module>], errors: &mut ConfigError) -> Option<Router> {
2160 let mut providers: Vec<&'static str> = Vec::new();
2161 let mut well_known = None;
2162 for module in modules {
2163 if let Some(router) = module.well_known() {
2164 providers.push(module.name());
2165 well_known = Some(router);
2166 }
2167 }
2168 if providers.len() > 1 {
2169 let listed = providers
2170 .iter()
2171 .map(|name| format!("`{name}`"))
2172 .collect::<Vec<_>>()
2173 .join(", ");
2174 errors.push(format!(
2175 "modules {listed} all provide a well-known router; at most one module may occupy /.well-known"
2176 ));
2177 }
2178 well_known
2179}
2180
2181#[cfg(test)]
2182mod acceptance_recording {
2183 use super::*;
2189 use crate::Migrations;
2190 use crate::config::MapConfig;
2191
2192 struct PubWriter;
2195
2196 impl Module for PubWriter {
2197 fn name(&self) -> &'static str {
2198 "writer"
2199 }
2200 fn version(&self) -> &'static str {
2201 "0.0.0-test"
2202 }
2203 fn requires(&self) -> &'static [Port] {
2204 &[]
2205 }
2206 fn public_writes(&self) -> bool {
2207 true
2208 }
2209 fn migrations(&self) -> Migrations {
2210 Migrations::default()
2211 }
2212 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2213 Ok(())
2214 }
2215 fn router(&self, _: ModuleContext) -> Router {
2216 Router::new()
2217 }
2218 }
2219
2220 struct AllPorts;
2223
2224 impl Runtime for AllPorts {
2225 fn provides(&self) -> Vec<Port> {
2226 Port::ALL.to_vec()
2227 }
2228 }
2229
2230 fn harness() -> Harness {
2231 Harness::builder()
2232 .venture(
2233 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2234 )
2235 .module(PubWriter)
2236 .runtime(AllPorts)
2237 .build()
2238 .expect("a development venture builds")
2239 }
2240
2241 fn deployed(unlimited: Option<&str>) -> MapConfig {
2242 let pairs = [("ENV", "production")].into_iter().chain(
2243 unlimited.map(|reason| (crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES, reason)),
2244 );
2245 MapConfig::from_pairs(pairs)
2246 }
2247
2248 #[test]
2249 fn the_waiver_is_recorded_once_and_only_when_it_waived() {
2250 let harness = harness();
2251 let config = deployed(Some("issue #437: limiter binding pending"));
2252 assert!(
2253 harness
2254 .production_readiness_now(VentureEnv::Production, &config, false, true)
2255 .is_empty(),
2256 "the recorded acceptance serves"
2257 );
2258 assert!(
2259 harness
2260 .unlimited_acceptance_recorded
2261 .load(Ordering::Relaxed)
2262 );
2263
2264 assert!(
2267 harness
2268 .production_readiness_now(VentureEnv::Production, &config, false, true)
2269 .is_empty()
2270 );
2271 }
2272
2273 #[test]
2274 fn nothing_is_recorded_when_the_port_was_fine_all_along() {
2275 let harness = harness();
2276 let config = deployed(Some("unused: the limiter resolved"));
2277 assert!(
2278 harness
2279 .production_readiness_now(VentureEnv::Production, &config, true, true)
2280 .is_empty()
2281 );
2282 assert!(
2283 !harness
2284 .unlimited_acceptance_recorded
2285 .load(Ordering::Relaxed),
2286 "a waiver that removed no refusal is not a recorded acceptance"
2287 );
2288 }
2289
2290 #[test]
2291 fn a_refusing_deployment_records_no_acceptance_at_all() {
2292 let harness = harness();
2293 let config = MapConfig::from_pairs([
2297 ("ENV", "production"),
2298 (
2299 crate::route_policy::ALLOW_UNPROTECTED_WRITES,
2300 "captcha pending",
2301 ),
2302 ]);
2303 assert_eq!(
2304 harness
2305 .production_readiness_now(VentureEnv::Production, &config, false, true)
2306 .len(),
2307 1
2308 );
2309 assert!(
2310 !harness
2311 .unprotected_acceptance_recorded
2312 .load(Ordering::Relaxed)
2313 );
2314 assert!(
2315 !harness
2316 .unlimited_acceptance_recorded
2317 .load(Ordering::Relaxed)
2318 );
2319 }
2320
2321 struct HmacWriter;
2325
2326 impl Module for HmacWriter {
2327 fn name(&self) -> &'static str {
2328 "webhooks"
2329 }
2330 fn version(&self) -> &'static str {
2331 "0.0.0-test"
2332 }
2333 fn requires(&self) -> &'static [Port] {
2334 &[]
2335 }
2336 fn public_writes(&self) -> bool {
2337 true
2338 }
2339 fn public_write_policy(&self) -> crate::route_policy::RoutePolicy {
2340 crate::route_policy::RoutePolicy::Signature
2341 }
2342 fn signature_verification(&self) -> crate::route_policy::SignatureVerification {
2343 crate::route_policy::SignatureVerification::Hmac {
2344 secret: "WEBHOOK_SECRET",
2345 }
2346 }
2347 fn migrations(&self) -> Migrations {
2348 Migrations::default()
2349 }
2350 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2351 Ok(())
2352 }
2353 fn router(&self, _: ModuleContext) -> Router {
2354 Router::new()
2355 }
2356 }
2357
2358 #[test]
2359 fn a_non_payments_signature_venture_boots_on_its_webhook_secret_alone() {
2360 let harness = Harness::builder()
2361 .venture(
2362 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2363 )
2364 .module(HmacWriter)
2365 .runtime(AllPorts)
2366 .build()
2367 .expect("a development venture builds");
2368 let deployed = MapConfig::from_pairs([("ENV", "production")]);
2369 let problems =
2370 harness.production_readiness_now(VentureEnv::Production, &deployed, true, true);
2371 assert_eq!(problems.len(), 1, "{problems:?}");
2372 assert!(
2373 problems[0].contains("WEBHOOKS_WEBHOOK_SECRET"),
2374 "{}",
2375 problems[0]
2376 );
2377 assert!(
2378 !problems[0].contains("Payments"),
2379 "a module verifying through the HMAC scheme is not the Payments leg's business: {}",
2380 problems[0]
2381 );
2382
2383 let ready = MapConfig::from_pairs([
2385 ("ENV", "production"),
2386 ("WEBHOOKS_WEBHOOK_SECRET", "whsec-test-dummy"),
2387 ]);
2388 assert!(
2389 harness
2390 .production_readiness_now(VentureEnv::Production, &ready, true, true)
2391 .is_empty()
2392 );
2393 }
2394}