diff --git a/crates/tinymcp/src/lib.rs b/crates/tinymcp/src/lib.rs index 5af654f..2366e69 100644 --- a/crates/tinymcp/src/lib.rs +++ b/crates/tinymcp/src/lib.rs @@ -82,7 +82,8 @@ pub use config_servers::{ pub use error::{Error, Result}; pub use registry::{ AuthDetection, AuthKind, Connections, McpRegistry, OAuthFlow, ProbeOutcome, - REMOTE_REQUEST_TIMEOUT, SecretRef, SecretVault, Store, Supervisor, SupervisorConfig, + REMOTE_REQUEST_TIMEOUT, SecretRef, SecretVault, ServerRef, Store, Supervisor, SupervisorConfig, + SupervisorEvent, TickReport, }; #[cfg(feature = "module")] pub use tinybus_module::{McpService, ModuleConfig, ServerDetail}; diff --git a/crates/tinymcp/src/registry/mod.rs b/crates/tinymcp/src/registry/mod.rs index 729e355..d2cc1a2 100644 --- a/crates/tinymcp/src/registry/mod.rs +++ b/crates/tinymcp/src/registry/mod.rs @@ -26,4 +26,4 @@ pub use ops::McpRegistry; pub use setup::{SecretRef, SecretVault}; pub use sources::{Registries, RegistrySource}; pub use store::Store; -pub use supervisor::{Supervisor, SupervisorConfig}; +pub use supervisor::{ServerRef, Supervisor, SupervisorConfig, SupervisorEvent, TickReport}; diff --git a/crates/tinymcp/src/registry/supervisor/mod.rs b/crates/tinymcp/src/registry/supervisor/mod.rs index 52bfdde..01b9033 100644 --- a/crates/tinymcp/src/registry/supervisor/mod.rs +++ b/crates/tinymcp/src/registry/supervisor/mod.rs @@ -26,10 +26,22 @@ //! unreachable MCP server as a process-level health failure would take a whole //! deployment out of rotation because one optional integration was down. It //! logs, and it keeps trying. +//! +//! # What it hands back instead +//! +//! Every [`Supervisor::tick`] returns a [`TickReport`]: one [`SupervisorEvent`] +//! per thing the cycle observed or did — a probe answered or timed out, a +//! session torn down, a reconnect that succeeded, failed, or was parked. A +//! host decides which of those its user should hear about, and where — an +//! event log, a notification for a server that stays down — without this +//! crate guessing at that policy. [`Supervisor::run`] drops the report; it is +//! for a host that drives the cycle itself. mod backoff; +mod report; mod types; +pub use report::{ServerRef, SupervisorEvent, TickReport}; pub use types::{Supervisor, SupervisorConfig}; #[cfg(test)] diff --git a/crates/tinymcp/src/registry/supervisor/report.rs b/crates/tinymcp/src/registry/supervisor/report.rs new file mode 100644 index 0000000..be12473 --- /dev/null +++ b/crates/tinymcp/src/registry/supervisor/report.rs @@ -0,0 +1,176 @@ +//! What one supervisor cycle observed, for a host to act on. +//! +//! The supervisor keeps servers connected and logs as it goes, but a log line +//! is not something a host can route: it cannot be filtered into an event log +//! a user reads, or turned into a notification when a server stays down. +//! [`Supervisor::tick`](super::Supervisor::tick) therefore hands back a +//! [`TickReport`] — one [`SupervisorEvent`] per thing it observed or did, in +//! the order it happened — and publishes nothing itself. The reasons it +//! publishes nothing have not changed: one unreachable integration is not a +//! process-level health failure, and which of these a user should hear about +//! is the host's decision to make. + +use std::time::Duration; + +use crate::registry::ProbeOutcome; +use tinymcp_bus::InstalledServer; + +/// Which install an event is about. +/// +/// The three names a host needs to route or render an event, copied out of +/// the install so the report owns its data and outlives the store read that +/// produced it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServerRef { + /// The install's identifier. + pub server_id: String, + /// The registry's qualified name, such as `@scope/server`. + pub qualified_name: String, + /// The registry's display name. + pub display_name: String, +} + +impl From<&InstalledServer> for ServerRef { + fn from(server: &InstalledServer) -> Self { + Self { + server_id: server.server_id.clone(), + qualified_name: server.qualified_name.clone(), + display_name: server.display_name.clone(), + } + } +} + +/// One thing the supervisor observed or did during a cycle. +/// +/// Non-exhaustive: a host matches with a wildcard, so a later cycle step can +/// report itself without breaking the hosts that do not care about it. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum SupervisorEvent { + /// A connected server answered its liveness probe. + /// + /// The nominal case, reported so a host can watch a server's latency drift + /// before it starts missing the window. + ProbeAnswered { + /// The server that answered. + server: ServerRef, + /// How long the round trip took. + elapsed: Duration, + }, + /// A connected server did not answer inside the probe window, and the + /// session was kept. + /// + /// Slow, not gone — see [`ProbeOutcome::TimedOut`]. The session ends only + /// once `consecutive` reaches `teardown_after`, and that cycle reports + /// [`Self::TransportDropped`] instead of this. + ProbeTimedOut { + /// The server that went quiet. + server: ServerRef, + /// The window that elapsed without an answer. + after: Duration, + /// How many probes in a row have now timed out, this one included. + consecutive: u32, + /// The streak length at which the session is torn down. + teardown_after: u32, + }, + /// A session was ended because its probe found it unusable, and a + /// reconnect follows in the same cycle. + /// + /// What that reconnect came to is reported separately, as + /// [`Self::Reconnected`], [`Self::ReconnectFailed`] or [`Self::Parked`]. + TransportDropped { + /// The server whose session ended. + server: ServerRef, + /// What the probe observed. Never [`ProbeOutcome::Alive`]. + outcome: ProbeOutcome, + /// The timeout streak that ended the session when the outcome was a + /// timeout; zero for a transport that was observed to fail. + consecutive_timeouts: u32, + }, + /// A server was connected, either freshly or after its session was ended. + Reconnected { + /// The server that connected. + server: ServerRef, + /// How many tools it advertises. + tools: usize, + /// How many consecutive attempts had failed before this one succeeded. + /// + /// Zero when the session was rebuilt in the same cycle that ended it, + /// which no user was around to notice; anything else means the server + /// had been unavailable across at least one whole cycle. + after_failures: u32, + }, + /// A connection attempt failed and will be retried after a backoff. + ReconnectFailed { + /// The server that could not be connected. + server: ServerRef, + /// What the attempt reported, already rendered. + error: String, + /// How many consecutive attempts have now failed, this one included. + failures: u32, + /// How long the supervisor waits before the next attempt. + retry_in: Duration, + }, + /// A connection attempt failed in a way retrying cannot fix, so the server + /// is parked until it is disabled and re-enabled. + /// + /// Today that is exactly [`Error::MissingRuntime`](crate::Error::MissingRuntime). + Parked { + /// The server that was parked. + server: ServerRef, + /// What the attempt reported, already rendered. + error: String, + }, +} + +impl SupervisorEvent { + /// The server this event is about. + #[must_use] + pub fn server(&self) -> &ServerRef { + match self { + Self::ProbeAnswered { server, .. } + | Self::ProbeTimedOut { server, .. } + | Self::TransportDropped { server, .. } + | Self::Reconnected { server, .. } + | Self::ReconnectFailed { server, .. } + | Self::Parked { server, .. } => server, + } + } + + /// A stable one-word label, for structured log fields. + #[must_use] + pub const fn kind(&self) -> &'static str { + match self { + Self::ProbeAnswered { .. } => "probe_answered", + Self::ProbeTimedOut { .. } => "probe_timed_out", + Self::TransportDropped { .. } => "transport_dropped", + Self::Reconnected { .. } => "reconnected", + Self::ReconnectFailed { .. } => "reconnect_failed", + Self::Parked { .. } => "parked", + } + } +} + +/// Everything one cycle observed, in the order it happened. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TickReport { + /// The events, in observation order. A server that was torn down and + /// reconnected in one cycle appears twice, the drop first. + pub events: Vec, +} + +impl TickReport { + /// Whether the cycle observed nothing at all. + /// + /// True for an empty store, and for one whose every install is disabled, + /// parked, or waiting out a backoff. A healthy connected server is *not* + /// nothing — it is a [`SupervisorEvent::ProbeAnswered`]. + #[must_use] + pub fn is_empty(&self) -> bool { + self.events.is_empty() + } + + pub(super) fn push(&mut self, event: SupervisorEvent) { + self.events.push(event); + } +} diff --git a/crates/tinymcp/src/registry/supervisor/test.rs b/crates/tinymcp/src/registry/supervisor/test.rs index 97d5595..3af5f57 100644 --- a/crates/tinymcp/src/registry/supervisor/test.rs +++ b/crates/tinymcp/src/registry/supervisor/test.rs @@ -15,8 +15,9 @@ use axum::{Json, Router}; use serde_json::{Value, json}; use super::backoff::{BACKOFF_BASE, BACKOFF_MAX, BackoffState, delay_after}; +use super::report::{ServerRef, SupervisorEvent, TickReport}; use super::types::{Supervisor, SupervisorConfig}; -use crate::registry::{Connections, OAuthFlow, Store}; +use crate::registry::{Connections, OAuthFlow, ProbeOutcome, Store}; use tinymcp_bus::{CommandKind, InstalledServer, McpClientIdentityConfig, Transport}; // --------------------------------------------------------------------------- @@ -626,6 +627,11 @@ struct ServerDials { list_delay: std::sync::atomic::AtomicU64, /// Whether `tools/list` answers with a JSON-RPC error instead of tools. list_errors: std::sync::atomic::AtomicBool, + /// Whether the *next* `tools/list` alone answers with an error. + /// + /// One failure, then back to normal: the shape of a transport that + /// hiccups once and is fine again by the time the reconnect lists tools. + fail_next_list: std::sync::atomic::AtomicBool, /// Whether `initialize` succeeds, i.e. whether a reconnect can work. initialize_ok: std::sync::atomic::AtomicBool, } @@ -635,10 +641,18 @@ impl ServerDials { std::sync::Arc::new(Self { list_delay: std::sync::atomic::AtomicU64::new(0), list_errors: std::sync::atomic::AtomicBool::new(false), + fail_next_list: std::sync::atomic::AtomicBool::new(false), initialize_ok: std::sync::atomic::AtomicBool::new(true), }) } + /// Fail the next `tools/list` only, so a probe finds a broken transport + /// but the reconnect that follows can still complete. + fn fail_next_list(&self) { + self.fail_next_list + .store(true, std::sync::atomic::Ordering::SeqCst); + } + fn set_list_delay(&self, delay: Duration) { self.list_delay.store( u64::try_from(delay.as_millis()).unwrap_or(u64::MAX), @@ -703,7 +717,10 @@ async fn serve_adjustable_server(dials: &std::sync::Arc) -> String tokio::time::sleep(Duration::from_millis(delay)).await; } - if dials.list_errors.load(std::sync::atomic::Ordering::SeqCst) { + let fail_once = dials + .fail_next_list + .swap(false, std::sync::atomic::Ordering::SeqCst); + if fail_once || dials.list_errors.load(std::sync::atomic::Ordering::SeqCst) { return Json(json!({ "jsonrpc": "2.0", "id": id, @@ -904,3 +921,367 @@ async fn a_transport_that_answers_with_an_error_is_torn_down_at_once() { ); assert_eq!(supervisor.backed_off_count(), 1); } + +// --------------------------------------------------------------------------- +// The report +// +// A tick hands back what it observed so a host can route it: a log line cannot +// be filtered into an event log or turned into a notification, and which +// observations a user should hear about is the host's call, not this crate's. +// These pin that every branch of the cycle reports itself, carrying the numbers +// a host needs to make that call — the streak, the failure count, whether a +// rebuild happened within the cycle or after the server had stayed down. +// --------------------------------------------------------------------------- + +fn kinds(report: &TickReport) -> Vec<&'static str> { + report.events.iter().map(SupervisorEvent::kind).collect() +} + +#[tokio::test] +async fn an_answered_probe_is_reported() { + let dials = ServerDials::new(); + let url = serve_adjustable_server(&dials).await; + let (store, connections, oauth, server) = connected_to(url).await; + let mut supervisor = probing_supervisor(); + + let report = supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + + assert!(!report.is_empty(), "a healthy server is an observation too"); + match report.events.as_slice() { + [ + SupervisorEvent::ProbeAnswered { + server: reported, .. + }, + ] => assert_eq!(*reported, ServerRef::from(&server)), + other => panic!("expected one probe_answered event, got {other:?}"), + } +} + +#[tokio::test] +async fn a_kept_timeout_is_reported_with_its_place_in_the_streak() { + let dials = ServerDials::new(); + let url = serve_adjustable_server(&dials).await; + let (store, connections, oauth, server) = connected_to(url).await; + dials.set_list_delay(SLOWER_THAN_THE_PROBE); + dials.refuse_reconnects(); + let mut supervisor = probing_supervisor(); + + let report = supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + + assert_eq!( + report.events, + [SupervisorEvent::ProbeTimedOut { + server: ServerRef::from(&server), + after: TEST_PROBE_TIMEOUT, + consecutive: 1, + teardown_after: 3, + }] + ); +} + +#[tokio::test] +async fn the_teardown_after_a_run_of_timeouts_reports_the_drop_and_the_refused_reconnect() { + let dials = ServerDials::new(); + let url = serve_adjustable_server(&dials).await; + let (store, connections, oauth, server) = connected_to(url).await; + dials.set_list_delay(SLOWER_THAN_THE_PROBE); + dials.refuse_reconnects(); + let mut supervisor = probing_supervisor(); + + for _ in 0..2 { + supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + } + let report = supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + + assert_eq!(kinds(&report), ["transport_dropped", "reconnect_failed"]); + match report.events.as_slice() { + [ + dropped, + SupervisorEvent::ReconnectFailed { + server: reported, + error, + failures, + retry_in, + }, + ] => { + assert_eq!( + *dropped, + SupervisorEvent::TransportDropped { + server: ServerRef::from(&server), + outcome: ProbeOutcome::TimedOut { + after: TEST_PROBE_TIMEOUT, + }, + consecutive_timeouts: 3, + } + ); + assert_eq!(*reported, ServerRef::from(&server)); + assert!(error.contains("not accepting sessions"), "{error}"); + assert_eq!(*failures, 1); + assert_eq!(*retry_in, BACKOFF_BASE); + } + other => panic!("expected a drop then a refused reconnect, got {other:?}"), + } +} + +#[tokio::test] +async fn a_broken_transport_is_reported_and_so_is_the_rebuild_that_follows() { + // The common field case: one request fails, the session is rebuilt within + // the same cycle, and no user was around to notice. The report still + // carries both halves, the drop first, with `after_failures` at zero so a + // host can tell this blip from a server that stayed down. + let dials = ServerDials::new(); + let url = serve_adjustable_server(&dials).await; + let (store, connections, oauth, server) = connected_to(url).await; + dials.fail_next_list(); + let mut supervisor = probing_supervisor(); + + let report = supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + + assert_eq!(kinds(&report), ["transport_dropped", "reconnected"]); + match report.events.as_slice() { + [ + SupervisorEvent::TransportDropped { + server: reported, + outcome: ProbeOutcome::Broken { error, .. }, + consecutive_timeouts, + }, + rebuilt, + ] => { + assert_eq!(*reported, ServerRef::from(&server)); + assert!(error.contains("the session is gone"), "{error}"); + assert_eq!(*consecutive_timeouts, 0); + assert_eq!( + *rebuilt, + SupervisorEvent::Reconnected { + server: ServerRef::from(&server), + tools: 1, + after_failures: 0, + } + ); + } + other => panic!("expected a broken drop then a rebuild, got {other:?}"), + } + assert_eq!(connections.connected_count().await, 1); +} + +#[tokio::test] +async fn a_failed_reconnect_is_reported_with_its_penalty_and_the_window_is_quiet() { + let store = Store::open_in_memory().unwrap(); + let server = install( + "srv-1", + Transport::HttpRemote { + url: "http://127.0.0.1:1/mcp".into(), + }, + true, + ); + store.insert_server(&server).unwrap(); + let connections = Connections::new(); + let oauth = OAuthFlow::new(None).unwrap(); + let mut supervisor = supervisor(); + let base = Instant::now(); + + let report = supervisor.tick(&store, &connections, &oauth, base).await; + + match report.events.as_slice() { + [ + SupervisorEvent::ReconnectFailed { + server: reported, + failures, + retry_in, + .. + }, + ] => { + assert_eq!(*reported, ServerRef::from(&server)); + assert_eq!(*failures, 1); + assert_eq!(*retry_in, BACKOFF_BASE); + } + other => panic!("expected one reconnect_failed event, got {other:?}"), + } + + // Inside the backoff window nothing is dialled, so there is nothing to + // report — a host must not read silence as recovery. + let quiet = supervisor + .tick(&store, &connections, &oauth, base + Duration::from_secs(1)) + .await; + assert!(quiet.is_empty()); +} + +#[tokio::test] +async fn a_reconnect_after_failures_reports_how_many_it_took() { + let url = serve_working_server().await; + let store = Store::open_in_memory().unwrap(); + store + .insert_server(&install( + "srv-1", + Transport::HttpRemote { + url: "http://127.0.0.1:1/mcp".into(), + }, + true, + )) + .unwrap(); + let connections = Connections::new(); + let oauth = OAuthFlow::new(None).unwrap(); + let mut supervisor = supervisor(); + let base = Instant::now(); + + supervisor.tick(&store, &connections, &oauth, base).await; + + // The server comes back, and the window has elapsed. + store.delete_server("srv-1").unwrap(); + let server = install("srv-1", Transport::HttpRemote { url }, true); + store.insert_server(&server).unwrap(); + let report = supervisor + .tick(&store, &connections, &oauth, base + Duration::from_secs(30)) + .await; + + assert_eq!( + report.events, + [SupervisorEvent::Reconnected { + server: ServerRef::from(&server), + tools: 1, + after_failures: 1, + }] + ); +} + +#[tokio::test] +async fn a_parked_server_is_reported_once_and_then_stays_quiet() { + let store = Store::open_in_memory().unwrap(); + let server = install("srv-1", Transport::Stdio, true); + store.insert_server(&server).unwrap(); + store + .set_env_values( + "srv-1", + &BTreeMap::from([( + "PATH".to_string(), + "/tinymcp/deliberately/does/not/exist".to_string(), + )]), + ) + .unwrap(); + let connections = Connections::new(); + let oauth = OAuthFlow::new(None).unwrap(); + let mut supervisor = supervisor(); + let base = Instant::now(); + + let report = supervisor.tick(&store, &connections, &oauth, base).await; + + match report.events.as_slice() { + [ + SupervisorEvent::Parked { + server: reported, + error, + }, + ] => { + assert_eq!(*reported, ServerRef::from(&server)); + assert!(!error.is_empty(), "the parking reason is carried"); + } + other => panic!("expected one parked event, got {other:?}"), + } + + let quiet = supervisor + .tick(&store, &connections, &oauth, base + BACKOFF_MAX * 2) + .await; + assert!( + quiet.is_empty(), + "a parked server is not retried, so there is nothing to report" + ); +} + +#[tokio::test] +async fn a_disabled_install_and_an_empty_store_report_nothing() { + let store = Store::open_in_memory().unwrap(); + let connections = Connections::new(); + let oauth = OAuthFlow::new(None).unwrap(); + let mut supervisor = supervisor(); + + let empty = supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + assert!(empty.is_empty()); + + store + .insert_server(&install("srv-off", Transport::Stdio, false)) + .unwrap(); + let disabled = supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + assert!(disabled.is_empty()); +} + +#[test] +fn a_server_ref_carries_the_install_identity() { + let server = install("srv-1", Transport::Stdio, true); + let reference = ServerRef::from(&server); + + assert_eq!(reference.server_id, "srv-1"); + assert_eq!(reference.qualified_name, "@test/srv-1"); + assert_eq!(reference.display_name, "srv-1"); +} + +#[test] +fn every_event_names_its_server_and_its_kind() { + let server = ServerRef { + server_id: "srv-1".into(), + qualified_name: "@test/srv-1".into(), + display_name: "srv-1".into(), + }; + let events = [ + SupervisorEvent::ProbeAnswered { + server: server.clone(), + elapsed: Duration::from_millis(5), + }, + SupervisorEvent::ProbeTimedOut { + server: server.clone(), + after: Duration::from_secs(8), + consecutive: 1, + teardown_after: 3, + }, + SupervisorEvent::TransportDropped { + server: server.clone(), + outcome: ProbeOutcome::Missing, + consecutive_timeouts: 0, + }, + SupervisorEvent::Reconnected { + server: server.clone(), + tools: 2, + after_failures: 0, + }, + SupervisorEvent::ReconnectFailed { + server: server.clone(), + error: "refused".into(), + failures: 1, + retry_in: Duration::from_secs(5), + }, + SupervisorEvent::Parked { + server: server.clone(), + error: "no runtime".into(), + }, + ]; + + let labels: Vec<_> = events.iter().map(SupervisorEvent::kind).collect(); + assert_eq!( + labels, + [ + "probe_answered", + "probe_timed_out", + "transport_dropped", + "reconnected", + "reconnect_failed", + "parked", + ] + ); + for event in &events { + assert_eq!(event.server(), &server); + } + assert!(TickReport::default().is_empty()); +} diff --git a/crates/tinymcp/src/registry/supervisor/types.rs b/crates/tinymcp/src/registry/supervisor/types.rs index c095830..e9d342c 100644 --- a/crates/tinymcp/src/registry/supervisor/types.rs +++ b/crates/tinymcp/src/registry/supervisor/types.rs @@ -4,6 +4,7 @@ use std::collections::{HashMap, HashSet}; use std::time::{Duration, Instant}; use super::backoff::BackoffState; +use super::report::{ServerRef, SupervisorEvent, TickReport}; use crate::registry::{Connections, OAuthFlow, ProbeOutcome, Store}; use tinymcp_bus::{InstalledServer, McpClientIdentityConfig, McpProxyConfig}; @@ -139,6 +140,9 @@ impl Supervisor { loop { interval.tick().await; + // The report is for a host that drives `tick` itself. This loop + // has no one to hand it to, and everything in it was logged as it + // happened. self.tick(store, connections, oauth, Instant::now()).await; } } @@ -147,18 +151,24 @@ impl Supervisor { /// /// `now` is supplied rather than read so backoff timing is deterministic /// under test. + /// + /// The report says what the cycle observed and did, install by install. + /// All of it was logged as it happened; the report exists for a host that + /// wants to put those observations somewhere a log line cannot go. pub async fn tick( &mut self, store: &Store, connections: &Connections, oauth: &OAuthFlow, now: Instant, - ) { + ) -> TickReport { + let mut report = TickReport::default(); + let servers = match store.list_servers() { Ok(servers) => servers, Err(error) => { tracing::warn!("the supervisor could not list installed servers: {error}"); - return; + return report; } }; @@ -179,10 +189,12 @@ impl Supervisor { continue; } - if connections.is_connected(&server_id).await - && self.judge_probe(connections, &server).await == AfterProbe::Keep - { - continue; + if connections.is_connected(&server_id).await { + let (verdict, event) = self.judge_probe(connections, &server).await; + report.push(event); + if verdict == AfterProbe::Keep { + continue; + } } // Checked after the liveness block, not before it: a live @@ -201,43 +213,88 @@ impl Supervisor { continue; } - match connections - .connect(store, oauth, &self.identity, self.proxy.as_ref(), &server) - .await - { - Ok(tools) => { - self.backoff.remove(&server_id); - self.timeouts.remove(&server_id); - self.terminal.remove(&server_id); - tracing::info!( - server_id = %server_id, - qualified_name = %server.qualified_name, - tools = tools.len(), - "reconnected" - ); + let event = self + .attempt_connect(store, connections, oauth, &server, now) + .await; + report.push(event); + } + + report + } + + /// Dials one install that is not connected, and records what came of it. + /// + /// Split out of [`Self::tick`] for the same reason as + /// [`Self::judge_probe`]: what an attempt's outcome *means* — a penalty, a + /// parking, a recovery — is the substance, and the loop is bookkeeping. + async fn attempt_connect( + &mut self, + store: &Store, + connections: &Connections, + oauth: &OAuthFlow, + server: &InstalledServer, + now: Instant, + ) -> SupervisorEvent { + let server_id = server.server_id.clone(); + + match connections + .connect(store, oauth, &self.identity, self.proxy.as_ref(), server) + .await + { + Ok(tools) => { + // Read before it is forgotten: how many attempts this success + // took is what tells a host whether the server had been + // unavailable across cycles or was merely rebuilt within one. + let after_failures = self + .backoff + .remove(&server_id) + .map_or(0, |state| state.failures); + self.timeouts.remove(&server_id); + self.terminal.remove(&server_id); + tracing::info!( + server_id = %server_id, + qualified_name = %server.qualified_name, + tools = tools.len(), + after_failures, + "reconnected" + ); + SupervisorEvent::Reconnected { + server: ServerRef::from(server), + tools: tools.len(), + after_failures, } - Err(error) if error.is_missing_runtime() => { - // No backoff entry: a penalty says "wait, then try again", - // and there is nothing to wait for. The server is parked - // until the user disables and re-enables it. - self.backoff.remove(&server_id); - self.terminal.insert(server_id.clone()); - tracing::warn!( - server_id = %server_id, - qualified_name = %server.qualified_name, - "connecting failed and will not be retried: {error}" - ); + } + Err(error) if error.is_missing_runtime() => { + // No backoff entry: a penalty says "wait, then try again", + // and there is nothing to wait for. The server is parked + // until the user disables and re-enables it. + self.backoff.remove(&server_id); + self.terminal.insert(server_id.clone()); + tracing::warn!( + server_id = %server_id, + qualified_name = %server.qualified_name, + "connecting failed and will not be retried: {error}" + ); + SupervisorEvent::Parked { + server: ServerRef::from(server), + error: error.to_string(), } - Err(error) => { - let state = self.backoff.entry(server_id.clone()).or_default(); - state.record_failure(now); - tracing::warn!( - server_id = %server_id, - qualified_name = %server.qualified_name, - failures = state.failures, - retry_in_seconds = state.current_delay().as_secs(), - "reconnecting failed: {error}" - ); + } + Err(error) => { + let state = self.backoff.entry(server_id.clone()).or_default(); + state.record_failure(now); + tracing::warn!( + server_id = %server_id, + qualified_name = %server.qualified_name, + failures = state.failures, + retry_in_seconds = state.current_delay().as_secs(), + "reconnecting failed: {error}" + ); + SupervisorEvent::ReconnectFailed { + server: ServerRef::from(server), + error: error.to_string(), + failures: state.failures, + retry_in: state.current_delay(), } } } @@ -246,18 +303,20 @@ impl Supervisor { /// Probes one connected server and decides what its answer means. /// /// Split out of [`Self::tick`] because the decision is the substance of - /// this type and the loop around it is bookkeeping. + /// this type and the loop around it is bookkeeping. Returns the verdict + /// and the event that records it; a `Rebuild` verdict has already ended + /// the session by the time this returns. async fn judge_probe( &mut self, connections: &Connections, server: &InstalledServer, - ) -> AfterProbe { + ) -> (AfterProbe, SupervisorEvent) { let server_id = server.server_id.clone(); let outcome = connections .probe_alive(&server_id, self.config.probe_timeout) .await; - match &outcome { + let (verdict, event) = match &outcome { ProbeOutcome::Alive { elapsed } => { tracing::trace!( server_id = %server_id, @@ -266,45 +325,19 @@ impl Supervisor { ); self.backoff.remove(&server_id); self.timeouts.remove(&server_id); - return AfterProbe::Keep; + ( + AfterProbe::Keep, + SupervisorEvent::ProbeAnswered { + server: ServerRef::from(server), + elapsed: *elapsed, + }, + ) } // Slow, not gone. Say so, count it, and leave the session // alone until a run of them says otherwise — the warning // reports what was observed rather than asserting a cause // nothing measured. - ProbeOutcome::TimedOut { after } => { - let streak = self - .timeouts - .entry(server_id.clone()) - .and_modify(|streak| *streak = streak.saturating_add(1)) - .or_insert(1); - let streak = *streak; - - if streak < CONSECUTIVE_TIMEOUTS_BEFORE_TEARDOWN { - tracing::warn!( - server_id = %server_id, - qualified_name = %server.qualified_name, - outcome = outcome.as_str(), - probe_timeout_seconds = after.as_secs(), - consecutive_timeouts = streak, - teardown_after = CONSECUTIVE_TIMEOUTS_BEFORE_TEARDOWN, - "the liveness probe did not answer in time; \ - keeping the session" - ); - return AfterProbe::Keep; - } - - tracing::warn!( - server_id = %server_id, - qualified_name = %server.qualified_name, - outcome = outcome.as_str(), - probe_timeout_seconds = after.as_secs(), - consecutive_timeouts = streak, - "the liveness probe has not answered for \ - {streak} consecutive ticks; reconnecting" - ); - self.timeouts.remove(&server_id); - } + ProbeOutcome::TimedOut { after } => self.judge_timeout(server, *after), // Observed to fail, so there is nothing to wait for: this is // the case the supervisor was built for, and it still acts // on the first sighting. @@ -318,17 +351,98 @@ impl Supervisor { reconnecting: {error}" ); self.timeouts.remove(&server_id); + ( + AfterProbe::Rebuild, + SupervisorEvent::TransportDropped { + server: ServerRef::from(server), + outcome: outcome.clone(), + consecutive_timeouts: 0, + }, + ) } // The entry went between the membership check and the probe. - // Nothing to report and nothing to tear down, but the caller still - // has to rebuild it. + // Nothing was observed to fail and there is nothing to tear down, + // but the caller still has to rebuild it, and a host still wants + // to know that it did. ProbeOutcome::Missing => { self.timeouts.remove(&server_id); + ( + AfterProbe::Rebuild, + SupervisorEvent::TransportDropped { + server: ServerRef::from(server), + outcome: ProbeOutcome::Missing, + consecutive_timeouts: 0, + }, + ) } + }; + + if verdict == AfterProbe::Rebuild { + connections.disconnect(&server_id).await; + } + + (verdict, event) + } + + /// Counts one probe timeout and decides whether the run has become a drop. + /// + /// The timeout half of [`Self::judge_probe`], on its own because it is the + /// one outcome with history: a single timeout keeps the session, and only + /// [`CONSECUTIVE_TIMEOUTS_BEFORE_TEARDOWN`] of them in a row end it. + fn judge_timeout( + &mut self, + server: &InstalledServer, + after: Duration, + ) -> (AfterProbe, SupervisorEvent) { + let server_id = server.server_id.clone(); + let outcome = ProbeOutcome::TimedOut { after }; + let streak = self + .timeouts + .entry(server_id.clone()) + .and_modify(|streak| *streak = streak.saturating_add(1)) + .or_insert(1); + let streak = *streak; + + if streak < CONSECUTIVE_TIMEOUTS_BEFORE_TEARDOWN { + tracing::warn!( + server_id = %server_id, + qualified_name = %server.qualified_name, + outcome = outcome.as_str(), + probe_timeout_seconds = after.as_secs(), + consecutive_timeouts = streak, + teardown_after = CONSECUTIVE_TIMEOUTS_BEFORE_TEARDOWN, + "the liveness probe did not answer in time; \ + keeping the session" + ); + return ( + AfterProbe::Keep, + SupervisorEvent::ProbeTimedOut { + server: ServerRef::from(server), + after, + consecutive: streak, + teardown_after: CONSECUTIVE_TIMEOUTS_BEFORE_TEARDOWN, + }, + ); } - connections.disconnect(&server_id).await; - AfterProbe::Rebuild + tracing::warn!( + server_id = %server_id, + qualified_name = %server.qualified_name, + outcome = outcome.as_str(), + probe_timeout_seconds = after.as_secs(), + consecutive_timeouts = streak, + "the liveness probe has not answered for \ + {streak} consecutive ticks; reconnecting" + ); + self.timeouts.remove(&server_id); + ( + AfterProbe::Rebuild, + SupervisorEvent::TransportDropped { + server: ServerRef::from(server), + outcome, + consecutive_timeouts: streak, + }, + ) } /// How many servers currently carry a backoff penalty.