diff --git a/crates/busbar-core-admin/src/keys.rs b/crates/busbar-core-admin/src/keys.rs index af68b1f183..f90f930d2a 100644 --- a/crates/busbar-core-admin/src/keys.rs +++ b/crates/busbar-core-admin/src/keys.rs @@ -222,7 +222,7 @@ fn is_valid_label_name(name: &str) -> bool { fn json_response(status: StatusCode, body: Value) -> Response { ( status, - [(CONTENT_TYPE, busbar_kernel::proxy::APPLICATION_JSON)], + [(CONTENT_TYPE, busbar_contract::protocol::APPLICATION_JSON)], body.to_string(), ) .into_response() diff --git a/crates/busbar-core-admin/src/v1/json/envelope.rs b/crates/busbar-core-admin/src/v1/json/envelope.rs index 24ca891424..572bde56ba 100644 --- a/crates/busbar-core-admin/src/v1/json/envelope.rs +++ b/crates/busbar-core-admin/src/v1/json/envelope.rs @@ -20,7 +20,7 @@ use crate::v1::contract::AdminError; pub fn ok_json(status: StatusCode, view: &T) -> Response { ( status, - [(CONTENT_TYPE, busbar_kernel::proxy::APPLICATION_JSON)], + [(CONTENT_TYPE, busbar_contract::protocol::APPLICATION_JSON)], serde_json::to_string(view).unwrap_or_else(|_| "{}".to_string()), ) .into_response() diff --git a/crates/busbar-core-admin/src/v1/json/handlers.rs b/crates/busbar-core-admin/src/v1/json/handlers.rs index 0551dc1eb6..a5ad916d3e 100644 --- a/crates/busbar-core-admin/src/v1/json/handlers.rs +++ b/crates/busbar-core-admin/src/v1/json/handlers.rs @@ -5199,7 +5199,10 @@ pub(crate) async fn openapi( if let Some(filtered) = openapi_for_configured_planes(&openapi_json(), configured) { return ( StatusCode::OK, - [(CONTENT_TYPE, busbar_kernel::proxy::APPLICATION_JSON), VARY], + [ + (CONTENT_TYPE, busbar_contract::protocol::APPLICATION_JSON), + VARY, + ], filtered, ) .into_response(); @@ -5208,7 +5211,7 @@ pub(crate) async fn openapi( ( StatusCode::OK, [ - (CONTENT_TYPE, busbar_kernel::proxy::APPLICATION_JSON), + (CONTENT_TYPE, busbar_contract::protocol::APPLICATION_JSON), (axum::http::header::CONTENT_ENCODING, "gzip"), VARY, ], @@ -5218,7 +5221,10 @@ pub(crate) async fn openapi( } else { ( StatusCode::OK, - [(CONTENT_TYPE, busbar_kernel::proxy::APPLICATION_JSON), VARY], + [ + (CONTENT_TYPE, busbar_contract::protocol::APPLICATION_JSON), + VARY, + ], openapi_json(), ) .into_response() diff --git a/crates/busbar-core-admin/src/v1/json/mod.rs b/crates/busbar-core-admin/src/v1/json/mod.rs index 26589e8b2e..c5b3d5ab7f 100644 --- a/crates/busbar-core-admin/src/v1/json/mod.rs +++ b/crates/busbar-core-admin/src/v1/json/mod.rs @@ -444,7 +444,7 @@ fn finish_admin_reply( // body was serialized handler-side by the SAME `serde_json::to_string(&view)` call. ( StatusCode::OK, - [(CONTENT_TYPE, busbar_kernel::proxy::APPLICATION_JSON)], + [(CONTENT_TYPE, busbar_contract::protocol::APPLICATION_JSON)], body, ) .into_response() diff --git a/crates/busbar-core-admin/src/v1/json/tests/tests.rs b/crates/busbar-core-admin/src/v1/json/tests/tests.rs index f5ddf22408..3348b1bd15 100644 --- a/crates/busbar-core-admin/src/v1/json/tests/tests.rs +++ b/crates/busbar-core-admin/src/v1/json/tests/tests.rs @@ -32,7 +32,7 @@ async fn parts(resp: Response) -> (StatusCode, String, serde_json::Value) { async fn err_json_uses_stable_envelope() { let (status, ct, body) = parts(err_json(&AdminError::not_found("hook"))).await; assert_eq!(status, StatusCode::NOT_FOUND); - assert_eq!(ct, busbar_kernel::proxy::APPLICATION_JSON); + assert_eq!(ct, busbar_contract::protocol::APPLICATION_JSON); assert_eq!(body["error"]["code"], "not_found"); assert!( body["error"]["message"] @@ -57,7 +57,7 @@ async fn ok_json_serializes_view_with_given_status() { } let (status, ct, body) = parts(ok_json(StatusCode::CREATED, &View { name: "x", n: 7 })).await; assert_eq!(status, StatusCode::CREATED); - assert_eq!(ct, busbar_kernel::proxy::APPLICATION_JSON); + assert_eq!(ct, busbar_contract::protocol::APPLICATION_JSON); assert_eq!(body, json!({"name": "x", "n": 7})); } diff --git a/crates/busbar-kernel/src/config/mod.rs b/crates/busbar-kernel/src/config/mod.rs index a38d2e7f8f..c74ffd8a61 100644 --- a/crates/busbar-kernel/src/config/mod.rs +++ b/crates/busbar-kernel/src/config/mod.rs @@ -884,12 +884,10 @@ pub use busbar_kernel::config::pools::{ STRATEGY_USAGE, }; -// The FAILOVER BUDGET numeric defaults/bounds are plain scalars with no config grammar attached, so -// they live in the neutral `busbar_kernel::failover` (a plane names the per-request failover -// budget without reaching into `busbar-core`); re-exported here at their historical -// `crate::config::*` paths so every core call site (`appbuild`, `config_validate`, `test_support`, -// the pools `default_failover_timeout`/`default_max_hops` serde defaults) resolves unchanged. -pub use busbar_kernel::failover::{ +// The FAILOVER BUDGET numeric defaults/bounds live beside the `FailoverCfg` shape in `pools`; +// re-exported here at their historical `crate::config::*` paths so every core call site +// (`appbuild`, `config_validate`, `test_support`) resolves unchanged. +pub use busbar_kernel::config::pools::{ DEFAULT_FAILOVER_CAP, DEFAULT_FAILOVER_DEADLINE_SECS, MAX_FAILOVER_DEADLINE_SECS, }; @@ -925,7 +923,7 @@ fn check_failover_pool( errors: &mut Vec, section: &str, pool: &str, - def: &crate::failover::CandidatePoolCfg, + def: &pools::CandidatePoolCfg, on_this_plane: impl Fn(&str) -> bool, on_other_plane: impl Fn(&str) -> bool, this_registry: &str, @@ -2256,7 +2254,7 @@ pub fn resolve( }; let members = pool.members.iter().map(|m| m.model.clone()).collect(); let repeatable = pool.repeatable.clone(); - let candidates = crate::failover::CandidatePoolCfg { + let candidates = pools::CandidatePoolCfg { members, repeatable, }; diff --git a/crates/busbar-kernel/src/config/pools.rs b/crates/busbar-kernel/src/config/pools.rs index 6939e0e04f..e0194e28af 100644 --- a/crates/busbar-kernel/src/config/pools.rs +++ b/crates/busbar-kernel/src/config/pools.rs @@ -10,7 +10,6 @@ use serde::Deserialize; use super::hooks::ON_ERROR_WEIGHTED; -use crate::failover::{DEFAULT_FAILOVER_CAP, DEFAULT_FAILOVER_DEADLINE_SECS}; #[derive(Debug, Clone, Default)] pub struct PoolCfg { @@ -615,3 +614,98 @@ pub struct AffinityCfg { #[serde(default)] pub header_name: Option, } + +// ── The FAILOVER BUDGET numeric defaults/bounds, beside the `FailoverCfg` shape whose serde +// defaults (`default_failover_timeout`/`default_max_hops`) read them. `config` re-exports each at +// its historical `crate::config::*` path, so `appbuild`/`config_validate`/`test_support` call +// sites are untouched. +/// Default failover wall-clock budget (seconds) when a pool doesn't set `failover.timeout_secs`. +pub const DEFAULT_FAILOVER_DEADLINE_SECS: u64 = 120; +/// Upper bound (seconds) on a pool's `failover.timeout_secs`. 24h is already absurdly long for a +/// per-request failover budget — anything larger is a fat-finger typo (extra zeros). Enforced at +/// `--validate`/boot so a merely-oversized value fails CLOSED with an actionable message instead of +/// being accepted and later feeding `RequestCtx::new` a duration large enough to overflow the +/// monotonic-clock `Instant` math. +pub const MAX_FAILOVER_DEADLINE_SECS: u64 = 86_400; +/// Default maximum failover hops per request when a pool doesn't set `failover.max_hops`. +pub const DEFAULT_FAILOVER_CAP: usize = 3; + +/// ONE POOL OF INTERCHANGEABLE UPSTREAMS, as the operator writes it — the ENTIRE config vocabulary +/// this feature adds, and it is CORE's rather than a plugin's. +/// +/// ```yaml +/// pool_name: # one upstream image, deployed twice +/// members: [instance-a, instance-b] +/// repeatable: [read_only_operation] # operations safe to perform TWICE. Default: none. +/// ``` +/// +/// ## Why it is ONE type shared by every plugin +/// +/// A pool already means *"these members are interchangeable for this request; use whichever is +/// healthy"*. That is the same sentence for every resident plugin, so an operator learns the concept +/// ONCE — a member list keyed by a pool name, referenced by bare name, never crossing a plugin +/// boundary. Plugin-local copies of this struct would be several grammars for one idea and would +/// diverge the first time any one of them grew a key; there is one, in core, and each plugin's section +/// merely says which registry the bare names are resolved against. +/// +/// ## It is OPT-IN and the default is UNCHANGED +/// +/// Owner's steer: *"maybe in a config we dont allow it or maybe we dont suggest it be done."* An +/// absent section is no pools, which is exactly the behaviour of every deployment that exists today: +/// one registration, one destination, no failover, nothing to reason about. Nothing here turns on by +/// itself. +/// +/// ## `repeatable:` is a LIST OF OPERATIONS and there is no key that disables the safety rule +/// +/// The dangerous half of this feature is repeating a call that already went out, so the declaration +/// is per OPERATION and enumerated by hand. There is deliberately NO `repeatable: all` and no +/// `retry: always`: an operator who wants `send_email` repeated has to write `send_email` down next +/// to the operations they thought about, which is a different act from flipping a switch (the failover walk's `Stage`). +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)] +#[serde(deny_unknown_fields)] // a typo'd key must fail boot, not silently un-declare a safety rule. +pub struct CandidatePoolCfg { + /// The interchangeable registrations, by bare name, resolved against whichever registry the + /// owning plugin keeps its own registrations in. ORDERED: the first is the + /// PRIMARY, and its approved fingerprint is the one every other member must match. + /// + /// Naming a member is NOT what makes two upstreams interchangeable — busbar checks the pins it + /// already computed and refuses the pool at dispatch if they disagree. The operator is asserting + /// *"these names are the same deployment"*, a claim busbar can and does verify. + #[serde(default)] + pub members: Vec, + /// The operations that may be performed TWICE — reads, searches, queries. An operation not named + /// here is never repeated after a dispatch has gone out. + /// + /// EMPTY BY DEFAULT, which is the fail-safe posture: an operator who says nothing gets + /// reroute-before-first-byte (which duplicates nothing) and no retries at all. + #[serde(default)] + pub repeatable: Vec, +} + +impl CandidatePoolCfg { + /// MAY THIS OPERATION BE PERFORMED TWICE? The one reader of `repeatable:`, so the default can + /// never be got wrong by a second caller spelling the lookup differently. + // Read only by the per-call dispatch path a protocol plane drives; a plane whose relay never + // repeats a dispatch has no caller here. Unconditional allow — the neutral seam names no plane + // feature. + #[allow(dead_code)] + pub fn repeatability(&self, operation: &str) -> Repeatable { + if self.repeatable.iter().any(|o| o == operation) { + Repeatable::Yes + } else { + Repeatable::No + } + } +} + +/// MAY THIS OPERATION BE PERFORMED TWICE? A property of the operation the caller named, declared by +/// the operator who vouched for it — never inferred, and never a property of the transport. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Repeatable { + /// The DEFAULT for everything an operator has not spoken about. A second execution may have a + /// second effect, so an after-dispatch hop (the failover walk's + /// `Stage::AfterDispatch`) is refused. + No, + /// The operator declared this operation safe to perform twice (a read, a search, a query). + Yes, +} diff --git a/crates/busbar-kernel/src/failover/mod.rs b/crates/busbar-kernel/src/failover/mod.rs index fb8d94aab7..9b8e038753 100644 --- a/crates/busbar-kernel/src/failover/mod.rs +++ b/crates/busbar-kernel/src/failover/mod.rs @@ -122,79 +122,17 @@ use busbar_kernel::store::LaneRuntime; -// Phase-B B1: the candidate/stage/refusal/admitted/attempt/order/walk_with FAMILY relocated to -// `busbar-substrate`; this glob keeps `crate::failover::X` resolving for every in-core caller. The -// serde config type (`CandidatePoolCfg`) and the disposition halves a plugin's dispatch engine drives -// (`walk`, `record_outcome`, `record_success`) stay here, over `busbar_kernel::store::LaneRuntime` -// and `crate::breaker`. Glob, so the re-export is never an unused import when a plane consumer is out. - -/// ONE POOL OF INTERCHANGEABLE UPSTREAMS, as the operator writes it — the ENTIRE config vocabulary -/// this feature adds, and it is CORE's rather than a plugin's. -/// -/// ```yaml -/// pool_name: # one upstream image, deployed twice -/// members: [instance-a, instance-b] -/// repeatable: [read_only_operation] # operations safe to perform TWICE. Default: none. -/// ``` -/// -/// ## Why it is ONE type shared by every plugin -/// -/// A pool already means *"these members are interchangeable for this request; use whichever is -/// healthy"*. That is the same sentence for every resident plugin, so an operator learns the concept -/// ONCE — a member list keyed by a pool name, referenced by bare name, never crossing a plugin -/// boundary. Plugin-local copies of this struct would be several grammars for one idea and would -/// diverge the first time any one of them grew a key; there is one, in core, and each plugin's section -/// merely says which registry the bare names are resolved against. -/// -/// ## It is OPT-IN and the default is UNCHANGED -/// -/// Owner's steer: *"maybe in a config we dont allow it or maybe we dont suggest it be done."* An -/// absent section is no pools, which is exactly the behaviour of every deployment that exists today: -/// one registration, one destination, no failover, nothing to reason about. Nothing here turns on by -/// itself. -/// -/// ## `repeatable:` is a LIST OF OPERATIONS and there is no key that disables the safety rule -/// -/// The dangerous half of this feature is repeating a call that already went out, so the declaration -/// is per OPERATION and enumerated by hand. There is deliberately NO `repeatable: all` and no -/// `retry: always`: an operator who wants `send_email` repeated has to write `send_email` down next -/// to the operations they thought about, which is a different act from flipping a switch. See [`Stage`]. -#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Deserialize, serde::Serialize)] -#[serde(deny_unknown_fields)] // a typo'd key must fail boot, not silently un-declare a safety rule. -pub struct CandidatePoolCfg { - /// The interchangeable registrations, by bare name, resolved against whichever registry the - /// owning plugin keeps its own registrations in. ORDERED: the first is the - /// PRIMARY, and its approved fingerprint is the one every other member must match. - /// - /// Naming a member is NOT what makes two upstreams interchangeable — busbar checks the pins it - /// already computed and refuses the pool at dispatch if they disagree. The operator is asserting - /// *"these names are the same deployment"*, a claim busbar can and does verify. - #[serde(default)] - pub members: Vec, - /// The operations that may be performed TWICE — reads, searches, queries. An operation not named - /// here is never repeated after a dispatch has gone out. - /// - /// EMPTY BY DEFAULT, which is the fail-safe posture: an operator who says nothing gets - /// reroute-before-first-byte (which duplicates nothing) and no retries at all. - #[serde(default)] - pub repeatable: Vec, -} - -impl CandidatePoolCfg { - /// MAY THIS OPERATION BE PERFORMED TWICE? The one reader of `repeatable:`, so the default can - /// never be got wrong by a second caller spelling the lookup differently. - // Read only by the per-call dispatch path a protocol plane drives; a plane whose relay never - // repeats a dispatch has no caller here. Unconditional allow — the neutral seam names no plane - // feature. - #[allow(dead_code)] - pub fn repeatability(&self, operation: &str) -> Repeatable { - if self.repeatable.iter().any(|o| o == operation) { - Repeatable::Yes - } else { - Repeatable::No - } - } -} +// The disposition halves a plugin's dispatch engine drives (`walk`, `record_outcome`, +// `record_success`) live here, over `busbar_kernel::store::LaneRuntime` and `crate::breaker`. The +// operator's pool vocabulary — the `CandidatePoolCfg` serde shape, `Repeatable`, and the failover +// budget's defaults and bound — is configuration, not walk machinery: it lives in +// `crate::config::pools`, and is re-exported here at its historical `failover::` path only for the +// callers that still spell it so (the legacy plane crates' walks and the root's door pools). The +// re-export goes with this module. +pub use crate::config::pools::{ + CandidatePoolCfg, Repeatable, DEFAULT_FAILOVER_CAP, DEFAULT_FAILOVER_DEADLINE_SECS, + MAX_FAILOVER_DEADLINE_SECS, +}; /// THE SEAM'S SPELLING OF [`walk_with`]: the operator's `members:` order, admitted breaker-only. /// @@ -308,23 +246,6 @@ mod failover_tests; use crate::audit::vocab; use crate::store::Unavailable; -// ── The FAILOVER BUDGET numeric defaults/bounds. Plain scalars with no config grammar attached -// (the serialized `FailoverCfg` shape stays in core, schema-frozen). They live HERE in the neutral -// substrate so a plane crate names the per-request failover budget without reaching into -// `busbar-core`; core's `config` re-exports each at its historical `crate::config::*` path (so -// `appbuild`/`config_validate`/`test_support` call sites are untouched), and the `serde` default -// fns (`default_failover_timeout`/`default_max_hops`) read the re-export. -/// Default failover wall-clock budget (seconds) when a pool doesn't set `failover.timeout_secs`. -pub const DEFAULT_FAILOVER_DEADLINE_SECS: u64 = 120; -/// Upper bound (seconds) on a pool's `failover.timeout_secs`. 24h is already absurdly long for a -/// per-request failover budget — anything larger is a fat-finger typo (extra zeros). Enforced at -/// `--validate`/boot so a merely-oversized value fails CLOSED with an actionable message instead of -/// being accepted and later feeding `RequestCtx::new` a duration large enough to overflow the -/// monotonic-clock `Instant` math. -pub const MAX_FAILOVER_DEADLINE_SECS: u64 = 86_400; -/// Default maximum failover hops per request when a pool doesn't set `failover.max_hops`. -pub const DEFAULT_FAILOVER_CAP: usize = 3; - /// ONE CANDIDATE: somewhere a request can be sent, on any plane. /// /// Implementing this is the ENTIRE cost of teaching a plane to fail over. The selection order, the @@ -364,17 +285,6 @@ pub enum Stage { AfterDispatch, } -/// MAY THIS OPERATION BE PERFORMED TWICE? A property of the operation the caller named, declared by -/// the operator who vouched for it — never inferred, and never a property of the transport. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Repeatable { - /// The DEFAULT for everything an operator has not spoken about. A second execution may have a - /// second effect, so an [`Stage::AfterDispatch`] hop is refused. - No, - /// The operator declared this operation safe to perform twice (a read, a search, a query). - Yes, -} - /// WHY NO FURTHER CANDIDATE MAY BE SELECTED. /// /// Core's refusal. Every variant carries the pool and enough to name a repair, and every one maps to diff --git a/crates/busbar-kernel/src/state.rs b/crates/busbar-kernel/src/state.rs index e9f17565a6..5d71f00ce2 100644 --- a/crates/busbar-kernel/src/state.rs +++ b/crates/busbar-kernel/src/state.rs @@ -112,7 +112,7 @@ pub struct App { /// carried resolved-verbatim onto the snapshot so that plane's own dispatch route builder /// reads the SAME generation the request was admitted on. Empty ⇒ every /// member keeps its degenerate single-member cell and no reroute exists to be had. - pub tool_pools: std::collections::BTreeMap, + pub tool_pools: std::collections::BTreeMap, /// THE PER-PLANE FAILOVER POOL MAPS reached through the GENERIC pool-member seam /// ([`busbar_kernel::plane_host::LanePoolHost::plane_pool_members`]), keyed by the plane's stable /// decl key (the opaque registry key) — a registry-keyed map in place of the former plane-named @@ -124,7 +124,7 @@ pub struct App { #[allow(dead_code)] pub plane_pools: std::collections::BTreeMap< &'static str, - std::collections::BTreeMap, + std::collections::BTreeMap, >, pub auth: Arc, /// GLOBAL rewrite hooks — the `prompt: rw` gates named in `global_hooks`, resolved to their @@ -682,7 +682,7 @@ impl App { pub fn plane_pools( &self, plane_key: &str, - ) -> Option<&std::collections::BTreeMap> { + ) -> Option<&std::collections::BTreeMap> { self.plane_pools.get(plane_key) } diff --git a/xtask/src/gates/config_schema/schema.rs b/xtask/src/gates/config_schema/schema.rs index ac9c6aa5df..7ae7fd8fe4 100644 --- a/xtask/src/gates/config_schema/schema.rs +++ b/xtask/src/gates/config_schema/schema.rs @@ -324,9 +324,10 @@ pub fn sources(cx: &Ctx) -> Result, String> { "crates/busbar-plane-mcp/src/tools_config.rs".to_string(), // `mcp:` — the plane's endpoint block (`McpCfg`), in the plane crate beside `tools:`. "crates/busbar-plane-mcp/src/endpoint.rs".to_string(), - // `tool_pools:` / `agent_pools:` — one type, two sections, and `repeatable:` is the SAFETY - // declaration that decides whether an operation with effects may be performed twice. - core_file(cx, &core, "failover/mod.rs")?, + // (`tool_pools:` / `agent_pools:` — `CandidatePoolCfg`, one type for two sections whose + // `repeatable:` is the SAFETY declaration that decides whether an operation with effects may + // be performed twice — lives in the kernel's `config/pools.rs`, which the grammar directory + // above tracks, so it carries no entry of its own here.) // `streams:` — the streaming plane's grammar, including the three plane-imposed session // CEILINGS that bound what a live-voice deployment may ever hold. "crates/busbar-plane-streaming/src/config.rs".to_string(),