use std::{
collections::{BTreeMap, BTreeSet, HashMap, HashSet},
fmt,
path::PathBuf,
sync::{Arc, Mutex},
time::{Duration, Instant as StdInstant},
};
use serde::{Deserialize, Serialize};
use subc_control::{
ops, CapabilityRequirementStatus, CatalogEntry, ClientControlPush, ClientControlRequest,
ClientControlResponse, ConsumerIdentity, DaemonBuildProvenance, DaemonObservedProcess,
ModuleDeclaredProvenance, ModuleProtocol, NotReadyReason, PollKind, RouteCloseReason,
SpawnCursor, StderrCaptureState, StderrTail, StderrTailEntry, SupervisorDaemonProvenance,
SupervisorEntry, SupervisorHealthEntry, SupervisorModuleProvenance, SupervisorObservedProcess,
SupervisorRescanResult, SupervisorRoute, SupervisorRouteConsumer, SupervisorRouteModule,
};
use subc_protocol::{
error_codes,
manifest::{
validate_hello_capability_grammar, validate_hello_self_signal_declarations,
CapabilityDeclarations, CapabilityNeed, Concurrency, ManifestProvenance, ModuleManifest,
ProviderRole,
},
session::{
HealthReport, ModuleControlPush, ModuleControlRequest, ModuleControlRequestFromModule,
ModuleControlResponse, ModuleControlResponseToModule, MODULE_CONTROL_OP_HEALTH_CHECK,
MODULE_TO_SUBC_OP_CATALOG_UPDATE,
},
BindIdentity, ErrorBody, Flags, FrameType, ModuleHelloAckBody, ModuleHelloBody, Principal,
Priority, RouteTarget, PROTOCOL_VERSION,
};
use tokio::time::{timeout_at, Instant};
use tracing::{debug, info, warn};
use crate::{
capability_requirements::{
log_duplicate_claim_events, log_requirement_events, CapabilityRequirementEvaluator,
CapabilityVerdict, DuplicateClaimSource, RegisteredModule, RequirementStatus,
RuntimeModule,
},
daemon_config::RestartRequiredSection,
forwarding::{
CloseReason, EndpointRoute, ForwardingError, ForwardingTable, GoodbyeTarget,
ModuleControlRpcCompletion, ModuleControlRpcOutcome, ModuleEndpointId,
PendingModuleControlRpc, RouteBindRelayOutcome, RoutePollSnapshot, RouteRelease,
},
observability::{
ROUTE_OPEN_REFUSED_DECLARED_NOT_READY, ROUTE_OPEN_REFUSED_REQUIRED_CAPABILITY_UNPROVIDED,
},
provenance::{
process_start_time, spawned_file_identity, ExecutableIdentityProbe, SpawnedFileIdentity,
},
registry::{ChannelState, ConnectionId, Registry, RegistryError},
router::{RouteCtx, RouterError},
server::MAX_PENDING_ROUTE_BINDS_PER_TARGET,
stderr_tail::{CaptureState, TailEntry},
supervise::{
validate_spec, ModuleProcessLiveness, ReservedHelloRejection, SpawnSubscribeRefusal,
SupervisorHandle, SwapHelloAdmission,
},
ConnectedClients, DaemonCounters, Frame, ProjectRootId, Supervisor,
};
pub const MIN_SUPPORTED_VERSION: u8 = PROTOCOL_VERSION;
const CAP_MANIFEST_REGISTRATION: &str = "manifest_registration_v1";
const CAP_CHANNEL_LIFECYCLE: &str = "channel_lifecycle_v1";
const CAP_PING_PONG: &str = "ping_pong_v1";
const CAP_SESSION_ATTACH: &str = "session_attach_v1";
const CAP_ADMISSION_FACTS_RELAY: &str = "admission_facts_relay_v1";
const SUBC_CONTROL_OPS: &[&str] = &[
ops::SERVER_DESCRIBE,
ops::CATALOG_LIST,
ops::ROUTE_OPEN,
ops::ROUTE_POLL,
ops::ROUTE_CLOSING,
ops::ROUTE_CLOSED,
ops::SUPERVISOR_LIST,
ops::SUPERVISOR_RESTART,
ops::SUPERVISOR_SWAP,
ops::SUPERVISOR_RELOAD,
ops::SUPERVISOR_RESCAN,
ops::SUPERVISOR_RELEASE_RESERVED,
ops::SUPERVISOR_SET_ENABLED,
ops::SUPERVISOR_HEALTH_PROBE,
ops::SUPERVISOR_HEALTH,
ops::SUPERVISOR_STDERR_TAIL,
ops::SUPERVISOR_TERMINALS,
ops::SUPERVISOR_ROUTES,
ops::SUPERVISOR_PROVENANCE,
ops::SUPERVISOR_SPAWN_SNAPSHOT,
ops::SUPERVISOR_SPAWN_SUBSCRIBE,
];
const MODULE_TO_SUBC_CONTROL_OPS: &[&str] =
&[MODULE_TO_SUBC_OP_CATALOG_UPDATE, "supervisor.live_roots"];
const MODULE_BASELINE_CONTROL_OPS: &[&str] = &["route.bind", "route.status"];
pub const DEFAULT_ROUTE_BIND_RELAY_TIMEOUT: Duration = Duration::from_secs(12);
pub const DEFAULT_ROUTE_BIND_BREAKER_THRESHOLD: u32 = 3;
pub const DEFAULT_ROUTE_BIND_BREAKER_COOLDOWN: Duration = Duration::from_secs(20);
const DEFAULT_HEALTH_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
const SLOW_CONTROL_DISPATCH_THRESHOLD: Duration = Duration::from_secs(1);
#[derive(Clone)]
struct DaemonProvenanceFacts {
build: DaemonBuildProvenance,
pid: Option<u32>,
started_at_ms: Option<u64>,
start_clock: Option<crate::clock::StartClock>,
executable_path: Option<PathBuf>,
executable_identity: Option<SpawnedFileIdentity>,
process_start_time: Option<u64>,
probe: ExecutableIdentityProbe,
}
impl Default for DaemonProvenanceFacts {
fn default() -> Self {
Self {
build: DaemonBuildProvenance {
build_git_sha: None,
build_lock_digest: None,
},
pid: None,
started_at_ms: None,
start_clock: None,
executable_path: None,
executable_identity: None,
process_start_time: None,
probe: ExecutableIdentityProbe::default(),
}
}
}
#[derive(Debug, Clone)]
struct SupervisorRescanContext {
supervisor: Supervisor,
config_path: PathBuf,
configured_port: Option<u16>,
storage_config: Option<crate::daemon_config::StorageConfig>,
admission_facts_carrier_module_id: Option<String>,
admission_facts_targets: Option<Vec<String>>,
}
#[derive(Clone)]
pub struct ControlHandler {
registry: Arc<Registry>,
forwarding: Arc<ForwardingTable>,
process_liveness: Option<Arc<dyn ModuleProcessLiveness>>,
supervisor: SupervisorHandle,
subc_capabilities: Arc<[String]>,
route_bind_relay_timeout: Duration,
route_bind_relay_timeouts: BTreeMap<String, Duration>,
route_bind_breakers: RouteBindBreakers,
route_bind_concurrency: RouteBindConcurrency,
route_bind_breaker_threshold: u32,
route_bind_breaker_cooldown: Duration,
health_probe_timeout: Duration,
storage_config: Option<crate::daemon_config::StorageConfig>,
machine_id: Option<crate::machine_id::MachineId>,
admission_facts_carrier_module_id: Option<String>,
admission_facts_targets: Option<Vec<String>>,
rescan: Option<SupervisorRescanContext>,
connected_clients: ConnectedClients,
counters: DaemonCounters,
capability_evaluator: Arc<CapabilityRequirementEvaluator>,
daemon_provenance: DaemonProvenanceFacts,
#[cfg(test)]
control_dispatch_delay: Option<Duration>,
#[cfg(test)]
provenance_probe_override: Option<subc_control::RunningImageAgreement>,
}
impl fmt::Debug for ControlHandler {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ControlHandler")
.field("registry", &self.registry)
.field("forwarding", &self.forwarding)
.field("process_liveness", &self.process_liveness.is_some())
.field("supervisor", &self.supervisor)
.field("subc_capabilities", &self.subc_capabilities)
.finish()
}
}
struct RouteOpenRequest {
target: RouteTarget,
identity: BindIdentity,
consumer_identity: Option<ConsumerIdentity>,
consumer_capabilities: Option<Vec<String>>,
admission_facts: Option<serde_json::Value>,
}
struct RouteBindReservationGuard {
forwarding: Arc<ForwardingTable>,
endpoint: ModuleEndpointId,
relay_corr: u64,
armed: bool,
}
struct ModuleControlRpcGuard {
forwarding: Arc<ForwardingTable>,
endpoint: ModuleEndpointId,
corr: u64,
armed: bool,
}
impl ModuleControlRpcGuard {
fn new(forwarding: Arc<ForwardingTable>, endpoint: ModuleEndpointId, corr: u64) -> Self {
Self {
forwarding,
endpoint,
corr,
armed: true,
}
}
fn disarm(&mut self) {
self.armed = false;
}
}
impl Drop for ModuleControlRpcGuard {
fn drop(&mut self) {
if self.armed {
let _ = self
.forwarding
.cancel_module_control_rpc(self.endpoint, self.corr);
}
}
}
impl RouteBindReservationGuard {
fn new(forwarding: Arc<ForwardingTable>, endpoint: ModuleEndpointId, relay_corr: u64) -> Self {
Self {
forwarding,
endpoint,
relay_corr,
armed: true,
}
}
fn release_and_disarm(&mut self) {
if !self.armed {
return;
}
if let Ok(Some(target)) = self.forwarding.abort_pending_relay(
self.endpoint,
self.relay_corr,
RouteBindRelayOutcome::ModuleGone("route.open handler canceled".to_string()),
) {
send_goodbye_target_best_effort(&target, "canceled route.bind");
}
self.armed = false;
}
fn disarm(&mut self) {
self.armed = false;
}
}
impl Drop for RouteBindReservationGuard {
fn drop(&mut self) {
self.release_and_disarm();
}
}
#[derive(Debug, Clone, Default)]
pub(crate) struct RouteBindBreakers {
modules: Arc<Mutex<HashMap<String, ModuleBreakerState>>>,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct RouteBindConcurrency {
modules: Arc<Mutex<HashMap<String, usize>>>,
}
struct RouteBindConcurrencyGuard {
concurrency: RouteBindConcurrency,
module_id: String,
}
impl RouteBindConcurrency {
fn try_admit(&self, module_id: &str, limit: usize) -> Result<RouteBindConcurrencyGuard, usize> {
let mut modules = self
.modules
.lock()
.expect("route.bind concurrency mutex poisoned");
let in_flight = modules.entry(module_id.to_string()).or_default();
if *in_flight >= limit {
return Err(*in_flight);
}
*in_flight += 1;
Ok(RouteBindConcurrencyGuard {
concurrency: self.clone(),
module_id: module_id.to_string(),
})
}
}
impl Drop for RouteBindConcurrencyGuard {
fn drop(&mut self) {
let mut modules = self
.concurrency
.modules
.lock()
.expect("route.bind concurrency mutex poisoned");
let remove = {
let in_flight = modules
.get_mut(&self.module_id)
.expect("admitted route.bind has a concurrency entry");
*in_flight -= 1;
*in_flight == 0
};
if remove {
modules.remove(&self.module_id);
}
}
}
#[derive(Debug, Default)]
struct ModuleBreakerState {
consecutive_timeouts: u32,
cooldown_until: Option<Instant>,
probe_in_flight: bool,
}
enum RouteBindAdmission<'a> {
Admitted {
guard: RouteBindBreakerGuard<'a>,
probe: bool,
},
Refused {
consecutive_timeouts: u32,
retry_in: Duration,
probe_in_flight: bool,
},
}
struct RouteBindBreakerGuard<'a> {
breakers: RouteBindBreakers,
module_id: &'a str,
settled: bool,
}
impl RouteBindBreakerGuard<'_> {
fn record_accepted(&mut self) -> bool {
self.settled = true;
self.breakers.record_accepted(self.module_id)
}
fn record_timeout(&mut self, threshold: u32, cooldown: Duration) -> Option<BreakerOpened> {
self.settled = true;
self.breakers
.record_timeout(self.module_id, threshold, cooldown)
}
fn record_inconclusive(&mut self) {
self.settled = true;
self.breakers.record_inconclusive(self.module_id);
}
}
impl Drop for RouteBindBreakerGuard<'_> {
fn drop(&mut self) {
if !self.settled {
self.breakers.record_inconclusive(self.module_id);
}
}
}
struct BreakerOpened {
consecutive_timeouts: u32,
reopened_after_probe: bool,
}
impl RouteBindBreakers {
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<String, ModuleBreakerState>> {
self.modules
.lock()
.expect("route.bind breaker mutex poisoned")
}
fn admit<'a>(&self, module_id: &'a str) -> RouteBindAdmission<'a> {
let admitted = |probe| RouteBindAdmission::Admitted {
guard: RouteBindBreakerGuard {
breakers: self.clone(),
module_id,
settled: false,
},
probe,
};
let mut modules = self.lock();
let Some(state) = modules.get_mut(module_id) else {
return admitted(false);
};
let Some(cooldown_until) = state.cooldown_until else {
return admitted(false);
};
if state.probe_in_flight {
return RouteBindAdmission::Refused {
consecutive_timeouts: state.consecutive_timeouts,
retry_in: Duration::ZERO,
probe_in_flight: true,
};
}
let now = Instant::now();
if now < cooldown_until {
return RouteBindAdmission::Refused {
consecutive_timeouts: state.consecutive_timeouts,
retry_in: cooldown_until - now,
probe_in_flight: false,
};
}
state.probe_in_flight = true;
admitted(true)
}
fn record_accepted(&self, module_id: &str) -> bool {
self.lock()
.remove(module_id)
.is_some_and(|state| state.cooldown_until.is_some())
}
fn record_timeout(
&self,
module_id: &str,
threshold: u32,
cooldown: Duration,
) -> Option<BreakerOpened> {
let mut modules = self.lock();
let state = modules.entry(module_id.to_string()).or_default();
let was_open = state.cooldown_until.is_some();
let was_probe = state.probe_in_flight;
state.probe_in_flight = false;
state.consecutive_timeouts = state.consecutive_timeouts.saturating_add(1);
if state.consecutive_timeouts < threshold {
return None;
}
state.cooldown_until = Some(Instant::now() + cooldown);
Some(BreakerOpened {
consecutive_timeouts: state.consecutive_timeouts,
reopened_after_probe: was_open && was_probe,
})
}
fn record_inconclusive(&self, module_id: &str) {
if let Some(state) = self.lock().get_mut(module_id) {
state.probe_in_flight = false;
}
}
pub(crate) fn reset_for_new_module_connection(&self, module_id: &str) -> Option<u32> {
self.lock()
.remove(module_id)
.map(|state| state.consecutive_timeouts)
.filter(|discarded| *discarded > 0)
}
fn open_snapshot(&self) -> Option<serde_json::Value> {
let now = Instant::now();
let modules = self.lock();
let open = modules
.iter()
.filter_map(|(module_id, state)| {
let cooldown_until = state.cooldown_until?;
Some((
module_id.clone(),
serde_json::json!({
"consecutive_timeouts": state.consecutive_timeouts,
"cooldown_remaining_ms":
cooldown_until.saturating_duration_since(now).as_millis() as u64,
"probe_in_flight": state.probe_in_flight,
}),
))
})
.collect::<serde_json::Map<String, serde_json::Value>>();
(!open.is_empty()).then_some(serde_json::Value::Object(open))
}
}
impl ControlHandler {
pub fn new(registry: Arc<Registry>) -> Self {
Self::with_forwarding(registry, Arc::new(ForwardingTable::default()))
}
pub fn with_forwarding(registry: Arc<Registry>, forwarding: Arc<ForwardingTable>) -> Self {
let counters = forwarding.counters();
let route_bind_breakers = forwarding.route_bind_breakers();
let route_bind_concurrency = forwarding.route_bind_concurrency();
Self {
registry,
forwarding,
process_liveness: None,
supervisor: SupervisorHandle::new(),
subc_capabilities: Arc::from([
CAP_MANIFEST_REGISTRATION.to_string(),
CAP_CHANNEL_LIFECYCLE.to_string(),
CAP_PING_PONG.to_string(),
CAP_SESSION_ATTACH.to_string(),
CAP_ADMISSION_FACTS_RELAY.to_string(),
]),
route_bind_relay_timeout: DEFAULT_ROUTE_BIND_RELAY_TIMEOUT,
route_bind_relay_timeouts: BTreeMap::new(),
route_bind_breakers,
route_bind_concurrency,
route_bind_breaker_threshold: DEFAULT_ROUTE_BIND_BREAKER_THRESHOLD,
route_bind_breaker_cooldown: DEFAULT_ROUTE_BIND_BREAKER_COOLDOWN,
health_probe_timeout: DEFAULT_HEALTH_PROBE_TIMEOUT,
storage_config: None,
machine_id: None,
admission_facts_carrier_module_id: None,
admission_facts_targets: None,
rescan: None,
connected_clients: ConnectedClients::new(),
counters,
capability_evaluator: Arc::new(CapabilityRequirementEvaluator::new()),
daemon_provenance: DaemonProvenanceFacts::default(),
#[cfg(test)]
control_dispatch_delay: None,
#[cfg(test)]
provenance_probe_override: None,
}
}
pub fn with_storage_config(
mut self,
storage_config: Option<crate::daemon_config::StorageConfig>,
) -> Self {
self.storage_config = storage_config;
self
}
pub fn with_machine_id(mut self, machine_id: Option<crate::machine_id::MachineId>) -> Self {
self.machine_id = machine_id;
self
}
pub fn with_admission_facts_config(
mut self,
carrier_module_id: Option<String>,
targets: Option<Vec<String>>,
) -> Self {
self.admission_facts_carrier_module_id = carrier_module_id;
self.admission_facts_targets = targets;
self
}
pub fn with_route_bind_relay_timeout(mut self, timeout: Duration) -> Self {
self.route_bind_relay_timeout = timeout;
self
}
pub fn with_route_bind_relay_timeouts(
mut self,
timeouts: impl IntoIterator<Item = (String, Duration)>,
) -> Self {
self.route_bind_relay_timeouts = timeouts.into_iter().collect();
self
}
pub fn route_bind_relay_timeout_for(&self, module_id: &str) -> Duration {
self.route_bind_relay_timeouts
.get(module_id)
.copied()
.unwrap_or(self.route_bind_relay_timeout)
}
pub fn with_route_bind_breaker(mut self, threshold: u32, cooldown: Duration) -> Self {
self.route_bind_breaker_threshold = threshold.max(1);
self.route_bind_breaker_cooldown = cooldown;
self
}
#[cfg(test)]
pub(crate) fn with_health_probe_timeout(mut self, timeout: Duration) -> Self {
self.health_probe_timeout = timeout;
self
}
#[cfg(test)]
pub(crate) fn with_control_dispatch_delay(mut self, delay: Duration) -> Self {
self.control_dispatch_delay = Some(delay);
self
}
pub fn with_process_liveness(
mut self,
process_liveness: Arc<dyn ModuleProcessLiveness>,
) -> Self {
self.process_liveness = Some(process_liveness);
self
}
pub fn with_supervisor(mut self, supervisor: SupervisorHandle) -> Self {
self.supervisor = supervisor;
self
}
pub fn with_daemon_provenance(
mut self,
pid: u32,
started_at_ms: u64,
executable_path: Option<PathBuf>,
build_git_sha: Option<String>,
build_lock_digest: Option<String>,
) -> Self {
let executable_identity = executable_path.as_deref().and_then(spawned_file_identity);
let process_start_time = process_start_time(pid);
self.daemon_provenance = DaemonProvenanceFacts {
build: DaemonBuildProvenance {
build_git_sha,
build_lock_digest,
},
pid: Some(pid),
started_at_ms: Some(started_at_ms),
start_clock: None,
executable_path,
executable_identity,
process_start_time,
probe: ExecutableIdentityProbe::default(),
};
self
}
pub(crate) fn with_daemon_start_clock(mut self, clock: crate::clock::StartClock) -> Self {
self.daemon_provenance.start_clock = Some(clock);
self
}
#[cfg(test)]
fn with_provenance_probe_result(mut self, result: subc_control::RunningImageAgreement) -> Self {
self.provenance_probe_override = Some(result);
self
}
pub fn with_capability_config(
self,
modules: impl IntoIterator<Item = (String, bool)>,
reserved_capabilities: BTreeMap<String, String>,
) -> Self {
self.capability_evaluator
.configure(modules, reserved_capabilities);
self
}
pub fn with_supervisor_rescan(
mut self,
supervisor: Supervisor,
config_path: impl Into<PathBuf>,
configured_port: Option<u16>,
) -> Self {
self.rescan = Some(SupervisorRescanContext {
supervisor,
config_path: config_path.into(),
configured_port,
storage_config: self.storage_config.clone(),
admission_facts_carrier_module_id: self.admission_facts_carrier_module_id.clone(),
admission_facts_targets: self.admission_facts_targets.clone(),
});
self
}
pub fn with_connected_clients(mut self, connected_clients: ConnectedClients) -> Self {
self.connected_clients = connected_clients;
self
}
pub fn forwarding(&self) -> Arc<ForwardingTable> {
Arc::clone(&self.forwarding)
}
pub(crate) fn counters(&self) -> DaemonCounters {
self.counters.clone()
}
pub fn spawn_capability_deadline_loop(self: Arc<Self>) {
tokio::spawn(async move {
loop {
self.capability_evaluator
.wait_for_change_or_deadline()
.await;
self.refresh_capability_requirements();
}
});
}
fn runtime_capability_snapshot(
&self,
) -> Result<(Vec<RuntimeModule>, Vec<RegisteredModule>), RouterError> {
let runtime = self
.supervisor
.list()
.into_iter()
.map(|module| {
let status = module.status().map_err(|err| {
RouterError::backend(0, 0, format!("failed to read capability status: {err}"))
})?;
Ok(RuntimeModule {
module_id: status.module_id,
state: status.state,
enabled: status.enabled,
})
})
.collect::<Result<Vec<_>, RouterError>>()?;
let (_, registrations) = self.registry.list_modules().map_err(|err| {
RouterError::backend(
0,
0,
format!("failed to list capability registrations: {err}"),
)
})?;
let registrations = registrations
.into_iter()
.map(|registration| RegisteredModule {
module_id: registration.manifest.module_id,
module_version: registration.manifest.module_version,
capabilities: registration.manifest.capabilities,
})
.collect();
Ok((runtime, registrations))
}
fn apply_registration_capabilities(&self, registration: &crate::registry::ModuleRegistration) {
let cached_registration = RegisteredModule {
module_id: registration.manifest.module_id.clone(),
module_version: registration.manifest.module_version.clone(),
capabilities: registration.manifest.capabilities.clone(),
};
if self.capability_evaluator.record_hello(&cached_registration) {
warn!(
module_id = %cached_registration.module_id,
"capability claims drifted from the cached manifest"
);
}
if capability_census_trigger(None, registration.manifest.capabilities.as_ref()) {
self.enforce_capability_denies();
}
self.refresh_capability_requirements();
}
pub(crate) fn install_swap_promotion_observer(self: &Arc<Self>) {
let observer: std::sync::Weak<dyn crate::supervise::SwapPromotionObserver> =
Arc::downgrade(self) as std::sync::Weak<ControlHandler>;
self.supervisor.set_swap_promotion_observer(observer);
}
pub fn refresh_capability_requirements(&self) {
match self.runtime_capability_snapshot() {
Ok((runtime, registrations)) => {
log_requirement_events(
self.capability_evaluator
.evaluate_now(&runtime, ®istrations),
);
}
Err(err) => warn!(error = %err, "failed to recompute capability requirements"),
}
}
fn enforce_capability_denies(&self) {
let (_, registrations) = match self.registry.list_modules() {
Ok(snapshot) => snapshot,
Err(err) => {
warn!(error = %err, "failed to read registrations for capability deny census");
return;
}
};
let manifests = registrations
.into_iter()
.map(|registration| {
(
registration.manifest.module_id.clone(),
registration.manifest,
)
})
.collect::<BTreeMap<_, _>>();
let census = match self.forwarding.route_census(None) {
Ok(census) => census,
Err(err) => {
warn!(error = %err, "failed to read route census for capability deny enforcement");
return;
}
};
for (target_module_id, routes) in census {
let Some(target_manifest) = manifests.get(&target_module_id) else {
continue;
};
let mut closed_routes = Vec::new();
let mut module_goodbyes = Vec::new();
for route in routes {
let Principal::Reserved {
module_id: opening_module_id,
} = &route.principal
else {
continue;
};
let Some(opening_manifest) = manifests.get(opening_module_id) else {
continue;
};
let Some(capability) = denied_capability(opening_manifest, target_manifest) else {
continue;
};
match self.forwarding.release_client_route(
route.goodbye_target.connection_id,
route.goodbye_target.channel,
route.goodbye_target.epoch,
) {
Ok(RouteRelease::Removed(module_goodbye)) => {
warn!(
opening_module_id,
target_module_id,
capability,
"force-closing route because an attested capability deny edge now matches"
);
closed_routes.push(route);
module_goodbyes.push(module_goodbye);
}
Ok(RouteRelease::Stale | RouteRelease::Absent) => {}
Err(err) => warn!(
opening_module_id,
target_module_id,
capability,
error = %err,
"failed to force-close capability-denied route"
),
}
}
if closed_routes.is_empty() {
continue;
}
send_route_control_pushes(
&self.forwarding,
closed_routes,
ClientControlPush::RouteClosed {
module_id: target_module_id,
reason: RouteCloseReason::CapabilityDenied,
drained: false,
abandoned: 0,
excluded_subscriptions: 0,
terminal: Some(false),
},
);
self.emit_route_goodbyes(module_goodbyes);
}
}
fn not_ready_reason(
&self,
registration: &crate::registry::ModuleRegistration,
) -> Option<NotReadyReason> {
if !registration.ready {
return Some(NotReadyReason {
reason: NotReadyReason::DECLARED_NOT_READY.to_string(),
capability: None,
});
}
self.first_unprovided_required_capability(registration)
.map(|capability| NotReadyReason {
reason: NotReadyReason::REQUIRED_CAPABILITY_UNPROVIDED.to_string(),
capability: Some(capability),
})
}
fn first_unprovided_required_capability(
&self,
registration: &crate::registry::ModuleRegistration,
) -> Option<String> {
let required = registration
.manifest
.capabilities
.iter()
.flat_map(|declarations| declarations.requires.iter())
.filter(|requirement| requirement.need == CapabilityNeed::Required)
.map(|requirement| requirement.capability.as_str())
.collect::<BTreeSet<_>>();
if required.is_empty() {
return None;
}
let module_id = registration.manifest.module_id.as_str();
let verdict = |capability: &str| self.capability_evaluator.verdict(module_id, capability);
if required
.iter()
.any(|capability| verdict(capability).is_none())
{
self.refresh_capability_requirements();
}
required
.into_iter()
.find(|capability| verdict(capability) != Some(CapabilityVerdict::Provided))
.map(str::to_string)
}
fn capability_requirement_statuses(&self) -> Vec<CapabilityRequirementStatus> {
self.capability_evaluator
.statuses()
.into_iter()
.map(capability_requirement_status)
.collect()
}
fn deregister_connection(
&self,
connection_id: ConnectionId,
) -> Result<Vec<crate::registry::ModuleRegistration>, RegistryError> {
self.registry.deregister_connection(connection_id)
}
pub(crate) fn route_open_target(&self, frame: &Frame) -> Option<String> {
if frame.header.channel != 0 || frame.header.ty != FrameType::Request {
return None;
}
let Ok(ClientControlRequest::RouteOpen { target, .. }) =
parse_client_control_request(&frame.body)
else {
return None;
};
Some(target_module_id(&target).to_string())
}
pub(crate) fn route_open_capacity_refusal(
&self,
ctx: &RouteCtx,
frame: &Frame,
target_module_id: &str,
limit: usize,
) -> Result<Frame, RouterError> {
self.route_open_admission_refusal_frame(
ctx,
frame,
target_module_id,
format!(
"connection already has {limit} route.open binds in flight; retry after one settles"
),
)
}
fn route_open_admission_refusal_frame(
&self,
ctx: &RouteCtx,
frame: &Frame,
target_module_id: &str,
message: impl Into<String>,
) -> Result<Frame, RouterError> {
self.route_open_refusal_frame(
ctx,
frame,
target_module_id,
"open_admission_full",
error_codes::TARGET_UNAVAILABLE,
message,
)
}
#[cfg(test)]
pub fn handle_control(
&self,
connection_id: ConnectionId,
frame: Frame,
) -> Result<Vec<Frame>, RouterError> {
match frame.header.ty {
FrameType::Ping => Ok(vec![pong(&frame)?]),
FrameType::Hello => self.handle_hello(connection_id, None, frame),
FrameType::Goodbye => self.handle_goodbye(connection_id),
ty => Ok(vec![control_error_frame(
&frame,
"unsupported_control_frame",
format!("unsupported channel-0 frame {ty:?}"),
)?]),
}
}
pub async fn handle_control_frame(
&self,
ctx: &RouteCtx,
frame: Frame,
) -> Result<Vec<Frame>, RouterError> {
self.handle_control_frame_timed(ctx, frame, None).await
}
pub(crate) async fn handle_control_frame_timed(
&self,
ctx: &RouteCtx,
frame: Frame,
dispatch_started_at: Option<StdInstant>,
) -> Result<Vec<Frame>, RouterError> {
match frame.header.ty {
FrameType::Ping => Ok(vec![pong(&frame)?]),
FrameType::Hello => {
self.handle_hello(ctx.connection_id, Some(ctx.egress.clone()), frame)
}
FrameType::Goodbye => self.handle_goodbye(ctx.connection_id),
FrameType::Cancel => {
if self
.supervisor
.cancel_spawn_subscription(ctx.connection_id, frame.header.corr)
{
Ok(Vec::new())
} else {
Ok(vec![control_error_frame(
&frame,
"unknown_subscription",
"no supervisor spawn subscription has this correlation id",
)?])
}
}
FrameType::Request => {
if self
.forwarding
.module_endpoint_for_connection(ctx.connection_id)
.map_err(RouterError::Forwarding)?
.is_some()
{
if !is_known_module_request_op(&frame.body) {
return Ok(vec![control_error_frame(
&frame,
"unsupported_control_frame",
"module-originated channel-0 REQUEST is not supported",
)?]);
}
let request = match parse_module_control_request_from_module(&frame.body) {
Ok(request) => request,
Err((err, ControlRequestBodyError::UnknownOp)) => {
return Ok(vec![control_error_frame(
&frame,
"unsupported_control_frame",
format!("unsupported module-originated channel-0 REQUEST: {err}"),
)?])
}
Err((err, ControlRequestBodyError::InvalidBody)) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_control_body",
format!("malformed module control body: {err}"),
)?])
}
};
let op = module_control_request_op(&request);
let corr = frame.header.corr;
log_control_dispatch_arrival(op, ctx.connection_id, corr);
let result =
self.handle_module_control_request(ctx.connection_id, frame, request);
log_slow_control_dispatch(dispatch_started_at, op, ctx.connection_id, corr);
return result;
}
if is_known_module_request_op(&frame.body) {
return Ok(vec![control_error_frame(
&frame,
"not_registered",
"catalog.update requires an active module registration owned by this connection",
)?]);
}
let request = match parse_client_control_request(&frame.body) {
Ok(request) => request,
Err((err, ControlRequestBodyError::UnknownOp)) => {
return Ok(vec![control_error_frame(
&frame,
"unknown_control_op",
format!("unknown client control op: {err}"),
)?])
}
Err((err, ControlRequestBodyError::InvalidBody)) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_control_body",
format!("malformed client control body: {err}"),
)?])
}
};
let op = client_control_request_op(&request);
let corr = frame.header.corr;
log_control_dispatch_arrival(op, ctx.connection_id, corr);
#[cfg(test)]
if let Some(delay) = self.control_dispatch_delay {
tokio::time::sleep(delay).await;
}
let result = self
.handle_client_control_request(ctx, frame, request)
.await;
log_slow_control_dispatch(dispatch_started_at, op, ctx.connection_id, corr);
result
}
FrameType::Push => {
let Some(endpoint) = self
.forwarding
.module_endpoint_for_connection(ctx.connection_id)
.map_err(RouterError::Forwarding)?
else {
return Ok(vec![control_error_frame(
&frame,
"unsupported_control_frame",
"client-originated channel-0 PUSH is not supported",
)?]);
};
self.handle_status_update(endpoint, frame)
}
FrameType::Response | FrameType::Error
if self
.forwarding
.module_endpoint_for_connection(ctx.connection_id)
.map_err(RouterError::Forwarding)?
.is_some() =>
{
self.handle_module_relay_response(ctx.connection_id, frame)
}
ty => Ok(vec![control_error_frame(
&frame,
"unsupported_control_frame",
format!("unsupported channel-0 frame {ty:?}"),
)?]),
}
}
pub fn cleanup_connection(
&self,
connection_id: ConnectionId,
) -> Result<Vec<crate::registry::ModuleRegistration>, RegistryError> {
let crash_closed = self
.registry
.get_module_by_connection(connection_id)?
.and_then(|registration| {
self.forwarding
.module_endpoint_for_connection(connection_id)
.ok()
.flatten()
.and_then(|endpoint| self.forwarding.endpoint_routes(endpoint).ok())
.map(|routes| (registration.manifest.module_id, routes))
});
if let Some((module_id, routes)) = crash_closed {
let terminal = match self.supervisor.get(&module_id) {
None => false,
Some(module) => match module.will_recover_after_connection_loss() {
Ok(will_recover) => !will_recover,
Err(err) => {
warn!(
%module_id,
error = %err,
"failed to read crash recovery verdict; reporting non-terminal conservatively"
);
false
}
},
};
send_route_control_pushes(
&self.forwarding,
routes,
ClientControlPush::RouteClosed {
module_id,
reason: RouteCloseReason::Crash,
drained: false,
abandoned: 0,
excluded_subscriptions: 0,
terminal: Some(terminal),
},
);
}
let registrations = self.deregister_connection(connection_id);
if let Ok(released_routes) = self.forwarding.cleanup_connection(connection_id) {
self.emit_route_goodbyes(released_routes);
}
if matches!(®istrations, Ok(r) if !r.is_empty()) {
crate::supervise::notify_registration_release();
self.capability_evaluator.wake_deadline_loop();
self.refresh_capability_requirements();
}
self.supervisor.remove_spawn_subscribers(connection_id);
registrations
}
pub(crate) fn handle_route_goodbye(
&self,
connection_id: ConnectionId,
route_channel: u16,
route_epoch: u32,
) -> Result<bool, RouterError> {
debug!(
connection_id = connection_id.get(),
route_channel, route_epoch, "handling route GOODBYE"
);
let RouteRelease::Removed(released_route) = self
.forwarding
.release_client_route(connection_id, route_channel, route_epoch)
.map_err(RouterError::Forwarding)?
else {
return Ok(false);
};
self.emit_route_goodbyes(vec![released_route]);
Ok(true)
}
fn emit_route_goodbyes(&self, released_routes: Vec<GoodbyeTarget>) {
for released in released_routes {
let frame = match Frame::build_with_version(
released.negotiated_ver,
FrameType::Goodbye,
control_flags(),
released.channel,
released.epoch,
0,
Vec::new(),
) {
Ok(frame) => frame,
Err(err) => {
warn!(
route_channel = released.channel,
error = %err,
"failed to build route GOODBYE frame"
);
continue;
}
};
if let Err(err) = released.sink.try_send(frame) {
if released.close_on_delivery_failure() {
warn!(
target_connection_id = released.connection_id.get(),
route_channel = released.channel,
error = %err,
"route GOODBYE was not delivered to client; closing target connection"
);
if self
.forwarding
.escalate_client_delivery_failure(
released.connection_id,
released.channel,
released.epoch,
CloseReason::new(
"route_goodbye_delivery_failed",
format!(
"failed to enqueue route GOODBYE for channel {}: {err}",
released.channel
),
),
)
.unwrap_or(false)
{
self.counters.increment_goodbye_relay_client_failed();
}
} else {
self.counters
.increment_goodbye_relay_module_dropped(released.module_id.as_deref());
warn!(
target_connection_id = released.connection_id.get(),
route_channel = released.channel,
error = %err,
"route GOODBYE to module dropped under backpressure; not closing shared module connection"
);
}
}
}
}
fn send_abandoned_route_bind_goodbye(
&self,
module_sink: &crate::FrameSink,
negotiated_ver: u8,
module_channel: u16,
module_epoch: u32,
) {
let frame = match Frame::build_with_version(
negotiated_ver,
FrameType::Goodbye,
control_flags(),
module_channel,
module_epoch,
0,
Vec::new(),
) {
Ok(frame) => frame,
Err(err) => {
warn!(
route_channel = module_channel,
error = %err,
"failed to build GOODBYE for abandoned route.bind"
);
return;
}
};
if let Err(err) = module_sink.try_send(frame) {
warn!(
route_channel = module_channel,
error = %err,
"GOODBYE for abandoned route.bind dropped; module idle reaper will reclaim the binding"
);
}
}
fn handle_hello(
&self,
connection_id: ConnectionId,
sink: Option<crate::FrameSink>,
frame: Frame,
) -> Result<Vec<Frame>, RouterError> {
debug!(
connection_id = connection_id.get(),
corr = frame.header.corr,
"handling HELLO"
);
let hello_value = match serde_json::from_slice::<serde_json::Value>(&frame.body) {
Ok(value) => value,
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_hello",
format!("malformed HELLO body: {err}"),
)?])
}
};
if let Err(err) = validate_hello_capability_grammar(&hello_value) {
return Ok(vec![control_error_frame(
&frame,
"invalid_capability_grammar",
err.to_string(),
)?]);
}
if let Err(err) = validate_hello_self_signal_declarations(&hello_value) {
return Ok(vec![control_error_frame(
&frame,
"invalid_manifest",
err.to_string(),
)?]);
}
if let Some(provenance) = hello_value
.get("manifest")
.and_then(|manifest| manifest.get("provenance"))
{
if let Err(err) = serde_json::from_value::<ManifestProvenance>(provenance.clone()) {
return Ok(vec![control_error_frame(
&frame,
"invalid_manifest",
format!("malformed manifest provenance: {err}"),
)?]);
}
}
let hello = match serde_json::from_value::<ModuleHelloBody>(hello_value) {
Ok(hello) => hello,
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_hello",
format!("malformed HELLO body: {err}"),
)?])
}
};
if hello.protocol_ver != hello.manifest.protocol_ver {
return Ok(vec![control_error_frame(
&frame,
"invalid_manifest",
format!(
"HELLO protocol_ver {} does not match manifest protocol_ver {}",
hello.protocol_ver, hello.manifest.protocol_ver
),
)?]);
}
if hello.manifest.module_id.trim().is_empty() {
return Ok(vec![control_error_frame(
&frame,
"invalid_manifest",
"manifest module_id must not be empty",
)?]);
}
let negotiated_ver = match negotiate_version(hello.protocol_ver) {
Ok(negotiated_ver) => negotiated_ver,
Err(message) => {
return Ok(vec![control_error_frame(
&frame,
"version_unsupported",
message,
)?])
}
};
let swap_admission = self
.supervisor
.swap_hello_admission(&hello.manifest.module_id, hello.launch_nonce.as_deref());
if swap_admission == SwapHelloAdmission::Refused {
warn!(
module_id = %hello.manifest.module_id,
connection_id = connection_id.get(),
"HELLO refused: a swap is open for this module_id and the launch nonce is not one the supervisor minted for it"
);
return Ok(vec![control_error_frame(
&frame,
"swap_token_invalid",
format!(
"module_id '{}' is being swapped; HELLO without the swap candidate's launch nonce is rejected",
hello.manifest.module_id
),
)?]);
}
let swap_candidate = swap_admission == SwapHelloAdmission::Candidate;
if let Some(rejection) = (!swap_candidate)
.then(|| {
self.supervisor.reserved_hello_rejection(
&hello.manifest.module_id,
hello.launch_nonce.as_deref(),
)
})
.flatten()
{
let message = match rejection {
ReservedHelloRejection::Exact { module_id } => format!(
"module_id '{module_id}' is reserved; HELLO without a valid launch nonce is rejected"
),
ReservedHelloRejection::Prefix {
prefix,
owner_module_id,
} => format!(
"module_id '{}' matches reserved prefix '{prefix}' owned by '{owner_module_id}'; HELLO without the owner launch nonce is rejected",
hello.manifest.module_id
),
};
return Ok(vec![control_error_frame(
&frame,
"reserved_module",
message,
)?]);
}
let reserved_capability_refusals = self.capability_evaluator.reserved_hello_refusals(
&hello.manifest.module_id,
hello.manifest.capabilities.as_ref(),
);
if let Some(refusal) = reserved_capability_refusals.first() {
let capability = refusal.capability.clone();
let bound_module = refusal.claimants[0].clone();
log_duplicate_claim_events(reserved_capability_refusals);
return Ok(vec![control_error_frame(
&frame,
"reserved_capability",
format!(
"capability '{}' is reserved for module_id '{}'; claimant '{}' was refused",
capability, bound_module, hello.manifest.module_id
),
)?]);
}
if self
.forwarding
.connection_has_client_routes(connection_id)
.map_err(RouterError::Forwarding)?
{
return Ok(vec![control_error_frame(
&frame,
"invalid_hello",
"connection has open client routes and cannot also register as a module",
)?]);
}
let control_ops = effective_module_control_ops(hello.control_ops);
if swap_candidate {
return self.register_swap_candidate(
connection_id,
sink,
&frame,
hello.manifest,
negotiated_ver,
control_ops,
);
}
let registration = match self.registry.register_with_control_ops(
hello.manifest,
negotiated_ver,
connection_id,
control_ops,
) {
Ok(registration) => registration,
Err(RegistryError::DuplicateModuleId { module_id }) => {
return Ok(vec![control_error_frame(
&frame,
"duplicate_module_id",
format!(
"module_id '{module_id}' is already registered; duplicate HELLO rejected"
),
)?])
}
Err(err @ RegistryError::PathHazardModuleId { .. }) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_module_id",
err.to_string(),
)?])
}
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
"registry_error",
err.to_string(),
)?])
}
};
if let Some(sink) = sink {
let concurrency = manifest_concurrency(®istration.manifest);
if let Err(err) = self.forwarding.register_module_connection(
connection_id,
registration.manifest.module_id.clone(),
negotiated_ver,
concurrency,
sink,
) {
if matches!(self.deregister_connection(connection_id), Ok(r) if !r.is_empty()) {
crate::supervise::notify_registration_release();
}
return Ok(vec![control_error_frame(
&frame,
forwarding_error_code(&err),
err.to_string(),
)?]);
}
}
if manifest_concurrency_was_defaulted(&frame.body, ®istration.manifest) {
info!(
module_id = %registration.manifest.module_id,
"management surface registered with DEFAULTED concurrency=module_managed (manifest predates the field; declare the real lane)"
);
}
self.apply_registration_capabilities(®istration);
info!(
module_id = %registration.manifest.module_id,
module_version = %registration.manifest.module_version,
negotiated_ver,
routable_provider = manifest_provides_routable_role(®istration.manifest),
connection_id = connection_id.get(),
"module registered"
);
self.hello_ack(&frame, negotiated_ver, ®istration.manifest.module_id)
}
fn register_swap_candidate(
&self,
connection_id: ConnectionId,
sink: Option<crate::FrameSink>,
frame: &Frame,
manifest: ModuleManifest,
negotiated_ver: u8,
control_ops: Vec<String>,
) -> Result<Vec<Frame>, RouterError> {
let module_id = manifest.module_id.clone();
let registration = match self.registry.register_candidate_with_control_ops(
manifest,
negotiated_ver,
connection_id,
control_ops,
) {
Ok(registration) => registration,
Err(RegistryError::DuplicateModuleId { module_id }) => {
return Ok(vec![control_error_frame(
frame,
"duplicate_module_id",
format!(
"module_id '{module_id}' already has a swap candidate registered; duplicate HELLO rejected"
),
)?])
}
Err(err @ RegistryError::PathHazardModuleId { .. }) => {
return Ok(vec![control_error_frame(
frame,
"invalid_module_id",
err.to_string(),
)?])
}
Err(err) => {
return Ok(vec![control_error_frame(
frame,
"registry_error",
err.to_string(),
)?])
}
};
if let Some(sink) = sink {
let concurrency = manifest_concurrency(®istration.manifest);
if let Err(err) = self.forwarding.register_candidate_module_connection(
connection_id,
module_id.clone(),
negotiated_ver,
concurrency,
sink,
) {
if matches!(self.deregister_connection(connection_id), Ok(r) if !r.is_empty()) {
crate::supervise::notify_registration_release();
}
return Ok(vec![control_error_frame(
frame,
forwarding_error_code(&err),
err.to_string(),
)?]);
}
}
self.supervisor.mark_swap_candidate_admitted(&module_id);
info!(
module_id = %module_id,
module_version = %registration.manifest.module_version,
negotiated_ver,
ready = registration.ready,
connection_id = connection_id.get(),
"swap candidate registered; not routable until cutover"
);
self.hello_ack(frame, negotiated_ver, &module_id)
}
fn hello_ack(
&self,
frame: &Frame,
negotiated_ver: u8,
module_id: &str,
) -> Result<Vec<Frame>, RouterError> {
let ack = ModuleHelloAckBody {
negotiated_ver,
subc_ops: module_subc_ops(),
subc_capabilities: self.subc_capabilities.as_ref().to_vec(),
storage: self
.storage_config
.as_ref()
.map(|cfg| cfg.descriptor_for(module_id)),
machine_id: self.machine_id.as_ref().map(|id| id.as_str().to_owned()),
};
let body = serde_json::to_vec(&ack).map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to encode HELLO_ACK: {err}"),
)
})?;
Ok(vec![Frame::build_with_version(
negotiated_ver,
FrameType::HelloAck,
control_flags(),
0,
0,
frame.header.corr,
body,
)
.map_err(RouterError::FrameBuild)?])
}
async fn handle_client_control_request(
&self,
ctx: &RouteCtx,
frame: Frame,
request: ClientControlRequest,
) -> Result<Vec<Frame>, RouterError> {
match request {
ClientControlRequest::ServerDescribe {} => self.handle_server_describe(frame),
ClientControlRequest::CatalogList { module_id } => {
self.handle_catalog_list(frame, module_id)
}
ClientControlRequest::RouteOpen {
target,
identity,
consumer_identity,
consumer_capabilities,
admission_facts,
} => {
self.handle_route_open(
ctx,
frame,
RouteOpenRequest {
target,
identity,
consumer_identity,
consumer_capabilities,
admission_facts,
},
)
.await
}
ClientControlRequest::RoutePoll {
route_channel,
route_epoch,
kind,
} => self.handle_route_poll(ctx, frame, route_channel, route_epoch, kind),
ClientControlRequest::SupervisorList {} => self.handle_supervisor_list(frame),
ClientControlRequest::SupervisorSpawnSnapshot {} => {
self.handle_supervisor_spawn_snapshot(frame)
}
ClientControlRequest::SupervisorSpawnSubscribe { since } => {
self.handle_supervisor_spawn_subscribe(ctx, frame, since)
}
ClientControlRequest::SupervisorRestart {
module_id,
drain_timeout_ms,
} => {
self.handle_supervisor_restart(frame, module_id, drain_timeout_ms)
.await
}
ClientControlRequest::SupervisorSwap {
module_id,
ready_timeout_ms,
} => {
self.handle_supervisor_swap(frame, module_id, ready_timeout_ms)
.await
}
ClientControlRequest::SupervisorReload { module_id } => {
self.handle_supervisor_reload(frame, module_id).await
}
ClientControlRequest::SupervisorRescan { preview } => {
self.handle_supervisor_rescan(frame, preview).await
}
ClientControlRequest::SupervisorReleaseReserved { module_id } => {
self.handle_supervisor_release_reserved(frame, module_id)
.await
}
ClientControlRequest::SupervisorSetEnabled { module_id, enabled } => {
self.handle_supervisor_set_enabled(frame, module_id, enabled)
.await
}
ClientControlRequest::SupervisorHealthProbe { module_id } => {
self.handle_supervisor_health_probe(frame, module_id).await
}
ClientControlRequest::SupervisorHealth {} => self.handle_supervisor_health(frame),
ClientControlRequest::SupervisorRoutes { module_id } => {
self.handle_supervisor_routes(frame, module_id)
}
ClientControlRequest::SupervisorProvenance { module_id } => {
self.handle_supervisor_provenance(frame, module_id).await
}
ClientControlRequest::SupervisorStderrTail {
module_id,
max_lines,
max_bytes,
} => self.handle_supervisor_stderr_tail(frame, module_id, max_lines, max_bytes),
ClientControlRequest::SupervisorTerminals { module_id } => {
self.handle_supervisor_terminals(frame, module_id)
}
}
}
fn handle_module_control_request(
&self,
connection_id: ConnectionId,
frame: Frame,
request: ModuleControlRequestFromModule,
) -> Result<Vec<Frame>, RouterError> {
match request {
ModuleControlRequestFromModule::CatalogUpdate {
provides,
capabilities,
ready,
} => self.handle_catalog_update(connection_id, frame, provides, capabilities, ready),
ModuleControlRequestFromModule::LiveRoots {} => {
let registered = self
.registry
.get_module_by_connection(connection_id)
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?;
let Some(registration) = registered else {
return Ok(vec![control_error_frame(&frame, "not_registered", "supervisor.live_roots requires an active module registration owned by this connection")?]);
};
let response = self
.forwarding
.live_roots(®istration.manifest.module_id)
.map_err(RouterError::Forwarding)?;
Ok(vec![control_response_body_frame(
&frame,
&response,
"ModuleControlResponseToModule::LiveRoots",
)?])
}
}
}
fn handle_catalog_update(
&self,
connection_id: ConnectionId,
frame: Frame,
provides: Vec<ProviderRole>,
capabilities: Option<CapabilityDeclarations>,
ready: Option<bool>,
) -> Result<Vec<Frame>, RouterError> {
self.refresh_capability_requirements();
let Some(registration) = self
.registry
.get_module_by_connection(connection_id)
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?
else {
return Ok(vec![control_error_frame(
&frame,
"not_registered",
"catalog.update requires an active module registration owned by this connection",
)?]);
};
if let Some(message) =
catalog_update_frozen_field_message(®istration.manifest, &provides)
{
return Ok(vec![control_error_frame(
&frame,
"catalog_update_frozen_field",
message,
)?]);
}
let mut candidate = registration.manifest.clone();
candidate.provides = provides.clone();
candidate.capabilities = capabilities
.clone()
.or_else(|| registration.manifest.capabilities.clone());
if let Err(err) = candidate.validate_capability_grammar() {
return Ok(vec![control_error_frame(
&frame,
"invalid_capability_grammar",
err.to_string(),
)?]);
}
let updated = self
.registry
.replace_catalog_for_connection(connection_id, provides, capabilities, ready)
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?;
if updated.is_none() {
return Ok(vec![control_error_frame(
&frame,
"not_registered",
"catalog.update requires an active module registration owned by this connection",
)?]);
}
if let Ok((_, registrations)) = self.runtime_capability_snapshot() {
log_duplicate_claim_events(
self.capability_evaluator
.duplicate_claims(DuplicateClaimSource::CatalogUpdate, ®istrations),
);
}
if capability_census_trigger(
registration.manifest.capabilities.as_ref(),
updated
.as_ref()
.and_then(|entry| entry.manifest.capabilities.as_ref()),
) {
self.enforce_capability_denies();
}
self.refresh_capability_requirements();
let response = ModuleControlResponseToModule::CatalogUpdate {};
control_response_body_frame(
&frame,
&response,
"ModuleControlResponseToModule::CatalogUpdate",
)
.map(|frame| vec![frame])
}
fn handle_server_describe(&self, frame: Frame) -> Result<Vec<Frame>, RouterError> {
self.refresh_capability_requirements();
let mut counters = self.counters.snapshot();
if let (Ok((connections_with_routes, max)), Some(obj)) = (
self.forwarding.client_route_concentration(),
counters.as_object_mut(),
) {
obj.insert(
"client_connections_with_routes".into(),
connections_with_routes.into(),
);
obj.insert("max_routes_on_one_connection".into(), max.into());
}
if let (Some(open_breakers), Some(obj)) = (
self.route_bind_breakers.open_snapshot(),
counters.as_object_mut(),
) {
obj.insert("route_bind_breakers_open".into(), open_breakers);
}
let response = ClientControlResponse::ServerDescribe {
protocol_ver: PROTOCOL_VERSION,
subc_ops: subc_ops(),
capabilities: self.subc_capabilities.as_ref().to_vec(),
connected_clients: self.connected_clients.count(),
counters: Some(counters),
build_git_sha: Some(env!("SUBC_BUILD_GIT_SHA").to_string()),
build_lock_digest: Some(env!("SUBC_BUILD_LOCK_DIGEST").to_string()),
capability_requirements: self.capability_requirement_statuses(),
machine_id: self.machine_id.as_ref().map(|id| id.as_str().to_owned()),
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::ServerDescribe",
)?])
}
fn handle_catalog_list(
&self,
frame: Frame,
module_id: Option<String>,
) -> Result<Vec<Frame>, RouterError> {
let (generation, modules) = self.registry.list_modules().map_err(|err| {
RouterError::backend(0, frame.header.corr, format!("registry error: {err}"))
})?;
let entries = modules
.into_iter()
.filter(|registration| {
module_id
.as_deref()
.map(|wanted| registration.manifest.module_id == wanted)
.unwrap_or(true)
})
.map(|registration| {
let not_ready = self.not_ready_reason(®istration);
let roles = registration.manifest.provides;
CatalogEntry {
module_id: registration.manifest.module_id,
ready: not_ready.is_none(),
not_ready,
module_version: Some(registration.manifest.module_version),
roles,
control_ops: registration.control_ops,
capabilities: registration.manifest.capabilities,
self_signals: registration.manifest.self_signals,
}
})
.collect();
let response = ClientControlResponse::CatalogList {
generation,
modules: entries,
subc_ops: subc_ops(),
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::CatalogList",
)?])
}
fn route_open_principal(
&self,
frame: &Frame,
consumer_identity: Option<ConsumerIdentity>,
) -> Result<Result<Principal, Frame>, RouterError> {
let Some(consumer_identity) = consumer_identity else {
return Ok(Ok(Principal::Direct));
};
if self.supervisor.spawned_consumer_authorized(
&consumer_identity.module_id,
&consumer_identity.launch_nonce,
) {
return Ok(Ok(Principal::Reserved {
module_id: consumer_identity.module_id,
}));
}
Ok(Err(control_error_frame(
frame,
"bad_consumer_identity",
format!(
"consumer_identity for module_id '{}' did not match a supervised launch nonce",
consumer_identity.module_id
),
)?))
}
fn route_open_refusal_frame(
&self,
ctx: &RouteCtx,
frame: &Frame,
module_id: &str,
reason: &'static str,
code: &'static str,
message: impl Into<String>,
) -> Result<Frame, RouterError> {
self.observe_route_open_refusal(ctx, module_id, reason, code);
control_error_frame(frame, code, message.into())
}
fn route_open_breaker_refusal_frame(
&self,
ctx: &RouteCtx,
frame: &Frame,
module_id: &str,
consecutive_timeouts: u32,
retry_in: Duration,
probe_in_flight: bool,
) -> Result<Frame, RouterError> {
self.counters
.increment_route_open_refused(crate::observability::ROUTE_OPEN_REFUSED_BREAKER_OPEN);
debug!(
target: "control",
code = "module_timeout",
module_id = ?module_id,
connection_id = ctx.connection_id.get(),
consecutive_timeouts,
retry_in_ms = retry_in.as_millis() as u64,
probe_in_flight,
"route.open refused by open bind-relay breaker"
);
let detail = if probe_in_flight {
"one probe bind is already in flight; retry once it settles".to_string()
} else {
format!("not relaying for another {retry_in:?}")
};
control_error_frame(
frame,
"module_timeout",
format!(
"module_id '{module_id}' failed {consecutive_timeouts} consecutive route.bind \
relays; {detail}"
),
)
}
fn observe_route_open_refusal(
&self,
ctx: &RouteCtx,
module_id: &str,
reason: &'static str,
code: &'static str,
) {
self.counters.increment_route_open_refused(code);
info!(
target: "control",
code,
reason,
module_id = ?module_id,
connection_id = ctx.connection_id.get(),
"route.open refused"
);
}
fn observe_route_open_accept(&self, ctx: &RouteCtx, module_id: &str, principal: &str) {
self.counters.increment_route_open_accepted(principal);
info!(
target: "control",
principal,
module_id,
connection_id = ctx.connection_id.get(),
"route.open accepted"
);
}
fn supervised_absent_route_open_refusal_frame(
&self,
ctx: &RouteCtx,
frame: &Frame,
module_id: &str,
code: &'static str,
status: &crate::supervise::ModuleStatus,
) -> Result<Frame, RouterError> {
self.counters.increment_route_open_refused(code);
info!(
target: "control",
code,
reason = "supervised_not_registered",
module_id = ?module_id,
connection_id = ctx.connection_id.get(),
state = %status.state,
enabled = status.enabled,
live = status.live,
"route.open refused"
);
control_error_frame(
frame,
code,
format!(
"module_id '{module_id}' is supervised but not available (state={}, enabled={}, live={})",
status.state, status.enabled, status.live
),
)
}
async fn handle_route_open(
&self,
ctx: &RouteCtx,
frame: Frame,
request: RouteOpenRequest,
) -> Result<Vec<Frame>, RouterError> {
let RouteOpenRequest {
target,
mut identity,
consumer_identity,
consumer_capabilities,
admission_facts,
} = request;
let target_module_id = target_module_id(&target).to_string();
debug!(
connection_id = ctx.connection_id.get(),
corr = frame.header.corr,
module_id = %target_module_id,
"handling route.open"
);
let Some(registration) = self
.registry
.get_module(&target_module_id)
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?
else {
if let Some((status, warming)) =
self.supervisor_status(&target_module_id, frame.header.corr)?
{
if status.protocol == ModuleProtocol::None {
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"protocol_none",
error_codes::MODULE_NO_PROTOCOL,
format!(
"module_id '{target_module_id}' is declared protocol: none; \
it speaks no subc wire and serves no routes"
),
)?]);
}
let code = if warming {
"module_warming"
} else {
"target_unavailable"
};
return Ok(vec![self.supervised_absent_route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
code,
&status,
)?]);
}
if let Some(removed_ago_ms) =
self.supervisor.removal_tombstone_age_ms(&target_module_id)
{
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"removed",
error_codes::MODULE_REMOVED,
format!("module_id '{target_module_id}' was removed {removed_ago_ms} ms ago"),
)?]);
}
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"not_registered",
error_codes::UNKNOWN_MODULE,
format!("module_id '{target_module_id}' is not registered"),
)?]);
};
if !registration.ready {
self.counters
.increment_route_open_refused(ROUTE_OPEN_REFUSED_DECLARED_NOT_READY);
info!(
target: "control",
code = error_codes::MODULE_WARMING,
module_id = ?target_module_id,
connection_id = ctx.connection_id.get(),
reason = "declared_not_ready",
"route.open refused"
);
return Ok(vec![control_error_body_frame(
&frame,
ErrorBody {
code: error_codes::MODULE_WARMING.to_string(),
message: format!(
"module_id '{target_module_id}' is registered and has declared itself not ready; retry"
),
detail: Some(serde_json::json!({
"reason": "declared_not_ready"
})),
},
)?]);
}
if let Some(capability) = self.first_unprovided_required_capability(®istration) {
self.counters
.increment_route_open_refused(ROUTE_OPEN_REFUSED_REQUIRED_CAPABILITY_UNPROVIDED);
info!(
target: "control",
code = error_codes::MODULE_WARMING,
module_id = ?target_module_id,
connection_id = ctx.connection_id.get(),
reason = NotReadyReason::REQUIRED_CAPABILITY_UNPROVIDED,
capability = %capability,
"route.open refused"
);
return Ok(vec![control_error_body_frame(
&frame,
ErrorBody {
code: error_codes::MODULE_WARMING.to_string(),
message: format!(
"module_id '{target_module_id}' requires capability '{capability}', \
which no registered module provides; retry"
),
detail: Some(serde_json::json!({
"reason": NotReadyReason::REQUIRED_CAPABILITY_UNPROVIDED,
"capability": capability,
})),
},
)?]);
}
if !target_has_required_role(&target, ®istration.manifest.provides) {
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"role_not_provided",
"target_unavailable",
format!("module_id '{target_module_id}' does not provide the requested target"),
)?]);
}
if registration.state != ChannelState::Active {
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"registration_not_active",
"target_unavailable",
format!("module_id '{target_module_id}' is not active"),
)?]);
}
if self
.forwarding
.module_is_draining(&target_module_id)
.map_err(RouterError::Forwarding)?
{
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"reloading",
"module_reloading",
format!("module_id '{target_module_id}' is reloading"),
)?]);
}
if self
.process_liveness
.as_ref()
.and_then(|process_liveness| process_liveness.process_live(&target_module_id))
== Some(false)
{
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"supervisor_not_live",
"target_unavailable",
format!("module_id '{target_module_id}' is not live"),
)?]);
}
if !self
.forwarding
.has_live_module_connection(&target_module_id)
.map_err(RouterError::Forwarding)?
{
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"no_forwarding_connection",
"target_unavailable",
format!("module_id '{target_module_id}' has no live forwarding connection"),
)?]);
}
if let Some(error) =
self.guard_module_control_op(&frame, &target_module_id, "route.bind")?
{
self.observe_route_open_refusal(
ctx,
&target_module_id,
"op_not_allowed",
"op_not_allowed",
);
return Ok(vec![error]);
}
let principal = match self.route_open_principal(&frame, consumer_identity)? {
Ok(principal) => principal,
Err(error) => {
self.observe_route_open_refusal(
ctx,
&target_module_id,
"bad_consumer_identity",
"bad_consumer_identity",
);
return Ok(vec![error]);
}
};
if let Principal::Reserved {
module_id: opening_module_id,
} = &principal
{
if let Some(opening_registration) = self
.registry
.get_module(opening_module_id)
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?
{
if let Some(capability) =
denied_capability(&opening_registration.manifest, ®istration.manifest)
{
warn!(
opening_module_id,
target_module_id,
capability,
"refusing route.open because an attested capability deny edge matches"
);
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"capability_deny_edge",
"capability_forbidden",
format!(
"module_id '{opening_module_id}' must never reach capability '{capability}' provided by '{target_module_id}'"
),
)?]);
}
}
}
if admission_facts.is_some() {
let carrier_matches = matches!(
&principal,
Principal::Reserved { module_id }
if self.admission_facts_carrier_module_id.as_deref() == Some(module_id)
);
if !carrier_matches {
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"admission_facts_carrier_not_permitted",
"admission_facts_not_permitted",
"admission facts may only be carried by the configured reserved module",
)?]);
}
let target_allowed = self
.admission_facts_targets
.as_ref()
.is_some_and(|targets| targets.iter().any(|id| id == &target_module_id));
if !target_allowed {
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"admission_facts_target_not_listed",
"admission_facts_target_not_allowed",
format!(
"admission facts are not permitted for target module_id '{target_module_id}'"
),
)?]);
}
}
let project_root = match ProjectRootId::from_path_allowing_missing(&identity.project_root) {
Ok(project_root) => project_root,
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_project_root",
err.to_string(),
)?])
}
};
identity.project_root = project_root.as_path().to_path_buf();
let _concurrency_guard = match self
.route_bind_concurrency
.try_admit(&target_module_id, MAX_PENDING_ROUTE_BINDS_PER_TARGET)
{
Ok(guard) => guard,
Err(in_flight) => {
return Ok(vec![self.route_open_admission_refusal_frame(
ctx,
&frame,
&target_module_id,
format!(
"module_id '{target_module_id}' already has {in_flight} route.bind relays in flight; retry after one settles"
),
)?]);
}
};
let mut breaker = match self.route_bind_breakers.admit(&target_module_id) {
RouteBindAdmission::Admitted { guard, probe } => {
if probe {
info!(
module_id = %target_module_id,
connection_id = ctx.connection_id.get(),
"route.bind breaker half-open: admitting one probe"
);
}
guard
}
RouteBindAdmission::Refused {
consecutive_timeouts,
retry_in,
probe_in_flight,
} => {
return Ok(vec![self.route_open_breaker_refusal_frame(
ctx,
&frame,
&target_module_id,
consecutive_timeouts,
retry_in,
probe_in_flight,
)?]);
}
};
let route_bind_relay_timeout = self.route_bind_relay_timeout_for(&target_module_id);
let relay_deadline = Instant::now() + route_bind_relay_timeout;
let pending = match self
.forwarding
.begin_route_bind_relay_for(
ctx.connection_id,
ctx.egress.clone(),
response_version(&frame),
frame.header.corr,
&target_module_id,
principal.clone(),
Some(project_root),
relay_deadline,
)
.await
{
Ok(pending) => pending,
Err(err) => {
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"relay_reservation_failed",
forwarding_error_code(&err),
err.to_string(),
)?])
}
};
let crate::forwarding::PendingRouteBindRelay {
endpoint,
module_sink,
negotiated_ver,
client_channel,
client_epoch,
module_channel,
module_epoch,
corr: relay_corr,
receiver,
} = pending;
let mut reservation =
RouteBindReservationGuard::new(Arc::clone(&self.forwarding), endpoint, relay_corr);
debug!(
connection_id = ctx.connection_id.get(),
client_channel,
client_epoch,
module_channel,
module_epoch,
"reserved route handle pair"
);
let principal_label = match &principal {
Principal::Reserved { module_id } => format!("reserved:{module_id}"),
Principal::Direct => "direct".to_string(),
other => format!("{other:?}"),
};
let relay = ModuleControlRequest::RouteBind {
route_channel: module_channel,
epoch: module_epoch,
target,
identity,
principal: Some(principal),
consumer_capabilities,
admission_facts,
};
let relay_body = serde_json::to_vec(&relay).map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to encode route.bind request: {err}"),
)
})?;
let relay_frame = Frame::build_with_version(
negotiated_ver,
FrameType::Request,
control_flags(),
0,
0,
relay_corr,
relay_body,
)
.map_err(RouterError::FrameBuild)?;
if let Err(err) = module_sink.send(relay_frame).await {
reservation.release_and_disarm();
return Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"relay_send_failed",
"target_unavailable",
err.to_string(),
)?]);
}
if !self
.forwarding
.mark_route_bind_relay_enqueued(endpoint, relay_corr)
.map_err(RouterError::Forwarding)?
{
self.send_abandoned_route_bind_goodbye(
&module_sink,
negotiated_ver,
module_channel,
module_epoch,
);
}
match timeout_at(relay_deadline, receiver).await {
Ok(Ok(RouteBindRelayOutcome::Accepted)) => {
reservation.disarm();
if breaker.record_accepted() {
info!(
module_id = %target_module_id,
"route.bind breaker closed: the probe was accepted"
);
}
self.observe_route_open_accept(ctx, &target_module_id, &principal_label);
Ok(Vec::new())
}
Ok(Ok(RouteBindRelayOutcome::Rejected(body))) => {
reservation.release_and_disarm();
breaker.record_inconclusive();
self.counters
.increment_route_open_refused("module_rejected");
info!(
target: "control",
code = "module_rejected",
module_code = ?body.code,
module_id = ?target_module_id,
connection_id = ctx.connection_id.get(),
"route.open refused"
);
Ok(vec![control_error_body_frame(&frame, body)?])
}
Ok(Ok(RouteBindRelayOutcome::ModuleGone(message))) => {
reservation.release_and_disarm();
breaker.record_inconclusive();
tracing::warn!(
module_id = %target_module_id,
"route.bind relay abandoned: {message}"
);
Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"relay_abandoned",
"target_unavailable",
message,
)?])
}
Ok(Err(_)) => {
reservation.release_and_disarm();
breaker.record_inconclusive();
Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"relay_waiter_canceled",
"target_unavailable",
"route.bind relay waiter was canceled before the module responded",
)?])
}
Err(_) => {
reservation.release_and_disarm();
if let Some(opened) = breaker.record_timeout(
self.route_bind_breaker_threshold,
self.route_bind_breaker_cooldown,
) {
warn!(
module_id = %target_module_id,
consecutive_timeouts = opened.consecutive_timeouts,
cooldown_ms = self.route_bind_breaker_cooldown.as_millis() as u64,
reopened_after_probe = opened.reopened_after_probe,
"route.bind breaker open: refusing route.open for this module without relaying until one probe says it recovered"
);
}
tracing::warn!(
module_id = %target_module_id,
timeout_ms = route_bind_relay_timeout.as_millis() as u64,
"route.bind relay timed out: module did not ack within budget"
);
Ok(vec![self.route_open_refusal_frame(
ctx,
&frame,
&target_module_id,
"relay_timed_out",
"module_timeout",
format!(
"module_id '{target_module_id}' did not answer route.bind within {:?}",
route_bind_relay_timeout
),
)?])
}
}
}
fn handle_supervisor_spawn_snapshot(&self, frame: Frame) -> Result<Vec<Frame>, RouterError> {
let response = ClientControlResponse::SupervisorSpawnSnapshot {
snapshot: self.supervisor.spawn_snapshot(),
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorSpawnSnapshot",
)?])
}
fn handle_supervisor_spawn_subscribe(
&self,
ctx: &RouteCtx,
frame: Frame,
since: Option<SpawnCursor>,
) -> Result<Vec<Frame>, RouterError> {
match self.supervisor.subscribe_spawns(
ctx.connection_id,
frame.header.corr,
response_version(&frame),
since,
ctx.egress.clone(),
) {
Ok(()) => Ok(Vec::new()),
Err(SpawnSubscribeRefusal::ForeignIncarnation { current }) => {
Ok(vec![control_error_body_frame(
&frame,
ErrorBody {
code: "spawn_cursor_incarnation_mismatch".to_string(),
message: "spawn cursor belongs to a different daemon incarnation"
.to_string(),
detail: Some(serde_json::json!({
"current_daemon_incarnation": current
})),
},
)?])
}
Err(SpawnSubscribeRefusal::TooOld { oldest }) => Ok(vec![control_error_body_frame(
&frame,
ErrorBody {
code: "spawn_cursor_too_old".to_string(),
message: "spawn cursor predates the retained event ring".to_string(),
detail: Some(serde_json::json!({
"oldest_retained_cursor": oldest
})),
},
)?]),
Err(SpawnSubscribeRefusal::Frame(error)) => Err(RouterError::backend(
0,
frame.header.corr,
format!("failed to open supervisor spawn subscription: {error}"),
)),
}
}
fn handle_supervisor_list(&self, frame: Frame) -> Result<Vec<Frame>, RouterError> {
let generation = self
.registry
.generation()
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?;
let modules = self
.supervisor
.list()
.into_iter()
.map(|module| {
let status = module.status_for_control("list").map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to read supervisor status: {err}"),
)
})?;
Ok(SupervisorEntry {
module_id: status.module_id,
state: status.state.to_string(),
enabled: status.enabled,
live: status.live,
protocol: status.protocol,
health: status.health.status,
last_probe_ms: status.health.last_probe_ms,
last_exit_code: status.last_exit.as_ref().and_then(|e| e.code),
last_exit_signal: status.last_exit.as_ref().and_then(|e| e.signal),
last_exit_ms: status.last_exit.as_ref().map(|e| e.at_ms),
last_exit_kind: status.last_exit.as_ref().map(|e| e.kind.into()),
restart_count: Some(status.restart_count),
max_restarts: Some(status.max_restarts),
lifetime_restarts: Some(status.lifetime_restarts),
spawn_generation: Some(status.spawn_generation),
restart_window_secs: Some(status.restart_window.as_secs()),
drain_timeout_ms: Some(status.drain_timeout.as_millis() as u64),
restart_backoff_ms: Some(status.restart_backoff.as_millis() as u64),
restart_max_backoff_ms: Some(status.restart_max_backoff.as_millis() as u64),
})
})
.collect::<Result<Vec<_>, RouterError>>()?;
let response = ClientControlResponse::SupervisorList {
generation,
modules,
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorList",
)?])
}
fn handle_supervisor_stderr_tail(
&self,
frame: Frame,
module_id: String,
max_lines: Option<u32>,
max_bytes: Option<u32>,
) -> Result<Vec<Frame>, RouterError> {
let Some(module) = self.supervisor.get(&module_id) else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not supervised"),
)?]);
};
let snapshot = module.stderr_tail(
max_lines.map(|value| value as usize),
max_bytes.map(|value| value as usize),
);
let response = ClientControlResponse::SupervisorStderrTail {
module_id,
tail: StderrTail {
capture: match snapshot.capture {
CaptureState::Captured => StderrCaptureState::Captured,
CaptureState::Incomplete { reason } => {
StderrCaptureState::Incomplete { reason }
}
CaptureState::NotCaptured { reason } => {
StderrCaptureState::NotCaptured { reason }
}
},
entries: snapshot
.entries
.into_iter()
.map(|entry| match entry {
TailEntry::Line { text, truncated } => {
StderrTailEntry::Line { text, truncated }
}
TailEntry::ProcessStart => StderrTailEntry::ProcessStart,
})
.collect(),
dropped_lines: snapshot.dropped_lines,
},
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorStderrTail",
)?])
}
fn handle_supervisor_terminals(
&self,
frame: Frame,
module_id: String,
) -> Result<Vec<Frame>, RouterError> {
let Some(module) = self.supervisor.get(&module_id) else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not supervised"),
)?]);
};
let response = ClientControlResponse::SupervisorTerminals {
module_id,
terminals: module.durable_terminal_history(),
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorTerminals",
)?])
}
fn handle_supervisor_routes(
&self,
frame: Frame,
module_id: Option<String>,
) -> Result<Vec<Frame>, RouterError> {
let modules = self
.forwarding
.route_census(module_id.as_deref())
.map_err(RouterError::Forwarding)?
.into_iter()
.map(|(module_id, routes)| SupervisorRouteModule {
module_id,
routes: routes
.into_iter()
.map(|route| SupervisorRoute {
consumer: match route.principal {
Principal::Reserved { module_id } => {
SupervisorRouteConsumer::Reserved { module_id }
}
Principal::Direct | Principal::Unverified => {
SupervisorRouteConsumer::Direct {
connection_id: route.goodbye_target.connection_id.get(),
}
}
},
age_ms: Instant::now()
.saturating_duration_since(route.bound_at)
.as_millis()
.try_into()
.unwrap_or(u64::MAX),
draining: route.draining,
drain_reason: route.drain_reason,
})
.collect(),
})
.collect();
let response = ClientControlResponse::SupervisorRoutes { modules };
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorRoutes",
)?])
}
async fn handle_supervisor_provenance(
&self,
frame: Frame,
module_id: Option<String>,
) -> Result<Vec<Frame>, RouterError> {
let mut selected = if let Some(module_id) = module_id {
let Some(module) = self.supervisor.get(&module_id) else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not supervised"),
)?]);
};
vec![module]
} else {
self.supervisor.list()
};
let mut modules = Vec::with_capacity(selected.len());
for module in selected.drain(..) {
let status = module.status().map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to read supervisor status: {err}"),
)
})?;
let module_declared = self
.registry
.get_module(&status.module_id)
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?
.and_then(|registration| registration.manifest.provenance)
.map(|build| ModuleDeclaredProvenance::Reported { build })
.unwrap_or(ModuleDeclaredProvenance::Unverifiable);
#[cfg(test)]
let running_image = match &self.provenance_probe_override {
Some(result) => result.clone(),
None => module.running_image_agreement().await,
};
#[cfg(not(test))]
let running_image = module.running_image_agreement().await;
modules.push(SupervisorModuleProvenance {
module_id: status.module_id,
module_declared,
daemon_observed: SupervisorObservedProcess {
pid: status.pid,
spawned_at_ms: status.spawned_at_ms,
spawned_from: status.spawned_from,
running_image,
},
});
}
let daemon = SupervisorDaemonProvenance {
daemon_build: self.daemon_provenance.build.clone(),
daemon_observed: DaemonObservedProcess {
pid: self.daemon_provenance.pid,
started_at_ms: self
.daemon_provenance
.start_clock
.map(|clock| clock.started_at_ms())
.or(self.daemon_provenance.started_at_ms),
running_image: self
.daemon_provenance
.probe
.observe(
self.daemon_provenance.pid,
self.daemon_provenance.executable_path.as_deref(),
self.daemon_provenance.executable_identity,
self.daemon_provenance.process_start_time,
)
.await,
},
};
let response = ClientControlResponse::SupervisorProvenance { daemon, modules };
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorProvenance",
)?])
}
fn handle_supervisor_health(&self, frame: Frame) -> Result<Vec<Frame>, RouterError> {
self.refresh_capability_requirements();
let generation = self
.registry
.generation()
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?;
let modules = self
.supervisor
.list()
.into_iter()
.map(|module| {
let status = module.status_for_control("health").map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to read supervisor health: {err}"),
)
})?;
let module_id = status.module_id;
let capability_detail = self
.capability_evaluator
.required_problem_detail(&module_id);
Ok(SupervisorHealthEntry {
module_id,
status: status.health.status,
detail: append_capability_problem_detail(
status.health.detail,
capability_detail,
),
metrics: status.health.metrics,
consecutive_failures: status.health.consecutive_failures,
late_answer_count: status.health.late_answer_count,
last_late_answer_latency_ms: status.health.last_late_answer_latency_ms,
last_action: status.health.last_action,
last_action_ms: status.health.last_action_ms,
last_probe_ms: status.health.last_probe_ms,
})
})
.collect::<Result<Vec<_>, RouterError>>()?;
let response = ClientControlResponse::SupervisorHealth {
generation,
modules,
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorHealth",
)?])
}
async fn handle_supervisor_restart(
&self,
frame: Frame,
module_id: String,
drain_timeout_ms: Option<u64>,
) -> Result<Vec<Frame>, RouterError> {
let operation_lock = self.supervisor.operation_lock();
let _operation_guard = operation_lock.lock().await;
let Some(module) = self.supervisor.get(&module_id) else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not supervised"),
)?]);
};
if let Err(err) = module.restart(drain_timeout_ms).await {
let (code, message) = match err {
crate::supervise::SuperviseError::Disabled { .. } => {
("module_disabled", err.to_string())
}
crate::supervise::SuperviseError::SwapInProgress { .. } => {
("swap_in_progress", err.to_string())
}
_ => (
"target_unavailable",
format!("failed to restart module_id '{module_id}': {err}"),
),
};
return Ok(vec![control_error_frame(&frame, code, message)?]);
}
let response = ClientControlResponse::SupervisorAck {
module_id,
applied: true,
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorAck",
)?])
}
async fn handle_supervisor_swap(
&self,
frame: Frame,
module_id: String,
ready_timeout_ms: Option<u64>,
) -> Result<Vec<Frame>, RouterError> {
let module = {
let operation_lock = self.supervisor.operation_lock();
let _operation_guard = operation_lock.lock().await;
self.supervisor.get(&module_id)
};
let Some(module) = module else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not supervised"),
)?]);
};
if let Err(err) = module
.swap(ready_timeout_ms.map(Duration::from_millis))
.await
{
use crate::supervise::SuperviseError;
let message = err.to_string();
let error = match err {
SuperviseError::Disabled { .. } => ErrorBody::new("module_disabled", message),
SuperviseError::SwapRefused { reason, .. } => ErrorBody {
code: "swap_refused".to_string(),
message,
detail: Some(serde_json::json!({ "reason": reason.as_str() })),
},
SuperviseError::SwapFailed {
arm,
candidate_exit,
..
} => ErrorBody {
code: "swap_failed".to_string(),
message,
detail: Some(serde_json::json!({
"arm": arm.as_str(),
"candidate_exit_code": candidate_exit.as_ref().and_then(|exit| exit.code),
"candidate_exit_signal": candidate_exit.as_ref().and_then(|exit| exit.signal),
})),
},
_ => ErrorBody::new(
"target_unavailable",
format!("failed to swap module_id '{module_id}': {message}"),
),
};
return Ok(vec![control_error_body_frame(&frame, error)?]);
}
let response = ClientControlResponse::SupervisorAck {
module_id,
applied: true,
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorAck",
)?])
}
async fn handle_supervisor_reload(
&self,
frame: Frame,
module_id: String,
) -> Result<Vec<Frame>, RouterError> {
let operation_lock = self.supervisor.operation_lock();
let _operation_guard = operation_lock.lock().await;
let Some(module) = self.supervisor.get(&module_id) else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not supervised"),
)?]);
};
if let Err(err) = module.reload().await {
let (code, message) = match err {
crate::supervise::SuperviseError::Disabled { .. } => {
("module_disabled", err.to_string())
}
crate::supervise::SuperviseError::SwapInProgress { .. } => {
("swap_in_progress", err.to_string())
}
_ => (
"reload_failed",
format!("failed to reload module_id '{module_id}': {err}"),
),
};
return Ok(vec![control_error_frame(&frame, code, message)?]);
}
let response = ClientControlResponse::SupervisorAck {
module_id,
applied: true,
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorAck",
)?])
}
async fn handle_supervisor_rescan(
&self,
frame: Frame,
preview: bool,
) -> Result<Vec<Frame>, RouterError> {
let Some(context) = self.rescan.clone() else {
return Ok(vec![control_error_frame(
&frame,
"rescan_unavailable",
"the daemon was not started with a reloadable config path".to_string(),
)?]);
};
let operation_lock = self.supervisor.operation_lock();
let _operation_guard = operation_lock.lock().await;
let loaded = match crate::daemon_config::load(&context.config_path) {
Ok(config) => config,
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_daemon_config",
format!("supervisor rescan rejected daemon config: {err}"),
)?])
}
};
let Some(config) = loaded else {
return Ok(vec![control_error_frame(
&frame,
"invalid_daemon_config",
format!(
"daemon config not found at {}; refusing to rescan (an absent config would \
retire every supervised module)",
context.config_path.display()
),
)?]);
};
let (
configured_port,
storage_config,
admission_facts_carrier_module_id,
admission_facts_targets,
modules,
reserved_capabilities,
) = (
config.port,
config.storage,
config.admission_facts_carrier_module_id,
config.admission_facts_targets,
config.modules,
config.reserved_capabilities,
);
let mut restart_required = Vec::new();
for section in RestartRequiredSection::ALL {
let changed = match section {
RestartRequiredSection::Port => configured_port != context.configured_port,
RestartRequiredSection::Storage => storage_config != context.storage_config,
RestartRequiredSection::AdmissionFactsCarrierModuleId => {
admission_facts_carrier_module_id != context.admission_facts_carrier_module_id
}
RestartRequiredSection::AdmissionFactsTargets => {
admission_facts_targets != context.admission_facts_targets
}
};
if changed {
restart_required.push(section.label().to_string());
}
}
if !restart_required.is_empty() {
warn!(
config_path = %context.config_path.display(),
sections = %restart_required.join(", "),
"daemon config changed outside the modules section; restart the daemon to apply those changes"
);
}
for configured in &modules {
if let Err(err) = validate_spec(&configured.module_spec()) {
return Ok(vec![control_error_frame(
&frame,
"invalid_daemon_config",
format!("supervisor rescan rejected daemon config: {err}"),
)?]);
}
}
let configured_capabilities = modules
.iter()
.map(|module| (module.module_id.clone(), module.enabled))
.collect::<Vec<_>>();
let preview_capability_warnings = if preview {
let (_, registrations) = self.runtime_capability_snapshot()?;
let current_modules = self
.supervisor
.list()
.into_iter()
.map(|module| module.module_id().to_string())
.collect::<BTreeSet<_>>();
let resulting_modules = configured_capabilities.clone();
let removed = current_modules
.into_iter()
.filter(|module_id| {
!resulting_modules
.iter()
.any(|(configured_id, _)| configured_id == module_id)
})
.collect::<Vec<_>>();
self.capability_evaluator.preview_removal_warnings(
resulting_modules,
&removed,
®istrations,
)
} else {
Vec::new()
};
let result = match self
.reconcile_supervised_modules(&context.supervisor, modules, preview)
.await
{
Ok(result) => result,
Err(message) => {
return Ok(vec![control_error_frame(&frame, "rescan_failed", message)?])
}
};
if !preview {
self.capability_evaluator
.configure(configured_capabilities, reserved_capabilities);
self.capability_evaluator.wake_deadline_loop();
self.refresh_capability_requirements();
}
let mut result = result;
result.restart_required = restart_required;
result.capability_warnings = preview_capability_warnings;
let response = ClientControlResponse::SupervisorRescan { result };
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorRescan",
)?])
}
async fn handle_supervisor_release_reserved(
&self,
frame: Frame,
module_id: String,
) -> Result<Vec<Frame>, RouterError> {
let Some(context) = self.rescan.clone() else {
return Ok(vec![control_error_frame(
&frame,
"release_unavailable",
"reserved-id release requires a daemon started with a reloadable config path",
)?]);
};
let operation_lock = self.supervisor.operation_lock();
let _operation_guard = operation_lock.lock().await;
let loaded = match crate::daemon_config::load(&context.config_path) {
Ok(Some(config)) => config,
Ok(None) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_daemon_config",
format!(
"daemon config not found at {}; refusing to release reserved module_id '{module_id}'",
context.config_path.display()
),
)?])
}
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
"invalid_daemon_config",
format!("unable to verify reserved-id release against daemon config: {err}"),
)?])
}
};
if loaded
.modules
.iter()
.any(|configured| configured.module_id == module_id)
{
return Ok(vec![control_error_frame(
&frame,
"reserved_module_configured",
format!(
"module_id '{module_id}' remains configured; remove its config entry and rescan before releasing its reserved id"
),
)?]);
}
if !self.supervisor.release_retained_reserved_gate(&module_id) {
return Ok(vec![control_error_frame(
&frame,
"reserved_gate_not_retained",
format!(
"module_id '{module_id}' has no retired reserved-id gate to release; rescan its removed reserved configuration first"
),
)?]);
}
let response = ClientControlResponse::SupervisorAck {
module_id,
applied: true,
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorAck",
)?])
}
async fn reconcile_supervised_modules(
&self,
supervisor: &Supervisor,
configured_modules: Vec<crate::daemon_config::ConfiguredModule>,
preview: bool,
) -> Result<SupervisorRescanResult, String> {
let mut current = BTreeMap::new();
for module in self.supervisor.list() {
let (spec, health) = module.configuration().map_err(|err| {
format!(
"failed to read configuration for module_id '{}': {err}",
module.module_id()
)
})?;
let enabled = module
.status()
.map_err(|err| {
format!(
"failed to read status for module_id '{}': {err}",
module.module_id()
)
})?
.enabled;
current.insert(
module.module_id().to_string(),
(module, spec, health, enabled),
);
}
let configured = configured_modules
.into_iter()
.map(|module| (module.module_id.clone(), module))
.collect::<BTreeMap<_, _>>();
let added = configured
.keys()
.filter(|module_id| !current.contains_key(*module_id))
.cloned()
.collect::<Vec<_>>();
let removed = current
.keys()
.filter(|module_id| !configured.contains_key(*module_id))
.cloned()
.collect::<Vec<_>>();
let mut changed_pending_reload = Vec::new();
let mut configuration_changes = BTreeSet::new();
let mut enabled_changes = BTreeSet::new();
let mut unchanged = 0_u32;
for (module_id, configured_module) in &configured {
let Some((_, current_spec, current_health, current_enabled)) = current.get(module_id)
else {
continue;
};
let configuration_changed = *current_spec != configured_module.module_spec()
|| *current_health != configured_module.health;
let enabled_changed = *current_enabled != configured_module.enabled;
if configuration_changed {
configuration_changes.insert(module_id.clone());
changed_pending_reload.push(module_id.clone());
}
if enabled_changed {
enabled_changes.insert(module_id.clone());
}
if !configuration_changed && !enabled_changed {
unchanged = unchanged.saturating_add(1);
}
}
if preview {
return Ok(SupervisorRescanResult {
added,
removed,
changed_pending_reload,
enabled_changes: enabled_changes.iter().cloned().collect(),
unchanged,
preview: true,
restart_required: Vec::new(),
capability_warnings: Vec::new(),
});
}
for module_id in &removed {
let module = ¤t
.get(module_id)
.expect("removed module came from current supervisor state")
.0;
module.retire().await.map_err(|err| {
format!("failed to retire module_id '{module_id}' during rescan: {err}")
})?;
self.supervisor.record_rescan_removal(module_id);
self.supervisor.retire(module_id);
}
for module_id in configured.keys() {
let Some((module, _, _, _)) = current.get(module_id) else {
continue;
};
let configured_module = configured
.get(module_id)
.expect("configured module id came from configured map");
if configuration_changes.contains(module_id) {
module
.update_configuration(
configured_module.module_spec(),
configured_module.health,
configured_module.drain_timeout_ms,
)
.await
.map_err(|err| {
format!(
"failed to update module_id '{module_id}' configuration during rescan: {err}"
)
})?;
}
if enabled_changes.contains(module_id) {
module
.set_enabled(configured_module.enabled)
.await
.map_err(|err| {
format!(
"failed to apply module_id '{module_id}' enabled={} during rescan: {err}",
configured_module.enabled
)
})?;
}
}
for module_id in &added {
let configured_module = configured
.get(module_id)
.expect("added module id came from configured map");
supervisor
.supervise_configured_with_health(
configured_module.module_spec(),
configured_module.enabled,
configured_module.health,
configured_module.drain_timeout_ms,
configured_module.restart,
)
.map_err(|err| {
format!("failed to add module_id '{module_id}' during rescan: {err}")
})?;
}
Ok(SupervisorRescanResult {
added,
removed,
changed_pending_reload,
enabled_changes: enabled_changes.iter().cloned().collect(),
unchanged,
preview: false,
restart_required: Vec::new(),
capability_warnings: Vec::new(),
})
}
async fn handle_supervisor_set_enabled(
&self,
frame: Frame,
module_id: String,
enabled: bool,
) -> Result<Vec<Frame>, RouterError> {
let operation_lock = self.supervisor.operation_lock();
let _operation_guard = operation_lock.lock().await;
let Some(module) = self.supervisor.get(&module_id) else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not supervised"),
)?]);
};
let applied = match module.set_enabled(enabled).await {
Ok(applied) => applied,
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
"target_unavailable",
format!("failed to set module_id '{module_id}' enabled={enabled}: {err}"),
)?])
}
};
self.capability_evaluator.wake_deadline_loop();
self.refresh_capability_requirements();
let response = ClientControlResponse::SupervisorAck { module_id, applied };
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorAck",
)?])
}
async fn handle_supervisor_health_probe(
&self,
frame: Frame,
module_id: String,
) -> Result<Vec<Frame>, RouterError> {
self.refresh_capability_requirements();
let Some(registration) = self
.registry
.get_module(&module_id)
.map_err(|err| RouterError::backend(0, frame.header.corr, err.to_string()))?
else {
return Ok(vec![control_error_frame(
&frame,
"unknown_module",
format!("module_id '{module_id}' is not registered"),
)?]);
};
if !module_registration_grants_op(®istration.control_ops, MODULE_CONTROL_OP_HEALTH_CHECK)
{
return Ok(vec![control_error_frame(
&frame,
"health_not_advertised",
format!("module_id '{module_id}' did not advertise health.check"),
)?]);
}
let deadline = Instant::now() + self.health_probe_timeout;
let pending = match self.forwarding.begin_module_control_rpc_for(
&module_id,
MODULE_CONTROL_OP_HEALTH_CHECK,
deadline,
) {
Ok(pending) => pending,
Err(err) => {
return Ok(vec![control_error_frame(
&frame,
forwarding_error_code(&err),
err.to_string(),
)?])
}
};
let PendingModuleControlRpc {
endpoint,
module_sink,
negotiated_ver,
corr: probe_corr,
receiver,
} = pending;
let mut guard =
ModuleControlRpcGuard::new(Arc::clone(&self.forwarding), endpoint, probe_corr);
let probe_body =
serde_json::to_vec(&ModuleControlRequest::HealthCheck {}).map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to encode health.check request: {err}"),
)
})?;
let probe_frame = Frame::build_with_version(
negotiated_ver,
FrameType::Request,
control_flags(),
0,
0,
probe_corr,
probe_body,
)
.map_err(RouterError::FrameBuild)?;
if let Err(err) = module_sink.send(probe_frame).await {
return Ok(vec![control_error_frame(
&frame,
"target_unavailable",
err.to_string(),
)?]);
}
match timeout_at(deadline, receiver).await {
Ok(Ok(ModuleControlRpcOutcome::Response(response))) => {
guard.disarm();
let Some(report) = response.health_report() else {
return Ok(vec![control_error_frame(
&frame,
"invalid_control_body",
"health.check RPC returned a non-health response",
)?]);
};
let HealthReport {
status,
detail,
metrics,
} = report;
let capability_detail = self
.capability_evaluator
.required_problem_detail(&module_id);
let response = ClientControlResponse::SupervisorHealthProbe {
module_id,
status,
detail: append_capability_problem_detail(detail, capability_detail),
metrics,
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::SupervisorHealthProbe",
)?])
}
Ok(Ok(ModuleControlRpcOutcome::Rejected(body))) => {
guard.disarm();
Ok(vec![control_error_body_frame(&frame, body)?])
}
Ok(Ok(ModuleControlRpcOutcome::ModuleGone(message))) => {
guard.disarm();
Ok(vec![control_error_frame(
&frame,
"target_unavailable",
message,
)?])
}
Ok(Ok(ModuleControlRpcOutcome::MalformedResponse(message))) => {
guard.disarm();
Ok(vec![control_error_frame(
&frame,
"invalid_control_body",
message,
)?])
}
Ok(Ok(ModuleControlRpcOutcome::UnexpectedOp { expected, actual })) => {
guard.disarm();
Ok(vec![control_error_frame(
&frame,
"invalid_control_body",
format!("expected module-control op '{expected}', got '{actual}'"),
)?])
}
Ok(Ok(ModuleControlRpcOutcome::DeadlineElapsed)) => {
guard.disarm();
Ok(vec![control_error_frame(
&frame,
"module_timeout",
format!(
"module_id '{module_id}' answered health.check after {:?}",
self.health_probe_timeout
),
)?])
}
Ok(Err(_)) => Ok(vec![control_error_frame(
&frame,
"target_unavailable",
"health.check waiter was canceled before the module responded",
)?]),
Err(_) => Ok(vec![control_error_frame(
&frame,
"module_timeout",
format!(
"module_id '{module_id}' did not answer health.check within {:?}",
self.health_probe_timeout
),
)?]),
}
}
fn supervisor_status(
&self,
module_id: &str,
corr: u64,
) -> Result<Option<(crate::supervise::ModuleStatus, bool)>, RouterError> {
self.supervisor
.get(module_id)
.map(|module| {
let warming = module.is_warming_for_control("status").map_err(|err| {
RouterError::backend(
0,
corr,
format!(
"failed to read supervisor warming state for module_id '{module_id}': {err}"
),
)
})?;
module.status_for_control("status").map_err(|err| {
RouterError::backend(
0,
corr,
format!(
"failed to read supervisor status for module_id '{module_id}': {err}"
),
)
}).map(|status| (status, warming))
})
.transpose()
}
fn guard_module_control_op(
&self,
frame: &Frame,
module_id: &str,
op: &str,
) -> Result<Option<Frame>, RouterError> {
if self.module_grants_op(module_id, op, frame.header.corr)? {
return Ok(None);
}
Ok(Some(control_error_frame(
frame,
"op_not_allowed",
format!("module_id '{module_id}' did not grant control op '{op}'"),
)?))
}
fn module_grants_op(&self, module_id: &str, op: &str, corr: u64) -> Result<bool, RouterError> {
let Some(registration) = self
.registry
.get_module(module_id)
.map_err(|err| RouterError::backend(0, corr, err.to_string()))?
else {
return Ok(false);
};
Ok(module_registration_grants_op(®istration.control_ops, op))
}
fn handle_status_update(
&self,
endpoint: ModuleEndpointId,
frame: Frame,
) -> Result<Vec<Frame>, RouterError> {
let update = match serde_json::from_slice::<ModuleControlPush>(&frame.body) {
Ok(update) => update,
Err(err) => {
if is_known_module_push_op(&frame.body) {
return Ok(vec![control_error_frame(
&frame,
"invalid_control_body",
format!("malformed module control push body: {err}"),
)?]);
}
return Ok(Vec::new());
}
};
match update {
ModuleControlPush::RouteStatus {
route_channel,
route_epoch,
status,
} => {
self.forwarding
.cache_status(endpoint, route_channel, route_epoch, status)
.map_err(RouterError::Forwarding)?;
}
}
Ok(Vec::new())
}
fn handle_route_poll(
&self,
ctx: &RouteCtx,
frame: Frame,
route_channel: u16,
route_epoch: u32,
kind: PollKind,
) -> Result<Vec<Frame>, RouterError> {
let snapshot = self
.forwarding
.route_poll_snapshot(ctx.connection_id, route_channel, route_epoch)
.map_err(RouterError::Forwarding)?;
let response = match (kind, snapshot) {
(PollKind::Status, RoutePollSnapshot::Bound { status, .. }) => {
ClientControlResponse::RoutePoll {
route_channel,
route_epoch,
status,
live: None,
}
}
(PollKind::Status, RoutePollSnapshot::Absent) => ClientControlResponse::RoutePoll {
route_channel,
route_epoch,
status: None,
live: None,
},
(PollKind::Liveness, RoutePollSnapshot::Bound { module_id, .. }) => {
let live = self
.process_liveness
.as_ref()
.and_then(|source| source.process_live(&module_id))
.unwrap_or(true);
ClientControlResponse::RoutePoll {
route_channel,
route_epoch,
status: None,
live: Some(live),
}
}
(PollKind::Liveness, RoutePollSnapshot::Absent) => ClientControlResponse::RoutePoll {
route_channel,
route_epoch,
status: None,
live: Some(false),
},
};
Ok(vec![control_response_body_frame(
&frame,
&response,
"ClientControlResponse::RoutePoll",
)?])
}
pub(crate) fn observe_module_control_completion(
&self,
completion: ModuleControlRpcCompletion,
) -> bool {
match completion {
ModuleControlRpcCompletion::Unknown => false,
ModuleControlRpcCompletion::Settled => true,
ModuleControlRpcCompletion::LateHealthAnswer { module_id, latency } => {
let latency_ms = latency.as_millis().min(u128::from(u64::MAX)) as u64;
info!(
module_id = %module_id,
latency_ms,
"late health.check answer proves the module is alive"
);
match self
.supervisor
.record_late_health_answer(&module_id, latency_ms)
{
Ok(true) => {}
Ok(false) => debug!(
module_id = %module_id,
latency_ms,
"late health.check answer has no active supervisor snapshot"
),
Err(err) => warn!(
module_id = %module_id,
latency_ms,
error = %err,
"failed to record late health.check answer"
),
}
true
}
}
}
fn refuse_to_end_module_connection_for_a_client(
&self,
module_connection_id: ConnectionId,
corr: u64,
err: ForwardingError,
) -> Result<(), RouterError> {
if let ForwardingError::ConnectionClosing { connection_id } = err {
if connection_id != module_connection_id {
warn!(
module_connection_id = module_connection_id.get(),
client_connection_id = connection_id.get(),
corr,
"dropping a route.bind response for a closing client; the module connection keeps serving"
);
return Ok(());
}
}
Err(RouterError::Forwarding(err))
}
fn handle_module_relay_response(
&self,
connection_id: ConnectionId,
frame: Frame,
) -> Result<Vec<Frame>, RouterError> {
let mut secondary_error = None;
let outcome = match frame.header.ty {
FrameType::Response => match serde_json::from_slice::<ControlOpProbe>(&frame.body) {
Ok(probe) if probe.op == "route.bind" => {
match serde_json::from_slice::<ModuleControlResponse>(&frame.body) {
Ok(ModuleControlResponse::RouteBindAck {}) => {
RouteBindRelayOutcome::Accepted
}
Ok(other) => {
let message =
format!("route.bind response carried unexpected body: {other:?}");
secondary_error = Some(control_error_frame(
&frame,
"invalid_control_body",
message.clone(),
)?);
RouteBindRelayOutcome::ModuleGone(message)
}
Err(err) => {
let message = format!("malformed route.bind response body: {err}");
secondary_error = Some(control_error_frame(
&frame,
"invalid_control_body",
message.clone(),
)?);
RouteBindRelayOutcome::ModuleGone(message)
}
}
}
Ok(probe) => {
let outcome = match serde_json::from_slice::<ModuleControlResponse>(&frame.body)
{
Ok(response) => ModuleControlRpcOutcome::Response(response),
Err(err) => ModuleControlRpcOutcome::MalformedResponse(format!(
"malformed {} response body: {err}",
probe.op
)),
};
let completion = self
.forwarding
.complete_module_control_rpc(
connection_id,
frame.header.corr,
Some(&probe.op),
outcome,
)
.map_err(RouterError::Forwarding)?;
if !self.observe_module_control_completion(completion) {
debug!(
connection_id = connection_id.get(),
corr = frame.header.corr,
op = %probe.op,
"dropping late or unknown module-control RPC response"
);
}
return Ok(Vec::new());
}
Err(err) => {
if let Some(expected_op) = self
.forwarding
.pending_module_control_op(connection_id, frame.header.corr)
.map_err(RouterError::Forwarding)?
{
let completion = self
.forwarding
.complete_module_control_rpc(
connection_id,
frame.header.corr,
None,
ModuleControlRpcOutcome::MalformedResponse(format!(
"malformed {expected_op} response body: {err}"
)),
)
.map_err(RouterError::Forwarding)?;
if !self.observe_module_control_completion(completion) {
debug!(
connection_id = connection_id.get(),
corr = frame.header.corr,
"dropping late malformed module-control RPC response"
);
}
return Ok(Vec::new());
}
let message = format!("malformed route.bind response body: {err}");
secondary_error = Some(control_error_frame(
&frame,
"invalid_control_body",
message.clone(),
)?);
RouteBindRelayOutcome::ModuleGone(message)
}
},
FrameType::Error => {
if self
.forwarding
.pending_module_control_op(connection_id, frame.header.corr)
.map_err(RouterError::Forwarding)?
.is_some()
{
let outcome = match serde_json::from_slice::<ErrorBody>(&frame.body) {
Ok(body) => ModuleControlRpcOutcome::Rejected(body),
Err(err) => ModuleControlRpcOutcome::MalformedResponse(format!(
"malformed module-control ERROR body: {err}"
)),
};
let completion = self
.forwarding
.complete_module_control_rpc(
connection_id,
frame.header.corr,
None,
outcome,
)
.map_err(RouterError::Forwarding)?;
if !self.observe_module_control_completion(completion) {
debug!(
connection_id = connection_id.get(),
corr = frame.header.corr,
"dropping late or unknown module-control RPC error"
);
}
return Ok(Vec::new());
}
match serde_json::from_slice::<ErrorBody>(&frame.body) {
Ok(body) => RouteBindRelayOutcome::Rejected(body),
Err(err) => {
let message = format!("malformed route.bind ERROR body: {err}");
secondary_error = Some(control_error_frame(
&frame,
"invalid_control_body",
message.clone(),
)?);
RouteBindRelayOutcome::ModuleGone(message)
}
}
}
ty => {
return Ok(vec![control_error_frame(
&frame,
"unsupported_control_frame",
format!("unsupported module channel-0 frame {ty:?}"),
)?])
}
};
let settled =
self.forwarding
.complete_pending_relay(connection_id, frame.header.corr, outcome);
let completion = match settled {
Ok(completion) => completion,
Err(err) => {
self.refuse_to_end_module_connection_for_a_client(
connection_id,
frame.header.corr,
err,
)?;
return Ok(secondary_error.into_iter().collect());
}
};
if let Some(target) = completion.abandoned.as_ref() {
send_goodbye_target_best_effort(target, "late accepted route.bind");
}
if !completion.settled {
debug!(
connection_id = connection_id.get(),
corr = frame.header.corr,
frame_type = ?frame.header.ty,
"dropping late or unknown route.bind relay response"
);
}
Ok(secondary_error.into_iter().collect())
}
fn handle_goodbye(&self, connection_id: ConnectionId) -> Result<Vec<Frame>, RouterError> {
debug!(connection_id = connection_id.get(), "handling GOODBYE");
let registrations = self
.deregister_connection(connection_id)
.map_err(|err| RouterError::backend(0, 0, err.to_string()))?;
let released_routes = self
.forwarding
.cleanup_connection(connection_id)
.map_err(RouterError::Forwarding)?;
self.emit_route_goodbyes(released_routes);
if !registrations.is_empty() {
crate::supervise::notify_registration_release();
}
Ok(Vec::new())
}
}
impl Default for ControlHandler {
fn default() -> Self {
Self::new(Arc::new(Registry::default()))
}
}
impl crate::supervise::SwapPromotionObserver for ControlHandler {
fn swap_promoted(&self, registration: &crate::registry::ModuleRegistration) {
self.apply_registration_capabilities(registration);
}
}
fn capability_requirement_status(status: RequirementStatus) -> CapabilityRequirementStatus {
CapabilityRequirementStatus {
consumer: status.consumer,
capability: status.capability,
need: match status.need {
subc_protocol::manifest::CapabilityNeed::Required => "required".to_string(),
subc_protocol::manifest::CapabilityNeed::Optional => "optional".to_string(),
},
verdict: status.verdict.as_str().to_string(),
episode_seq: status.episode_seq,
config_satisfiable: status.config_satisfiable,
runtime_available: status.runtime_available,
detail: status.detail,
}
}
fn append_capability_problem_detail(
detail: Option<String>,
capability_detail: Option<String>,
) -> Option<String> {
match (detail, capability_detail) {
(Some(detail), Some(capability_detail)) => Some(format!("{detail}; {capability_detail}")),
(Some(detail), None) => Some(detail),
(None, Some(capability_detail)) => Some(capability_detail),
(None, None) => None,
}
}
fn subc_ops() -> Vec<String> {
SUBC_CONTROL_OPS
.iter()
.map(|op| (*op).to_string())
.collect()
}
fn module_subc_ops() -> Vec<String> {
SUBC_CONTROL_OPS
.iter()
.chain(MODULE_TO_SUBC_CONTROL_OPS.iter())
.map(|op| (*op).to_string())
.collect()
}
#[cfg(test)]
fn module_baseline_control_ops() -> Vec<String> {
MODULE_BASELINE_CONTROL_OPS
.iter()
.map(|op| (*op).to_string())
.collect()
}
fn effective_module_control_ops(declared: Option<Vec<String>>) -> Vec<String> {
let mut seen = HashSet::new();
let mut effective = Vec::new();
for op in MODULE_BASELINE_CONTROL_OPS {
if seen.insert((*op).to_string()) {
effective.push((*op).to_string());
}
}
for op in declared.unwrap_or_default() {
if seen.insert(op.clone()) {
effective.push(op);
}
}
effective
}
fn module_registration_grants_op(control_ops: &[String], op: &str) -> bool {
MODULE_BASELINE_CONTROL_OPS.contains(&op) || control_ops.iter().any(|granted| granted == op)
}
fn target_module_id(target: &RouteTarget) -> &str {
match target {
RouteTarget::ToolProvider { module_id }
| RouteTarget::ManagementSurface { module_id }
| RouteTarget::InternalService { module_id, .. } => module_id,
}
}
fn target_has_required_role(target: &RouteTarget, roles: &[ProviderRole]) -> bool {
roles.iter().any(|role| match (target, role) {
(RouteTarget::ToolProvider { .. }, ProviderRole::ToolProvider { .. }) => true,
(RouteTarget::ManagementSurface { .. }, ProviderRole::ManagementSurface { .. }) => true,
(
RouteTarget::InternalService { service_id, .. },
ProviderRole::InternalService {
service_id: provided,
..
},
) => service_id == provided,
_ => false,
})
}
fn is_routable_role(role: &ProviderRole) -> bool {
matches!(
role,
ProviderRole::ToolProvider { .. }
| ProviderRole::ManagementSurface { .. }
| ProviderRole::InternalService { .. }
)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ControlRequestBodyError {
UnknownOp,
InvalidBody,
}
#[derive(Debug, Deserialize)]
struct ControlOpProbe {
op: String,
}
const MODULE_PUSH_OPS: &[&str] = &["route.status"];
fn is_known_module_push_op(body: &[u8]) -> bool {
serde_json::from_slice::<ControlOpProbe>(body)
.map(|probe| MODULE_PUSH_OPS.contains(&probe.op.as_str()))
.unwrap_or(false)
}
fn is_known_module_request_op(body: &[u8]) -> bool {
serde_json::from_slice::<ControlOpProbe>(body)
.map(|probe| MODULE_TO_SUBC_CONTROL_OPS.contains(&probe.op.as_str()))
.unwrap_or(false)
}
fn log_control_dispatch_arrival(op: &'static str, connection_id: ConnectionId, corr: u64) {
debug!(
op = %op,
connection_id = connection_id.get(),
corr,
"control dispatch"
);
}
fn log_slow_control_dispatch(
dispatch_started_at: Option<StdInstant>,
op: &'static str,
connection_id: ConnectionId,
corr: u64,
) {
let Some(dispatch_started_at) = dispatch_started_at else {
return;
};
let elapsed = dispatch_started_at.elapsed();
if elapsed >= SLOW_CONTROL_DISPATCH_THRESHOLD {
warn!(
op = %op,
connection_id = connection_id.get(),
corr,
elapsed_ms = elapsed.as_millis() as u64,
"slow control dispatch"
);
}
}
fn client_control_request_op(request: &ClientControlRequest) -> &'static str {
match request {
ClientControlRequest::ServerDescribe {} => ops::SERVER_DESCRIBE,
ClientControlRequest::SupervisorProvenance { .. } => ops::SUPERVISOR_PROVENANCE,
ClientControlRequest::CatalogList { .. } => ops::CATALOG_LIST,
ClientControlRequest::RouteOpen { .. } => ops::ROUTE_OPEN,
ClientControlRequest::RoutePoll { .. } => ops::ROUTE_POLL,
ClientControlRequest::SupervisorList {} => ops::SUPERVISOR_LIST,
ClientControlRequest::SupervisorSpawnSnapshot {} => ops::SUPERVISOR_SPAWN_SNAPSHOT,
ClientControlRequest::SupervisorSpawnSubscribe { .. } => ops::SUPERVISOR_SPAWN_SUBSCRIBE,
ClientControlRequest::SupervisorRestart { .. } => ops::SUPERVISOR_RESTART,
ClientControlRequest::SupervisorSwap { .. } => ops::SUPERVISOR_SWAP,
ClientControlRequest::SupervisorReload { .. } => ops::SUPERVISOR_RELOAD,
ClientControlRequest::SupervisorRescan { .. } => ops::SUPERVISOR_RESCAN,
ClientControlRequest::SupervisorReleaseReserved { .. } => ops::SUPERVISOR_RELEASE_RESERVED,
ClientControlRequest::SupervisorSetEnabled { .. } => ops::SUPERVISOR_SET_ENABLED,
ClientControlRequest::SupervisorHealthProbe { .. } => ops::SUPERVISOR_HEALTH_PROBE,
ClientControlRequest::SupervisorHealth {} => ops::SUPERVISOR_HEALTH,
ClientControlRequest::SupervisorRoutes { .. } => ops::SUPERVISOR_ROUTES,
ClientControlRequest::SupervisorStderrTail { .. } => ops::SUPERVISOR_STDERR_TAIL,
ClientControlRequest::SupervisorTerminals { .. } => ops::SUPERVISOR_TERMINALS,
}
}
fn module_control_request_op(request: &ModuleControlRequestFromModule) -> &'static str {
match request {
ModuleControlRequestFromModule::CatalogUpdate { .. } => MODULE_TO_SUBC_OP_CATALOG_UPDATE,
ModuleControlRequestFromModule::LiveRoots {} => "supervisor.live_roots",
}
}
fn parse_client_control_request(
body: &[u8],
) -> Result<ClientControlRequest, (serde_json::Error, ControlRequestBodyError)> {
serde_json::from_slice::<ClientControlRequest>(body).map_err(|err| {
let classification = match serde_json::from_slice::<ControlOpProbe>(body) {
Ok(probe) if SUBC_CONTROL_OPS.contains(&probe.op.as_str()) => {
ControlRequestBodyError::InvalidBody
}
Ok(_) => ControlRequestBodyError::UnknownOp,
Err(_) => ControlRequestBodyError::InvalidBody,
};
(err, classification)
})
}
fn parse_module_control_request_from_module(
body: &[u8],
) -> Result<ModuleControlRequestFromModule, (serde_json::Error, ControlRequestBodyError)> {
serde_json::from_slice::<ModuleControlRequestFromModule>(body).map_err(|err| {
let classification = match serde_json::from_slice::<ControlOpProbe>(body) {
Ok(probe) if MODULE_TO_SUBC_CONTROL_OPS.contains(&probe.op.as_str()) => {
ControlRequestBodyError::InvalidBody
}
Ok(_) => ControlRequestBodyError::UnknownOp,
Err(_) => ControlRequestBodyError::InvalidBody,
};
(err, classification)
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum ProviderRoleKind {
ToolProvider,
PipelineStage,
ManagementSurface,
InternalService,
}
fn provider_role_kind(role: &ProviderRole) -> ProviderRoleKind {
match role {
ProviderRole::ToolProvider { .. } => ProviderRoleKind::ToolProvider,
ProviderRole::PipelineStage { .. } => ProviderRoleKind::PipelineStage,
ProviderRole::ManagementSurface { .. } => ProviderRoleKind::ManagementSurface,
ProviderRole::InternalService { .. } => ProviderRoleKind::InternalService,
}
}
fn provider_role_kind_set(roles: &[ProviderRole]) -> BTreeSet<ProviderRoleKind> {
roles.iter().map(provider_role_kind).collect()
}
fn capability_census_trigger(
old: Option<&CapabilityDeclarations>,
new: Option<&CapabilityDeclarations>,
) -> bool {
let old_provides = old
.map(|capabilities| capabilities.provides.iter().collect::<HashSet<_>>())
.unwrap_or_default();
let old_denies = old
.map(|capabilities| capabilities.must_never_reach.iter().collect::<HashSet<_>>())
.unwrap_or_default();
let new = new.cloned().unwrap_or(CapabilityDeclarations {
provides: Vec::new(),
requires: Vec::new(),
must_never_reach: Vec::new(),
});
new.provides
.iter()
.any(|capability| !old_provides.contains(capability))
|| new
.must_never_reach
.iter()
.any(|capability| !old_denies.contains(capability))
}
fn denied_capability<'a>(
opening_manifest: &'a ModuleManifest,
target_manifest: &ModuleManifest,
) -> Option<&'a str> {
let opening_capabilities = opening_manifest.capabilities.as_ref()?;
let target_capabilities = target_manifest.capabilities.as_ref()?;
opening_capabilities
.must_never_reach
.iter()
.find(|denied| {
target_capabilities
.provides
.iter()
.any(|provided| provided == *denied)
})
.map(String::as_str)
}
fn catalog_update_frozen_field_message(
registered: &ModuleManifest,
provides: &[ProviderRole],
) -> Option<String> {
let old_has_provides = !registered.provides.is_empty();
let new_has_provides = !provides.is_empty();
if old_has_provides != new_has_provides {
return Some(format!(
"catalog.update cannot change module '{}' between supervision-only and routable; routability is fixed at HELLO",
registered.module_id
));
}
if provider_role_kind_set(®istered.provides) != provider_role_kind_set(provides) {
return Some(format!(
"catalog.update cannot change provider role kinds for module '{}'; role kinds are fixed at HELLO",
registered.module_id
));
}
let registered_concurrency = manifest_concurrency(registered);
let mut candidate = registered.clone();
candidate.provides = provides.to_vec();
let candidate_concurrency = manifest_concurrency(&candidate);
if candidate_concurrency != registered_concurrency {
return Some(format!(
"catalog.update cannot change module '{}' concurrency from {:?} to {:?}; concurrency is fixed at HELLO",
registered.module_id, registered_concurrency, candidate_concurrency
));
}
None
}
fn manifest_provides_routable_role(manifest: &ModuleManifest) -> bool {
manifest.provides.iter().any(is_routable_role)
}
fn manifest_concurrency(manifest: &ModuleManifest) -> Concurrency {
manifest
.provides
.iter()
.find_map(|provider| match provider {
ProviderRole::ToolProvider { concurrency, .. }
| ProviderRole::ManagementSurface { concurrency, .. } => Some(concurrency.clone()),
ProviderRole::PipelineStage { .. } | ProviderRole::InternalService { .. } => None,
})
.unwrap_or(Concurrency::ModuleManaged)
}
fn manifest_concurrency_was_defaulted(raw_hello: &[u8], manifest: &ModuleManifest) -> bool {
let has_management_surface = manifest
.provides
.iter()
.any(|provider| matches!(provider, ProviderRole::ManagementSurface { .. }));
if !has_management_surface {
return false;
}
let Ok(raw) = serde_json::from_slice::<serde_json::Value>(raw_hello) else {
return false;
};
let Some(provides) = raw
.get("manifest")
.and_then(|manifest| manifest.get("provides"))
.and_then(serde_json::Value::as_array)
else {
return false;
};
provides.iter().any(|role| {
role.get("role").and_then(serde_json::Value::as_str) == Some("management_surface")
&& role.get("concurrency").is_none()
})
}
fn negotiate_version(peer_version: u8) -> Result<u8, String> {
if peer_version != PROTOCOL_VERSION {
return Err(format!(
"protocol_ver {peer_version} is unsupported; this daemon requires exactly {PROTOCOL_VERSION}"
));
}
Ok(PROTOCOL_VERSION)
}
fn pong(frame: &Frame) -> Result<Frame, RouterError> {
Frame::build_with_version(
response_version(frame),
FrameType::Pong,
frame.header.flags,
0,
0,
frame.header.corr,
Vec::new(),
)
.map_err(RouterError::FrameBuild)
}
fn control_error_frame(
frame: &Frame,
code: &'static str,
message: impl Into<String>,
) -> Result<Frame, RouterError> {
control_error_body_frame(
frame,
ErrorBody {
code: code.to_string(),
message: message.into(),
detail: None,
},
)
}
fn control_error_body_frame(frame: &Frame, error: ErrorBody) -> Result<Frame, RouterError> {
let body = serde_json::to_vec(&error).map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to encode control ERROR: {err}"),
)
})?;
Frame::build_with_version(
response_version(frame),
FrameType::Error,
control_flags(),
0,
0,
frame.header.corr,
body,
)
.map_err(RouterError::FrameBuild)
}
fn control_response_body_frame<T: Serialize>(
frame: &Frame,
reply: &T,
label: &'static str,
) -> Result<Frame, RouterError> {
let body = serde_json::to_vec(reply).map_err(|err| {
RouterError::backend(
0,
frame.header.corr,
format!("failed to encode {label}: {err}"),
)
})?;
Frame::build_with_version(
response_version(frame),
FrameType::Response,
control_flags(),
0,
0,
frame.header.corr,
body,
)
.map_err(RouterError::FrameBuild)
}
fn forwarding_error_code(err: &ForwardingError) -> &'static str {
match err {
ForwardingError::NoModuleConnection => "target_unavailable",
ForwardingError::ModuleReloading { .. } => "module_reloading",
ForwardingError::ClientRouteChannelExhausted { .. }
| ForwardingError::ModuleRouteChannelExhausted { .. } => "route_limit",
ForwardingError::StaleModuleEndpoint
| ForwardingError::UnknownReservation { .. }
| ForwardingError::ConnectionClosing { .. }
| ForwardingError::ClientEgressClosed { .. } => "target_unavailable",
ForwardingError::CandidateSlotOccupied { .. } => "duplicate_module_id",
ForwardingError::RelayCorrelationExhausted
| ForwardingError::RouteOpenBuild(_)
| ForwardingError::Poisoned => "forwarding_error",
}
}
fn response_version(frame: &Frame) -> u8 {
if (MIN_SUPPORTED_VERSION..=PROTOCOL_VERSION).contains(&frame.header.ver) {
frame.header.ver
} else {
PROTOCOL_VERSION
}
}
fn control_flags() -> Flags {
Flags::new(false, Priority::Passive, false)
}
fn send_goodbye_target_best_effort(target: &GoodbyeTarget, context: &str) {
let Ok(frame) = Frame::build_with_version(
target.negotiated_ver,
FrameType::Goodbye,
control_flags(),
target.channel,
target.epoch,
0,
Vec::new(),
) else {
return;
};
if let Err(err) = target.sink.try_send(frame) {
warn!(
route_channel = target.channel,
route_epoch = target.epoch,
error = %err,
%context,
"route GOODBYE dropped under backpressure"
);
}
}
pub(crate) fn send_route_control_pushes(
forwarding: &ForwardingTable,
routes: Vec<EndpointRoute>,
push: ClientControlPush,
) {
let body = match serde_json::to_vec(&push) {
Ok(body) => body,
Err(err) => {
warn!(error = %err, "failed to serialize route lifecycle control PUSH");
return;
}
};
let mut targets = Vec::new();
for route in routes {
let target = route.goodbye_target;
if let Some(existing) = targets
.iter()
.find(|existing: &&GoodbyeTarget| existing.connection_id == target.connection_id)
{
debug_assert_eq!(
existing.negotiated_ver, target.negotiated_ver,
"one connection cannot negotiate multiple frame versions"
);
continue;
}
targets.push(target);
}
for target in targets {
let frame = match Frame::build_with_version(
target.negotiated_ver,
FrameType::Push,
control_flags(),
0,
0,
0,
body.clone(),
) {
Ok(frame) => frame,
Err(err) => {
warn!(
route_channel = target.channel,
error = %err,
"failed to build route lifecycle control PUSH frame"
);
continue;
}
};
if let Err(err) = target.sink.try_send(frame) {
if target.close_on_delivery_failure() {
warn!(
target_connection_id = target.connection_id.get(),
route_channel = target.channel,
error = %err,
"route lifecycle control PUSH was not delivered to client; closing target connection"
);
let _ = forwarding.escalate_client_delivery_failure(
target.connection_id,
target.channel,
target.epoch,
CloseReason::new(
"route_lifecycle_push_delivery_failed",
format!(
"failed to enqueue route lifecycle control PUSH for channel {}: {err}",
target.channel
),
),
);
}
}
}
}
#[cfg(test)]
mod tests {
use std::{
collections::BTreeMap,
fmt,
path::PathBuf,
sync::{Arc, Mutex},
time::Duration,
};
use serde_json::{json, Value};
use subc_protocol::{
manifest::{
Concurrency, ExecutionMode, IdentityScope, ManagementOperation,
ManagementOperationKind, ObservabilityKind, ObservabilitySurface, ProviderRole, Tool,
},
session::HealthStatus,
FrameType,
};
use super::*;
use crate::{
forwarding::{DataRoute, DataRouteState},
registry::ChannelState,
router::FrameSink,
stderr_tail::DEFAULT_MAX_LINE_BYTES,
supervise::{ModuleSpec, ModuleState, RestartPolicy, Supervisor, SupervisorHandle},
test_support::TestTempDir,
RouteCtx, Router,
};
use tokio::{
sync::mpsc,
time::{sleep, Instant},
};
use tracing::{
field::{Field, Visit},
Event, Subscriber,
};
use tracing_subscriber::{layer::Context, prelude::*, Layer};
fn fake_aft_stub_path() -> PathBuf {
let mut path = std::env::current_exe().expect("current_exe available in tests");
path.pop(); path.pop(); path.push(if cfg!(windows) {
"fake-aft-stub.exe"
} else {
"fake-aft-stub"
});
assert!(
path.exists(),
"fake-aft-stub not built at {}: run `cargo test -p subc-core` (which builds \
[[bin]] targets) rather than `cargo test -p subc-core --lib` (which does not)",
path.display()
);
path
}
fn client_retries(code: &str) -> bool {
subc_protocol::error_codes::is_retryable_route_open(code)
}
#[test]
fn retryability_of_forwarding_codes_matches_the_failure() {
let transient = [
ForwardingError::NoModuleConnection,
ForwardingError::ModuleReloading {
module_id: "m".into(),
},
ForwardingError::StaleModuleEndpoint,
ForwardingError::UnknownReservation {
client_channel: 1,
module_channel: 1,
},
ForwardingError::ConnectionClosing {
connection_id: ConnectionId::new(1),
},
ForwardingError::ClientEgressClosed {
connection_id: ConnectionId::new(1),
},
];
for err in transient {
let code = forwarding_error_code(&err);
assert!(
client_retries(code),
"{err:?} is transient but publishes {code:?}, which clients treat as permanent"
);
}
let permanent = [
ForwardingError::ClientRouteChannelExhausted {
connection_id: ConnectionId::new(1),
},
ForwardingError::ModuleRouteChannelExhausted {
endpoint: ModuleEndpointId {
connection_id: ConnectionId::new(1),
generation: 1,
},
},
ForwardingError::RelayCorrelationExhausted,
ForwardingError::RouteOpenBuild("x".into()),
ForwardingError::Poisoned,
];
for err in permanent {
let code = forwarding_error_code(&err);
assert!(
!client_retries(code),
"{err:?} cannot be fixed by retrying but publishes {code:?}, which clients retry"
);
}
}
#[tokio::test]
async fn an_unattested_caller_is_never_stamped_as_a_supervised_module() {
let handler = ControlHandler::default();
let frame =
Frame::build(FrameType::Request, control_flags(), 0, 0, 900, Vec::new()).unwrap();
let stamped = handler.route_open_principal(&frame, None).unwrap().unwrap();
assert_eq!(
stamped,
Principal::Direct,
"a caller that proved nothing must not be stamped as a supervised module"
);
let forged = handler
.route_open_principal(
&frame,
Some(ConsumerIdentity {
module_id: "aft".to_string(),
launch_nonce: "not-a-real-nonce".to_string(),
}),
)
.unwrap();
let refusal = forged.expect_err("an unmatched launch nonce must not yield a principal");
assert_eq!(parse_error(&refusal)["code"], "bad_consumer_identity");
}
#[test]
fn a_wire_body_actually_yields_the_consumer_identity_the_daemon_stamps_from() {
let body = br#"{"op":"route.open","target":{"kind":"tool_provider","module_id":"m"},"identity":{"session":"s","project_root":"/p","harness":"h"},"consumer_identity":{"module_id":"aft","launch_nonce":"n"}}"#;
let parsed: ClientControlRequest = serde_json::from_slice(body).unwrap();
let ClientControlRequest::RouteOpen {
consumer_identity, ..
} = parsed
else {
panic!("route.open body must parse as RouteOpen");
};
assert_eq!(
consumer_identity,
Some(ConsumerIdentity {
module_id: "aft".to_string(),
launch_nonce: "n".to_string(),
}),
"the wire field name must reach the value route_open_principal reads"
);
}
fn manifest(module_id: &str, protocol_ver: u8) -> ModuleManifest {
ModuleManifest::builder(module_id, "0.1.0")
.protocol_ver(protocol_ver)
.provides(vec![ProviderRole::ToolProvider {
tools: vec![Tool {
name: "read".to_string(),
description: None,
execution_mode: ExecutionMode::Pure,
schema: json!({"type": "object"}),
}],
identity_scope: vec![IdentityScope::Project, IdentityScope::Session],
concurrency: Concurrency::ModuleManaged,
emits_push: true,
sub_supervises: true,
}])
.build()
}
fn hello_frame(module_id: &str, protocol_ver: u8, corr: u64) -> Frame {
hello_frame_with_control_ops(module_id, protocol_ver, corr, None)
}
fn hello_frame_with_control_ops(
module_id: &str,
protocol_ver: u8,
corr: u64,
control_ops: Option<Vec<String>>,
) -> Frame {
hello_frame_full(module_id, protocol_ver, corr, control_ops, None)
}
fn hello_frame_with_nonce(
module_id: &str,
protocol_ver: u8,
corr: u64,
launch_nonce: Option<&str>,
) -> Frame {
hello_frame_full(
module_id,
protocol_ver,
corr,
None,
launch_nonce.map(ToOwned::to_owned),
)
}
fn hello_frame_full(
module_id: &str,
protocol_ver: u8,
corr: u64,
control_ops: Option<Vec<String>>,
launch_nonce: Option<String>,
) -> Frame {
let body = serde_json::to_vec(&ModuleHelloBody {
manifest: manifest(module_id, protocol_ver),
protocol_ver,
control_ops,
launch_nonce,
})
.unwrap();
Frame::build(FrameType::Hello, control_flags(), 0, 0, corr, body).unwrap()
}
fn non_routable_hello_frame_with_control_ops(
module_id: &str,
corr: u64,
control_ops: Option<Vec<String>>,
) -> Frame {
let mut manifest = manifest(module_id, PROTOCOL_VERSION);
manifest.provides.clear();
let body = serde_json::to_vec(&ModuleHelloBody {
manifest,
protocol_ver: PROTOCOL_VERSION,
control_ops,
launch_nonce: None,
})
.unwrap();
Frame::build(FrameType::Hello, control_flags(), 0, 0, corr, body).unwrap()
}
fn capability_grammar_hello_frame(
capabilities: Value,
runtime_computed: Option<Value>,
corr: u64,
) -> Frame {
let mut body = serde_json::to_value(ModuleHelloBody {
manifest: manifest("capability-grammar-test", PROTOCOL_VERSION),
protocol_ver: PROTOCOL_VERSION,
control_ops: None,
launch_nonce: None,
})
.expect("HELLO body serializes");
body["manifest"]["capabilities"] = capabilities;
if let Some(runtime_computed) = runtime_computed {
body["runtime_computed"] = runtime_computed;
}
Frame::build(
FrameType::Hello,
control_flags(),
0,
0,
corr,
serde_json::to_vec(&body).expect("HELLO body reserializes"),
)
.expect("HELLO frame builds")
}
fn channel_request(channel: u16, corr: u64) -> Frame {
Frame::build(
FrameType::Request,
Flags::new(true, Priority::Interactive, false),
channel,
0,
corr,
b"opaque".to_vec(),
)
.unwrap()
}
fn route_ctx(
connection_id: ConnectionId,
) -> (RouteCtx, mpsc::Receiver<crate::router::OutboundFrame>) {
let (tx, rx) = mpsc::channel(8);
(
RouteCtx {
connection_id,
egress: FrameSink::new(tx),
},
rx,
)
}
fn parse_ack(frame: &Frame) -> ModuleHelloAckBody {
serde_json::from_slice(&frame.body).unwrap()
}
fn parse_error(frame: &Frame) -> Value {
serde_json::from_slice(&frame.body).unwrap()
}
fn parse_route_poll(frame: &Frame) -> ClientControlResponse {
serde_json::from_slice(&frame.body).unwrap()
}
fn route_poll_frame(corr: u64, kind: PollKind, route_channel: u16) -> Frame {
let body = serde_json::to_vec(&ClientControlRequest::RoutePoll {
route_channel,
route_epoch: 0,
kind,
})
.unwrap();
Frame::build(FrameType::Request, control_flags(), 0, 0, corr, body).unwrap()
}
fn supervisor_health_probe_frame(corr: u64, module_id: &str) -> Frame {
let body = serde_json::to_vec(&ClientControlRequest::SupervisorHealthProbe {
module_id: module_id.to_string(),
})
.unwrap();
Frame::build(FrameType::Request, control_flags(), 0, 0, corr, body).unwrap()
}
fn route_open_frame(corr: u64, module_id: &str, project_root: TestTempDir) -> Frame {
route_open_frame_with_consumer_capabilities(corr, module_id, project_root, None)
}
fn route_open_frame_with_consumer_capabilities(
corr: u64,
module_id: &str,
project_root: TestTempDir,
consumer_capabilities: Option<Vec<String>>,
) -> Frame {
let body = serde_json::to_vec(&ClientControlRequest::RouteOpen {
target: RouteTarget::ToolProvider {
module_id: module_id.to_string(),
},
identity: BindIdentity::new(
project_root.path().to_path_buf(),
"unit".to_string(),
"session".to_string(),
),
consumer_identity: None,
consumer_capabilities,
admission_facts: None,
})
.unwrap();
Frame::build(FrameType::Request, control_flags(), 0, 0, corr, body).unwrap()
}
fn route_open_frame_with_admission_facts(
corr: u64,
module_id: &str,
project_root: TestTempDir,
consumer_identity: Option<subc_control::ConsumerIdentity>,
facts: Option<Value>,
) -> Frame {
let body = serde_json::to_vec(&ClientControlRequest::RouteOpen {
target: RouteTarget::ToolProvider {
module_id: module_id.to_string(),
},
identity: BindIdentity::new(
project_root.path().to_path_buf(),
"unit".to_string(),
format!("session-{corr}"),
),
consumer_identity,
consumer_capabilities: None,
admission_facts: facts,
})
.unwrap();
Frame::build(FrameType::Request, control_flags(), 0, 0, corr, body).unwrap()
}
#[derive(Clone, Default)]
struct EventCapture {
events: Arc<Mutex<Vec<CapturedEvent>>>,
}
#[derive(Clone, Debug)]
struct CapturedEvent {
target: String,
fields: BTreeMap<String, String>,
}
impl EventCapture {
fn events(&self) -> Vec<CapturedEvent> {
self.events.lock().unwrap().clone()
}
}
impl<S> Layer<S> for EventCapture
where
S: Subscriber,
{
fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) {
let mut visitor = EventFieldVisitor::default();
event.record(&mut visitor);
self.events.lock().unwrap().push(CapturedEvent {
target: event.metadata().target().to_string(),
fields: visitor.fields,
});
}
}
#[derive(Default)]
struct EventFieldVisitor {
fields: BTreeMap<String, String>,
}
impl Visit for EventFieldVisitor {
fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
self.fields
.insert(field.name().to_string(), format!("{value:?}"));
}
}
fn health_response(corr: u64, status: HealthStatus) -> Frame {
let body = serde_json::to_vec(&ModuleControlResponse::HealthCheck {
status,
detail: Some("warming".to_string()),
metrics: Some(json!({"queue_depth": 3})),
})
.unwrap();
Frame::build(FrameType::Response, control_flags(), 0, 0, corr, body).unwrap()
}
fn route_bind_ack(corr: u64) -> Frame {
let body = serde_json::to_vec(&ModuleControlResponse::RouteBindAck {}).unwrap();
Frame::build(FrameType::Response, control_flags(), 0, 0, corr, body).unwrap()
}
fn unique_project_root(label: &str) -> TestTempDir {
TestTempDir::new(label)
}
fn assert_route_poll_liveness(frame: &Frame, expected_live: bool) {
match parse_route_poll(frame) {
ClientControlResponse::RoutePoll {
status: None,
live: Some(live),
..
} => assert_eq!(live, expected_live),
other => panic!("unexpected route.poll response: {other:?}"),
}
}
fn bind_liveness_route(
registry: &Registry,
forwarding: &ForwardingTable,
module_id: &str,
) -> (RouteCtx, u16, u32) {
let module_connection = ConnectionId::new(101);
let client_connection = ConnectionId::new(202);
let registration = registry
.register_with_control_ops(
manifest(module_id, PROTOCOL_VERSION),
PROTOCOL_VERSION,
module_connection,
module_baseline_control_ops(),
)
.unwrap();
let (module_tx, _module_rx) = mpsc::channel(8);
let endpoint = forwarding
.register_module_connection(
module_connection,
module_id.to_string(),
PROTOCOL_VERSION,
manifest_concurrency(®istration.manifest),
FrameSink::new(module_tx),
)
.unwrap();
let (client_ctx, _client_rx) = route_ctx(client_connection);
let pending = forwarding
.begin_route_bind_relay_for_test(
client_connection,
client_ctx.egress.clone(),
1,
module_id,
)
.unwrap();
assert_eq!(pending.endpoint, endpoint);
let route_channel = pending.client_channel;
let route_epoch = pending.client_epoch;
forwarding
.complete_pending_relay(
module_connection,
pending.corr,
RouteBindRelayOutcome::Accepted,
)
.unwrap();
(client_ctx, route_channel, route_epoch)
}
struct FakeProcessLiveness {
live: Option<bool>,
}
impl ModuleProcessLiveness for FakeProcessLiveness {
fn process_live(&self, _module_id: &str) -> Option<bool> {
self.live
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn supervisor_stderr_tail_converts_a_real_truncated_ring_entry_to_prefix_only_wire_data()
{
let registry = Arc::new(Registry::default());
let supervisor_handle = SupervisorHandle::new();
let supervisor = Supervisor::new(
Arc::clone(®istry),
RestartPolicy::new(1, Duration::from_millis(10)),
)
.with_handle(supervisor_handle.clone());
let source_line = format!("config error: {}", "x".repeat(DEFAULT_MAX_LINE_BYTES));
let module = supervisor
.spawn(ModuleSpec {
module_id: "stderr-tail-wire".to_string(),
program: fake_aft_stub_path(),
args: Vec::new(),
env: vec![
("FAKE_AFT_STDERR_LINE".to_string(), source_line.clone()),
("FAKE_AFT_EXIT_CODE".to_string(), "1".to_string()),
],
reserved: false,
reserved_prefixes: Vec::new(),
protocol: ModuleProtocol::Subc,
overlap: Default::default(),
})
.unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
loop {
let tail = module.stderr_tail(None, None);
if tail
.entries
.iter()
.any(|entry| matches!(entry, TailEntry::ProcessStart))
&& tail.entries.iter().any(|entry| {
matches!(
entry,
TailEntry::Line {
truncated: true,
..
}
)
})
{
break;
}
assert!(
Instant::now() < deadline,
"module did not produce a truncated line and restart boundary: {tail:?}"
);
sleep(Duration::from_millis(10)).await;
}
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor_handle);
let request = ClientControlRequest::SupervisorStderrTail {
module_id: "stderr-tail-wire".to_string(),
max_lines: None,
max_bytes: None,
};
let frame = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
1,
serde_json::to_vec(&request).unwrap(),
)
.unwrap();
let (ctx, _egress) = route_ctx(ConnectionId::new(1));
let responses = handler.handle_control_frame(&ctx, frame).await.unwrap();
let ClientControlResponse::SupervisorStderrTail { tail, .. } =
serde_json::from_slice(&responses[0].body).unwrap()
else {
panic!("expected supervisor.stderr_tail response");
};
assert!(
tail.entries
.iter()
.any(|entry| matches!(entry, StderrTailEntry::ProcessStart)),
"the control response lost the restart boundary"
);
let Some(StderrTailEntry::Line { text, truncated }) = tail.entries.iter().find(|entry| {
matches!(
entry,
StderrTailEntry::Line {
truncated: true,
..
}
)
}) else {
panic!("the control response lost the truncated line");
};
assert_eq!(text, &source_line[..DEFAULT_MAX_LINE_BYTES]);
assert!(*truncated);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn supervisor_terminals_golden_is_generated_through_the_real_handler() {
let registry = Arc::new(Registry::default());
let supervisor_handle = SupervisorHandle::new();
let supervisor =
Supervisor::new(Arc::clone(®istry), RestartPolicy::new(1, Duration::ZERO))
.with_handle(supervisor_handle.clone());
let module = supervisor
.spawn(ModuleSpec {
module_id: "terminal-golden".to_string(),
program: fake_aft_stub_path(),
args: Vec::new(),
env: vec![("FAKE_AFT_EXIT_CODE".to_string(), "23".to_string())],
reserved: false,
reserved_prefixes: Vec::new(),
protocol: ModuleProtocol::Subc,
overlap: Default::default(),
})
.unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
while module.terminal_history().entries.len() != 2 {
assert!(
Instant::now() < deadline,
"module did not retain two terminal exits: {:?}",
module.terminal_history()
);
sleep(Duration::from_millis(10)).await;
}
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor_handle);
let request = ClientControlRequest::SupervisorTerminals {
module_id: "terminal-golden".to_string(),
};
let frame = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
1,
serde_json::to_vec(&request).unwrap(),
)
.unwrap();
let (ctx, _egress) = route_ctx(ConnectionId::new(1));
let responses = handler.handle_control_frame(&ctx, frame).await.unwrap();
let response: ClientControlResponse = serde_json::from_slice(&responses[0].body).unwrap();
let ClientControlResponse::SupervisorTerminals { terminals, .. } = &response else {
panic!("expected supervisor.terminals response");
};
assert_eq!(terminals.entries.len(), 2);
assert_eq!(terminals.dropped, 0);
let mut rendered = serde_json::to_value(response).unwrap();
rendered["daemon_started_at_ms"] = json!(1_700_000_000_000u64);
for (index, entry) in rendered["entries"]
.as_array_mut()
.expect("terminal response entries array")
.iter_mut()
.enumerate()
{
entry["at_ms"] = json!(1_700_000_000_001u64 + index as u64);
}
let golden_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../subc-control/tests/golden/client_control_response_supervisor_terminals.json");
let serialized = serde_json::to_string_pretty(&rendered).unwrap() + "\n";
if std::env::var_os("UPDATE_GOLDEN").is_some() {
std::fs::write(&golden_path, &serialized).unwrap();
}
let expected: Value =
serde_json::from_str(&std::fs::read_to_string(&golden_path).unwrap()).unwrap();
assert_eq!(rendered, expected);
}
#[test]
fn hello_registers_manifest_and_returns_ack() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let conn = ConnectionId::new(1);
let responses = handler
.handle_control(conn, hello_frame("aft", PROTOCOL_VERSION, 7))
.unwrap();
assert_eq!(responses.len(), 1);
assert_eq!(responses[0].header.ty, FrameType::HelloAck);
assert_eq!(responses[0].header.channel, 0);
assert_eq!(responses[0].header.corr, 7);
let ack = parse_ack(&responses[0]);
assert_eq!(ack.negotiated_ver, PROTOCOL_VERSION);
assert!(ack
.subc_capabilities
.contains(&CAP_MANIFEST_REGISTRATION.to_string()));
assert!(ack.subc_ops.contains(&ops::SUPERVISOR_LIST.to_string()));
assert!(ack.subc_ops.contains(&ops::SUPERVISOR_RESTART.to_string()));
assert!(ack
.subc_ops
.contains(&ops::SUPERVISOR_SET_ENABLED.to_string()));
assert!(ack
.subc_ops
.contains(&MODULE_TO_SUBC_OP_CATALOG_UPDATE.to_string()));
let registration = registry.get_module("aft").unwrap().unwrap();
assert_eq!(registration.negotiated_ver, PROTOCOL_VERSION);
assert_eq!(registration.state, ChannelState::Active);
assert_eq!(registration.connection_id, conn);
assert_eq!(registration.control_ops, module_baseline_control_ops());
}
#[test]
fn capability_grammar_refusals_name_the_field_and_leave_no_catalog_entry() {
let invalid_identifiers = [
("case_change", "credentials-Provider/v1"),
("leading_zero", "credentials-provider/v01"),
("trailing_hyphen", "credentials-provider-/v1"),
("consecutive_hyphens", "credentials--provider/v1"),
("uppercase", "Credentials-provider/v1"),
("missing_v", "credentials-provider/1"),
("whitespace", "credentials provider/v1"),
("zero_version", "credentials-provider/v0"),
("out_of_range_version", "credentials-provider/v4294967296"),
(
"overlength_name",
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa/v1",
),
];
let mut cases = invalid_identifiers
.into_iter()
.map(|(name, identifier)| {
(
format!("identifier_{name}"),
"capabilities.provides[0]".to_string(),
identifier.to_string(),
json!({ "provides": [identifier] }),
None,
)
})
.collect::<Vec<_>>();
cases.extend([
(
"unknown_need".to_string(),
"capabilities.requires[0].need".to_string(),
"deferred".to_string(),
json!({ "requires": [{ "capability": "credentials-provider/v1", "need": "deferred" }] }),
None,
),
(
"duplicate_provides".to_string(),
"capabilities.provides[1]".to_string(),
"credentials-provider/v1".to_string(),
json!({ "provides": ["credentials-provider/v1", "credentials-provider/v1"] }),
None,
),
(
"duplicate_must_never_reach".to_string(),
"capabilities.must_never_reach[1]".to_string(),
"credentials-provider/v1".to_string(),
json!({ "must_never_reach": ["credentials-provider/v1", "credentials-provider/v1"] }),
None,
),
(
"duplicate_requires_same_need".to_string(),
"capabilities.requires[1]".to_string(),
"credentials-provider/v1".to_string(),
json!({ "requires": [
{ "capability": "credentials-provider/v1", "need": "required" },
{ "capability": "credentials-provider/v1", "need": "required" }
] }),
None,
),
(
"duplicate_requires_conflicting_need".to_string(),
"capabilities.requires[1]".to_string(),
"credentials-provider/v1".to_string(),
json!({ "requires": [
{ "capability": "credentials-provider/v1", "need": "required" },
{ "capability": "credentials-provider/v1", "need": "optional" }
] }),
None,
),
(
"capabilities_root_pointer".to_string(),
"runtime_computed[0]".to_string(),
"/capabilities".to_string(),
json!({}),
Some(json!(["/capabilities"])),
),
(
"capabilities_descendant_pointer".to_string(),
"runtime_computed[0]".to_string(),
"/capabilities/provides".to_string(),
json!({}),
Some(json!(["/capabilities/provides"])),
),
(
"malformed_pointer_without_leading_slash".to_string(),
"runtime_computed[0]".to_string(),
"capabilities".to_string(),
json!({}),
Some(json!(["capabilities"])),
),
(
"malformed_pointer_escape".to_string(),
"runtime_computed[0]".to_string(),
"/roles/~2/tools".to_string(),
json!({}),
Some(json!(["/roles/~2/tools"])),
),
(
"unknown_capabilities_field".to_string(),
"capabilities.future".to_string(),
"<array>".to_string(),
json!({ "future": [] }),
None,
),
]);
for (index, (name, field, value, capabilities, runtime_computed)) in
cases.into_iter().enumerate()
{
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let response = handler
.handle_control(
ConnectionId::new((index + 1) as u64),
capability_grammar_hello_frame(
capabilities,
runtime_computed,
index as u64 + 1,
),
)
.expect("invalid HELLO returns a refusal");
assert_eq!(response.len(), 1, "{name} must emit one refusal");
let error = parse_error(&response[0]);
assert_eq!(error["code"], "invalid_capability_grammar", "{name}");
let message = error["message"]
.as_str()
.expect("error message is a string");
assert!(
message.contains(&field),
"{name}: field missing from {message}"
);
assert!(
message.contains(&value),
"{name}: value missing from {message}"
);
assert_eq!(
registry
.active_registration_count()
.expect("registry reads"),
0,
"{name}: refused HELLO must not create a catalog entry"
);
}
}
#[test]
fn legal_runtime_pointer_and_capabilities_are_mirrored_in_catalog_list() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let capabilities = json!({
"provides": ["credentials-provider/v1"],
"requires": [{ "capability": "context-transform/v1", "need": "optional" }],
"must_never_reach": ["federation-transport/v1"]
});
let response = handler
.handle_control(
ConnectionId::new(99),
capability_grammar_hello_frame(
capabilities.clone(),
Some(json!(["/roles/0/tools"])),
99,
),
)
.expect("valid HELLO registers");
assert_eq!(response[0].header.ty, FrameType::HelloAck);
let request = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
100,
serde_json::to_vec(&ClientControlRequest::CatalogList { module_id: None })
.expect("catalog request serializes"),
)
.expect("catalog request frame builds");
let response = handler
.handle_catalog_list(request, None)
.expect("catalog list succeeds");
let ClientControlResponse::CatalogList { modules, .. } =
serde_json::from_slice(&response[0].body).expect("catalog response decodes")
else {
panic!("catalog request must return catalog.list");
};
assert_eq!(modules.len(), 1);
assert_eq!(
serde_json::to_value(&modules[0].capabilities).expect("catalog capabilities serialize"),
capabilities
);
}
#[test]
fn catalog_list_mirrors_management_operation_description() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let description = "List managed records and return their identifiers and metadata.";
let mut manifest = manifest("described-management", PROTOCOL_VERSION);
manifest.provides = vec![ProviderRole::ManagementSurface {
operations: vec![ManagementOperation {
name: "records.list".to_string(),
kind: ManagementOperationKind::Query,
description: Some(description.to_string()),
}],
config_schema: json!({"type": "object"}),
observability: vec![ObservabilitySurface {
name: "records.stats".to_string(),
kind: ObservabilityKind::Snapshot,
}],
identity_scope: vec![IdentityScope::Project],
concurrency: Concurrency::ModuleManaged,
}];
registry
.register_with_control_ops(
manifest,
PROTOCOL_VERSION,
ConnectionId::new(99),
Vec::new(),
)
.expect("described management manifest registers");
let request = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
100,
serde_json::to_vec(&ClientControlRequest::CatalogList { module_id: None })
.expect("catalog request serializes"),
)
.expect("catalog request frame builds");
let response = handler
.handle_catalog_list(request, None)
.expect("catalog list succeeds");
let body: Value = serde_json::from_slice(&response[0].body).expect("catalog response JSON");
assert_eq!(
body["modules"][0]["roles"][0]["operations"][0]["description"], description,
"catalog.list must preserve the declared operation description verbatim"
);
}
#[test]
fn reserved_capability_refusal_mutation_proof_leaves_no_catalog_entry() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry)).with_capability_config(
[("vault".to_string(), true), ("squatter".to_string(), true)],
BTreeMap::from([("credentials-provider/v1".to_string(), "vault".to_string())]),
);
let mut squatter = manifest("squatter", PROTOCOL_VERSION);
squatter.capabilities = Some(subc_protocol::manifest::CapabilityDeclarations {
provides: vec!["credentials-provider/v1".to_string()],
requires: Vec::new(),
must_never_reach: Vec::new(),
});
let frame = Frame::build(
FrameType::Hello,
control_flags(),
0,
0,
77,
serde_json::to_vec(&ModuleHelloBody {
manifest: squatter,
protocol_ver: PROTOCOL_VERSION,
control_ops: None,
launch_nonce: None,
})
.expect("HELLO serializes"),
)
.expect("HELLO frame builds");
let response = handler
.handle_control(ConnectionId::new(77), frame)
.expect("reserved claim receives a typed refusal");
assert_eq!(parse_error(&response[0])["code"], "reserved_capability");
assert_eq!(
registry
.active_registration_count()
.expect("registry reads"),
0,
"a reserved capability refusal must not leave a catalog entry"
);
}
#[test]
fn server_describe_surfaces_required_capability_verdict_fields() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry)).with_capability_config(
[
("consumer".to_string(), true),
("provider".to_string(), false),
],
BTreeMap::new(),
);
let mut consumer = manifest("consumer", PROTOCOL_VERSION);
consumer.capabilities = Some(subc_protocol::manifest::CapabilityDeclarations {
provides: Vec::new(),
requires: vec![subc_protocol::manifest::CapabilityRequirement {
capability: "credentials-provider/v1".to_string(),
need: subc_protocol::manifest::CapabilityNeed::Required,
}],
must_never_reach: Vec::new(),
});
let hello = Frame::build(
FrameType::Hello,
control_flags(),
0,
0,
78,
serde_json::to_vec(&ModuleHelloBody {
manifest: consumer,
protocol_ver: PROTOCOL_VERSION,
control_ops: None,
launch_nonce: None,
})
.expect("HELLO serializes"),
)
.expect("HELLO frame builds");
handler
.handle_control(ConnectionId::new(78), hello)
.expect("consumer registers");
let describe = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
79,
serde_json::to_vec(&ClientControlRequest::ServerDescribe {})
.expect("request serializes"),
)
.expect("describe frame builds");
let response = handler
.handle_server_describe(describe)
.expect("server.describe succeeds");
let rendered: Value = serde_json::from_slice(&response[0].body).expect("response JSON");
let requirement = &rendered["capability_requirements"][0];
assert_eq!(requirement["consumer"], "consumer");
assert_eq!(requirement["verdict"], "never_provided");
assert_eq!(requirement["episode_seq"], 1);
assert_eq!(requirement["config_satisfiable"], false);
assert_eq!(requirement["runtime_available"], false);
assert!(requirement["detail"]
.as_str()
.expect("detail string")
.contains("credentials-provider/v1"));
}
#[test]
fn catalog_list_omits_capabilities_for_legacy_manifest() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let hello = handler
.handle_control(
ConnectionId::new(101),
hello_frame("legacy-capability-manifest", PROTOCOL_VERSION, 101),
)
.expect("legacy HELLO registers");
assert_eq!(hello[0].header.ty, FrameType::HelloAck);
let request = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
102,
serde_json::to_vec(&ClientControlRequest::CatalogList { module_id: None })
.expect("catalog request serializes"),
)
.expect("catalog request frame builds");
let response = handler
.handle_catalog_list(request, None)
.expect("catalog list succeeds");
let body: Value = serde_json::from_slice(&response[0].body).expect("catalog response JSON");
assert!(
body["modules"][0].get("capabilities").is_none(),
"legacy manifest must retain an absent capabilities field on catalog.list"
);
}
#[test]
fn hello_ack_omits_storage_when_no_storage_config() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let responses = handler
.handle_control(
ConnectionId::new(1),
hello_frame("aft", PROTOCOL_VERSION, 7),
)
.unwrap();
let ack = parse_ack(&responses[0]);
assert_eq!(ack.storage, None, "no storage config -> no descriptor");
assert_eq!(ack.machine_id, None, "no machine id configured -> no field");
}
#[tokio::test]
async fn hello_ack_and_server_describe_carry_the_configured_machine_id() {
let id = crate::machine_id::MachineId::parse("0123456789abcdef0123456789abcdef").unwrap();
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry)).with_machine_id(Some(id.clone()));
let responses = handler
.handle_control(
ConnectionId::new(1),
hello_frame("aft", PROTOCOL_VERSION, 7),
)
.unwrap();
let ack = parse_ack(&responses[0]);
assert_eq!(ack.machine_id.as_deref(), Some(id.as_str()));
let described = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(2)).0,
Frame::build(
FrameType::Request,
control_flags(),
0,
0,
9,
serde_json::to_vec(&ClientControlRequest::ServerDescribe {}).unwrap(),
)
.unwrap(),
)
.await
.unwrap();
let ClientControlResponse::ServerDescribe { machine_id, .. } =
serde_json::from_slice(&described[0].body).unwrap()
else {
panic!("server.describe answered with another shape");
};
assert_eq!(machine_id.as_deref(), Some(id.as_str()));
}
#[test]
fn hello_ack_delivers_resolved_storage_descriptor_per_module() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry)).with_storage_config(Some(
crate::daemon_config::StorageConfig::Sqlite {
data_home: std::path::PathBuf::from("/data"),
},
));
let responses = handler
.handle_control(
ConnectionId::new(1),
hello_frame("alfonso-routing", PROTOCOL_VERSION, 7),
)
.unwrap();
let ack = parse_ack(&responses[0]);
assert_eq!(
ack.storage,
Some(serde_json::json!({
"module_id": "alfonso-routing",
"storage_namespace": "default",
"isolation": { "kind": "module" },
"backend": {
"backend": "sqlite",
"path": "/data/cortexkit/alfonso-routing/store.db"
}
})),
"the delivered descriptor is the module's own sqlite store path"
);
}
#[test]
fn hello_control_ops_none_is_baseline_and_guard_rejects_synthetic_gated_op() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let conn = ConnectionId::new(1);
let responses = handler
.handle_control(
conn,
hello_frame_with_control_ops("aft", PROTOCOL_VERSION, 7, None),
)
.unwrap();
assert_eq!(responses[0].header.ty, FrameType::HelloAck);
let registration = registry.get_module("aft").unwrap().unwrap();
assert_eq!(registration.control_ops, module_baseline_control_ops());
let frame =
Frame::build(FrameType::Request, control_flags(), 0, 0, 77, Vec::new()).unwrap();
assert!(handler
.guard_module_control_op(&frame, "aft", "route.bind")
.unwrap()
.is_none());
let error = handler
.guard_module_control_op(&frame, "aft", "test.synthetic")
.unwrap()
.expect("synthetic ungranted op should be rejected");
assert_eq!(error.header.ty, FrameType::Error);
assert_eq!(parse_error(&error)["code"], "op_not_allowed");
}
#[test]
fn hello_control_ops_some_adds_optional_grants() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
handler
.handle_control(
ConnectionId::new(1),
hello_frame_with_control_ops(
"aft",
PROTOCOL_VERSION,
7,
Some(vec![
"future.synthetic".to_string(),
"route.bind".to_string(),
]),
),
)
.unwrap();
let registration = registry.get_module("aft").unwrap().unwrap();
assert_eq!(
registration.control_ops,
vec![
"route.bind".to_string(),
"route.status".to_string(),
"future.synthetic".to_string(),
]
);
let frame =
Frame::build(FrameType::Request, control_flags(), 0, 0, 78, Vec::new()).unwrap();
assert!(handler
.guard_module_control_op(&frame, "aft", "future.synthetic")
.unwrap()
.is_none());
}
#[tokio::test]
async fn health_probe_refuses_unadvertised_module_without_sending_frame() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let (module_ctx, mut module_rx) = route_ctx(ConnectionId::new(10));
let responses = handler
.handle_control_frame(
&module_ctx,
hello_frame_with_control_ops("aft", PROTOCOL_VERSION, 7, None),
)
.await
.unwrap();
assert_eq!(responses[0].header.ty, FrameType::HelloAck);
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(20));
let responses = handler
.handle_control_frame(&client_ctx, supervisor_health_probe_frame(77, "aft"))
.await
.unwrap();
assert_eq!(responses.len(), 1);
assert_eq!(responses[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&responses[0])["code"], "health_not_advertised");
assert!(module_rx.try_recv().is_err());
}
#[tokio::test]
async fn health_probe_demuxes_while_route_bind_relay_is_in_flight() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let (module_ctx, mut module_rx) = route_ctx(ConnectionId::new(30));
handler
.handle_control_frame(
&module_ctx,
hello_frame_with_control_ops(
"aft",
PROTOCOL_VERSION,
7,
Some(vec![MODULE_CONTROL_OP_HEALTH_CHECK.to_string()]),
),
)
.await
.unwrap();
let project_root = unique_project_root("demux");
let (route_client_ctx, mut route_client_rx) = route_ctx(ConnectionId::new(31));
let route_handler = handler.clone();
let route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(
&route_client_ctx,
route_open_frame(100, "aft", project_root),
)
.await
.unwrap()
});
let bind_frame = tokio::time::timeout(Duration::from_secs(1), module_rx.recv())
.await
.unwrap()
.unwrap();
assert!(matches!(
serde_json::from_slice::<ModuleControlRequest>(&bind_frame.body).unwrap(),
ModuleControlRequest::RouteBind { .. }
));
let (health_client_ctx, _health_client_rx) = route_ctx(ConnectionId::new(32));
let health_handler = handler.clone();
let health_task = tokio::spawn(async move {
health_handler
.handle_control_frame(
&health_client_ctx,
supervisor_health_probe_frame(101, "aft"),
)
.await
.unwrap()
});
let health_frame = tokio::time::timeout(Duration::from_secs(1), module_rx.recv())
.await
.unwrap()
.unwrap();
assert_eq!(
serde_json::from_slice::<ModuleControlRequest>(&health_frame.body).unwrap(),
ModuleControlRequest::HealthCheck {}
);
handler
.handle_control_frame(
&module_ctx,
health_response(health_frame.header.corr, HealthStatus::Degraded),
)
.await
.unwrap();
let health_response = health_task.await.unwrap();
assert_eq!(health_response.len(), 1);
match serde_json::from_slice::<ClientControlResponse>(&health_response[0].body).unwrap() {
ClientControlResponse::SupervisorHealthProbe {
module_id,
status,
detail,
metrics,
} => {
assert_eq!(module_id, "aft");
assert_eq!(status, HealthStatus::Degraded);
assert_eq!(detail.as_deref(), Some("warming"));
assert_eq!(metrics, Some(json!({"queue_depth": 3})));
}
other => panic!("unexpected health response: {other:?}"),
}
handler
.handle_control_frame(&module_ctx, route_bind_ack(bind_frame.header.corr))
.await
.unwrap();
let route_response = route_task.await.unwrap();
assert!(route_response.is_empty());
let published = route_client_rx.recv().await.unwrap();
assert!(matches!(
serde_json::from_slice::<ClientControlResponse>(&published.body).unwrap(),
ClientControlResponse::RouteOpen { .. }
));
}
async fn relay_route_open(
handler: &ControlHandler,
client_connection: ConnectionId,
client_egress: &FrameSink,
module_rx: &mut mpsc::Receiver<crate::router::OutboundFrame>,
corr: u64,
module_id: &str,
project_root_label: &str,
) -> (tokio::task::JoinHandle<Vec<Frame>>, Frame) {
let ctx = RouteCtx {
connection_id: client_connection,
egress: client_egress.clone(),
};
let handler = handler.clone();
let project_root = unique_project_root(project_root_label);
let module_id = module_id.to_string();
let dispatch = tracing::dispatcher::get_default(|dispatch| dispatch.clone());
let task = tokio::spawn(async move {
let _guard = tracing::dispatcher::set_default(&dispatch);
handler
.handle_control_frame(&ctx, route_open_frame(corr, &module_id, project_root))
.await
.unwrap()
});
let bind = tokio::time::timeout(Duration::from_secs(2), module_rx.recv())
.await
.expect("module receives the relayed route.bind")
.expect("module egress is open");
(task, bind.frame)
}
fn route_bind_channel(frame: &Frame) -> (u16, u32) {
match serde_json::from_slice::<ModuleControlRequest>(&frame.body).unwrap() {
ModuleControlRequest::RouteBind {
route_channel,
epoch,
..
} => (route_channel, epoch),
other => panic!("expected a route.bind request, got {other:?}"),
}
}
fn published_route(frame: &Frame) -> (u16, u32) {
match serde_json::from_slice::<ClientControlResponse>(&frame.body).unwrap() {
ClientControlResponse::RouteOpen {
route_channel,
route_epoch,
} => (route_channel, route_epoch),
other => panic!("expected a route.open response, got {other:?}"),
}
}
#[tokio::test]
async fn late_bind_ack_for_a_closing_client_keeps_the_module_connection_serving() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let module_connection = ConnectionId::new(30);
let (module_ctx, mut module_rx) = route_ctx(module_connection);
handler
.handle_control_frame(&module_ctx, hello_frame("aft", PROTOCOL_VERSION, 7))
.await
.unwrap();
let dying_client = ConnectionId::new(31);
let (dying_ctx, mut dying_rx) = route_ctx(dying_client);
let (first_task, first_bind) = relay_route_open(
&handler,
dying_client,
&dying_ctx.egress,
&mut module_rx,
100,
"aft",
"closing-first",
)
.await;
handler
.handle_control_frame(&module_ctx, route_bind_ack(first_bind.header.corr))
.await
.unwrap();
assert!(first_task.await.unwrap().is_empty());
let (first_channel, first_epoch) = published_route(&dying_rx.recv().await.unwrap());
let (second_task, second_bind) = relay_route_open(
&handler,
dying_client,
&dying_ctx.egress,
&mut module_rx,
101,
"aft",
"closing-second",
)
.await;
let (abandoned_channel, abandoned_epoch) = route_bind_channel(&second_bind);
assert!(forwarding
.escalate_client_delivery_failure(
dying_client,
first_channel,
first_epoch,
CloseReason::new(
"module_to_client_delivery_failed",
"client egress refused a module frame",
),
)
.unwrap());
assert!(!dying_ctx.egress.is_closed());
let ack = handler
.handle_control_frame(&module_ctx, route_bind_ack(second_bind.header.corr))
.await;
let module_loop_error = ack.as_ref().err().map(ToString::to_string);
if module_loop_error.is_some() {
handler.cleanup_connection(module_connection).unwrap();
}
let post_ack_module_frame = tokio::time::timeout(Duration::from_secs(1), module_rx.recv())
.await
.ok()
.flatten();
assert!(
registry
.get_module_by_connection(module_connection)
.unwrap()
.is_some(),
"one client's closing connection ended the shared module connection: \
{module_loop_error:?}"
);
let cotenant = ConnectionId::new(32);
let (cotenant_ctx, mut cotenant_rx) = route_ctx(cotenant);
let (cotenant_task, cotenant_bind) = relay_route_open(
&handler,
cotenant,
&cotenant_ctx.egress,
&mut module_rx,
102,
"aft",
"closing-cotenant",
)
.await;
handler
.handle_control_frame(&module_ctx, route_bind_ack(cotenant_bind.header.corr))
.await
.unwrap();
assert!(cotenant_task.await.unwrap().is_empty());
let (cotenant_channel, cotenant_epoch) =
published_route(&cotenant_rx.recv().await.unwrap());
assert!(matches!(
forwarding
.lookup_data_route(cotenant, cotenant_channel, cotenant_epoch)
.unwrap(),
DataRoute::Client(DataRouteState::Bound(_))
));
let goodbye = post_ack_module_frame
.expect("module receives a GOODBYE for the abandoned route channel");
assert_eq!(goodbye.header.ty, FrameType::Goodbye);
assert_eq!(goodbye.header.channel, abandoned_channel);
assert_eq!(goodbye.header.epoch, abandoned_epoch);
assert!(dying_rx.try_recv().is_err());
let second_response = second_task.await.unwrap();
assert_eq!(second_response.len(), 1);
assert_eq!(
parse_error(&second_response[0])["code"],
"target_unavailable"
);
}
#[test]
fn only_the_modules_own_closing_connection_ends_the_module_loop() {
let handler = ControlHandler::default();
let module_connection = ConnectionId::new(30);
let client_connection = ConnectionId::new(31);
handler
.refuse_to_end_module_connection_for_a_client(
module_connection,
77,
ForwardingError::ConnectionClosing {
connection_id: client_connection,
},
)
.expect("a closing client must never end the module connection");
assert!(matches!(
handler.refuse_to_end_module_connection_for_a_client(
module_connection,
78,
ForwardingError::ConnectionClosing {
connection_id: module_connection,
},
),
Err(RouterError::Forwarding(ForwardingError::ConnectionClosing {
connection_id
})) if connection_id == module_connection
));
assert!(matches!(
handler.refuse_to_end_module_connection_for_a_client(
module_connection,
79,
ForwardingError::Poisoned,
),
Err(RouterError::Forwarding(ForwardingError::Poisoned))
));
assert!(matches!(
handler.refuse_to_end_module_connection_for_a_client(
module_connection,
80,
ForwardingError::StaleModuleEndpoint,
),
Err(RouterError::Forwarding(
ForwardingError::StaleModuleEndpoint
))
));
}
#[tokio::test]
async fn route_open_refuses_consumer_identity_that_fails_spawn_attestation() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("fed", "fed-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, _target_rx) = route_ctx(ConnectionId::new(90));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let wrong_nonce = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(91)).0,
route_open_frame_with_admission_facts(
20,
"target",
unique_project_root("admission-facts"),
Some(subc_control::ConsumerIdentity {
module_id: "fed".to_string(),
launch_nonce: "not-the-real-nonce".to_string(),
}),
None,
),
)
.await
.unwrap();
assert_eq!(
parse_error(&wrong_nonce[0])["code"],
"bad_consumer_identity",
"a mismatched launch nonce must be refused, not stamped Reserved"
);
let never_spawned = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(92)).0,
route_open_frame_with_admission_facts(
21,
"target",
unique_project_root("admission-facts"),
Some(subc_control::ConsumerIdentity {
module_id: "never-spawned".to_string(),
launch_nonce: "any-nonce".to_string(),
}),
None,
),
)
.await
.unwrap();
assert_eq!(
parse_error(&never_spawned[0])["code"],
"bad_consumer_identity",
"an unspawned module_id must be refused rather than accepted for lack of a record"
);
}
#[tokio::test]
async fn route_open_stamps_reserved_for_a_correctly_attested_consumer() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("fed", "fed-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(95));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let (client_ctx, mut client_rx) = route_ctx(ConnectionId::new(96));
let route_handler = handler.clone();
let route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(
&client_ctx,
route_open_frame_with_admission_facts(
30,
"target",
unique_project_root("admission-facts"),
Some(subc_control::ConsumerIdentity {
module_id: "fed".to_string(),
launch_nonce: "fed-nonce".to_string(),
}),
None,
),
)
.await
.unwrap()
});
let bind_frame = tokio::time::timeout(Duration::from_secs(5), target_rx.recv())
.await
.expect("no route.bind within 5s: the consumer-identity guard refused a correctly attested consumer")
.expect("module control channel closed before route.bind");
let bind: ModuleControlRequest = serde_json::from_slice(&bind_frame.body).unwrap();
let ModuleControlRequest::RouteBind { principal, .. } = bind else {
panic!("expected route.bind")
};
assert_eq!(
principal,
Some(Principal::Reserved {
module_id: "fed".to_string()
}),
"a correctly attested consumer must be stamped Reserved for its own id"
);
handler
.handle_control_frame(&target_ctx, route_bind_ack(bind_frame.header.corr))
.await
.unwrap();
assert!(route_task.await.unwrap().is_empty());
assert!(
matches!(
serde_json::from_slice::<ClientControlResponse>(
&client_rx.recv().await.unwrap().body
)
.unwrap(),
ClientControlResponse::RouteOpen { .. }
),
"the route must actually open, not merely avoid an error"
);
}
#[tokio::test(start_paused = true)]
async fn supervisor_routes_serializes_live_draining_bindings_from_the_real_handler() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("fed", "fed-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(101));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let (direct_ctx, mut direct_rx) = route_ctx(ConnectionId::new(102));
let direct_handler = handler.clone();
let direct_open = tokio::spawn(async move {
direct_handler
.handle_control_frame(
&direct_ctx,
route_open_frame(2, "target", unique_project_root("route-census-direct")),
)
.await
.unwrap()
});
let direct_bind = tokio::time::timeout(Duration::from_secs(5), target_rx.recv())
.await
.expect("no direct route.bind within 5s")
.expect("target control channel closed before direct route.bind");
handler
.handle_control_frame(&target_ctx, route_bind_ack(direct_bind.header.corr))
.await
.unwrap();
assert!(direct_open.await.unwrap().is_empty());
let _ = direct_rx.recv().await.unwrap();
let (reserved_ctx, mut reserved_rx) = route_ctx(ConnectionId::new(103));
let reserved_handler = handler.clone();
let reserved_open = tokio::spawn(async move {
reserved_handler
.handle_control_frame(
&reserved_ctx,
route_open_frame_with_admission_facts(
3,
"target",
unique_project_root("admission-facts"),
Some(ConsumerIdentity {
module_id: "fed".to_string(),
launch_nonce: "fed-nonce".to_string(),
}),
None,
),
)
.await
.unwrap()
});
let reserved_bind = tokio::time::timeout(Duration::from_secs(5), target_rx.recv())
.await
.expect("no reserved route.bind within 5s")
.expect("target control channel closed before reserved route.bind");
handler
.handle_control_frame(&target_ctx, route_bind_ack(reserved_bind.header.corr))
.await
.unwrap();
assert!(reserved_open.await.unwrap().is_empty());
let _ = reserved_rx.recv().await.unwrap();
forwarding
.begin_module_drain("target", subc_control::RouteCloseReason::Reload)
.unwrap();
let (census_ctx, _census_rx) = route_ctx(ConnectionId::new(104));
let census_body = serde_json::to_vec(&ClientControlRequest::SupervisorRoutes {
module_id: Some("target".to_string()),
})
.unwrap();
let census_frame =
Frame::build(FrameType::Request, control_flags(), 0, 0, 4, census_body).unwrap();
let response = handler
.handle_control_frame(&census_ctx, census_frame)
.await
.unwrap()
.pop()
.unwrap();
let actual: Value = serde_json::from_slice(&response.body).unwrap();
let decoded: ClientControlResponse = serde_json::from_value(actual.clone()).unwrap();
assert!(matches!(
decoded,
ClientControlResponse::SupervisorRoutes { .. }
));
let routes = actual["modules"][0]["routes"].as_array().unwrap();
assert_eq!(routes.len(), 2);
assert!(routes.iter().all(|route| route["draining"] == true));
assert!(
routes.iter().all(|route| route["drain_reason"] == "reload"),
"draining routes must name the drain's reason: {routes:?}"
);
assert!(routes.iter().any(|route| {
route["consumer"] == serde_json::json!({"kind": "direct", "connection_id": 102})
}));
assert!(routes.iter().any(|route| {
route["consumer"] == serde_json::json!({"kind": "reserved", "module_id": "fed"})
}));
let golden_path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../subc-control/tests/golden/client_control_response_supervisor_routes.json");
if std::env::var_os("UPDATE_GOLDEN").is_some() {
std::fs::write(
&golden_path,
format!("{}\n", serde_json::to_string_pretty(&actual).unwrap()),
)
.unwrap();
}
let expected: Value =
serde_json::from_str(&std::fs::read_to_string(golden_path).unwrap()).unwrap();
assert_eq!(actual, expected);
}
async fn query_live_roots(
handler: &ControlHandler,
module_ctx: &RouteCtx,
) -> ModuleControlResponseToModule {
let body = serde_json::to_vec(&ModuleControlRequestFromModule::LiveRoots {}).unwrap();
let frame = Frame::build(FrameType::Request, control_flags(), 0, 0, 900, body).unwrap();
let response = handler
.handle_control_frame(module_ctx, frame)
.await
.unwrap()
.pop()
.unwrap();
serde_json::from_slice(&response.body).unwrap()
}
#[tokio::test(start_paused = true)]
async fn supervisor_live_roots_root_known_arm_counts_bound_and_pending_from_real_handler() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler = ControlHandler::with_forwarding(registry, forwarding);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(301));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let root = unique_project_root("live-roots-known");
let path = ProjectRootId::from_path_allowing_missing(root.path())
.unwrap()
.as_path()
.to_path_buf();
let (client_ctx, mut client_rx) = route_ctx(ConnectionId::new(302));
let open_handler = handler.clone();
let opened = tokio::spawn(async move {
open_handler
.handle_control_frame(&client_ctx, route_open_frame(2, "target", root))
.await
.unwrap()
});
let bind = tokio::time::timeout(Duration::from_secs(5), target_rx.recv())
.await
.unwrap()
.unwrap();
handler
.handle_control_frame(&target_ctx, route_bind_ack(bind.header.corr))
.await
.unwrap();
assert!(opened.await.unwrap().is_empty());
let _ = client_rx.recv().await.unwrap();
let root = unique_project_root("live-roots-pending");
let pending_path = ProjectRootId::from_path_allowing_missing(root.path())
.unwrap()
.as_path()
.to_path_buf();
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(303));
let open_handler = handler.clone();
let pending = tokio::spawn(async move {
open_handler
.handle_control_frame(&client_ctx, route_open_frame(3, "target", root))
.await
.unwrap()
});
let pending_bind = tokio::time::timeout(Duration::from_secs(5), target_rx.recv())
.await
.unwrap()
.unwrap();
let actual = query_live_roots(&handler, &target_ctx).await;
let ModuleControlResponseToModule::LiveRoots {
roots,
unknown_root_bindings,
total_bindings,
} = actual
else {
panic!("expected live roots")
};
assert_eq!(total_bindings, 2, "root-known arm must count live routes");
assert_eq!(unknown_root_bindings, 0);
assert_eq!(
roots.len(),
2,
"root-known arm must retain each canonical root"
);
assert_eq!(
total_bindings,
roots.iter().map(|r| r.bound + r.pending).sum::<u64>() + unknown_root_bindings
);
let counts = roots
.iter()
.map(|root| (root.project_root.clone(), root.bound, root.pending))
.collect::<Vec<_>>();
let mut expected = vec![(path, 1, 0), (pending_path, 0, 1)];
expected.sort_by(|a, b| a.0.cmp(&b.0));
assert_eq!(
counts, expected,
"roots must sort by path and count pending separately"
);
handler
.handle_control_frame(&target_ctx, route_bind_ack(pending_bind.header.corr))
.await
.unwrap();
assert!(pending.await.unwrap().is_empty());
}
#[tokio::test(start_paused = true)]
async fn supervisor_live_roots_unknown_root_arm_is_not_no_bindings() {
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::new(Registry::default()), Arc::clone(&forwarding));
let (target_ctx, _target_rx) = route_ctx(ConnectionId::new(311));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(312));
let pending = forwarding
.begin_route_bind_relay_for_test(
client_ctx.connection_id,
client_ctx.egress.clone(),
2,
"target",
)
.unwrap();
forwarding
.complete_pending_relay(
target_ctx.connection_id,
pending.corr,
RouteBindRelayOutcome::Accepted,
)
.unwrap();
let actual = query_live_roots(&handler, &target_ctx).await;
let ModuleControlResponseToModule::LiveRoots {
roots,
unknown_root_bindings,
total_bindings,
} = actual
else {
panic!("expected live roots")
};
assert!(roots.is_empty(), "unknown-root arm must not invent a root");
assert_eq!(
unknown_root_bindings, 1,
"unknown-root arm must not read as no bindings"
);
assert_eq!(total_bindings, 1, "unknown-root arm has a live binding");
assert_eq!(
total_bindings,
roots.iter().map(|r| r.bound + r.pending).sum::<u64>() + unknown_root_bindings
);
}
#[tokio::test(start_paused = true)]
async fn supervisor_live_roots_cross_module_scope_uses_requesting_connection() {
let handler = ControlHandler::with_forwarding(
Arc::new(Registry::default()),
Arc::new(ForwardingTable::default()),
);
let (first_ctx, _first_rx) = route_ctx(ConnectionId::new(315));
let (second_ctx, mut second_rx) = route_ctx(ConnectionId::new(316));
handler
.handle_control_frame(&first_ctx, hello_frame("first", PROTOCOL_VERSION, 1))
.await
.unwrap();
handler
.handle_control_frame(&second_ctx, hello_frame("second", PROTOCOL_VERSION, 2))
.await
.unwrap();
let root = unique_project_root("second-only");
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(317));
let cloned = handler.clone();
let open = tokio::spawn(async move {
cloned
.handle_control_frame(&client_ctx, route_open_frame(3, "second", root))
.await
.unwrap()
});
let bind = tokio::time::timeout(Duration::from_secs(5), second_rx.recv())
.await
.unwrap()
.unwrap();
let first = query_live_roots(&handler, &first_ctx).await;
let second = query_live_roots(&handler, &second_ctx).await;
assert!(
matches!(
first,
ModuleControlResponseToModule::LiveRoots {
total_bindings: 0,
..
}
),
"cross-module scope must not expose another module's roots"
);
assert!(
matches!(
second,
ModuleControlResponseToModule::LiveRoots {
total_bindings: 1,
..
}
),
"second module must see its pending route"
);
handler
.handle_control_frame(&second_ctx, route_bind_ack(bind.header.corr))
.await
.unwrap();
assert!(open.await.unwrap().is_empty());
}
#[tokio::test(start_paused = true)]
async fn supervisor_live_roots_no_bindings_arm_is_empty() {
let handler = ControlHandler::with_forwarding(
Arc::new(Registry::default()),
Arc::new(ForwardingTable::default()),
);
let (target_ctx, _target_rx) = route_ctx(ConnectionId::new(321));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let actual = query_live_roots(&handler, &target_ctx).await;
let ModuleControlResponseToModule::LiveRoots {
roots,
unknown_root_bindings,
total_bindings,
} = actual
else {
panic!("expected live roots")
};
assert!(roots.is_empty());
assert_eq!(unknown_root_bindings, 0);
assert_eq!(total_bindings, 0);
assert_eq!(
total_bindings,
roots.iter().map(|r| r.bound + r.pending).sum::<u64>() + unknown_root_bindings
);
}
fn fed_admission_facts_vectors() -> Vec<(String, Value)> {
let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("tests/fixtures/fed/admission-facts-emit.jsonl");
let text = std::fs::read_to_string(&path)
.unwrap_or_else(|err| panic!("vendored fed corpus unreadable at {path:?}: {err}"));
let vectors: Vec<(String, Value)> = text
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| {
let entry: Value = serde_json::from_str(line).expect("corpus line must be JSON");
let id = entry["corpus_id"]
.as_str()
.expect("every vector carries a corpus_id")
.to_string();
(id, entry["package"].clone())
})
.collect();
assert_eq!(
vectors.len(),
3,
"vendored fed corpus changed size; re-sync from subc-federation"
);
const COMMONLY_MODELLED: [&str; 3] = ["schema", "verified_class", "org"];
let richest = vectors
.iter()
.filter_map(|(_, package)| package.as_object())
.map(|object| {
object
.keys()
.filter(|key| !COMMONLY_MODELLED.contains(&key.as_str()))
.count()
})
.max()
.unwrap_or(0);
assert!(
richest >= 2,
"vendored corpus no longer carries a package with unmodelled fields, \
so the relay test can no longer distinguish a verbatim relay from a lossy one"
);
vectors
}
#[tokio::test]
async fn admission_facts_relay_carries_vendored_packages_verbatim() {
for (corpus_id, package) in fed_admission_facts_vectors() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("fed", "fed-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor)
.with_admission_facts_config(
Some("fed".to_string()),
Some(vec!["target".to_string()]),
);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(90));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(91));
let route_handler = handler.clone();
let expected = package.clone();
let route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(
&client_ctx,
route_open_frame_with_admission_facts(
20,
"target",
unique_project_root("admission-facts"),
Some(subc_control::ConsumerIdentity {
module_id: "fed".to_string(),
launch_nonce: "fed-nonce".to_string(),
}),
Some(package),
),
)
.await
.unwrap()
});
let bind_frame = target_rx.recv().await.unwrap();
let bind: ModuleControlRequest = serde_json::from_slice(&bind_frame.body).unwrap();
let ModuleControlRequest::RouteBind {
admission_facts, ..
} = bind
else {
panic!("{corpus_id}: expected route.bind")
};
assert_eq!(
admission_facts,
Some(expected),
"{corpus_id}: relay must not add, drop or reshape any field"
);
handler
.handle_control_frame(&target_ctx, route_bind_ack(bind_frame.header.corr))
.await
.unwrap();
route_task.await.unwrap();
}
}
#[tokio::test]
async fn admission_facts_gate_checks_carrier_target_and_precedence() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("fed", "fed-nonce".to_string());
supervisor.set_spawn_nonce("other", "other-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor)
.with_admission_facts_config(
Some("fed".to_string()),
Some(vec!["target".to_string()]),
);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(70));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let (other_ctx, _other_rx) = route_ctx(ConnectionId::new(71));
handler
.handle_control_frame(&other_ctx, hello_frame("other", PROTOCOL_VERSION, 2))
.await
.unwrap();
let facts = json!({"schema": 1, "verified_class": "member", "org": "01H"});
let expected_facts = facts.clone();
let (client_ctx, mut client_rx) = route_ctx(ConnectionId::new(72));
let route_handler = handler.clone();
let route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(
&client_ctx,
route_open_frame_with_admission_facts(
10,
"target",
unique_project_root("admission-facts"),
Some(subc_control::ConsumerIdentity {
module_id: "fed".to_string(),
launch_nonce: "fed-nonce".to_string(),
}),
Some(facts.clone()),
),
)
.await
.unwrap()
});
let bind_frame = target_rx.recv().await.unwrap();
let bind: ModuleControlRequest = serde_json::from_slice(&bind_frame.body).unwrap();
let ModuleControlRequest::RouteBind {
admission_facts, ..
} = bind
else {
panic!("expected route.bind")
};
assert_eq!(admission_facts, Some(expected_facts));
handler
.handle_control_frame(&target_ctx, route_bind_ack(bind_frame.header.corr))
.await
.unwrap();
assert!(route_task.await.unwrap().is_empty());
assert!(matches!(
serde_json::from_slice::<ClientControlResponse>(&client_rx.recv().await.unwrap().body)
.unwrap(),
ClientControlResponse::RouteOpen { .. }
));
let direct = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(73)).0,
route_open_frame_with_admission_facts(
11,
"target",
unique_project_root("admission-facts"),
None,
Some(json!({"x": 1})),
),
)
.await
.unwrap();
assert_eq!(
parse_error(&direct[0])["code"],
"admission_facts_not_permitted"
);
let different_reserved = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(77)).0,
route_open_frame_with_admission_facts(
15,
"target",
unique_project_root("admission-facts"),
Some(subc_control::ConsumerIdentity {
module_id: "other".to_string(),
launch_nonce: "other-nonce".to_string(),
}),
Some(json!({"x": 1})),
),
)
.await
.unwrap();
assert_eq!(
parse_error(&different_reserved[0])["code"],
"admission_facts_not_permitted"
);
let other_target = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(74)).0,
route_open_frame_with_admission_facts(
12,
"other",
unique_project_root("admission-facts"),
Some(subc_control::ConsumerIdentity {
module_id: "fed".to_string(),
launch_nonce: "fed-nonce".to_string(),
}),
Some(json!({"x": 1})),
),
)
.await
.unwrap();
assert_eq!(
parse_error(&other_target[0])["code"],
"admission_facts_target_not_allowed"
);
let nonexistent = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(75)).0,
route_open_frame_with_admission_facts(
13,
"missing",
unique_project_root("admission-facts"),
None,
Some(json!({"x": 1})),
),
)
.await
.unwrap();
assert_eq!(parse_error(&nonexistent[0])["code"], "unknown_module");
let described = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(76)).0,
Frame::build(
FrameType::Request,
control_flags(),
0,
0,
14,
serde_json::to_vec(&ClientControlRequest::ServerDescribe {}).unwrap(),
)
.unwrap(),
)
.await
.unwrap();
let ClientControlResponse::ServerDescribe { capabilities, .. } =
serde_json::from_slice(&described[0].body).unwrap()
else {
panic!("expected server.describe response")
};
assert!(capabilities
.iter()
.any(|cap| cap == "admission_facts_relay_v1"));
}
#[tokio::test]
async fn admission_facts_without_configured_carrier_are_rejected() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler = ControlHandler::with_forwarding(registry, forwarding);
let (target_ctx, _) = route_ctx(ConnectionId::new(78));
handler
.handle_control_frame(&target_ctx, hello_frame("target", PROTOCOL_VERSION, 1))
.await
.unwrap();
let responses = handler
.handle_control_frame(
&route_ctx(ConnectionId::new(79)).0,
route_open_frame_with_admission_facts(
16,
"target",
unique_project_root("admission-facts"),
None,
Some(json!({"x": 1})),
),
)
.await
.unwrap();
assert_eq!(
parse_error(&responses[0])["code"],
"admission_facts_not_permitted"
);
}
#[tokio::test]
async fn route_open_relays_consumer_capabilities_verbatim() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let (module_ctx, mut module_rx) = route_ctx(ConnectionId::new(37));
handler
.handle_control_frame(&module_ctx, hello_frame("aft", PROTOCOL_VERSION, 7))
.await
.unwrap();
let expected = vec!["elicitation".to_string(), "roots".to_string()];
let expected_for_request = expected.clone();
let project_root = unique_project_root("consumer-capabilities-present");
let (client_ctx, mut client_rx) = route_ctx(ConnectionId::new(38));
let route_handler = handler.clone();
let route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(
&client_ctx,
route_open_frame_with_consumer_capabilities(
401,
"aft",
project_root,
Some(expected_for_request),
),
)
.await
.unwrap()
});
let bind_frame = tokio::time::timeout(Duration::from_secs(1), module_rx.recv())
.await
.unwrap()
.unwrap();
let bind: ModuleControlRequest = serde_json::from_slice(&bind_frame.body).unwrap();
let ModuleControlRequest::RouteBind {
consumer_capabilities,
..
} = bind
else {
panic!("expected route.bind request, got {bind:?}");
};
assert_eq!(consumer_capabilities, Some(expected.clone()));
handler
.handle_control_frame(&module_ctx, route_bind_ack(bind_frame.header.corr))
.await
.unwrap();
let route_response = route_task.await.unwrap();
assert!(route_response.is_empty());
let published = client_rx.recv().await.unwrap();
assert!(matches!(
serde_json::from_slice::<ClientControlResponse>(&published.body).unwrap(),
ClientControlResponse::RouteOpen { .. }
));
}
#[tokio::test]
async fn route_open_without_consumer_capabilities_relays_none() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let (module_ctx, mut module_rx) = route_ctx(ConnectionId::new(39));
handler
.handle_control_frame(&module_ctx, hello_frame("aft", PROTOCOL_VERSION, 7))
.await
.unwrap();
let project_root = unique_project_root("consumer-capabilities-absent");
let (client_ctx, mut client_rx) = route_ctx(ConnectionId::new(40));
let route_handler = handler.clone();
let route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(&client_ctx, route_open_frame(402, "aft", project_root))
.await
.unwrap()
});
let bind_frame = tokio::time::timeout(Duration::from_secs(1), module_rx.recv())
.await
.unwrap()
.unwrap();
let bind: ModuleControlRequest = serde_json::from_slice(&bind_frame.body).unwrap();
let ModuleControlRequest::RouteBind {
consumer_capabilities,
..
} = bind
else {
panic!("expected route.bind request, got {bind:?}");
};
assert_eq!(consumer_capabilities, None);
handler
.handle_control_frame(&module_ctx, route_bind_ack(bind_frame.header.corr))
.await
.unwrap();
let route_response = route_task.await.unwrap();
assert!(route_response.is_empty());
let published = client_rx.recv().await.unwrap();
assert!(matches!(
serde_json::from_slice::<ClientControlResponse>(&published.body).unwrap(),
ClientControlResponse::RouteOpen { .. }
));
}
#[tokio::test]
async fn supervision_only_module_health_probe_does_not_enable_route_open_and_cleans_up() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_health_probe_timeout(Duration::from_secs(5));
let (module_ctx, mut module_rx) = route_ctx(ConnectionId::new(35));
let responses = handler
.handle_control_frame(
&module_ctx,
non_routable_hello_frame_with_control_ops(
"mcp",
300,
Some(vec![MODULE_CONTROL_OP_HEALTH_CHECK.to_string()]),
),
)
.await
.unwrap();
assert_eq!(responses[0].header.ty, FrameType::HelloAck);
assert!(registry
.get_module("mcp")
.unwrap()
.unwrap()
.manifest
.provides
.is_empty());
let (route_client_ctx, _route_client_rx) = route_ctx(ConnectionId::new(36));
let route_response = handler
.handle_control_frame(
&route_client_ctx,
route_open_frame(301, "mcp", unique_project_root("non-routable-mcp")),
)
.await
.unwrap();
assert_eq!(route_response[0].header.ty, FrameType::Error);
assert_eq!(
parse_error(&route_response[0])["code"],
"target_unavailable"
);
assert!(parse_error(&route_response[0])["message"]
.as_str()
.unwrap()
.contains("does not provide the requested target"));
assert!(module_rx.try_recv().is_err());
let (health_client_ctx, _health_client_rx) = route_ctx(ConnectionId::new(37));
let health_handler = handler.clone();
let health_task = tokio::spawn(async move {
health_handler
.handle_control_frame(
&health_client_ctx,
supervisor_health_probe_frame(302, "mcp"),
)
.await
.unwrap()
});
let health_frame = tokio::time::timeout(Duration::from_secs(1), module_rx.recv())
.await
.unwrap()
.unwrap();
assert_eq!(
serde_json::from_slice::<ModuleControlRequest>(&health_frame.body).unwrap(),
ModuleControlRequest::HealthCheck {}
);
handler
.handle_control_frame(
&module_ctx,
health_response(health_frame.header.corr, HealthStatus::Ok),
)
.await
.unwrap();
let health_response = health_task.await.unwrap();
assert_eq!(health_response[0].header.ty, FrameType::Response);
match serde_json::from_slice::<ClientControlResponse>(&health_response[0].body).unwrap() {
ClientControlResponse::SupervisorHealthProbe {
module_id, status, ..
} => {
assert_eq!(module_id, "mcp");
assert_eq!(status, HealthStatus::Ok);
}
other => panic!("unexpected health response: {other:?}"),
}
forwarding
.cleanup_connection(module_ctx.connection_id)
.unwrap();
let (cleanup_probe_ctx, _cleanup_probe_rx) = route_ctx(ConnectionId::new(38));
let cleanup_response = tokio::time::timeout(
Duration::from_millis(200),
handler.handle_control_frame(
&cleanup_probe_ctx,
supervisor_health_probe_frame(303, "mcp"),
),
)
.await
.expect("probe should fail immediately when the control lane is gone")
.unwrap();
assert_eq!(cleanup_response[0].header.ty, FrameType::Error);
assert_eq!(
parse_error(&cleanup_response[0])["code"],
"target_unavailable"
);
assert!(parse_error(&cleanup_response[0])["message"]
.as_str()
.unwrap()
.contains("no module connection"));
handler
.cleanup_connection(module_ctx.connection_id)
.unwrap();
}
#[tokio::test]
async fn route_open_classifies_unregistered_running_supervised_module_as_warming() {
let registry = Arc::new(Registry::default());
let supervisor_handle = SupervisorHandle::new();
let supervisor =
Supervisor::new(Arc::clone(®istry), RestartPolicy::new(0, Duration::ZERO))
.with_handle(supervisor_handle.clone())
.with_connection_file_path(
std::env::temp_dir()
.join(format!("subc-route-open-warming-{}", std::process::id())),
);
let module = supervisor
.supervise_configured(
ModuleSpec {
module_id: "warming".to_string(),
program: fake_aft_stub_path(),
args: Vec::new(),
env: Vec::new(),
reserved: false,
reserved_prefixes: Vec::new(),
protocol: ModuleProtocol::Subc,
overlap: Default::default(),
},
true,
)
.unwrap();
assert_eq!(module.state().unwrap(), ModuleState::Running);
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor_handle);
let (ctx, _rx) = route_ctx(ConnectionId::new(39));
let response = handler
.handle_control_frame(
&ctx,
route_open_frame(304, "warming", unique_project_root("warming")),
)
.await
.unwrap();
module.stop().await.unwrap();
assert_eq!(response[0].header.ty, FrameType::Error);
let error = parse_error(&response[0]);
assert_eq!(error["code"], "module_warming");
assert!(error["message"]
.as_str()
.unwrap()
.contains("state=running, enabled=true, live=false"));
}
#[tokio::test]
async fn route_open_refusal_names_the_check_that_refused() {
let handler = ControlHandler::new(Arc::new(Registry::default()));
let capture = EventCapture::default();
let _subscriber =
tracing::subscriber::set_default(tracing_subscriber::registry().with(capture.clone()));
let (ctx, _rx) = route_ctx(ConnectionId::new(95));
let response = handler
.handle_control_frame(
&ctx,
route_open_frame(395, "nobody", unique_project_root("refusal-reason")),
)
.await
.unwrap();
assert_eq!(parse_error(&response[0])["code"], "unknown_module");
let event = capture
.events()
.into_iter()
.find(|event| {
event.target == "control"
&& event.fields.get("code") == Some(&"\"unknown_module\"".to_string())
})
.expect("route.open refusal event");
assert_eq!(
event.fields.get("reason"),
Some(&"\"not_registered\"".to_string())
);
}
#[tokio::test]
async fn route_open_supervised_absence_emits_refusal_fields_and_counts_code() {
let registry = Arc::new(Registry::default());
let supervisor_handle = SupervisorHandle::new();
let supervisor =
Supervisor::new(Arc::clone(®istry), RestartPolicy::new(0, Duration::ZERO))
.with_handle(supervisor_handle.clone())
.with_connection_file_path(std::env::temp_dir().join(format!(
"subc-route-open-refusal-info-{}",
std::process::id()
)));
let module = supervisor
.supervise_configured(
ModuleSpec {
module_id: "warming".to_string(),
program: fake_aft_stub_path(),
args: Vec::new(),
env: Vec::new(),
reserved: false,
reserved_prefixes: Vec::new(),
protocol: ModuleProtocol::Subc,
overlap: Default::default(),
},
true,
)
.unwrap();
assert_eq!(module.state().unwrap(), ModuleState::Running);
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor_handle);
assert!(handler
.counters()
.snapshot()
.get("route_open_refused_by_code")
.is_none());
let capture = EventCapture::default();
let _subscriber =
tracing::subscriber::set_default(tracing_subscriber::registry().with(capture.clone()));
let (ctx, _rx) = route_ctx(ConnectionId::new(94));
let response = handler
.handle_control_frame(
&ctx,
route_open_frame(394, "warming", unique_project_root("refusal-info")),
)
.await
.unwrap();
module.stop().await.unwrap();
assert_eq!(parse_error(&response[0])["code"], "module_warming");
let event = capture
.events()
.into_iter()
.find(|event| {
event.target == "control"
&& event.fields.get("code") == Some(&"\"module_warming\"".to_string())
})
.expect("route.open refusal event");
assert_eq!(
event.fields.get("module_id"),
Some(&"\"warming\"".to_string())
);
assert_eq!(event.fields.get("connection_id"), Some(&"94".to_string()));
assert_eq!(
event.fields.get("reason"),
Some(&"\"supervised_not_registered\"".to_string())
);
assert_eq!(event.fields.get("state"), Some(&"running".to_string()));
assert_eq!(event.fields.get("enabled"), Some(&"true".to_string()));
assert_eq!(event.fields.get("live"), Some(&"false".to_string()));
assert_eq!(
handler.counters().snapshot()["route_open_refused_by_code"],
json!({ "module_warming": 1 })
);
}
#[tokio::test(flavor = "current_thread")]
async fn route_open_unknown_module_escapes_target_module_id() {
let handler = ControlHandler::new(Arc::new(Registry::default()));
let capture = EventCapture::default();
let _subscriber =
tracing::subscriber::set_default(tracing_subscriber::registry().with(capture.clone()));
let hostile_module_id = "\u{1b}]52;c;AAAA\u{07}";
let (ctx, _rx) = route_ctx(ConnectionId::new(95));
let response = handler
.handle_control_frame(
&ctx,
route_open_frame(
395,
hostile_module_id,
unique_project_root("hostile-target-module-id"),
),
)
.await
.unwrap();
assert_eq!(parse_error(&response[0])["code"], "unknown_module");
let event = capture
.events()
.into_iter()
.find(|event| {
event.target == "control"
&& event.fields.get("code") == Some(&"\"unknown_module\"".to_string())
})
.expect("route.open unknown-module refusal event");
let logged = event.fields.get("module_id").expect("module_id field");
assert!(!logged.bytes().any(|byte| byte < 0x20));
assert_eq!(logged, r#""\u{1b}]52;c;AAAA\u{7}""#);
}
#[tokio::test(flavor = "current_thread")]
async fn route_open_module_rejection_uses_daemon_counter_key() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let module_connection = ConnectionId::new(95);
let (module_ctx, mut module_rx) = route_ctx(module_connection);
handler
.handle_control_frame(&module_ctx, hello_frame("aft", PROTOCOL_VERSION, 395))
.await
.unwrap();
let client_connection = ConnectionId::new(96);
let (client_ctx, _client_rx) = route_ctx(client_connection);
let capture = EventCapture::default();
let _subscriber =
tracing::subscriber::set_default(tracing_subscriber::registry().with(capture.clone()));
let (route_task, bind) = relay_route_open(
&handler,
client_connection,
&client_ctx.egress,
&mut module_rx,
396,
"aft",
"hostile-module-code",
)
.await;
let hostile_code = "\u{1b}]52;c;AAAA\u{07}";
let rejection = Frame::build(
FrameType::Error,
control_flags(),
0,
0,
bind.header.corr,
serde_json::to_vec(&ErrorBody::new(hostile_code, "module refused route.bind")).unwrap(),
)
.unwrap();
handler
.handle_control_frame(&module_ctx, rejection)
.await
.unwrap();
let response = route_task.await.unwrap();
assert_eq!(parse_error(&response[0])["code"], hostile_code);
let counters = handler.counters().snapshot();
assert_eq!(
counters["route_open_refused_by_code"],
json!({ "module_rejected": 1 })
);
assert!(counters["route_open_refused_by_code"]
.get(hostile_code)
.is_none());
let event = capture
.events()
.into_iter()
.find(|event| {
event.target == "control"
&& event.fields.get("code") == Some(&"\"module_rejected\"".to_string())
})
.expect("route.open module-rejection refusal event");
let logged = event.fields.get("module_code").expect("module_code field");
assert!(!logged.bytes().any(|byte| byte < 0x20));
assert_eq!(logged, r#""\u{1b}]52;c;AAAA\u{7}""#);
}
#[tokio::test]
async fn route_open_keeps_failed_unregistered_supervised_module_unavailable() {
let registry = Arc::new(Registry::default());
let supervisor_handle = SupervisorHandle::new();
let missing_program = std::env::temp_dir().join(format!(
"subc-route-open-missing-program-{}",
std::process::id()
));
let supervisor =
Supervisor::new(Arc::clone(®istry), RestartPolicy::new(0, Duration::ZERO))
.with_handle(supervisor_handle.clone());
let module = supervisor
.supervise_configured(
ModuleSpec {
module_id: "failed".to_string(),
program: missing_program,
args: Vec::new(),
env: Vec::new(),
reserved: false,
reserved_prefixes: Vec::new(),
protocol: ModuleProtocol::Subc,
overlap: Default::default(),
},
true,
)
.unwrap();
assert_eq!(module.state().unwrap(), ModuleState::Failed);
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor_handle);
let (ctx, _rx) = route_ctx(ConnectionId::new(40));
let response = handler
.handle_control_frame(
&ctx,
route_open_frame(305, "failed", unique_project_root("failed")),
)
.await
.unwrap();
assert_eq!(response[0].header.ty, FrameType::Error);
let error = parse_error(&response[0]);
assert_eq!(error["code"], "target_unavailable");
assert!(error["message"]
.as_str()
.unwrap()
.contains("state=failed, enabled=true, live=false"));
}
#[tokio::test]
async fn route_open_role_mismatch_remains_target_unavailable() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
handler
.handle_control(
ConnectionId::new(41),
non_routable_hello_frame_with_control_ops("health-only", 306, None),
)
.unwrap();
let (ctx, _rx) = route_ctx(ConnectionId::new(42));
let response = handler
.handle_control_frame(
&ctx,
route_open_frame(307, "health-only", unique_project_root("role-mismatch")),
)
.await
.unwrap();
assert_eq!(parse_error(&response[0])["code"], "target_unavailable");
assert!(parse_error(&response[0])["message"]
.as_str()
.unwrap()
.contains("does not provide the requested target"));
}
#[tokio::test]
async fn route_open_inactive_registration_remains_target_unavailable() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
handler
.handle_control(
ConnectionId::new(43),
hello_frame("inactive", PROTOCOL_VERSION, 308),
)
.unwrap();
assert!(registry
.set_module_state_for_test("inactive", ChannelState::Closed)
.unwrap());
let (ctx, _rx) = route_ctx(ConnectionId::new(44));
let response = handler
.handle_control_frame(
&ctx,
route_open_frame(309, "inactive", unique_project_root("inactive")),
)
.await
.unwrap();
assert_eq!(parse_error(&response[0])["code"], "target_unavailable");
assert!(parse_error(&response[0])["message"]
.as_str()
.unwrap()
.contains("is not active"));
}
#[tokio::test]
async fn late_health_reply_is_recorded_through_the_module_response_path() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor_handle = SupervisorHandle::new();
let supervisor = Supervisor::new(Arc::clone(®istry), crate::RestartPolicy::default())
.with_forwarding(Arc::clone(&forwarding))
.with_handle(supervisor_handle.clone());
let module = supervisor
.supervise_configured(
crate::ModuleSpec {
module_id: "late-health-response".to_string(),
program: PathBuf::from("disabled-module"),
args: Vec::new(),
env: Vec::new(),
reserved: false,
reserved_prefixes: Vec::new(),
protocol: ModuleProtocol::Subc,
overlap: Default::default(),
},
false,
)
.unwrap();
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor_handle);
let (module_ctx, _module_rx) = route_ctx(ConnectionId::new(39));
handler
.handle_control_frame(
&module_ctx,
hello_frame_with_control_ops(
"late-health-response",
PROTOCOL_VERSION,
7,
Some(vec![MODULE_CONTROL_OP_HEALTH_CHECK.to_string()]),
),
)
.await
.unwrap();
let probe_started_at = Instant::now() - Duration::from_millis(80);
let pending = forwarding
.begin_health_probe_rpc_for(
"late-health-response",
MODULE_CONTROL_OP_HEALTH_CHECK,
probe_started_at,
Instant::now() - Duration::from_millis(1),
)
.unwrap();
assert!(forwarding
.tombstone_health_probe_rpc(pending.endpoint, pending.corr)
.unwrap());
let responses = handler
.handle_control_frame(&module_ctx, health_response(pending.corr, HealthStatus::Ok))
.await
.unwrap();
assert!(responses.is_empty());
let health = module.status().unwrap().health;
assert_eq!(health.late_answer_count, 1);
assert!(health.last_late_answer_latency_ms.unwrap() >= 80);
}
#[tokio::test]
async fn health_probe_timeout_and_module_death_are_typed() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_health_probe_timeout(Duration::from_millis(50));
let (module_ctx, mut module_rx) = route_ctx(ConnectionId::new(40));
handler
.handle_control_frame(
&module_ctx,
hello_frame_with_control_ops(
"aft",
PROTOCOL_VERSION,
7,
Some(vec![MODULE_CONTROL_OP_HEALTH_CHECK.to_string()]),
),
)
.await
.unwrap();
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(41));
let responses = handler
.handle_control_frame(&client_ctx, supervisor_health_probe_frame(201, "aft"))
.await
.unwrap();
assert_eq!(responses[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&responses[0])["code"], "module_timeout");
let _ = module_rx.try_recv();
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(42));
let health_handler = handler.clone();
let death_task = tokio::spawn(async move {
health_handler
.handle_control_frame(&client_ctx, supervisor_health_probe_frame(202, "aft"))
.await
.unwrap()
});
tokio::time::timeout(Duration::from_secs(1), module_rx.recv())
.await
.unwrap()
.unwrap();
handler
.cleanup_connection(module_ctx.connection_id)
.unwrap();
let responses = death_task.await.unwrap();
assert_eq!(responses[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&responses[0])["code"], "target_unavailable");
}
#[test]
fn hello_requires_exact_protocol_version() {
for (connection, offered) in [(1, PROTOCOL_VERSION - 1), (2, PROTOCOL_VERSION + 1)] {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let responses = handler
.handle_control(
ConnectionId::new(connection),
hello_frame("aft", offered, 9),
)
.unwrap();
assert_eq!(responses.len(), 1);
assert_eq!(responses[0].header.ty, FrameType::Error);
let error = parse_error(&responses[0]);
assert_eq!(error["code"], "version_unsupported");
assert!(registry.get_module("aft").unwrap().is_none());
assert_eq!(registry.active_registration_count().unwrap(), 0);
}
}
#[test]
fn unknown_module_push_op_is_ignored_but_malformed_known_op_errors() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let module_connection = ConnectionId::new(301);
let registration = registry
.register_with_control_ops(
manifest("aft-push", PROTOCOL_VERSION),
PROTOCOL_VERSION,
module_connection,
module_baseline_control_ops(),
)
.unwrap();
let (module_tx, _module_rx) = mpsc::channel(8);
let endpoint = forwarding
.register_module_connection(
module_connection,
"aft-push".to_string(),
PROTOCOL_VERSION,
manifest_concurrency(®istration.manifest),
FrameSink::new(module_tx),
)
.unwrap();
let unknown = Frame::build(
FrameType::Push,
control_flags(),
0,
0,
5,
serde_json::to_vec(&json!({"op": "route.future.v2", "extra": 1})).unwrap(),
)
.unwrap();
let out = handler.handle_status_update(endpoint, unknown).unwrap();
assert!(
out.is_empty(),
"unknown push op must be ignored, got {out:?}"
);
let malformed = Frame::build(
FrameType::Push,
control_flags(),
0,
0,
6,
serde_json::to_vec(&json!({"op": "route.status"})).unwrap(),
)
.unwrap();
let out = handler.handle_status_update(endpoint, malformed).unwrap();
assert_eq!(out.len(), 1);
assert_eq!(out[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&out[0])["code"], "invalid_control_body");
}
#[test]
fn hello_rejected_when_connection_already_owns_client_routes() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let _ = bind_liveness_route(®istry, &forwarding, "aft-module");
let client_connection = ConnectionId::new(202);
let responses = handler
.handle_control(
client_connection,
hello_frame("aft-second", PROTOCOL_VERSION, 9),
)
.unwrap();
assert_eq!(responses[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&responses[0])["code"], "invalid_hello");
assert!(registry.get_module("aft-second").unwrap().is_none());
}
#[test]
fn reserved_module_hello_requires_matching_launch_nonce() {
let registry = Arc::new(Registry::default());
let supervisor = SupervisorHandle::new();
supervisor.set_reserved_nonce("vault", "the-real-nonce".to_string());
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor);
let no_nonce = handler
.handle_control(
ConnectionId::new(1),
hello_frame("vault", PROTOCOL_VERSION, 1),
)
.unwrap();
assert_eq!(no_nonce[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&no_nonce[0])["code"], "reserved_module");
assert!(registry.get_module("vault").unwrap().is_none());
let wrong = handler
.handle_control(
ConnectionId::new(2),
hello_frame_with_nonce("vault", PROTOCOL_VERSION, 2, Some("forged")),
)
.unwrap();
assert_eq!(wrong[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&wrong[0])["code"], "reserved_module");
assert!(registry.get_module("vault").unwrap().is_none());
let ok = handler
.handle_control(
ConnectionId::new(3),
hello_frame_with_nonce("vault", PROTOCOL_VERSION, 3, Some("the-real-nonce")),
)
.unwrap();
assert_eq!(ok[0].header.ty, FrameType::HelloAck);
assert!(registry.get_module("vault").unwrap().is_some());
}
#[test]
fn reserved_prefix_hello_uses_delimiter_sensitive_owner_nonce() {
let registry = Arc::new(Registry::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("federation", "owner-nonce".to_string());
supervisor.set_reserved_prefixes("federation", &["fed:".to_string()]);
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor);
let squat = handler
.handle_control(
ConnectionId::new(1),
hello_frame("fed:peerA:tool", PROTOCOL_VERSION, 1),
)
.unwrap();
assert_eq!(squat[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&squat[0])["code"], "reserved_module");
assert!(parse_error(&squat[0])["message"]
.as_str()
.unwrap()
.contains("fed:"));
let accepted_peer = handler
.handle_control(
ConnectionId::new(2),
hello_frame_with_nonce("fed:peerA:tool", PROTOCOL_VERSION, 2, Some("owner-nonce")),
)
.unwrap();
assert_eq!(accepted_peer[0].header.ty, FrameType::HelloAck);
let accepted_short = handler
.handle_control(
ConnectionId::new(3),
hello_frame_with_nonce("fed:x", PROTOCOL_VERSION, 3, Some("owner-nonce")),
)
.unwrap();
assert_eq!(accepted_short[0].header.ty, FrameType::HelloAck);
for (conn, module_id) in [(4, "fedx:tool"), (5, "fed"), (6, "FED:x")] {
let response = handler
.handle_control(
ConnectionId::new(conn),
hello_frame(module_id, PROTOCOL_VERSION, conn),
)
.unwrap();
assert_eq!(response[0].header.ty, FrameType::HelloAck, "{module_id}");
}
}
#[test]
fn exact_reserved_module_takes_precedence_over_reserved_prefix() {
let registry = Arc::new(Registry::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("federation", "owner-nonce".to_string());
supervisor.set_reserved_prefixes("federation", &["fed:".to_string()]);
supervisor.set_reserved_nonce("fed:special", "exact-nonce".to_string());
let handler = ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor);
let owner_nonce = handler
.handle_control(
ConnectionId::new(1),
hello_frame_with_nonce("fed:special", PROTOCOL_VERSION, 1, Some("owner-nonce")),
)
.unwrap();
assert_eq!(owner_nonce[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&owner_nonce[0])["code"], "reserved_module");
assert!(registry.get_module("fed:special").unwrap().is_none());
let exact_nonce = handler
.handle_control(
ConnectionId::new(2),
hello_frame_with_nonce("fed:special", PROTOCOL_VERSION, 2, Some("exact-nonce")),
)
.unwrap();
assert_eq!(exact_nonce[0].header.ty, FrameType::HelloAck);
assert!(registry.get_module("fed:special").unwrap().is_some());
}
#[test]
fn non_reserved_module_ignores_launch_nonce() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
let no_nonce = handler
.handle_control(
ConnectionId::new(1),
hello_frame("aft-no-nonce", PROTOCOL_VERSION, 1),
)
.unwrap();
assert_eq!(no_nonce[0].header.ty, FrameType::HelloAck);
assert!(registry.get_module("aft-no-nonce").unwrap().is_some());
let echoed_nonce = handler
.handle_control(
ConnectionId::new(2),
hello_frame_with_nonce("aft-with-nonce", PROTOCOL_VERSION, 2, Some("spawn-nonce")),
)
.unwrap();
assert_eq!(echoed_nonce[0].header.ty, FrameType::HelloAck);
assert!(registry.get_module("aft-with-nonce").unwrap().is_some());
}
#[test]
fn malformed_hello_returns_error_and_handler_still_answers_ping() {
let handler = ControlHandler::default();
let conn = ConnectionId::new(1);
let malformed = Frame::build(
FrameType::Hello,
control_flags(),
0,
0,
3,
b"{not json".to_vec(),
)
.unwrap();
let error = handler.handle_control(conn, malformed).unwrap();
assert_eq!(error[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&error[0])["code"], "invalid_hello");
let ping = Frame::build(FrameType::Ping, control_flags(), 0, 0, 4, Vec::new()).unwrap();
let pong = handler.handle_control(conn, ping).unwrap();
assert_eq!(pong[0].header.ty, FrameType::Pong);
assert_eq!(pong[0].header.corr, 4);
}
#[test]
fn duplicate_module_id_is_rejected_without_replacing_active_registration() {
let registry = Arc::new(Registry::default());
let handler = ControlHandler::new(Arc::clone(®istry));
handler
.handle_control(
ConnectionId::new(1),
hello_frame("aft", PROTOCOL_VERSION, 1),
)
.unwrap();
let duplicate = handler
.handle_control(
ConnectionId::new(2),
hello_frame("aft", PROTOCOL_VERSION, 2),
)
.unwrap();
assert_eq!(duplicate[0].header.ty, FrameType::Error);
assert_eq!(parse_error(&duplicate[0])["code"], "duplicate_module_id");
let registration = registry.get_module("aft").unwrap().unwrap();
assert_eq!(registration.connection_id, ConnectionId::new(1));
}
#[test]
fn liveness_poll_reports_false_when_process_liveness_reports_dead() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let process_liveness = Arc::new(FakeProcessLiveness { live: Some(false) });
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_process_liveness(process_liveness);
let (ctx, route_channel, route_epoch) =
bind_liveness_route(®istry, &forwarding, "aft-dead");
let responses = handler
.handle_route_poll(
&ctx,
route_poll_frame(41, PollKind::Liveness, route_channel),
route_channel,
route_epoch,
PollKind::Liveness,
)
.unwrap();
assert_eq!(responses.len(), 1);
assert_eq!(responses[0].header.ty, FrameType::Response);
assert_route_poll_liveness(&responses[0], false);
}
#[test]
fn liveness_poll_without_process_source_uses_bound_route() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let (ctx, route_channel, route_epoch) =
bind_liveness_route(®istry, &forwarding, "aft-bound-only");
let responses = handler
.handle_route_poll(
&ctx,
route_poll_frame(42, PollKind::Liveness, route_channel),
route_channel,
route_epoch,
PollKind::Liveness,
)
.unwrap();
assert_route_poll_liveness(&responses[0], true);
}
#[test]
fn liveness_poll_untracked_process_source_uses_bound_route() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let process_liveness = Arc::new(FakeProcessLiveness { live: None });
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_process_liveness(process_liveness);
let (ctx, route_channel, route_epoch) =
bind_liveness_route(®istry, &forwarding, "aft-untracked");
let responses = handler
.handle_route_poll(
&ctx,
route_poll_frame(43, PollKind::Liveness, route_channel),
route_channel,
route_epoch,
PollKind::Liveness,
)
.unwrap();
assert_route_poll_liveness(&responses[0], true);
}
#[tokio::test]
async fn unknown_op_returns_unknown_control_op() {
let handler = ControlHandler::default();
let (ctx, _rx) = route_ctx(ConnectionId::new(77));
let request = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
55,
br#"{"op":"route.nope","route_channel":1}"#.to_vec(),
)
.unwrap();
let response = handler.handle_control_frame(&ctx, request).await.unwrap();
assert_eq!(response.len(), 1);
assert_eq!(response[0].header.ty, FrameType::Error);
assert_eq!(response[0].header.corr, 55);
assert_eq!(parse_error(&response[0])["code"], "unknown_control_op");
}
#[tokio::test]
async fn supervisor_provenance_rejects_unknown_exact_module() {
let handler = ControlHandler::default();
let (ctx, _rx) = route_ctx(ConnectionId::new(79));
let request = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
57,
br#"{"op":"supervisor.provenance","module_id":"missing"}"#.to_vec(),
)
.unwrap();
let response = handler.handle_control_frame(&ctx, request).await.unwrap();
assert_eq!(response.len(), 1);
assert_eq!(response[0].header.ty, FrameType::Error);
assert_eq!(response[0].header.corr, 57);
let error = parse_error(&response[0]);
assert_eq!(error["code"], "unknown_module");
assert_eq!(error["message"], "module_id 'missing' is not supervised");
}
#[test]
fn provenance_probe_override_keeps_handler_tests_deterministic() {
let expected = subc_control::RunningImageAgreement::Unavailable {
reason: subc_control::RunningImageUnavailableReason::HashFailed,
};
let handler = ControlHandler::default().with_provenance_probe_result(expected.clone());
assert_eq!(handler.provenance_probe_override, Some(expected));
}
#[tokio::test]
async fn malformed_control_bodies_return_invalid_control_body() {
let handler = ControlHandler::default();
let (ctx, _rx) = route_ctx(ConnectionId::new(78));
for (corr, body) in [
(56, br#"{"route_channel":1}"#.as_slice()),
(57, br#"{"op":17,"route_channel":1}"#.as_slice()),
(
58,
br#"{"op":"route.poll","route_channel":"bad","kind":"status"}"#.as_slice(),
),
] {
let request = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
corr,
body.to_vec(),
)
.unwrap();
let response = handler.handle_control_frame(&ctx, request).await.unwrap();
assert_eq!(response.len(), 1);
assert_eq!(response[0].header.ty, FrameType::Error);
assert_eq!(response[0].header.corr, corr);
assert_eq!(parse_error(&response[0])["code"], "invalid_control_body");
}
}
#[tokio::test]
async fn goodbye_tears_down_registration_and_later_channel_is_unknown() {
let registry = Arc::new(Registry::default());
let control = Arc::new(ControlHandler::new(Arc::clone(®istry)));
let router = Router::with_control_handler(Arc::clone(&control));
let connection = router.begin_connection();
let (ctx, mut rx) = route_ctx(connection.id());
router
.route_for_connection(&ctx, hello_frame("aft", PROTOCOL_VERSION, 11))
.await
.unwrap();
let response = rx.recv().await.unwrap();
let ack = parse_ack(&response);
assert_eq!(ack.negotiated_ver, PROTOCOL_VERSION);
let channel = 1;
let goodbye =
Frame::build(FrameType::Goodbye, control_flags(), 0, 0, 12, Vec::new()).unwrap();
router.route_for_connection(&ctx, goodbye).await.unwrap();
assert!(rx.try_recv().is_err());
assert!(registry.get_module("aft").unwrap().is_none());
router
.route_for_connection(&ctx, channel_request(channel, 13))
.await
.unwrap();
let error_frame = rx.recv().await.unwrap();
assert_eq!(error_frame.header.ty, FrameType::Error);
assert_eq!(error_frame.header.channel, channel);
}
#[tokio::test]
async fn dropping_router_connection_releases_registration() {
let registry = Arc::new(Registry::default());
let control = Arc::new(ControlHandler::new(Arc::clone(®istry)));
let router = Router::with_control_handler(control);
let connection = router.begin_connection();
let (ctx, mut rx) = route_ctx(connection.id());
router
.route_for_connection(&ctx, hello_frame("aft", PROTOCOL_VERSION, 31))
.await
.unwrap();
let response = rx.recv().await.unwrap();
let ack = parse_ack(&response);
assert_eq!(ack.negotiated_ver, PROTOCOL_VERSION);
assert!(registry.get_module("aft").unwrap().is_some());
drop(connection);
assert!(registry.get_module("aft").unwrap().is_none());
assert_eq!(registry.active_registration_count().unwrap(), 0);
}
fn capability_manifest(
module_id: &str,
provides: &[&str],
must_never_reach: &[&str],
) -> ModuleManifest {
let mut manifest = manifest(module_id, PROTOCOL_VERSION);
manifest.capabilities = Some(CapabilityDeclarations {
provides: provides
.iter()
.map(|capability| (*capability).to_string())
.collect(),
requires: Vec::new(),
must_never_reach: must_never_reach
.iter()
.map(|capability| (*capability).to_string())
.collect(),
});
manifest
}
fn hello_frame_with_manifest(manifest: ModuleManifest, corr: u64) -> Frame {
Frame::build(
FrameType::Hello,
control_flags(),
0,
0,
corr,
serde_json::to_vec(&ModuleHelloBody {
protocol_ver: manifest.protocol_ver,
manifest,
control_ops: None,
launch_nonce: None,
})
.expect("capability test HELLO serializes"),
)
.expect("capability test HELLO frame builds")
}
fn catalog_update_with_capabilities_frame(
corr: u64,
capabilities: CapabilityDeclarations,
) -> Frame {
Frame::build(
FrameType::Request,
control_flags(),
0,
0,
corr,
serde_json::to_vec(&ModuleControlRequestFromModule::CatalogUpdate {
provides: manifest("catalog-update-placeholder", PROTOCOL_VERSION).provides,
capabilities: Some(capabilities),
ready: None,
})
.expect("capability catalog.update serializes"),
)
.expect("capability catalog.update frame builds")
}
async fn register_capability_manifest(
handler: &ControlHandler,
ctx: &RouteCtx,
manifest: ModuleManifest,
corr: u64,
) {
let replies = handler
.handle_control_frame(ctx, hello_frame_with_manifest(manifest, corr))
.await
.expect("capability test HELLO succeeds");
assert!(
matches!(replies.as_slice(), [Frame { header, .. }] if header.ty == FrameType::HelloAck),
"capability test HELLO must register"
);
}
async fn open_route_for_capability_test(
handler: &ControlHandler,
target_ctx: &RouteCtx,
target_rx: &mut mpsc::Receiver<crate::router::OutboundFrame>,
client_connection_id: u64,
corr: u64,
target_module_id: &str,
consumer_identity: Option<ConsumerIdentity>,
) -> (
mpsc::Receiver<crate::router::OutboundFrame>,
ModuleControlRequest,
) {
let (client_ctx, mut client_rx) = route_ctx(ConnectionId::new(client_connection_id));
let route_handler = handler.clone();
let target_module_id = target_module_id.to_string();
let route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(
&client_ctx,
route_open_frame_with_admission_facts(
corr,
&target_module_id,
unique_project_root("admission-facts"),
consumer_identity,
None,
),
)
.await
.expect("capability test route.open succeeds")
});
let bind = tokio::time::timeout(Duration::from_secs(1), target_rx.recv())
.await
.expect("capability test route.open must reach route.bind")
.expect("target control receiver stays open");
let bind_request: ModuleControlRequest =
serde_json::from_slice(&bind.body).expect("route.bind decodes");
handler
.handle_control_frame(target_ctx, route_bind_ack(bind.header.corr))
.await
.expect("capability test route.bind ACK succeeds");
assert!(route_task.await.expect("route.open task joins").is_empty());
let opened = client_rx
.recv()
.await
.expect("successful route.open publishes a response");
assert!(matches!(
serde_json::from_slice::<ClientControlResponse>(&opened.body),
Ok(ClientControlResponse::RouteOpen { .. })
));
(client_rx, bind_request)
}
fn assert_capability_denied_push(frame: Frame, target_module_id: &str) {
assert_eq!(frame.header.ty, FrameType::Push);
assert_eq!(frame.header.channel, 0);
assert_eq!(
serde_json::from_slice::<ClientControlPush>(&frame.body)
.expect("route.closed control push decodes"),
ClientControlPush::RouteClosed {
module_id: target_module_id.to_string(),
reason: RouteCloseReason::CapabilityDenied,
drained: false,
abandoned: 0,
excluded_subscriptions: 0,
terminal: Some(false),
}
);
}
#[tokio::test]
async fn route_open_capability_forbidden_mutation_proof_creates_no_route() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("opener", "opener-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(700));
let (opener_ctx, _opener_rx) = route_ctx(ConnectionId::new(701));
register_capability_manifest(
&handler,
&target_ctx,
capability_manifest("target", &["credentials-provider/v1"], &[]),
1,
)
.await;
register_capability_manifest(
&handler,
&opener_ctx,
capability_manifest("opener", &[], &["credentials-provider/v1"]),
2,
)
.await;
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(702));
let replies = handler
.handle_control_frame(
&client_ctx,
route_open_frame_with_admission_facts(
3,
"target",
unique_project_root("admission-facts"),
Some(ConsumerIdentity {
module_id: "opener".to_string(),
launch_nonce: "opener-nonce".to_string(),
}),
None,
),
)
.await
.expect("denied route.open returns a typed frame");
assert_eq!(parse_error(&replies[0])["code"], "capability_forbidden");
assert_eq!(forwarding.active_binding_count().unwrap(), 0);
assert!(
target_rx.try_recv().is_err(),
"forbidden route.open must not relay route.bind"
);
}
#[tokio::test]
async fn capability_deny_edge_hello_mutation_proof_force_closes_existing_route() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("opener", "opener-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(710));
let (old_opener_ctx, _old_opener_rx) = route_ctx(ConnectionId::new(711));
register_capability_manifest(
&handler,
&target_ctx,
capability_manifest("target", &["credentials-provider/v1"], &[]),
1,
)
.await;
register_capability_manifest(
&handler,
&old_opener_ctx,
capability_manifest("opener", &[], &[]),
2,
)
.await;
let (mut client_rx, _) = open_route_for_capability_test(
&handler,
&target_ctx,
&mut target_rx,
712,
3,
"target",
Some(ConsumerIdentity {
module_id: "opener".to_string(),
launch_nonce: "opener-nonce".to_string(),
}),
)
.await;
assert_eq!(forwarding.active_binding_count().unwrap(), 1);
handler
.cleanup_connection(old_opener_ctx.connection_id)
.expect("old opener registration cleans up");
let (new_opener_ctx, _new_opener_rx) = route_ctx(ConnectionId::new(713));
register_capability_manifest(
&handler,
&new_opener_ctx,
capability_manifest("opener", &[], &["credentials-provider/v1"]),
4,
)
.await;
assert_capability_denied_push(
client_rx
.try_recv()
.expect("HELLO deny addition must emit route.closed")
.frame,
"target",
);
assert_eq!(forwarding.active_binding_count().unwrap(), 0);
assert!(matches!(
target_rx.try_recv(),
Ok(outbound) if outbound.header.ty == FrameType::Goodbye
));
}
#[tokio::test]
async fn capability_claim_catalog_update_mutation_proof_force_closes_existing_route() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("opener", "opener-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(720));
let (opener_ctx, _opener_rx) = route_ctx(ConnectionId::new(721));
register_capability_manifest(
&handler,
&target_ctx,
capability_manifest("target", &[], &[]),
1,
)
.await;
register_capability_manifest(
&handler,
&opener_ctx,
capability_manifest("opener", &[], &["credentials-provider/v1"]),
2,
)
.await;
let (mut client_rx, _) = open_route_for_capability_test(
&handler,
&target_ctx,
&mut target_rx,
722,
3,
"target",
Some(ConsumerIdentity {
module_id: "opener".to_string(),
launch_nonce: "opener-nonce".to_string(),
}),
)
.await;
assert_eq!(forwarding.active_binding_count().unwrap(), 1);
let replies = handler
.handle_control_frame(
&target_ctx,
catalog_update_with_capabilities_frame(
4,
CapabilityDeclarations {
provides: vec!["credentials-provider/v1".to_string()],
requires: Vec::new(),
must_never_reach: Vec::new(),
},
),
)
.await
.expect("claim catalog.update succeeds");
assert!(matches!(
serde_json::from_slice::<ModuleControlResponseToModule>(&replies[0].body),
Ok(ModuleControlResponseToModule::CatalogUpdate {})
));
assert_capability_denied_push(
client_rx
.try_recv()
.expect("claim addition must emit route.closed")
.frame,
"target",
);
assert_eq!(forwarding.active_binding_count().unwrap(), 0);
assert!(matches!(
target_rx.try_recv(),
Ok(outbound) if outbound.header.ty == FrameType::Goodbye
));
}
#[tokio::test]
async fn capability_claim_removal_mutation_proof_keeps_route_open_without_close_frame() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("opener", "opener-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(730));
let (opener_ctx, _opener_rx) = route_ctx(ConnectionId::new(731));
register_capability_manifest(
&handler,
&target_ctx,
capability_manifest("target", &["credentials-provider/v1"], &[]),
1,
)
.await;
register_capability_manifest(
&handler,
&opener_ctx,
capability_manifest("opener", &[], &[]),
2,
)
.await;
let (mut client_rx, _) = open_route_for_capability_test(
&handler,
&target_ctx,
&mut target_rx,
732,
3,
"target",
Some(ConsumerIdentity {
module_id: "opener".to_string(),
launch_nonce: "opener-nonce".to_string(),
}),
)
.await;
assert_eq!(forwarding.active_binding_count().unwrap(), 1);
handler
.handle_control_frame(
&target_ctx,
catalog_update_with_capabilities_frame(
4,
CapabilityDeclarations {
provides: Vec::new(),
requires: Vec::new(),
must_never_reach: Vec::new(),
},
),
)
.await
.expect("claim removal catalog.update succeeds");
assert_eq!(
forwarding.active_binding_count().unwrap(),
1,
"removing an attested target claim must leave the route census unchanged"
);
assert!(
client_rx.try_recv().is_err(),
"claim removal must not emit route.closed capability_denied"
);
assert!(
target_rx.try_recv().is_err(),
"claim removal must not send the target a route GOODBYE"
);
}
#[tokio::test]
async fn direct_client_scope_honesty_mutation_proof_opens_denied_capability_provider() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("opener", "opener-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (target_ctx, mut target_rx) = route_ctx(ConnectionId::new(740));
let (opener_ctx, _opener_rx) = route_ctx(ConnectionId::new(741));
register_capability_manifest(
&handler,
&target_ctx,
capability_manifest("target", &["credentials-provider/v1"], &[]),
1,
)
.await;
register_capability_manifest(
&handler,
&opener_ctx,
capability_manifest("opener", &[], &["credentials-provider/v1"]),
2,
)
.await;
let (_client_rx, bind) = open_route_for_capability_test(
&handler,
&target_ctx,
&mut target_rx,
742,
3,
"target",
None,
)
.await;
let ModuleControlRequest::RouteBind { principal, .. } = bind else {
panic!("direct scope-honesty route must bind");
};
assert_eq!(principal, Some(Principal::Direct));
assert_eq!(forwarding.active_binding_count().unwrap(), 1);
}
#[tokio::test]
async fn must_never_reach_self_route_is_capability_forbidden() {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce("self-provider", "self-nonce".to_string());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding))
.with_supervisor(supervisor);
let (self_ctx, mut self_rx) = route_ctx(ConnectionId::new(750));
register_capability_manifest(
&handler,
&self_ctx,
capability_manifest(
"self-provider",
&["credentials-provider/v1"],
&["credentials-provider/v1"],
),
1,
)
.await;
let (client_ctx, _client_rx) = route_ctx(ConnectionId::new(751));
let replies = handler
.handle_control_frame(
&client_ctx,
route_open_frame_with_admission_facts(
2,
"self-provider",
unique_project_root("admission-facts"),
Some(ConsumerIdentity {
module_id: "self-provider".to_string(),
launch_nonce: "self-nonce".to_string(),
}),
None,
),
)
.await
.expect("self-route refusal returns a typed frame");
assert_eq!(parse_error(&replies[0])["code"], "capability_forbidden");
assert_eq!(forwarding.active_binding_count().unwrap(), 0);
assert!(
self_rx.try_recv().is_err(),
"self denial must not relay route.bind"
);
}
#[test]
fn unsupported_channel_zero_frame_returns_error() {
let handler = ControlHandler::default();
let request = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
21,
b"opaque".to_vec(),
)
.unwrap();
let response = handler
.handle_control(ConnectionId::new(1), request)
.unwrap();
assert_eq!(response[0].header.ty, FrameType::Error);
assert_eq!(
parse_error(&response[0])["code"],
"unsupported_control_frame"
);
}
mod swap {
use super::*;
const INCUMBENT: ConnectionId = ConnectionId::new(30);
const CANDIDATE: ConnectionId = ConnectionId::new(40);
struct Swap {
registry: Arc<Registry>,
forwarding: Arc<ForwardingTable>,
handler: ControlHandler,
incumbent_ctx: RouteCtx,
incumbent_rx: mpsc::Receiver<crate::router::OutboundFrame>,
candidate_ctx: RouteCtx,
candidate_rx: mpsc::Receiver<crate::router::OutboundFrame>,
}
async fn swap_with_incumbent() -> Swap {
let registry = Arc::new(Registry::default());
let forwarding = Arc::new(ForwardingTable::default());
let handler =
ControlHandler::with_forwarding(Arc::clone(®istry), Arc::clone(&forwarding));
let (incumbent_ctx, incumbent_rx) = route_ctx(INCUMBENT);
handler
.handle_control_frame(&incumbent_ctx, hello_frame("aft", PROTOCOL_VERSION, 7))
.await
.unwrap();
let (candidate_ctx, candidate_rx) = route_ctx(CANDIDATE);
Swap {
registry,
forwarding,
handler,
incumbent_ctx,
incumbent_rx,
candidate_ctx,
candidate_rx,
}
}
fn register_candidate(swap: &Swap, ready: Option<bool>) {
let mut candidate_manifest = manifest("aft", PROTOCOL_VERSION);
candidate_manifest.ready = ready;
let registration = swap
.registry
.register_candidate_with_control_ops(
candidate_manifest,
PROTOCOL_VERSION,
CANDIDATE,
module_baseline_control_ops(),
)
.unwrap();
swap.forwarding
.register_candidate_module_connection(
CANDIDATE,
"aft".to_string(),
PROTOCOL_VERSION,
manifest_concurrency(®istration.manifest),
swap.candidate_ctx.egress.clone(),
)
.unwrap();
}
fn cutover(swap: &Swap) -> crate::forwarding::ModuleEndpointId {
let cutover = swap.forwarding.cutover_candidate("aft").unwrap().unwrap();
swap.registry.promote_candidate("aft").unwrap().unwrap();
cutover.incumbent.unwrap()
}
fn keyed_total(counters: &Value, key: &str) -> u64 {
counters[key]
.as_object()
.map(|counts| counts.values().filter_map(Value::as_u64).sum())
.unwrap_or(0)
}
#[tokio::test]
async fn incumbent_ack_between_promotion_and_drain_keeps_the_incumbent_serving() {
let mut swap = swap_with_incumbent().await;
let handler = swap.handler.clone();
let cotenant = ConnectionId::new(31);
let (cotenant_ctx, mut cotenant_rx) = route_ctx(cotenant);
let (cotenant_task, cotenant_bind) = relay_route_open(
&handler,
cotenant,
&cotenant_ctx.egress,
&mut swap.incumbent_rx,
100,
"aft",
"swap-cotenant",
)
.await;
handler
.handle_control_frame(
&swap.incumbent_ctx,
route_bind_ack(cotenant_bind.header.corr),
)
.await
.unwrap();
assert!(cotenant_task.await.unwrap().is_empty());
let (cotenant_channel, cotenant_epoch) =
published_route(&cotenant_rx.recv().await.unwrap());
let caller = ConnectionId::new(32);
let (caller_ctx, mut caller_rx) = route_ctx(caller);
let (caller_task, caller_bind) = relay_route_open(
&handler,
caller,
&caller_ctx.egress,
&mut swap.incumbent_rx,
101,
"aft",
"swap-caller",
)
.await;
let (abandoned_channel, abandoned_epoch) = route_bind_channel(&caller_bind);
register_candidate(&swap, None);
cutover(&swap);
let ack = handler
.handle_control_frame(&swap.incumbent_ctx, route_bind_ack(caller_bind.header.corr))
.await;
let module_loop_error = ack.as_ref().err().map(ToString::to_string);
if module_loop_error.is_some() {
handler.cleanup_connection(INCUMBENT).unwrap();
}
assert!(
cotenant_rx.try_recv().is_err(),
"the co-tenant route on the incumbent was torn down by one late ack: \
{module_loop_error:?}"
);
assert!(matches!(
swap.forwarding
.lookup_data_route(cotenant, cotenant_channel, cotenant_epoch)
.unwrap(),
DataRoute::Client(DataRouteState::Bound(_))
));
assert_eq!(module_loop_error, None);
assert!(swap
.registry
.get_module_by_connection(INCUMBENT)
.unwrap()
.is_some());
let goodbye = tokio::time::timeout(Duration::from_secs(1), swap.incumbent_rx.recv())
.await
.expect("the incumbent is told to drop the abandoned binding")
.unwrap()
.frame;
assert_eq!(goodbye.header.ty, FrameType::Goodbye);
assert_eq!(goodbye.header.channel, abandoned_channel);
assert_eq!(goodbye.header.epoch, abandoned_epoch);
assert!(swap.incumbent_rx.try_recv().is_err());
let response = caller_task.await.unwrap();
assert_eq!(response.len(), 1);
assert_eq!(parse_error(&response[0])["code"], "module_reloading");
assert!(caller_rx.try_recv().is_err());
assert_eq!(swap.forwarding.reserved_route_count().unwrap(), (0, 0));
let counters = handler.counters().snapshot();
assert_eq!(
keyed_total(&counters, "route_open_accepted_by_principal"),
1
);
assert_eq!(keyed_total(&counters, "route_open_refused_by_code"), 1);
assert_eq!(counters["route_open_refused_by_code"]["module_rejected"], 1);
}
#[tokio::test]
async fn route_open_after_cutover_and_incumbent_drain_is_relayed_to_the_candidate() {
let mut swap = swap_with_incumbent().await;
register_candidate(&swap, None);
let incumbent = cutover(&swap);
swap.forwarding
.begin_endpoint_drain(incumbent, RouteCloseReason::Restart)
.unwrap()
.expect("the incumbent is still registered");
let client = ConnectionId::new(33);
let (client_ctx, mut client_rx) = route_ctx(client);
let route_handler = swap.handler.clone();
let open_ctx = RouteCtx {
connection_id: client,
egress: client_ctx.egress.clone(),
};
let mut route_task = tokio::spawn(async move {
route_handler
.handle_control_frame(
&open_ctx,
route_open_frame(90, "aft", unique_project_root("swap-after-drain")),
)
.await
.unwrap()
});
let bind = tokio::select! {
bind = swap.candidate_rx.recv() => bind.expect("candidate egress is open").frame,
response = &mut route_task => {
let response = response.unwrap();
panic!(
"post-cutover route.open was refused instead of relayed to the candidate: {}",
parse_error(&response[0])["code"]
);
}
};
swap.handler
.handle_control_frame(&swap.candidate_ctx, route_bind_ack(bind.header.corr))
.await
.unwrap();
assert!(route_task.await.unwrap().is_empty());
let (channel, epoch) = published_route(&client_rx.recv().await.unwrap());
match swap
.forwarding
.lookup_data_route(client, channel, epoch)
.unwrap()
{
DataRoute::Client(DataRouteState::Bound(route)) => {
assert_eq!(route.module_endpoint.connection_id, CANDIDATE)
}
other => panic!("expected a bound route on the candidate, got {other:?}"),
}
assert!(swap.incumbent_rx.try_recv().is_err());
}
#[tokio::test]
async fn candidate_catalog_update_ready_reaches_the_candidate_registration() {
let swap = swap_with_incumbent().await;
register_candidate(&swap, Some(false));
let update = Frame::build(
FrameType::Request,
control_flags(),
0,
0,
55,
serde_json::to_vec(&ModuleControlRequestFromModule::CatalogUpdate {
provides: manifest("aft", PROTOCOL_VERSION).provides,
capabilities: None,
ready: Some(true),
})
.unwrap(),
)
.unwrap();
let replies = swap
.handler
.handle_control_frame(&swap.candidate_ctx, update)
.await
.unwrap();
assert_eq!(replies.len(), 1);
assert_eq!(
replies[0].header.ty,
FrameType::Response,
"candidate catalog.update was refused: {:?}",
serde_json::from_slice::<Value>(&replies[0].body).ok()
);
assert!(swap.registry.get_candidate("aft").unwrap().unwrap().ready);
assert_eq!(
swap.registry
.get_module("aft")
.unwrap()
.unwrap()
.connection_id,
INCUMBENT
);
}
}
mod swap_admission {
use super::*;
const INCUMBENT_NONCE: &str = "incumbent-nonce";
const CANDIDATE_NONCE: &str = "candidate-nonce";
fn handler_with_incumbent(
module_id: &str,
reserved: bool,
) -> (Arc<Registry>, SupervisorHandle, ControlHandler) {
let registry = Arc::new(Registry::default());
let supervisor = SupervisorHandle::new();
supervisor.set_spawn_nonce(module_id, INCUMBENT_NONCE.to_string());
if reserved {
supervisor.set_reserved_nonce(module_id, INCUMBENT_NONCE.to_string());
}
let handler =
ControlHandler::new(Arc::clone(®istry)).with_supervisor(supervisor.clone());
let incumbent = handler
.handle_control(
ConnectionId::new(1),
hello_frame_with_nonce(module_id, PROTOCOL_VERSION, 1, Some(INCUMBENT_NONCE)),
)
.unwrap();
assert_eq!(incumbent[0].header.ty, FrameType::HelloAck);
supervisor.open_swap(module_id, CANDIDATE_NONCE.to_string());
(registry, supervisor, handler)
}
#[test]
fn unminted_nonce_on_an_unreserved_id_with_an_open_swap_is_refused() {
let (registry, _supervisor, handler) = handler_with_incumbent("aft", false);
for (connection, nonce) in [(2, Some("forged")), (3, None)] {
let replies = handler
.handle_control(
ConnectionId::new(connection),
hello_frame_with_nonce("aft", PROTOCOL_VERSION, connection, nonce),
)
.unwrap();
assert_eq!(replies[0].header.ty, FrameType::Error);
assert_eq!(
parse_error(&replies[0])["code"],
"swap_token_invalid",
"nonce {nonce:?}"
);
}
assert!(registry.get_candidate("aft").unwrap().is_none());
assert_eq!(
registry.get_module("aft").unwrap().unwrap().connection_id,
ConnectionId::new(1)
);
let admitted = handler
.handle_control(
ConnectionId::new(4),
hello_frame_with_nonce("aft", PROTOCOL_VERSION, 4, Some(CANDIDATE_NONCE)),
)
.unwrap();
assert_eq!(admitted[0].header.ty, FrameType::HelloAck);
assert_eq!(
registry
.get_candidate("aft")
.unwrap()
.unwrap()
.connection_id,
ConnectionId::new(4)
);
assert_eq!(
registry.get_module("aft").unwrap().unwrap().connection_id,
ConnectionId::new(1),
"the candidate must not take the active slot"
);
let replayed = handler
.handle_control(
ConnectionId::new(5),
hello_frame_with_nonce("aft", PROTOCOL_VERSION, 5, Some(CANDIDATE_NONCE)),
)
.unwrap();
assert_eq!(parse_error(&replayed[0])["code"], "swap_token_invalid");
handler.cleanup_connection(ConnectionId::new(1)).unwrap();
let squatter = handler
.handle_control(
ConnectionId::new(6),
hello_frame_with_nonce("aft", PROTOCOL_VERSION, 6, Some("forged")),
)
.unwrap();
assert_eq!(parse_error(&squatter[0])["code"], "swap_token_invalid");
assert!(
registry.get_module("aft").unwrap().is_none(),
"a squatter took the active slot of an id being swapped"
);
}
#[test]
fn reserved_module_candidate_is_admitted_ahead_of_the_reserved_gate() {
let (registry, _supervisor, handler) = handler_with_incumbent("vault", true);
let replies = handler
.handle_control(
ConnectionId::new(2),
hello_frame_with_nonce("vault", PROTOCOL_VERSION, 2, Some(CANDIDATE_NONCE)),
)
.unwrap();
assert_eq!(
replies[0].header.ty,
FrameType::HelloAck,
"reserved candidate refused: {:?}",
serde_json::from_slice::<Value>(&replies[0].body).ok()
);
assert_eq!(
registry
.get_candidate("vault")
.unwrap()
.unwrap()
.connection_id,
ConnectionId::new(2)
);
}
#[test]
fn without_an_open_swap_the_ordinary_gates_decide() {
let (registry, supervisor, handler) = handler_with_incumbent("vault", true);
supervisor.close_swap("vault");
let candidate = handler
.handle_control(
ConnectionId::new(2),
hello_frame_with_nonce("vault", PROTOCOL_VERSION, 2, Some(CANDIDATE_NONCE)),
)
.unwrap();
assert_eq!(parse_error(&candidate[0])["code"], "reserved_module");
let duplicate = handler
.handle_control(
ConnectionId::new(3),
hello_frame_with_nonce("vault", PROTOCOL_VERSION, 3, Some(INCUMBENT_NONCE)),
)
.unwrap();
assert_eq!(parse_error(&duplicate[0])["code"], "duplicate_module_id");
assert!(registry.get_candidate("vault").unwrap().is_none());
}
}
}
#[cfg(test)]
mod concurrency_default_exposure_tests {
use super::*;
fn hello_body(role_json: &str) -> Vec<u8> {
format!(
r#"{{"protocol_ver":2,"module_id":"m","manifest":{{"module_id":"m","module_version":"1.0.0","protocol_ver":2,"trust_tier":"first_party","provides":[{role_json}],"consumes":[],"bindings":{{"storage":{{"kind":"sqlite","scope":"project","owns_schema":false}},"vault_grants":[],"identity":{{"requires":[],"optional":[]}}}}}}}}"#
)
.into_bytes()
}
fn manifest_from(body: &[u8]) -> ModuleManifest {
let value: serde_json::Value = serde_json::from_slice(body).expect("hello parses");
serde_json::from_value(value.get("manifest").expect("manifest key").clone())
.expect("manifest parses")
}
const SURFACE_TAIL: &str = r#""operations":[],"config_schema":{"type":"object"},"observability":[],"identity_scope":[]"#;
#[test]
fn absent_concurrency_on_management_surface_is_reported_as_defaulted() {
let body = hello_body(&format!(
r#"{{"role":"management_surface",{SURFACE_TAIL}}}"#
));
let manifest = manifest_from(&body);
assert_eq!(manifest_concurrency(&manifest), Concurrency::ModuleManaged);
assert!(manifest_concurrency_was_defaulted(&body, &manifest));
}
#[test]
fn declared_concurrency_is_not_reported_even_when_it_equals_the_default() {
let body = hello_body(&format!(
r#"{{"role":"management_surface",{SURFACE_TAIL},"concurrency":"module_managed"}}"#
));
let manifest = manifest_from(&body);
assert_eq!(manifest_concurrency(&manifest), Concurrency::ModuleManaged);
assert!(!manifest_concurrency_was_defaulted(&body, &manifest));
}
#[test]
fn non_management_roles_are_never_reported() {
let body = hello_body(
r#"{"role":"internal_service","service_id":"s","transport":"bulk","agent_facing":false,"operations":[]}"#,
);
let manifest = manifest_from(&body);
assert!(!manifest_concurrency_was_defaulted(&body, &manifest));
}
}