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 url_safety: _,
912 http: _,
913 clock,
914 id_gen,
915 defer,
916 actors: _,
917 dispatcher: _,
918 tenants: _,
919 } = ports;
920
921 let scope_state = ScopeState {
922 defer: defer.unwrap_or_else(|| Arc::new(crate::ports::NoopDefer)),
923 id_gen: id_gen.unwrap_or_else(|| Arc::new(crate::ports::UlidIdGen)),
924 module: (self.modules.len() == 1).then(|| self.modules[0].name()),
927 };
928 let ready_state = ReadyState {
929 db,
930 clock: clock.unwrap_or_else(|| Arc::new(SystemClock)),
931 missing_ports,
932 };
933
934 let surface_state = SurfaceState {
935 config,
936 source: surface_source,
937 };
938
939 let root = self.root_router(api, ui, gateway, health_state, ready_state, surface_state);
940
941 root.layer(from_fn_with_state(
950 Arc::new(readiness),
951 production_readiness_guard,
952 ))
953 .layer(from_fn_with_state(gateway_state, gateway_guard))
954 .layer(from_fn_with_state(scope_state, scope_layer))
955 .layer(axum::middleware::from_fn(token_response_layer))
956 .layer(cors_layer(&self.venture.cors_origins))
957 .layer(from_fn_with_state(problem_type_base, problem_type_layer))
966 }
967}
968
969fn module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
977 let path = match path.split_once(['?', '#']) {
978 Some((path, _)) => path,
979 None => path,
980 };
981 let mut segments = path.split('/').filter(|segment| !segment.is_empty());
982 if segments.next() != Some("v1") {
983 return None;
984 }
985 let name = segments.next()?;
986 Some((name, segments.collect()))
987}
988
989fn exact_module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
996 let path = match path.split_once(['?', '#']) {
997 Some((path, _)) => path,
998 None => path,
999 };
1000 let mut segments = path.split('/');
1001 if segments.next() != Some("") || segments.next() != Some("v1") {
1003 return None;
1004 }
1005 let name = segments.next()?;
1006 if name.is_empty() {
1007 return None;
1008 }
1009 Some((name, segments.collect()))
1010}
1011
1012fn match_streaming(routes: &[StreamRoute], method: &Method, segments: &[&str]) -> Option<usize> {
1015 routes.iter().find_map(|route| {
1016 (route.method == *method && pattern_matches(route.path, segments))
1017 .then_some(route.max_bytes)
1018 })
1019}
1020
1021fn pattern_matches(pattern: &str, segments: &[&str]) -> bool {
1025 let mut pattern_segments = pattern.split('/').filter(|segment| !segment.is_empty());
1026 let mut rest = segments.iter();
1027 loop {
1028 match pattern_segments.next() {
1029 Some("{*rest}") => {
1030 return pattern_segments.next().is_none() && rest.next().is_some();
1031 }
1032 Some(param) if param.starts_with('{') && param.ends_with('}') => {
1033 if rest.next().is_none() {
1034 return false;
1035 }
1036 }
1037 Some(literal) => {
1038 if rest.next() != Some(&literal) {
1039 return false;
1040 }
1041 }
1042 None => return rest.next().is_none(),
1043 }
1044 }
1045}
1046
1047#[derive(Clone)]
1050struct StreamingState {
1051 modules: Arc<Vec<(&'static str, &'static [StreamRoute])>>,
1052}
1053
1054impl StreamingState {
1055 fn build(modules: &[Arc<dyn Module>]) -> Self {
1056 let routes = modules
1059 .iter()
1060 .filter(|module| !module.streaming_routes().is_empty())
1061 .map(|module| (module.name(), module.streaming_routes()))
1062 .collect();
1063 Self {
1064 modules: Arc::new(routes),
1065 }
1066 }
1067
1068 fn max_bytes(&self, path: &str, method: &Method) -> Option<usize> {
1069 let (name, segments) = exact_module_and_segments(path)?;
1070 let (_, routes) = self.modules.iter().find(|(module, _)| *module == name)?;
1071 match_streaming(routes, method, &segments)
1072 }
1073}
1074
1075async fn streaming_request_layer(
1087 State(state): State<StreamingState>,
1088 mut request: axum::extract::Request,
1089 next: Next,
1090) -> Response {
1091 if state.modules.is_empty() {
1092 return next.run(request).await;
1093 }
1094 let Some(max_bytes) = state.max_bytes(request.uri().path(), request.method()) else {
1095 return next.run(request).await;
1096 };
1097 if request.extensions().get::<RequestStream>().is_some() {
1098 return next.run(request).await;
1099 }
1100 if let Some(declared) = crate::ports::declared_content_length(request.headers())
1103 && declared > max_bytes
1104 {
1105 let problem = Problem::request_too_large();
1106 let problem = match request.extensions().get::<Scope>() {
1107 Some(scope) => problem.instance(&scope.request_id),
1108 None => problem,
1109 };
1110 return problem.into_response();
1111 }
1112 let body = std::mem::replace(request.body_mut(), axum::body::Body::empty());
1113 request
1114 .extensions_mut()
1115 .insert(RequestStream::new(body.into_data_stream(), max_bytes));
1116 next.run(request).await
1117}
1118
1119fn with_abuse_floor(
1124 api: Router,
1125 rate_limiter: Option<Arc<dyn RateLimiter>>,
1126 tenant_layer: Arc<TenantLayer>,
1127) -> Router {
1128 api.layer(from_fn_with_state(rate_limiter, admin_rate_limit_layer))
1129 .layer(axum::middleware::from_fn_with_state(
1130 tenant_layer,
1131 resolve_tenant_layer,
1132 ))
1133 .layer(axum::middleware::from_fn(security_headers_layer))
1134 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
1135}
1136
1137async fn problem_type_layer(
1142 State(base): State<Arc<str>>,
1143 request: axum::extract::Request,
1144 next: Next,
1145) -> Response {
1146 let mut response = next.run(request).await;
1147 let Some(problem) = response.extensions().get::<Problem>().cloned() else {
1148 return response;
1149 };
1150 let body = problem.body(&base).to_string();
1151 *response.body_mut() = axum::body::Body::from(body);
1152 response.headers_mut().remove(header::CONTENT_LENGTH);
1153 response.extensions_mut().remove::<Problem>();
1157 response
1158}
1159
1160type ProbeCache = Option<(i64, Vec<serde_json::Value>)>;
1163
1164fn new_probe_cache() -> Arc<futures_util::lock::Mutex<ProbeCache>> {
1167 Arc::new(futures_util::lock::Mutex::new(None))
1168}
1169
1170#[derive(Clone)]
1171struct HealthState {
1172 venture: Arc<Venture>,
1173 modules: Vec<Arc<dyn Module>>,
1174 harness_build: Option<String>,
1176 mailer: Option<Arc<dyn Mailer>>,
1179 captcha_configured: bool,
1180 sidecars: Vec<SidecarMount>,
1181 dispatcher: Option<Arc<dyn Dispatcher>>,
1182 clock: Arc<dyn Clock>,
1183 probe_cache: Arc<futures_util::lock::Mutex<ProbeCache>>,
1194}
1195
1196const SIDECAR_PROBE_TTL_SECS: i64 = 30;
1199
1200const SIDECAR_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
1203
1204async fn health_handler(State(state): State<HealthState>) -> impl IntoResponse {
1205 let modules: Vec<serde_json::Value> = state
1206 .modules
1207 .iter()
1208 .map(|module| {
1209 json!({
1210 "name": module.name(),
1211 "version": module.version(),
1212 "requires": module.requires().iter().map(Port::name).collect::<Vec<_>>(),
1213 "optional": module.optional().iter().map(Port::name).collect::<Vec<_>>(),
1214 "tables": module.tables(),
1215 "emits": module.emits(),
1216 })
1217 })
1218 .collect();
1219 let sidecars = probe_sidecars(&state).await;
1220 let mailers = state
1224 .mailer
1225 .as_ref()
1226 .map(|mailer| mailer.providers())
1227 .unwrap_or_default();
1228 Json(json!({
1229 "venture": state.venture.name,
1230 "env": state.venture.env.as_str(),
1231 "harness_api": HARNESS_API,
1232 "harness_build": state.harness_build,
1233 "mailer": if state.mailer.is_some() { "configured" } else { "not_configured" },
1239 "captcha": if state.captcha_configured { "configured" } else { "absent" },
1240 "mailers": mailers,
1241 "modules": modules,
1242 "sidecars": sidecars,
1243 }))
1244}
1245
1246async fn probe_sidecars(state: &HealthState) -> Vec<serde_json::Value> {
1258 if state.sidecars.is_empty() {
1259 return Vec::new();
1260 }
1261 let mut cache = state.probe_cache.lock().await;
1262 let now_ms =
1263 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1264 if let Some((at_ms, entries)) = cache.as_ref()
1265 && now_ms - at_ms < SIDECAR_PROBE_TTL_SECS * 1000
1266 {
1267 return entries.clone();
1268 }
1269
1270 let mut entries = Vec::new();
1271 for mount in &state.sidecars {
1272 entries.push(probe_one_sidecar(state, mount).await);
1273 }
1274 let stamped_ms =
1275 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1276 *cache = Some((stamped_ms, entries.clone()));
1277 entries
1278}
1279
1280async fn probe_one_sidecar(state: &HealthState, mount: &SidecarMount) -> serde_json::Value {
1281 let mut missing = json!({
1282 "name": mount.name,
1283 "binding": mount.binding,
1284 "probe": "unreachable",
1285 "contract": serde_json::Value::Null,
1286 "module": serde_json::Value::Null,
1287 "version": serde_json::Value::Null,
1288 });
1289 let Some(dispatcher) = state.dispatcher.as_ref() else {
1290 return missing;
1291 };
1292 let request = http::Request::builder()
1293 .method(http::Method::GET)
1294 .uri("/__health")
1295 .body(Bytes::new());
1296 let Ok(request) = request else {
1297 return missing;
1298 };
1299 let dispatcher = Arc::clone(dispatcher);
1304 let binding = mount.binding.clone();
1305 let probe = async move { dispatcher.dispatch(&binding, request).await };
1306 let Some(Ok(response)) =
1307 crate::ports::timeout(&*state.clock, probe, SIDECAR_PROBE_TIMEOUT).await
1308 else {
1309 return missing;
1310 };
1311 let contract = response
1312 .headers()
1313 .get(crate::sidecar::X_HARNESS_API)
1314 .and_then(|v| v.to_str().ok())
1315 .and_then(|v| v.parse::<u32>().ok());
1316 let body = serde_json::from_slice::<serde_json::Value>(response.body()).ok();
1317 let module = response
1318 .headers()
1319 .get(crate::sidecar::X_HARNESS_MODULE)
1320 .and_then(|v| v.to_str().ok())
1321 .map(str::to_owned)
1322 .or_else(|| first_module_string(body.as_ref(), "name"));
1323 let version = first_module_string(body.as_ref(), "version");
1324 let tables = declared_tables(body.as_ref());
1325 let contract = contract.or_else(|| {
1326 body.as_ref()
1327 .and_then(|b| b.get("harness_api"))
1328 .and_then(serde_json::Value::as_u64)
1329 .and_then(|api| u32::try_from(api).ok())
1330 });
1331 missing["contract"] = contract.map_or(serde_json::Value::Null, |api| json!(api));
1332 missing["module"] = module.map_or(serde_json::Value::Null, |name| json!(name));
1333 missing["version"] = version.map_or(serde_json::Value::Null, |version| json!(version));
1334 missing["tables"] = json!(tables);
1335 match contract {
1336 Some(api) if api != HARNESS_API => missing["probe"] = json!("mismatch"),
1337 Some(_) => missing["probe"] = json!("ok"),
1338 None => missing["probe"] = json!("unreachable"),
1339 }
1340 let clashes = collisions_with(&tables, state, &mount.name);
1352 if !clashes.is_empty() {
1353 let detail = format!(
1354 "sidecar `{}` claims table(s) {} already claimed in this deployment",
1355 mount.name,
1356 clashes.join(", ")
1357 );
1358 tracing::error!(sidecar = %mount.name, tables = %clashes.join(","), "{detail}");
1359 crate::logging::forward_internal_error(&detail);
1360 missing["probe"] = json!("table-collision");
1361 missing["table_collisions"] = json!(clashes);
1362 }
1363 missing
1364}
1365
1366fn declared_tables(body: Option<&serde_json::Value>) -> Vec<String> {
1370 let Some(modules) = body
1371 .and_then(|b| b.get("modules"))
1372 .and_then(|m| m.as_array())
1373 else {
1374 return Vec::new();
1375 };
1376 let mut tables: Vec<String> = modules
1377 .iter()
1378 .filter_map(|module| module.get("tables")?.as_array())
1379 .flatten()
1380 .filter_map(|table| table.as_str().map(str::to_owned))
1381 .collect();
1382 tables.sort_unstable();
1383 tables.dedup();
1384 tables
1385}
1386
1387fn collisions_with(claimed: &[String], state: &HealthState, mount: &str) -> Vec<String> {
1390 let mut clashes: Vec<String> = Vec::new();
1391 for module in &state.modules {
1392 if module.name() == mount {
1393 continue;
1394 }
1395 for table in module.tables() {
1396 if claimed.iter().any(|claimed| claimed == table)
1397 && !clashes.iter().any(|seen| seen == table)
1398 {
1399 clashes.push(format!("`{table}` (also module `{}`)", module.name()));
1400 }
1401 }
1402 }
1403 clashes
1404}
1405
1406fn first_module_string(body: Option<&serde_json::Value>, key: &str) -> Option<String> {
1409 body?
1410 .get("modules")?
1411 .get(0)?
1412 .get(key)?
1413 .as_str()
1414 .map(str::to_owned)
1415}
1416
1417struct TenantLayer {
1420 routing: Option<Arc<dyn crate::tenant::TenantRouting>>,
1421 fallback: Option<Arc<dyn Database>>,
1427}
1428
1429async fn resolve_tenant_layer(
1436 State(layer): State<Arc<TenantLayer>>,
1437 scope: Scope,
1438 mut request: axum::extract::Request,
1439 next: Next,
1440) -> Response {
1441 let host = request
1442 .headers()
1443 .get(header::HOST)
1444 .and_then(|value| value.to_str().ok())
1445 .unwrap_or_default()
1446 .to_owned();
1447
1448 let resolution = match layer.routing.as_ref() {
1453 Some(routing) => routing.resolve(&host),
1454 None => <crate::tenant::ImplicitTenant as crate::tenant::ResolveTenant>::resolve(
1455 &crate::tenant::ImplicitTenant,
1456 &host,
1457 ),
1458 };
1459
1460 let tenant = match resolution.admit() {
1461 Ok(tenant) => tenant,
1462 Err(problem) => {
1463 return Problem::new(problem)
1464 .instance(&scope.request_id)
1465 .into_response();
1466 }
1467 };
1468
1469 let db = match layer.routing.as_ref() {
1470 Some(routing) => match routing.database(&tenant).await {
1471 Ok(db) => Some(db),
1472 Err(err) => {
1473 tracing::error!(tenant = %tenant.id(), "{err}");
1476 crate::logging::forward_internal_error(&err.to_string());
1477 return Problem::new(&SLUGS.tenant_degraded)
1478 .instance(&scope.request_id)
1479 .into_response();
1480 }
1481 },
1482 None => layer.fallback.clone(),
1483 };
1484
1485 if let Some(db) = db {
1486 let tenancy = if layer.routing.is_some() {
1491 crate::tenant::Tenancy::FromRegistry
1492 } else {
1493 crate::tenant::Tenancy::Sole
1494 };
1495 request
1496 .extensions_mut()
1497 .insert(crate::tenant_conn::ResolvedTenant {
1498 tenant,
1499 tenancy,
1500 db,
1501 });
1502 }
1503 next.run(request).await
1504}
1505
1506#[derive(Clone)]
1507struct ReadyState {
1508 db: Option<Arc<dyn Database>>,
1509 clock: Arc<dyn Clock>,
1510 missing_ports: Vec<Port>,
1514}
1515
1516fn missing_module_ports(modules: &[Arc<dyn Module>], ports: &Ports) -> Vec<Port> {
1521 const PROBED: &[Port] = &[Port::VectorIndex, Port::Embedder];
1522 PROBED
1523 .iter()
1524 .copied()
1525 .filter(|port| {
1526 modules
1527 .iter()
1528 .any(|module| module.requires().contains(port))
1529 && !ports.has(*port)
1530 })
1531 .collect()
1532}
1533
1534async fn ready_handler(State(state): State<ReadyState>) -> impl IntoResponse {
1538 if !state.missing_ports.is_empty() {
1539 let detail = state
1540 .missing_ports
1541 .iter()
1542 .map(|port| format!("{} port is not configured", port.name()))
1543 .collect::<Vec<_>>()
1544 .join("; ");
1545 return Problem::not_ready(detail).into_response();
1546 }
1547 let Some(db) = state.db else {
1548 return Problem::not_ready("database port is not configured").into_response();
1549 };
1550 let stmt = Statement::new("SELECT 1");
1551 let query = async move { db.query(&stmt).await };
1552 match crate::ports::timeout(&*state.clock, query, Duration::from_secs(2)).await {
1553 Some(Ok(_rows)) => Json(json!({ "ok": true })).into_response(),
1554 Some(Err(err)) => {
1555 error!(error = %err, "readiness probe query failed");
1556 crate::logging::forward_internal_error(&format!("readiness probe query failed: {err}"));
1557 Problem::not_ready("database query failed").into_response()
1558 }
1559 None => Problem::not_ready("database did not answer within 2 s").into_response(),
1560 }
1561}
1562
1563#[derive(Clone)]
1564struct SurfaceState {
1565 config: Arc<dyn Config>,
1566 source: Arc<dyn SurfaceSource>,
1567}
1568
1569struct MergedSurface {
1576 base: Arc<SurfaceVariants>,
1577 mounts: Vec<SidecarMount>,
1578 dispatcher: Option<Arc<dyn Dispatcher>>,
1579 gateway: Option<Arc<HmacSigner>>,
1582 env: VentureEnv,
1583 captcha_configured: bool,
1584}
1585
1586impl MergedSurface {
1587 async fn sidecar_modules(&self) -> Vec<crate::surface::ModuleSurface> {
1589 let mut extra = Vec::new();
1590 let Some(dispatcher) = &self.dispatcher else {
1591 return extra;
1592 };
1593 for mount in &self.mounts {
1594 if !dispatcher.has(&mount.binding) {
1595 continue;
1596 }
1597 let mut builder = axum::http::Request::builder()
1598 .method(axum::http::Method::GET)
1599 .uri("/__surface")
1600 .header(header::ACCEPT, "application/json");
1601 if let Some(signer) = &self.gateway {
1602 builder =
1604 builder.header(X_HARNESS_GATEWAY, mint_gateway(signer, &mount.name, false));
1605 }
1606 let request = builder
1607 .body(bytes::Bytes::new())
1608 .expect("static request builds");
1609 let answer = match dispatcher.dispatch(&mount.binding, request).await {
1610 Ok(response) if response.status().is_success() => response,
1611 Ok(response) => {
1612 tracing::warn!(module = mount.name, status = %response.status(), "sidecar surface not available");
1613 continue;
1614 }
1615 Err(err) => {
1616 tracing::warn!(module = mount.name, error = %err, "sidecar surface fetch failed");
1617 continue;
1618 }
1619 };
1620 if answer.body().len() > MAX_SIDECAR_SURFACE_BYTES {
1623 tracing::warn!(
1624 module = mount.name,
1625 "sidecar surface exceeds {MAX_SIDECAR_SURFACE_BYTES} bytes and was not merged"
1626 );
1627 continue;
1628 }
1629 match serde_json::from_slice::<SurfaceDocument>(answer.body()) {
1630 Ok(document) => match sanitize_sidecar_document(&document, &mount.name) {
1631 Ok(entries) => extra.extend(entries.into_iter().filter_map(|mut entry| {
1632 if self.env == VentureEnv::Production
1636 && !self.captcha_configured
1637 && strip_unguarded_captcha_actions(&mut entry.surface) > 0
1638 {
1639 tracing::warn!(
1640 module = entry.name,
1641 "merged sidecar actions requiring a captcha were dropped: no Captcha port here"
1642 );
1643 }
1644 (!entry.surface.is_empty()).then_some(entry)
1645 })),
1646 Err(problems) => {
1647 for problem in problems {
1648 tracing::warn!(module = mount.name, error = problem, "sidecar surface rejected at merge");
1649 }
1650 }
1651 },
1652 Err(err) => {
1653 tracing::warn!(module = mount.name, error = %err, "sidecar surface is not a surface document");
1654 }
1655 }
1656 }
1657 extra
1658 }
1659}
1660
1661#[async_trait::async_trait]
1662impl SurfaceSource for MergedSurface {
1663 async fn current(&self) -> Arc<SurfaceDocument> {
1664 if self.mounts.is_empty() {
1665 return Arc::clone(&self.base.document);
1666 }
1667 let mut document = (*self.base.document).clone();
1668 document.modules.extend(self.sidecar_modules().await);
1669 Arc::new(document)
1670 }
1671
1672 fn built(&self) -> Arc<SurfaceDocument> {
1673 Arc::clone(&self.base.document)
1674 }
1675
1676 fn rendered(&self, admin: bool) -> Option<&RenderedSurface> {
1677 Some(if admin {
1678 &self.base.full
1679 } else {
1680 &self.base.public
1681 })
1682 }
1683}
1684
1685async fn surface_handler(
1701 State(state): State<SurfaceState>,
1702 headers: HeaderMap,
1703) -> impl IntoResponse {
1704 let admin = require_admin(&*state.config, &headers).is_ok();
1705 let current = state.source.current().await;
1709 let built = state.source.built();
1710 let prerendered = Arc::ptr_eq(¤t, &built).then(|| state.source.rendered(admin));
1711 let fresh;
1712 let rendered: &RenderedSurface = if let Some(rendered) = prerendered.flatten() {
1713 rendered
1714 } else {
1715 fresh = if admin {
1716 RenderedSurface::render(¤t)
1717 } else {
1718 RenderedSurface::render(¤t.public())
1719 };
1720 &fresh
1721 };
1722 let matches = headers
1723 .get(header::IF_NONE_MATCH)
1724 .and_then(|value| value.to_str().ok())
1725 .is_some_and(|value| {
1726 value
1727 .split(',')
1728 .map(str::trim)
1729 .any(|tag| tag == "*" || tag == rendered.etag)
1730 });
1731 let mut response = if matches {
1732 StatusCode::NOT_MODIFIED.into_response()
1733 } else {
1734 (
1735 [(header::CONTENT_TYPE, "application/json")],
1736 rendered.json.clone(),
1737 )
1738 .into_response()
1739 };
1740 let response_headers = response.headers_mut();
1741 response_headers.insert(
1742 header::ETAG,
1743 header::HeaderValue::from_str(&rendered.etag).expect("hex etag is a valid header"),
1744 );
1745 response_headers.insert(
1746 header::CACHE_CONTROL,
1747 if admin {
1748 header::HeaderValue::from_static("private, no-store")
1749 } else {
1750 header::HeaderValue::from_static("no-cache")
1751 },
1752 );
1753 response_headers.insert(
1754 header::VARY,
1755 header::HeaderValue::from_static("Authorization"),
1756 );
1757 response
1758}
1759
1760#[derive(Default)]
1763pub struct HarnessBuilder {
1764 venture: Option<Venture>,
1765 modules: Vec<Arc<dyn Module>>,
1766 provides: Vec<Port>,
1767 runtime: Option<Arc<dyn Runtime>>,
1768 module_templates: Vec<(String, Box<dyn Template>)>,
1769 overrides: Vec<(String, Box<dyn Template>)>,
1770 ui: Option<Arc<dyn UiMount>>,
1771 actor_handlers: Vec<(String, Arc<dyn ActorHandler>)>,
1772}
1773
1774impl HarnessBuilder {
1775 #[must_use]
1776 pub fn venture(mut self, venture: Venture) -> Self {
1777 self.venture = Some(venture);
1778 self
1779 }
1780
1781 #[must_use]
1784 pub fn module(mut self, module: impl Module) -> Self {
1785 self.modules.push(Arc::new(module));
1786 self
1787 }
1788
1789 #[must_use]
1792 pub fn module_arc(mut self, module: Arc<dyn Module>) -> Self {
1793 self.modules.push(module);
1794 self
1795 }
1796
1797 #[must_use]
1801 pub fn runtime(mut self, runtime: impl Runtime) -> Self {
1802 let runtime: Arc<dyn Runtime> = Arc::new(runtime);
1803 self.provides = runtime.provides();
1804 self.runtime = Some(runtime);
1805 self
1806 }
1807
1808 #[must_use]
1811 pub fn templates(
1812 mut self,
1813 templates: impl IntoIterator<Item = (String, Box<dyn Template>)>,
1814 ) -> Self {
1815 self.module_templates.extend(templates);
1816 self
1817 }
1818
1819 #[must_use]
1822 pub fn ui(mut self, ui: impl UiMount) -> Self {
1823 self.ui = Some(Arc::new(ui));
1824 self
1825 }
1826
1827 #[must_use]
1830 pub fn template(mut self, id: impl Into<String>, template: Box<dyn Template>) -> Self {
1831 self.overrides.push((id.into(), template));
1832 self
1833 }
1834
1835 #[must_use]
1840 pub fn actor(mut self, kind: &str, handler: Arc<dyn ActorHandler>) -> Self {
1841 self.actor_handlers.push((kind.to_owned(), handler));
1842 self
1843 }
1844
1845 pub fn build(self) -> Result<Harness, ConfigError> {
1856 let mut errors = ConfigError::default();
1857
1858 let venture = if let Some(venture) = self.venture {
1859 venture.validate(&mut errors);
1860 venture
1861 } else {
1862 errors.push("missing venture: call .venture(Venture::new(..)) before .build()");
1863 Venture::new("invalid", "invalid.invalid")
1864 };
1865
1866 let mut names: HashMap<&'static str, usize> = HashMap::new();
1867 let mut tables: HashMap<&'static str, &'static str> = HashMap::new();
1868
1869 for module in &self.modules {
1870 if module.harness_api() != HARNESS_API {
1871 errors.push(harness_api_mismatch(module.as_ref()));
1872 }
1873
1874 let name = module.name();
1875 if name.is_empty() || !is_module_name(name) {
1876 errors.push(format!(
1877 "module name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
1878 ));
1879 }
1880 if names.insert(name, 1).is_some() {
1881 errors.push(format!("duplicate module name `{name}`"));
1882 }
1883
1884 for port in module.requires().iter().chain(module.optional()) {
1885 if !Port::ALL.contains(port) {
1886 errors.push(format!(
1887 "module `{name}` declares unknown port {}",
1888 port.name()
1889 ));
1890 }
1891 }
1892 for port in module.requires() {
1893 if module.optional().contains(port) {
1894 errors.push(format!(
1895 "module `{name}` lists port {} in both requires() and optional()",
1896 port.name()
1897 ));
1898 }
1899 }
1900
1901 module.surface().validate(name, &mut errors);
1902
1903 check_tables(module.as_ref(), &mut tables, &mut errors);
1904 }
1905
1906 let actor_handlers = collect_actors(&self.actor_handlers, &self.modules, &mut errors);
1907
1908 let well_known = collect_well_known(&self.modules, &mut errors);
1909
1910 for module in &self.modules {
1911 for port in module.requires() {
1912 if !self.provides.contains(port) {
1913 errors.push(format!(
1914 "module `{}` requires port {} which the runtime does not provide",
1915 module.name(),
1916 port.name()
1917 ));
1918 }
1919 }
1920 warn_undeclared_ports(module.as_ref(), &self.provides);
1921 }
1922
1923 let ordered = resolve_dependency_order(&self.modules, &mut errors);
1924
1925 append_production_readiness(&venture, &self.modules, self.runtime.as_ref(), &mut errors);
1926
1927 check_template_ids(
1928 self.overrides.iter().chain(self.module_templates.iter()),
1929 &names,
1930 &mut errors,
1931 );
1932
1933 let surface = Arc::new(SurfaceVariants::compose(
1934 &venture,
1935 &self.modules,
1936 self.ui.as_ref(),
1937 ));
1938 if let Some(ui) = &self.ui {
1939 ui.validate(&surface.document, &mut errors);
1940 }
1941
1942 errors.into_result()?;
1943
1944 let mut registry = TemplateRegistry::new();
1945 registry.register_all(self.module_templates);
1946 registry.register_all(self.overrides);
1947
1948 let mut events = EventBus::new();
1949 for module in &self.modules {
1950 for (name, handler) in module.events() {
1951 events = events.on(name, handler);
1952 }
1953 }
1954
1955 Ok(Harness {
1956 venture: Arc::new(venture),
1957 unprotected_acceptance_recorded: AtomicBool::new(false),
1958 unlimited_acceptance_recorded: AtomicBool::new(false),
1959 personal_data: Arc::new(crate::PersonalDataCatalog::compose(
1965 ordered.iter().map(|m| (m.name(), m.personal_data())),
1966 )),
1967 modules: ordered,
1968 templates: Arc::new(registry),
1969 events,
1970 runtime: self.runtime,
1971 well_known,
1972 surface,
1973 ui: self.ui,
1974 actor_handlers,
1975 })
1976 }
1977}
1978
1979fn resolve_dependency_order(
1992 modules: &[Arc<dyn Module>],
1993 errors: &mut ConfigError,
1994) -> Vec<Arc<dyn Module>> {
1995 let names: BTreeSet<&str> = modules.iter().map(|module| module.name()).collect();
1996 let mut unknown = false;
1997 for module in modules {
1998 for needed in module.depends_on() {
1999 if !names.contains(needed) {
2000 unknown = true;
2001 errors.push(format!(
2002 "module `{}` depends on `{needed}`, which this venture does not compose: \
2003 add the module or drop the dependency",
2004 module.name()
2005 ));
2006 }
2007 }
2008 }
2009 if unknown {
2010 return modules.to_vec();
2011 }
2012
2013 let mut remaining: Vec<Option<Arc<dyn Module>>> = modules
2017 .iter()
2018 .map(|module| Some(Arc::clone(module)))
2019 .collect();
2020 let mut placed: BTreeSet<&str> = BTreeSet::new();
2021 let mut ordered: Vec<Arc<dyn Module>> = Vec::with_capacity(modules.len());
2022
2023 while ordered.len() < modules.len() {
2024 let mut progressed = false;
2025 for slot in &mut remaining {
2026 let ready = slot.as_ref().is_some_and(|module| {
2027 module
2028 .depends_on()
2029 .iter()
2030 .all(|needed| placed.contains(needed))
2031 });
2032 if ready {
2033 let module = slot.take().expect("checked above");
2034 placed.insert(module.name());
2035 ordered.push(module);
2036 progressed = true;
2037 }
2038 }
2039 if !progressed {
2040 let stuck: Vec<&str> = remaining
2042 .iter()
2043 .filter_map(|slot| slot.as_ref().map(|module| module.name()))
2044 .collect();
2045 errors.push(format!(
2046 "modules [{}] depend on each other in a cycle: migrations cannot be ordered so \
2047 that every module follows the ones it sits on top of",
2048 stuck.join(", ")
2049 ));
2050 return modules.to_vec();
2051 }
2052 }
2053 ordered
2054}
2055
2056fn check_template_ids<'a>(
2057 ids: impl Iterator<Item = &'a (String, Box<dyn Template>)>,
2058 names: &HashMap<&'static str, usize>,
2059 errors: &mut ConfigError,
2060) {
2061 for (id, _) in ids {
2062 let Some(module_name) = id.split('/').next() else {
2063 continue;
2064 };
2065 if !names.contains_key(module_name) {
2066 errors.push(format!(
2067 "template `{id}` names module `{module_name}` which is not registered"
2068 ));
2069 }
2070 }
2071}
2072
2073fn is_module_name(name: &str) -> bool {
2074 name.split('-').all(|part| {
2075 !part.is_empty()
2076 && part
2077 .chars()
2078 .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
2079 })
2080}
2081
2082async fn production_readiness_guard(
2092 State(problems): State<Arc<Vec<String>>>,
2093 request: axum::extract::Request,
2094 next: Next,
2095) -> Response {
2096 if problems.is_empty() || !request.uri().path().starts_with("/v1/") {
2097 return next.run(request).await;
2098 }
2099 let problem = Problem::new(&SLUGS.not_production_ready).with_detail(problems.join("; "));
2100 match request.extensions().get::<Scope>() {
2101 Some(scope) => problem.instance(&scope.request_id.clone()),
2102 None => problem,
2103 }
2104 .into_response()
2105}
2106
2107async fn admin_rate_limit_layer(
2116 State(limiter): State<Option<Arc<dyn RateLimiter>>>,
2117 request: axum::extract::Request,
2118 next: Next,
2119) -> Response {
2120 if !crate::sidecar::is_admin_path(request.uri().path()) {
2121 return next.run(request).await;
2122 }
2123 let keys: Vec<String> = crate::rate_limit::rate_limit_keys(
2124 crate::rate_limit::client_ip(request.headers()).as_deref(),
2125 None,
2126 )
2127 .into_iter()
2128 .map(|key| format!("admin:{key}"))
2129 .collect();
2130 if let RateLimit::Denied { decision } =
2131 crate::rate_limit::check_rate_limit(limiter.as_ref(), &keys, RateLimitFailure::FailClosed)
2132 .await
2133 {
2134 return rate_limited(&decision);
2135 }
2136 next.run(request).await
2137}
2138
2139fn append_production_readiness(
2151 venture: &Venture,
2152 modules: &[Arc<dyn Module>],
2153 runtime: Option<&Arc<dyn Runtime>>,
2154 errors: &mut ConfigError,
2155) {
2156 for error in crate::route_policy::production_readiness(
2157 venture.env,
2158 &crate::route_policy::WriteGuards::collect(modules),
2159 runtime,
2160 crate::route_policy::rate_limiter_effective(runtime),
2166 crate::route_policy::signer_effective(runtime),
2167 None,
2168 None,
2169 ) {
2170 errors.push(error);
2171 }
2172}
2173
2174fn collect_well_known(modules: &[Arc<dyn Module>], errors: &mut ConfigError) -> Option<Router> {
2175 let mut providers: Vec<&'static str> = Vec::new();
2176 let mut well_known = None;
2177 for module in modules {
2178 if let Some(router) = module.well_known() {
2179 providers.push(module.name());
2180 well_known = Some(router);
2181 }
2182 }
2183 if providers.len() > 1 {
2184 let listed = providers
2185 .iter()
2186 .map(|name| format!("`{name}`"))
2187 .collect::<Vec<_>>()
2188 .join(", ");
2189 errors.push(format!(
2190 "modules {listed} all provide a well-known router; at most one module may occupy /.well-known"
2191 ));
2192 }
2193 well_known
2194}
2195
2196#[cfg(test)]
2197mod acceptance_recording {
2198 use super::*;
2204 use crate::Migrations;
2205 use crate::config::MapConfig;
2206
2207 struct PubWriter;
2210
2211 impl Module for PubWriter {
2212 fn name(&self) -> &'static str {
2213 "writer"
2214 }
2215 fn version(&self) -> &'static str {
2216 "0.0.0-test"
2217 }
2218 fn requires(&self) -> &'static [Port] {
2219 &[]
2220 }
2221 fn public_writes(&self) -> bool {
2222 true
2223 }
2224 fn migrations(&self) -> Migrations {
2225 Migrations::default()
2226 }
2227 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2228 Ok(())
2229 }
2230 fn router(&self, _: ModuleContext) -> Router {
2231 Router::new()
2232 }
2233 }
2234
2235 struct AllPorts;
2238
2239 impl Runtime for AllPorts {
2240 fn provides(&self) -> Vec<Port> {
2241 Port::ALL.to_vec()
2242 }
2243 }
2244
2245 fn harness() -> Harness {
2246 Harness::builder()
2247 .venture(
2248 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2249 )
2250 .module(PubWriter)
2251 .runtime(AllPorts)
2252 .build()
2253 .expect("a development venture builds")
2254 }
2255
2256 fn deployed(unlimited: Option<&str>) -> MapConfig {
2257 let pairs = [("ENV", "production")].into_iter().chain(
2258 unlimited.map(|reason| (crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES, reason)),
2259 );
2260 MapConfig::from_pairs(pairs)
2261 }
2262
2263 #[test]
2264 fn the_waiver_is_recorded_once_and_only_when_it_waived() {
2265 let harness = harness();
2266 let config = deployed(Some("issue #437: limiter binding pending"));
2267 assert!(
2268 harness
2269 .production_readiness_now(VentureEnv::Production, &config, false, true)
2270 .is_empty(),
2271 "the recorded acceptance serves"
2272 );
2273 assert!(
2274 harness
2275 .unlimited_acceptance_recorded
2276 .load(Ordering::Relaxed)
2277 );
2278
2279 assert!(
2282 harness
2283 .production_readiness_now(VentureEnv::Production, &config, false, true)
2284 .is_empty()
2285 );
2286 }
2287
2288 #[test]
2289 fn nothing_is_recorded_when_the_port_was_fine_all_along() {
2290 let harness = harness();
2291 let config = deployed(Some("unused: the limiter resolved"));
2292 assert!(
2293 harness
2294 .production_readiness_now(VentureEnv::Production, &config, true, true)
2295 .is_empty()
2296 );
2297 assert!(
2298 !harness
2299 .unlimited_acceptance_recorded
2300 .load(Ordering::Relaxed),
2301 "a waiver that removed no refusal is not a recorded acceptance"
2302 );
2303 }
2304
2305 #[test]
2306 fn a_refusing_deployment_records_no_acceptance_at_all() {
2307 let harness = harness();
2308 let config = MapConfig::from_pairs([
2312 ("ENV", "production"),
2313 (
2314 crate::route_policy::ALLOW_UNPROTECTED_WRITES,
2315 "captcha pending",
2316 ),
2317 ]);
2318 assert_eq!(
2319 harness
2320 .production_readiness_now(VentureEnv::Production, &config, false, true)
2321 .len(),
2322 1
2323 );
2324 assert!(
2325 !harness
2326 .unprotected_acceptance_recorded
2327 .load(Ordering::Relaxed)
2328 );
2329 assert!(
2330 !harness
2331 .unlimited_acceptance_recorded
2332 .load(Ordering::Relaxed)
2333 );
2334 }
2335
2336 struct HmacWriter;
2340
2341 impl Module for HmacWriter {
2342 fn name(&self) -> &'static str {
2343 "webhooks"
2344 }
2345 fn version(&self) -> &'static str {
2346 "0.0.0-test"
2347 }
2348 fn requires(&self) -> &'static [Port] {
2349 &[]
2350 }
2351 fn public_writes(&self) -> bool {
2352 true
2353 }
2354 fn public_write_policy(&self) -> crate::route_policy::RoutePolicy {
2355 crate::route_policy::RoutePolicy::Signature
2356 }
2357 fn signature_verification(&self) -> crate::route_policy::SignatureVerification {
2358 crate::route_policy::SignatureVerification::Hmac {
2359 secret: "WEBHOOK_SECRET",
2360 }
2361 }
2362 fn migrations(&self) -> Migrations {
2363 Migrations::default()
2364 }
2365 fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2366 Ok(())
2367 }
2368 fn router(&self, _: ModuleContext) -> Router {
2369 Router::new()
2370 }
2371 }
2372
2373 #[test]
2374 fn a_non_payments_signature_venture_boots_on_its_webhook_secret_alone() {
2375 let harness = Harness::builder()
2376 .venture(
2377 Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2378 )
2379 .module(HmacWriter)
2380 .runtime(AllPorts)
2381 .build()
2382 .expect("a development venture builds");
2383 let deployed = MapConfig::from_pairs([("ENV", "production")]);
2384 let problems =
2385 harness.production_readiness_now(VentureEnv::Production, &deployed, true, true);
2386 assert_eq!(problems.len(), 1, "{problems:?}");
2387 assert!(
2388 problems[0].contains("WEBHOOKS_WEBHOOK_SECRET"),
2389 "{}",
2390 problems[0]
2391 );
2392 assert!(
2393 !problems[0].contains("Payments"),
2394 "a module verifying through the HMAC scheme is not the Payments leg's business: {}",
2395 problems[0]
2396 );
2397
2398 let ready = MapConfig::from_pairs([
2400 ("ENV", "production"),
2401 ("WEBHOOKS_WEBHOOK_SECRET", "whsec-test-dummy"),
2402 ]);
2403 assert!(
2404 harness
2405 .production_readiness_now(VentureEnv::Production, &ready, true, true)
2406 .is_empty()
2407 );
2408 }
2409}