From 8a4d21c76435e30287468340f356a9f92a7cd608 Mon Sep 17 00:00:00 2001 From: iceteaSA <171169159+iceteaSA@users.noreply.github.com> Date: Sun, 16 Aug 2026 10:27:29 +0200 Subject: [PATCH] forwarding: a read-only route census so blast radius is knowable before a restart Retain the attested principal and bind Instant on every RouteBinding so the census can distinguish reserved consumers from direct connections and report route age.\n\nDraining bindings stay visible with their draining state. supervisor.routes remains control-plane-only and is not exposed through the MCP facade or any agent-tool surface. --- Cargo.lock | 4 +- crates/subc-control/Cargo.toml | 2 +- crates/subc-control/src/lib.rs | 61 +++++++ .../client_control_response_catalog_list.json | 3 +- ...ient_control_response_server_describe.json | 3 +- ...esponse_server_describe_with_counters.json | 3 +- ...nt_control_response_supervisor_routes.json | 26 +++ crates/subc-control/tests/golden_json.rs | 26 ++- crates/subc-core/Cargo.toml | 2 +- crates/subc-core/src/bench_harness.rs | 3 +- crates/subc-core/src/bin/ck.rs | 109 +++++++++++- crates/subc-core/src/control.rs | 162 +++++++++++++++++- crates/subc-core/src/forwarding.rs | 120 ++++++++++++- crates/subc-core/tests/closure.rs | 24 +++ 14 files changed, 535 insertions(+), 13 deletions(-) create mode 100644 crates/subc-control/tests/golden/client_control_response_supervisor_routes.json diff --git a/Cargo.lock b/Cargo.lock index c1b8ef8a..def3f8a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -703,7 +703,7 @@ dependencies = [ [[package]] name = "subc-control" -version = "0.1.1" +version = "0.1.2" dependencies = [ "serde", "serde_json", @@ -712,7 +712,7 @@ dependencies = [ [[package]] name = "subc-core" -version = "0.3.1" +version = "0.3.2" dependencies = [ "cortexkit-paths", "fs4", diff --git a/crates/subc-control/Cargo.toml b/crates/subc-control/Cargo.toml index 2957f2f0..eb6cfbf7 100644 --- a/crates/subc-control/Cargo.toml +++ b/crates/subc-control/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "subc-control" -version = "0.1.1" +version = "0.1.2" edition = "2021" publish = true description = "Client-facing subc control-plane wire shapes." diff --git a/crates/subc-control/src/lib.rs b/crates/subc-control/src/lib.rs index 590c0f28..7f6d43a3 100644 --- a/crates/subc-control/src/lib.rs +++ b/crates/subc-control/src/lib.rs @@ -46,6 +46,7 @@ pub mod ops { pub const SUPERVISOR_HEALTH_PROBE: &str = "supervisor.health_probe"; pub const SUPERVISOR_HEALTH: &str = "supervisor.health"; pub const SUPERVISOR_STDERR_TAIL: &str = "supervisor.stderr_tail"; + pub const SUPERVISOR_ROUTES: &str = "supervisor.routes"; } /// Client-originated channel-0 control RPC body. @@ -135,6 +136,23 @@ pub enum ClientControlRequest { SupervisorHealthProbe { module_id: String }, #[serde(rename = "supervisor.health")] SupervisorHealth {}, + /// Enumerate the routes currently served by one supervised module, or every + /// module when omitted. + /// + /// This privileged census is control-plane-only. It is deliberately not an + /// MCP facade or agent-tool operation: callers holding the daemon control + /// connection may inspect live route ownership, while agent-facing modules + /// must not be able to address that surface at all. + /// + /// The daemon answers from its forwarding table under a read lock and never + /// consults a module. That makes the read safe during a drain, when a module + /// cannot be queried without recreating the hang/restart hazard that route + /// status reads avoid. + #[serde(rename = "supervisor.routes")] + SupervisorRoutes { + #[serde(default, skip_serializing_if = "Option::is_none")] + module_id: Option, + }, /// Retained stderr for one module. /// /// A separate op rather than a field on `supervisor.list`: the tail is @@ -224,6 +242,8 @@ pub enum ClientControlResponse { generation: u64, modules: Vec, }, + #[serde(rename = "supervisor.routes")] + SupervisorRoutes { modules: Vec }, #[serde(rename = "supervisor.stderr_tail")] SupervisorStderrTail { module_id: String, @@ -249,6 +269,36 @@ pub struct StderrTail { pub dropped_lines: u64, } +/// Live routes served by one module. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct SupervisorRouteModule { + pub module_id: String, + pub routes: Vec, +} + +/// One live consumer route in a [`SupervisorRouteModule`]. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct SupervisorRoute { + pub consumer: SupervisorRouteConsumer, + /// Milliseconds since the daemon bound this route. + pub age_ms: u64, + /// True once the endpoint began draining. Draining routes remain visible so + /// a census does not misreport an already-closing route as live. + pub draining: bool, +} + +/// The identity tier the daemon can honestly report for a route consumer. +/// +/// A caller that proved a live daemon-issued launch nonce is named `reserved`. +/// A direct key-holder has no such attestation, so it is reported as `direct` +/// with its connection counter instead of an invented module name. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum SupervisorRouteConsumer { + Reserved { module_id: String }, + Direct { connection_id: u64 }, +} + /// Whether stderr is being captured for a module, and if not, why not. /// /// A typed state rather than an empty-tail convention. "The module printed @@ -554,4 +604,15 @@ mod tests { assert_eq!(consumer_capabilities, None); assert_eq!(admission_facts, None); } + + #[test] + fn supervisor_routes_is_a_control_plane_request() { + let body = serde_json::json!({ + "op": "supervisor.routes", + "module_id": "aft" + }); + + let request: ClientControlRequest = serde_json::from_value(body.clone()).unwrap(); + assert_eq!(serde_json::to_value(request).unwrap(), body); + } } diff --git a/crates/subc-control/tests/golden/client_control_response_catalog_list.json b/crates/subc-control/tests/golden/client_control_response_catalog_list.json index 8a0e34d1..d1d82394 100644 --- a/crates/subc-control/tests/golden/client_control_response_catalog_list.json +++ b/crates/subc-control/tests/golden/client_control_response_catalog_list.json @@ -91,6 +91,7 @@ "supervisor.set_enabled", "supervisor.health_probe", "supervisor.health", - "supervisor.stderr_tail" + "supervisor.stderr_tail", + "supervisor.routes" ] } diff --git a/crates/subc-control/tests/golden/client_control_response_server_describe.json b/crates/subc-control/tests/golden/client_control_response_server_describe.json index ae4aaab2..ec04fd1c 100644 --- a/crates/subc-control/tests/golden/client_control_response_server_describe.json +++ b/crates/subc-control/tests/golden/client_control_response_server_describe.json @@ -17,6 +17,7 @@ "supervisor.set_enabled", "supervisor.health_probe", "supervisor.health", - "supervisor.stderr_tail" + "supervisor.stderr_tail", + "supervisor.routes" ] } diff --git a/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json b/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json index 22bc36b2..00fd8ca8 100644 --- a/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json +++ b/crates/subc-control/tests/golden/client_control_response_server_describe_with_counters.json @@ -28,6 +28,7 @@ "supervisor.set_enabled", "supervisor.health_probe", "supervisor.health", - "supervisor.stderr_tail" + "supervisor.stderr_tail", + "supervisor.routes" ] } diff --git a/crates/subc-control/tests/golden/client_control_response_supervisor_routes.json b/crates/subc-control/tests/golden/client_control_response_supervisor_routes.json new file mode 100644 index 00000000..dd41c0f1 --- /dev/null +++ b/crates/subc-control/tests/golden/client_control_response_supervisor_routes.json @@ -0,0 +1,26 @@ +{ + "modules": [ + { + "module_id": "target", + "routes": [ + { + "age_ms": 0, + "consumer": { + "connection_id": 102, + "kind": "direct" + }, + "draining": true + }, + { + "age_ms": 0, + "consumer": { + "kind": "reserved", + "module_id": "fed" + }, + "draining": true + } + ] + } + ], + "op": "supervisor.routes" +} diff --git a/crates/subc-control/tests/golden_json.rs b/crates/subc-control/tests/golden_json.rs index 6a6abaf6..8a265574 100644 --- a/crates/subc-control/tests/golden_json.rs +++ b/crates/subc-control/tests/golden_json.rs @@ -5,7 +5,8 @@ use serde_json::Value; use subc_control::{ CatalogEntry, ClientControlRequest, ClientControlResponse, ConsumerIdentity, PollKind, StderrCaptureState, StderrTail, StderrTailEntry, SupervisorEntry, SupervisorHealthEntry, - SupervisorHealthStatus, SupervisorRescanResult, + SupervisorHealthStatus, SupervisorRescanResult, SupervisorRoute, SupervisorRouteConsumer, + SupervisorRouteModule, }; use subc_protocol::{ manifest::{ @@ -276,6 +277,28 @@ fn client_control_responses() -> Vec<(&'static str, ClientControlResponse)> { modules: vec![supervisor_health_entry()], }, ), + ( + "client_control_response_supervisor_routes", + ClientControlResponse::SupervisorRoutes { + modules: vec![SupervisorRouteModule { + module_id: "target".to_string(), + routes: vec![ + SupervisorRoute { + consumer: SupervisorRouteConsumer::Direct { connection_id: 102 }, + age_ms: 0, + draining: true, + }, + SupervisorRoute { + consumer: SupervisorRouteConsumer::Reserved { + module_id: "fed".to_string(), + }, + age_ms: 0, + draining: true, + }, + ], + }], + }, + ), ( "client_control_response_supervisor_stderr_tail", ClientControlResponse::SupervisorStderrTail { @@ -346,6 +369,7 @@ fn thin_core_ops() -> Vec { "supervisor.health_probe".to_string(), "supervisor.health".to_string(), "supervisor.stderr_tail".to_string(), + "supervisor.routes".to_string(), ] } diff --git a/crates/subc-core/Cargo.toml b/crates/subc-core/Cargo.toml index f182283b..2fba44a9 100644 --- a/crates/subc-core/Cargo.toml +++ b/crates/subc-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "subc-core" -version = "0.3.1" +version = "0.3.2" edition = "2021" publish = false description = "subc daemon core: loopback TCP transport and opaque-byte splice router." diff --git a/crates/subc-core/src/bench_harness.rs b/crates/subc-core/src/bench_harness.rs index de692fe7..5dbe1c15 100644 --- a/crates/subc-core/src/bench_harness.rs +++ b/crates/subc-core/src/bench_harness.rs @@ -7,7 +7,7 @@ use subc_protocol::{ Bindings, Concurrency, ExecutionMode, IdentityBinding, IdentityScope, ModuleManifest, ProviderRole, StorageBinding, StorageKind, StorageScope, Tool, TrustTier, }, - Flags, FrameType, Priority, PROTOCOL_VERSION, + Flags, FrameType, Principal, Priority, PROTOCOL_VERSION, }; use tokio::sync::mpsc; @@ -177,6 +177,7 @@ pub async fn build_bench_forwarding_setup( PROTOCOL_VERSION, route_index as u64 + 1, &module_id, + Principal::Direct, tokio::time::Instant::now() + std::time::Duration::from_secs(30), ) .await diff --git a/crates/subc-core/src/bin/ck.rs b/crates/subc-core/src/bin/ck.rs index 27a02d42..5580546b 100644 --- a/crates/subc-core/src/bin/ck.rs +++ b/crates/subc-core/src/bin/ck.rs @@ -41,7 +41,7 @@ const PROD_CONNECTION_RELATIVE_PATH: &[&str] = const QUOTA_MODULE_ID: &str = "insula"; const CK_HARNESS: &str = "ck"; -const TOP_HELP_BASE: &str = "ck — CortexKit operator CLI\n\nusage:\n ck [--subc ] [--json] [] []\n\ndomains:\n module supervised modules: list, status, stderr, restart, stop, start, rescan\n health one-line health for every supervised module\n quota AI-provider quota and usage windows\n daemon daemon version, uptime, and connection info"; +const TOP_HELP_BASE: &str = "ck — CortexKit operator CLI\n\nusage:\n ck [--subc ] [--json] [] []\n\ndomains:\n module supervised modules: list, status, stderr, restart, stop, start, rescan\n routes live consumers for one module or the whole daemon\n health one-line health for every supervised module\n quota AI-provider quota and usage windows\n daemon daemon version, uptime, and connection info"; const TOP_HELP_TAIL: &str = "flags:\n --subc use a specific connection file (default: auto-discover)\n --json raw JSON output instead of tables\n\nrun 'ck ' with no verb to see that domain's commands"; @@ -108,6 +108,8 @@ fn discover_external_domains() -> Vec { const MODULE_HELP: &str = "ck module — inspect and control supervised modules\n\nusage: ck [--json] module []\n\nverbs:\n ck module list all modules with state and health\n ck module status one module in detail ck module stderr retained stderr for a module (-n to limit)\n ck module restart drain-restart a module\n ck module stop disable and stop a module (persists until start)\n ck module start enable and spawn a module\n ck module rescan re-read subc.jsonc and reconcile the module set\n ck module rescan --dry-run show what a rescan would change, without changing it"; +const ROUTES_HELP: &str = "ck routes — inspect live route consumers\n\nusage: ck [--json] routes []\n\n ck routes live consumers for every connected module\n ck routes live consumers for one connected module"; + const QUOTA_HELP: &str = "ck quota - AI-provider quota and usage windows\n\nusage: ck [--json] quota [--verbose] []\n\n ck quota connected providers and their usage windows\n ck quota --verbose all tracked providers, including unavailable ones\n ck quota claude one provider's windows and status in detail"; const HEALTH_HELP: &str = "ck health — module health\n\nusage: ck [--json] health []\n\n ck health one-line health for every supervised module (cached)\n ck health fresh health.check probe with FULL metrics — bypasses\n the supervisor cache and its size truncation"; @@ -163,6 +165,9 @@ async fn run(argv: impl IntoIterator) -> Result<(), CkError> { Command::Module(ModuleCommand::Start { module_id }) => { module_set_enabled(&mut client, &module_id, true, args.json).await } + Command::Routes { module_id } => { + supervisor_routes(&mut client, module_id.as_deref(), args.json).await + } Command::Health => health(&mut client, args.json).await, Command::HealthDetail { module_id } => { health_detail(&mut client, &module_id, args.json).await @@ -200,6 +205,9 @@ struct CkArgs { enum Command { Module(ModuleCommand), + Routes { + module_id: Option, + }, Health, HealthDetail { module_id: String, @@ -668,6 +676,76 @@ async fn module_set_enabled( print_ack_with_state(client, module_id, ack, verb, json_output).await } +async fn supervisor_routes( + client: &mut CkClient, + module_id: Option<&str>, + json_output: bool, +) -> Result<(), CkError> { + let response = client + .rpc_value(ClientControlRequest::SupervisorRoutes { + module_id: module_id.map(str::to_owned), + }) + .await?; + if json_output { + print_json(&response)?; + return Ok(()); + } + + let mut rows = Vec::new(); + for module in modules_array(&response) { + let module_id = display_field(module, "module_id"); + for route in module + .get("routes") + .and_then(Value::as_array) + .into_iter() + .flatten() + { + let consumer = match route.get("consumer").and_then(Value::as_object) { + Some(consumer) + if consumer.get("kind").and_then(Value::as_str) == Some("reserved") => + { + consumer + .get("module_id") + .and_then(Value::as_str) + .unwrap_or("reserved") + .to_string() + } + Some(consumer) => format!( + "direct ({})", + consumer + .get("connection_id") + .and_then(Value::as_u64) + .map(|id| format!("connection {id}")) + .unwrap_or_else(|| "connection unavailable".to_string()) + ), + None => "direct (connection unavailable)".to_string(), + }; + let age = route + .get("age_ms") + .and_then(Value::as_u64) + .map(|age| format_duration(Duration::from_millis(age))) + .unwrap_or_else(|| "?".to_string()); + let state = if route + .get("draining") + .and_then(Value::as_bool) + .unwrap_or(false) + { + "draining" + } else { + "live" + }; + rows.push(vec![module_id.clone(), consumer, age, state.to_string()]); + } + } + + if rows.is_empty() { + println!("(no live routes)"); + } else { + print_table(&["MODULE", "CONSUMER", "AGE", "STATE"], rows); + } + Ok(()) +} + async fn print_ack_with_state( client: &mut CkClient, module_id: &str, @@ -2672,7 +2750,10 @@ fn parse_args(argv: impl IntoIterator) -> Result bool { - matches!(domain, "module" | "health" | "daemon" | "quota" | "help") + matches!( + domain, + "module" | "routes" | "health" | "daemon" | "quota" | "help" + ) } fn parse_command(domain: &str, tail: &[OsString]) -> Result { @@ -2683,6 +2764,7 @@ fn parse_command(domain: &str, tail: &[OsString]) -> Result { let topic = tail.first().map(|t| t.to_string_lossy()); Ok(Command::Help(match topic.as_deref() { Some("module") => MODULE_HELP.into(), + Some("routes") => ROUTES_HELP.into(), Some("quota") => QUOTA_HELP.into(), Some("health") => HEALTH_HELP.into(), Some("daemon") => DAEMON_HELP.into(), @@ -2746,6 +2828,15 @@ fn parse_command(domain: &str, tail: &[OsString]) -> Result { }; Ok(Command::Module(command)) } + "routes" => match tail { + [] => Ok(Command::Routes { module_id: None }), + [module_id] if module_id != "-h" && module_id != "--help" && module_id != "help" => { + Ok(Command::Routes { + module_id: Some(module_id.to_string_lossy().into_owned()), + }) + } + _ => Ok(Command::Help(ROUTES_HELP.into())), + }, "health" => match tail.first() { None => Ok(Command::Health), Some(argument) => { @@ -3573,4 +3664,18 @@ mod tests { )); assert!(QUOTA_HELP.contains("--verbose")); } + + #[test] + fn routes_command_accepts_an_optional_module_id() { + let all = parse_command("routes", &[]).unwrap(); + assert!(matches!(all, Command::Routes { module_id: None })); + + let one = parse_command("routes", &[OsString::from("aft")]).unwrap(); + assert!(matches!( + one, + Command::Routes { + module_id: Some(module_id) + } if module_id == "aft" + )); + } } diff --git a/crates/subc-core/src/control.rs b/crates/subc-core/src/control.rs index a9f9adee..eb65ca8c 100644 --- a/crates/subc-core/src/control.rs +++ b/crates/subc-core/src/control.rs @@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize}; use subc_control::{ ops, CatalogEntry, ClientControlRequest, ClientControlResponse, ConsumerIdentity, PollKind, StderrCaptureState, StderrTail, StderrTailEntry, SupervisorEntry, SupervisorHealthEntry, - SupervisorRescanResult, + SupervisorRescanResult, SupervisorRoute, SupervisorRouteConsumer, SupervisorRouteModule, }; use subc_protocol::{ manifest::{Concurrency, ModuleManifest, ProviderRole}, @@ -64,6 +64,7 @@ const SUBC_CONTROL_OPS: &[&str] = &[ ops::SUPERVISOR_HEALTH_PROBE, ops::SUPERVISOR_HEALTH, ops::SUPERVISOR_STDERR_TAIL, + ops::SUPERVISOR_ROUTES, ]; const MODULE_TO_SUBC_CONTROL_OPS: &[&str] = &[MODULE_TO_SUBC_OP_CATALOG_UPDATE]; @@ -841,6 +842,9 @@ impl ControlHandler { 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::SupervisorStderrTail { module_id, max_lines, @@ -1210,6 +1214,7 @@ impl ControlHandler { response_version(&frame), frame.header.corr, &target_module_id, + principal.clone(), relay_deadline, ) .await @@ -1427,6 +1432,49 @@ impl ControlHandler { )?]) } + fn handle_supervisor_routes( + &self, + frame: Frame, + module_id: Option, + ) -> Result, 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, + }) + .collect(), + }) + .collect(); + let response = ClientControlResponse::SupervisorRoutes { modules }; + Ok(vec![control_response_body_frame( + &frame, + &response, + "ClientControlResponse::SupervisorRoutes", + )?]) + } + fn handle_supervisor_health(&self, frame: Frame) -> Result, RouterError> { let generation = self .registry @@ -3868,6 +3916,118 @@ mod tests { ); } + #[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", + 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").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().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); + } + /// Read the vendored fed corpus rather than hand-building a package. /// /// A hand-built object encodes what the test author believed the carrier diff --git a/crates/subc-core/src/forwarding.rs b/crates/subc-core/src/forwarding.rs index c58b3c0f..29bc0641 100644 --- a/crates/subc-core/src/forwarding.rs +++ b/crates/subc-core/src/forwarding.rs @@ -11,7 +11,8 @@ use std::{ use subc_control::ClientControlResponse; use subc_protocol::{ - manifest::Concurrency, session::ModuleControlResponse, ErrorBody, Flags, FrameType, Priority, + manifest::Concurrency, session::ModuleControlResponse, ErrorBody, Flags, FrameType, Principal, + Priority, }; use tokio::sync::{mpsc, oneshot, Semaphore}; use tokio::time::Instant; @@ -67,6 +68,8 @@ pub(crate) struct RouteBinding { pub module_negotiated_ver: u8, pub module_channel: u16, pub module_epoch: u32, + pub principal: Principal, + pub bound_at: Instant, pub flow: Arc, } @@ -134,6 +137,19 @@ impl GoodbyeTarget { } } +/// One route currently served by a module endpoint. +/// +/// `goodbye_target` is deliberately retained alongside the census projection so +/// a draining caller can address exactly the same route set that this read +/// reports, without a second forwarding-table pass. +#[derive(Debug, Clone)] +pub(crate) struct EndpointRoute { + pub goodbye_target: GoodbyeTarget, + pub principal: Principal, + pub bound_at: Instant, + pub draining: bool, +} + #[derive(Debug)] pub(crate) struct PendingRouteBindRelay { pub endpoint: ModuleEndpointId, @@ -228,6 +244,7 @@ struct PendingRouteBindRelayEntry { client_negotiated_ver: u8, client_permit: mpsc::OwnedPermit, route_open_frame: Frame, + principal: Principal, deadline: Instant, relay_enqueued: bool, sender: oneshot::Sender, @@ -400,6 +417,7 @@ impl ForwardingTable { Ok(endpoint) } + #[allow(clippy::too_many_arguments)] pub(crate) async fn begin_route_bind_relay_for( &self, client_connection_id: ConnectionId, @@ -407,6 +425,7 @@ impl ForwardingTable { client_negotiated_ver: u8, client_corr: u64, module_id: &str, + principal: Principal, deadline: Instant, ) -> Result { // Reserve egress capacity before taking the forwarding lock. The permit is @@ -425,6 +444,7 @@ impl ForwardingTable { client_negotiated_ver, client_corr, module_id, + principal, deadline, client_permit, ) @@ -450,6 +470,7 @@ impl ForwardingTable { subc_protocol::PROTOCOL_VERSION, client_corr, module_id, + Principal::Direct, Instant::now() + std::time::Duration::from_secs(60), permit, ) @@ -555,6 +576,7 @@ impl ForwardingTable { client_negotiated_ver: u8, client_corr: u64, expected_module_id: &str, + principal: Principal, deadline: Instant, client_permit: mpsc::OwnedPermit, ) -> Result { @@ -639,6 +661,7 @@ impl ForwardingTable { client_negotiated_ver, client_permit, route_open_frame, + principal, deadline, relay_enqueued: false, sender, @@ -1317,6 +1340,39 @@ impl ForwardingTable { Ok(released) } + /// Enumerate one endpoint's current routes without contacting the module. + /// + /// The read lock makes this safe while the endpoint drains: the returned + /// `draining` marker describes the same table state that owns the route, + /// rather than inferring liveness from a module that may be stopping. + #[allow(dead_code)] // Reserved for route.closing, which needs these exact goodbye targets. + pub(crate) fn endpoint_routes( + &self, + endpoint: ModuleEndpointId, + ) -> Result, ForwardingError> { + let inner = self.read_inner()?; + Ok(endpoint_routes_locked(&inner, endpoint)) + } + + /// Snapshot all live endpoint route sets under one forwarding-table read lock. + pub(crate) fn route_census( + &self, + module_id: Option<&str>, + ) -> Result)>, ForwardingError> { + let inner = self.read_inner()?; + let mut endpoints = inner + .modules_by_id + .iter() + .filter(|(id, _)| module_id.is_none_or(|requested| requested == id.as_str())) + .map(|(id, module)| (id.clone(), module.endpoint)) + .collect::>(); + endpoints.sort_by(|left, right| left.0.cmp(&right.0)); + Ok(endpoints + .into_iter() + .map(|(id, endpoint)| (id, endpoint_routes_locked(&inner, endpoint))) + .collect()) + } + /// True if this connection already owns committed or reserved CLIENT routes. /// A module registers (HELLO) before serving and never opens client routes, so /// a connection that has client routes must not also become a module endpoint @@ -1565,6 +1621,39 @@ fn next_channel(channel: u16) -> u16 { } } +fn endpoint_routes_locked( + inner: &ForwardingInner, + endpoint: ModuleEndpointId, +) -> Vec { + let draining = inner.draining_endpoints.contains(&endpoint); + let mut routes = inner + .module_to_client + .iter() + .filter(|(module_key, _)| module_key.endpoint == endpoint) + .map(|(_, route)| EndpointRoute { + goodbye_target: GoodbyeTarget { + connection_id: route.client_connection_id, + sink: route.client_sink.clone(), + negotiated_ver: route.client_negotiated_ver, + channel: route.client_channel, + epoch: route.client_epoch, + kind: GoodbyeTargetKind::Client, + }, + principal: route.principal.clone(), + bound_at: route.bound_at, + draining, + }) + .collect::>(); + routes.sort_by_key(|route| { + ( + route.goodbye_target.connection_id.get(), + route.goodbye_target.channel, + route.goodbye_target.epoch, + ) + }); + routes +} + fn release_reserved_route_locked( inner: &mut ForwardingInner, client_key: ClientRouteKey, @@ -1692,6 +1781,8 @@ fn commit_route_locked( module_negotiated_ver: module.negotiated_ver, module_channel: reservation.module_key.channel, module_epoch: reservation.module_epoch, + principal: pending.principal, + bound_at: Instant::now(), flow: Arc::new(ChannelFlow::new(window_for(&module.concurrency))), }); inner @@ -2258,6 +2349,33 @@ mod tests { .unwrap() } + #[test] + fn endpoint_routes_keep_goodbye_targets_and_mark_draining_routes() { + let (forwarding, module_connection, endpoint, client, sink, _client_rx) = + route_fixture("census"); + let pending = begin_test_route(&forwarding, client, sink, 1, "census"); + forwarding + .complete_pending_relay( + module_connection, + pending.corr, + RouteBindRelayOutcome::Accepted, + ) + .unwrap(); + + let routes = forwarding.endpoint_routes(endpoint).unwrap(); + assert_eq!(routes.len(), 1); + assert!(matches!(routes[0].principal, Principal::Direct)); + assert_eq!(routes[0].goodbye_target.connection_id, client); + assert_eq!(routes[0].goodbye_target.channel, pending.client_channel); + assert_eq!(routes[0].goodbye_target.epoch, pending.client_epoch); + assert!(!routes[0].draining); + + forwarding.begin_module_drain("census").unwrap(); + let draining_routes = forwarding.endpoint_routes(endpoint).unwrap(); + assert_eq!(draining_routes.len(), 1); + assert!(draining_routes[0].draining); + } + #[test] fn aborted_reservation_consumes_both_epochs_and_reuse_advances_them() { let (forwarding, _, endpoint, client, sink, _client_rx) = route_fixture("epoch-abort"); diff --git a/crates/subc-core/tests/closure.rs b/crates/subc-core/tests/closure.rs index e6367c94..75a0161f 100644 --- a/crates/subc-core/tests/closure.rs +++ b/crates/subc-core/tests/closure.rs @@ -82,6 +82,8 @@ async fn foreseeable_modules_close_over_existing_control_primitives() { let mut observed_frame_types = Vec::new(); assert_server_describe_uses_only_thin_core_ops(&server, &mut used_channel0_ops).await; + assert_supervisor_routes_reads_the_empty_forwarding_table(&server, &mut used_channel0_ops) + .await; let embedding_payload = embedding_payload(); modules.push( @@ -366,6 +368,27 @@ async fn assert_server_describe_uses_only_thin_core_ops( } } +async fn assert_supervisor_routes_reads_the_empty_forwarding_table( + server: &TestServer, + used_channel0_ops: &mut BTreeSet<&'static str>, +) { + let mut client = connect_authed_client(&server.connection_file_path) + .await + .unwrap(); + let response = control_round_trip( + &mut client, + 11, + ClientControlRequest::SupervisorRoutes { module_id: None }, + ops::SUPERVISOR_ROUTES, + used_channel0_ops, + ) + .await; + match response { + ClientControlResponse::SupervisorRoutes { modules } => assert!(modules.is_empty()), + other => panic!("unexpected supervisor.routes response: {other:?}"), + } +} + async fn assert_catalog_lists_every_archetype_with_only_thin_core_ops( server: &TestServer, used_channel0_ops: &mut BTreeSet<&'static str>, @@ -701,6 +724,7 @@ fn thin_core_ops() -> BTreeSet<&'static str> { ops::SUPERVISOR_HEALTH_PROBE, ops::SUPERVISOR_HEALTH, ops::SUPERVISOR_STDERR_TAIL, + ops::SUPERVISOR_ROUTES, ]) }