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, ]) }