diff --git a/crates/tinymcp/src/lib.rs b/crates/tinymcp/src/lib.rs index f1c221d..5af654f 100644 --- a/crates/tinymcp/src/lib.rs +++ b/crates/tinymcp/src/lib.rs @@ -81,8 +81,8 @@ pub use config_servers::{ }; pub use error::{Error, Result}; pub use registry::{ - AuthDetection, AuthKind, Connections, McpRegistry, OAuthFlow, SecretRef, SecretVault, Store, - Supervisor, SupervisorConfig, + AuthDetection, AuthKind, Connections, McpRegistry, OAuthFlow, ProbeOutcome, + REMOTE_REQUEST_TIMEOUT, SecretRef, SecretVault, Store, Supervisor, SupervisorConfig, }; #[cfg(feature = "module")] pub use tinybus_module::{McpService, ModuleConfig, ServerDetail}; diff --git a/crates/tinymcp/src/registry/connections/mod.rs b/crates/tinymcp/src/registry/connections/mod.rs index a363312..c8d074d 100644 --- a/crates/tinymcp/src/registry/connections/mod.rs +++ b/crates/tinymcp/src/registry/connections/mod.rs @@ -42,7 +42,7 @@ mod types; /// flow's connection test so a test dials exactly as a real connect would. pub(crate) use dial::build_http_auth; -pub use types::Connections; +pub use types::{Connections, ProbeOutcome, REMOTE_REQUEST_TIMEOUT}; #[cfg(test)] mod test; diff --git a/crates/tinymcp/src/registry/connections/test.rs b/crates/tinymcp/src/registry/connections/test.rs index b2c2c9d..9b48477 100644 --- a/crates/tinymcp/src/registry/connections/test.rs +++ b/crates/tinymcp/src/registry/connections/test.rs @@ -16,7 +16,7 @@ use serde_json::{Value, json}; use super::dial::{build_http_auth, credential_safe_dial_url, is_internal_key}; use super::status::{ConnectFailure, classify}; -use super::types::Connections; +use super::types::{Connections, ProbeOutcome}; use crate::Error; use crate::registry::Store; use crate::registry::oauth::{OAUTH_BUNDLE_KEY, OAuthFlow}; @@ -462,15 +462,19 @@ async fn a_connected_server_answers_a_probe() { connections .probe_alive("srv-1", Duration::from_secs(5)) .await + .is_alive() ); } #[tokio::test] -async fn a_server_that_was_never_connected_fails_its_probe() { - assert!( - !Connections::new() +async fn a_server_that_was_never_connected_reports_a_missing_probe() { + // Missing rather than merely "not alive": there was no transport to judge, + // which is a different thing from one that answered badly. + assert_eq!( + Connections::new() .probe_alive("srv-1", Duration::from_secs(1)) - .await + .await, + ProbeOutcome::Missing ); } @@ -916,6 +920,7 @@ mod stdio { connections .probe_alive("srv-1", Duration::from_secs(5)) .await + .is_alive() ); } @@ -972,3 +977,39 @@ mod stdio { ); } } + +#[test] +fn every_probe_outcome_has_a_stable_label_and_only_one_is_alive() { + // The label rides on the supervisor's warning, so it is part of what an + // operator greps for; and `is_alive` is what every caller branches on. + let cases = [ + ( + ProbeOutcome::Alive { + elapsed: Duration::from_millis(12), + }, + "alive", + true, + ), + (ProbeOutcome::Missing, "missing", false), + ( + ProbeOutcome::Broken { + error: "closed".into(), + elapsed: Duration::from_millis(3), + }, + "broken", + false, + ), + ( + ProbeOutcome::TimedOut { + after: Duration::from_secs(30), + }, + "timed_out", + false, + ), + ]; + + for (outcome, label, alive) in cases { + assert_eq!(outcome.as_str(), label); + assert_eq!(outcome.is_alive(), alive, "{label}"); + } +} diff --git a/crates/tinymcp/src/registry/connections/types.rs b/crates/tinymcp/src/registry/connections/types.rs index 075d21d..88aef3b 100644 --- a/crates/tinymcp/src/registry/connections/types.rs +++ b/crates/tinymcp/src/registry/connections/types.rs @@ -25,6 +25,74 @@ use tinymcp_bus::{ /// that passes the test behaves the same once installed. const REMOTE_TIMEOUT_SECS: u64 = 30; +/// How long a request to a connected server is allowed to take. +/// +/// Public because it is the budget every other deadline in this crate is +/// reconciled against: a server answering inside this window is *usable*. A +/// liveness probe deliberately uses a shorter window +/// ([`SupervisorConfig::probe_timeout`](crate::SupervisorConfig)) so a server +/// that has gone quiet is noticed early — which is exactly why one probe +/// timeout is not a verdict, and why it takes a run of them before a session is +/// torn down. The reconciliation is that run, not an equal number. +pub const REMOTE_REQUEST_TIMEOUT: Duration = Duration::from_secs(REMOTE_TIMEOUT_SECS); + +/// What a liveness probe observed. +/// +/// The failing outcomes are kept apart because a caller has a genuinely +/// different correct response to each, which is the whole reason this is not a +/// `bool`. A transport that answered with an error is broken now and there is +/// nothing to wait for. A transport that did not answer inside the probe window +/// may simply be slower than that window. The window is at most +/// [`REMOTE_REQUEST_TIMEOUT`] and is normally configured shorter — it is an +/// early signal, not the budget a real call gets — so exceeding it does not +/// mean the server would have failed a real call. Collapsing the two lets a +/// supervisor tear down a working session and then report a drop that never +/// happened. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub enum ProbeOutcome { + /// The server answered inside the probe window. + Alive { + /// How long the round trip took. + elapsed: Duration, + }, + /// There is no entry for this server, so there was nothing to probe. + Missing, + /// The transport answered with an error. + Broken { + /// What the transport reported, already rendered. + error: String, + /// How long it took to fail. + elapsed: Duration, + }, + /// The server did not answer inside the probe window. + /// + /// Not the same as broken: nothing was observed to fail, only to be slow. + TimedOut { + /// The window that elapsed without an answer. + after: Duration, + }, +} + +impl ProbeOutcome { + /// Whether the server answered. + #[must_use] + pub const fn is_alive(&self) -> bool { + matches!(self, Self::Alive { .. }) + } + + /// A stable one-word label, for structured log fields. + #[must_use] + pub const fn as_str(&self) -> &'static str { + match self { + Self::Alive { .. } => "alive", + Self::Missing => "missing", + Self::Broken { .. } => "broken", + Self::TimedOut { .. } => "timed_out", + } + } +} + /// A live transport for one connected install. #[derive(Debug)] enum ActiveClient { @@ -314,28 +382,47 @@ impl Connections { self.live.read().await.contains_key(server_id) } - /// Whether a connected server still answers, within `timeout`. + /// What a connected server does when asked a question, within `timeout`. /// /// Issues a real round trip. This is how a dead transport becomes visible /// before a user's next tool call finds it. /// - /// A missing entry, a transport error, and a timeout all report `false`: - /// each means "not usable, reconnect", and distinguishing them would give a - /// caller a choice it has no different response to. - pub async fn probe_alive(&self, server_id: &str, timeout: Duration) -> bool { + /// Reports [`ProbeOutcome`] rather than a bool because the failures are not + /// interchangeable. A missing entry and a transport error mean the session + /// is unusable now. A timeout means only that the answer did not arrive + /// inside `timeout`, which is normally shorter than + /// [`REMOTE_REQUEST_TIMEOUT`] — so a server that would have served a real + /// call perfectly well can still exceed it. A caller that treats the two + /// alike disconnects working sessions and then reports a drop that never + /// happened. + pub async fn probe_alive(&self, server_id: &str, timeout: Duration) -> ProbeOutcome { let Some(connection) = self.get(server_id).await else { - return false; + return ProbeOutcome::Missing; }; + // Measured rather than inferred: a caller deciding whether a slow server + // is worth keeping needs the number, and a log that carries it turns a + // repeated warning into evidence instead of a restatement. + let started = tokio::time::Instant::now(); match tokio::time::timeout(timeout, connection.client.list_tools()).await { - Ok(Ok(_)) => true, + Ok(Ok(_)) => ProbeOutcome::Alive { + elapsed: started.elapsed(), + }, Ok(Err(error)) => { - tracing::debug!(server_id, "probe found a broken transport: {error}"); - false + let elapsed = started.elapsed(); + tracing::debug!( + server_id, + ?elapsed, + "probe found a broken transport: {error}" + ); + ProbeOutcome::Broken { + error: error.to_string(), + elapsed, + } } Err(_) => { tracing::debug!(server_id, ?timeout, "probe timed out"); - false + ProbeOutcome::TimedOut { after: timeout } } } } diff --git a/crates/tinymcp/src/registry/mod.rs b/crates/tinymcp/src/registry/mod.rs index 0db8f4c..729e355 100644 --- a/crates/tinymcp/src/registry/mod.rs +++ b/crates/tinymcp/src/registry/mod.rs @@ -20,7 +20,7 @@ pub mod store; pub mod supervisor; pub use boot::{BootOutcome, connect_installed_servers}; -pub use connections::Connections; +pub use connections::{Connections, ProbeOutcome, REMOTE_REQUEST_TIMEOUT}; pub use oauth::{AuthDetection, AuthKind, OAuthFlow}; pub use ops::McpRegistry; pub use setup::{SecretRef, SecretVault}; diff --git a/crates/tinymcp/src/registry/supervisor/test.rs b/crates/tinymcp/src/registry/supervisor/test.rs index 7fef507..f01d053 100644 --- a/crates/tinymcp/src/registry/supervisor/test.rs +++ b/crates/tinymcp/src/registry/supervisor/test.rs @@ -349,10 +349,19 @@ async fn a_tick_over_an_empty_store_does_nothing() { } #[test] -fn the_default_pacing_is_a_minute_with_an_eight_second_probe() { +fn the_default_probe_window_is_shorter_than_a_real_request_budget() { let config = SupervisorConfig::default(); assert_eq!(config.tick_interval, Duration::from_secs(60)); assert_eq!(config.probe_timeout, Duration::from_secs(8)); + // The probe is an early signal, not a verdict, so its window is + // deliberately tighter than the budget a real call gets. What keeps a + // merely-slow server from being torn down is the consecutive-timeout run, + // not an equal deadline — and the window also bounds the worst-case cycle, + // since `tick` probes installs in sequence. + assert!( + config.probe_timeout < crate::REMOTE_REQUEST_TIMEOUT, + "the probe window must stay tighter than the request budget" + ); } // --------------------------------------------------------------------------- @@ -536,3 +545,304 @@ async fn a_server_that_cannot_be_reconnected_earns_a_growing_backoff() { supervisor.tick(&store, &connections, &oauth, now).await; assert_eq!(connections.connected_count().await, 0); } + +// --------------------------------------------------------------------------- +// Probe outcomes +// +// The supervisor used to collapse every failed probe into one bool and then +// report it as "the transport dropped". A server that is up but answers a +// `tools/list` more slowly than the probe window was therefore disconnected by +// the supervisor itself, and the reconnect that followed was repairing damage +// the supervisor had caused. These cover the three outcomes separately. +// +// Each test makes the teardown *observable* by refusing the reconnect: with +// `initialize` failing, a session that is torn down cannot come back, so the +// connected count distinguishes "left alone" from "dropped and rebuilt" — +// which a count alone cannot do while reconnects succeed. +// --------------------------------------------------------------------------- + +/// How a [`serve_adjustable_server`] answers. +#[derive(Debug)] +struct ServerDials { + /// How long `tools/list` takes before answering. + 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 `initialize` succeeds, i.e. whether a reconnect can work. + initialize_ok: std::sync::atomic::AtomicBool, +} + +impl ServerDials { + fn new() -> std::sync::Arc { + std::sync::Arc::new(Self { + list_delay: std::sync::atomic::AtomicU64::new(0), + list_errors: std::sync::atomic::AtomicBool::new(false), + initialize_ok: std::sync::atomic::AtomicBool::new(true), + }) + } + + fn set_list_delay(&self, delay: Duration) { + self.list_delay.store( + u64::try_from(delay.as_millis()).unwrap_or(u64::MAX), + std::sync::atomic::Ordering::SeqCst, + ); + } + + fn set_list_errors(&self, errors: bool) { + self.list_errors + .store(errors, std::sync::atomic::Ordering::SeqCst); + } + + /// Stop answering `initialize`, so a torn-down session cannot be rebuilt. + fn refuse_reconnects(&self) { + self.initialize_ok + .store(false, std::sync::atomic::Ordering::SeqCst); + } +} + +/// Binds a loopback port and serves an MCP server that can be made slow, +/// broken, or unreconnectable while the test runs. +/// +/// `initialize` is separate from `tools/list` on purpose: a connect has to be +/// able to succeed before the probe behaviour under test matters. +async fn serve_adjustable_server(dials: &std::sync::Arc) -> String { + let dials = std::sync::Arc::clone(dials); + let app = Router::new().route( + "/", + post(move |Json(body): Json| { + let dials = std::sync::Arc::clone(&dials); + async move { + let method = body + .get("method") + .and_then(Value::as_str) + .unwrap_or_default(); + let id = body["id"].clone(); + + if method == "initialize" { + if !dials + .initialize_ok + .load(std::sync::atomic::Ordering::SeqCst) + { + return Json(json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32000, "message": "not accepting sessions" }, + })); + } + return Json(json!({ + "jsonrpc": "2.0", + "id": id, + "result": { + "protocolVersion": tinymcp_bus::LATEST_PROTOCOL_VERSION, + "capabilities": {}, + "serverInfo": { "name": "adjustable", "version": "1" }, + }, + })); + } + + let delay = dials.list_delay.load(std::sync::atomic::Ordering::SeqCst); + if delay > 0 { + tokio::time::sleep(Duration::from_millis(delay)).await; + } + + if dials.list_errors.load(std::sync::atomic::Ordering::SeqCst) { + return Json(json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32001, "message": "the session is gone" }, + })); + } + + Json(json!({ + "jsonrpc": "2.0", + "id": id, + "result": { "tools": [{ "name": "forecast" }] }, + })) + } + }), + ); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{addr}/") +} + +/// A probe window short enough to exceed on purpose. +const TEST_PROBE_TIMEOUT: Duration = Duration::from_millis(100); + +/// Comfortably past [`TEST_PROBE_TIMEOUT`], and still quick to schedule. +const SLOWER_THAN_THE_PROBE: Duration = Duration::from_millis(1_500); + +fn probing_supervisor() -> Supervisor { + Supervisor::new( + SupervisorConfig { + tick_interval: Duration::from_millis(10), + probe_timeout: TEST_PROBE_TIMEOUT, + }, + McpClientIdentityConfig::default(), + None, + ) +} + +/// Connects `server` and hands back everything a tick needs. +async fn connected_to(url: String) -> (Store, Connections, OAuthFlow, InstalledServer) { + let server = install("srv-1", Transport::HttpRemote { url }, true); + let store = Store::open_in_memory().unwrap(); + store.insert_server(&server).unwrap(); + + let connections = Connections::new(); + let oauth = OAuthFlow::new(None).unwrap(); + connections + .connect( + &store, + &oauth, + &McpClientIdentityConfig::default(), + None, + &server, + ) + .await + .expect("the first connect"); + + (store, connections, oauth, server) +} + +#[tokio::test] +async fn one_slow_probe_leaves_a_working_session_alone() { + // The regression this whole change is about. A single probe that runs out + // of window is evidence of slowness, not of a drop, and acting on it makes + // the supervisor the cause of the outage it goes on to report. + 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(); + supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + + assert_eq!( + connections.connected_count().await, + 1, + "a single slow probe must not end the session" + ); + assert_eq!( + supervisor.consecutive_timeouts("srv-1"), + 1, + "the timeout should be counted rather than acted on" + ); + assert_eq!( + supervisor.backed_off_count(), + 0, + "nothing was reconnected, so nothing should carry a reconnect penalty" + ); +} + +#[tokio::test] +async fn a_run_of_slow_probes_does_eventually_end_the_session() { + // The other half of the trade-off: a server that never answers has to be + // recovered, or "do not act on one timeout" becomes "never act at all". + 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 tick in 1..=2 { + supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + assert_eq!( + connections.connected_count().await, + 1, + "the session should survive timeout {tick} of 3" + ); + } + + supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + + assert_eq!( + connections.connected_count().await, + 0, + "the third consecutive timeout should end the session" + ); + assert_eq!( + supervisor.consecutive_timeouts("srv-1"), + 0, + "the streak is spent once it has been acted on" + ); + assert_eq!( + supervisor.backed_off_count(), + 1, + "the refused reconnect should earn a backoff penalty" + ); +} + +#[tokio::test] +async fn an_answered_probe_clears_the_timeout_streak() { + // Otherwise timeouts accumulate across hours of healthy operation and a + // server is eventually torn down for three slow answers that were nowhere + // near each other. + 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(); + + dials.set_list_delay(SLOWER_THAN_THE_PROBE); + supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + assert_eq!(supervisor.consecutive_timeouts("srv-1"), 1); + + dials.set_list_delay(Duration::ZERO); + supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + assert_eq!( + supervisor.consecutive_timeouts("srv-1"), + 0, + "one answer should reset the run" + ); + assert_eq!(connections.connected_count().await, 1); +} + +#[tokio::test] +async fn a_transport_that_answers_with_an_error_is_torn_down_at_once() { + // No regression to the case the supervisor was built for: a transport that + // was *observed* to fail has nothing left to wait for, so it is not put + // behind the consecutive-timeout threshold. + let dials = ServerDials::new(); + let url = serve_adjustable_server(&dials).await; + let (store, connections, oauth, _server) = connected_to(url).await; + + dials.set_list_errors(true); + dials.refuse_reconnects(); + + let mut supervisor = probing_supervisor(); + supervisor + .tick(&store, &connections, &oauth, Instant::now()) + .await; + + assert_eq!( + connections.connected_count().await, + 0, + "a broken transport should be dropped on the first sighting" + ); + assert_eq!( + supervisor.consecutive_timeouts("srv-1"), + 0, + "a transport error is not a timeout and must not fill the streak" + ); + assert_eq!(supervisor.backed_off_count(), 1); +} diff --git a/crates/tinymcp/src/registry/supervisor/types.rs b/crates/tinymcp/src/registry/supervisor/types.rs index aaf28b1..4eafbf7 100644 --- a/crates/tinymcp/src/registry/supervisor/types.rs +++ b/crates/tinymcp/src/registry/supervisor/types.rs @@ -4,18 +4,56 @@ use std::collections::HashMap; use std::time::{Duration, Instant}; use super::backoff::BackoffState; -use crate::registry::{Connections, OAuthFlow, Store}; -use tinymcp_bus::{McpClientIdentityConfig, McpProxyConfig}; +use crate::registry::{Connections, OAuthFlow, ProbeOutcome, Store}; +use tinymcp_bus::{InstalledServer, McpClientIdentityConfig, McpProxyConfig}; + +/// How many consecutive probe timeouts end the session. +/// +/// One timeout is not evidence of a drop — see [`ProbeOutcome::TimedOut`]. It +/// takes a run of them, spread across [`SupervisorConfig::tick_interval`], for +/// "slow" to become "gone". Tearing down on the first would make the supervisor +/// the cause of the outage it then reports. +const CONSECUTIVE_TIMEOUTS_BEFORE_TEARDOWN: u32 = 3; + +/// What [`Supervisor::judge_probe`] concluded about a connected server. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum AfterProbe { + /// The session is usable; leave it alone. + Keep, + /// The session is finished; it has been ended and needs connecting again. + Rebuild, +} /// How the supervisor is paced. #[derive(Debug, Clone)] pub struct SupervisorConfig { /// How often to walk the installed servers. pub tick_interval: Duration, - /// How long a liveness probe may take before the server counts as dropped. + /// How long a liveness probe may take before it is recorded as a timeout. + /// + /// Deliberately shorter than [`REMOTE_REQUEST_TIMEOUT`](crate::REMOTE_REQUEST_TIMEOUT), + /// the budget a real + /// call gets: this is an early signal that a server has gone quiet, not a + /// verdict on whether it is usable. Exceeding it therefore means *slow*, + /// not *dead*, which is why a single timeout costs nothing and it takes a + /// run of consecutive ones to tear a session down. That run is the + /// reconciliation between the two deadlines — by the time one is acted on, + /// the server has had far longer than a real request would ever get. + /// + /// # Why not simply widen it to the transport budget + /// + /// Because [`Self::tick`](super::Supervisor::tick) probes installs in + /// sequence and each probe can consume the whole window, so the window + /// bounds the worst-case cycle. Widening it to 30s multiplies that by + /// ~3.75 and, past a handful of unresponsive installs, a cycle outruns + /// `tick_interval`. [`Supervisor::run`](super::Supervisor::run) sets + /// `MissedTickBehavior::Delay` so that does not become a burst of + /// catch-up ticks — but a host that drives `tick` from its own timer gets + /// no such protection, and at least one does. Raising this default is + /// therefore not a local decision; it needs the probe loop bounded first + /// (concurrent probes, or a cycle deadline). /// - /// A tool listing is a cheap round trip; a server that cannot answer one - /// within this window is not usable for a real call either. + /// A host that knows its servers are slow can still raise it explicitly. pub probe_timeout: Duration, } @@ -38,6 +76,12 @@ pub struct Supervisor { identity: McpClientIdentityConfig, proxy: Option, backoff: HashMap, + /// Consecutive probe timeouts per server. + /// + /// Held here rather than in [`Connections`] on purpose: `disconnect` clears + /// that map, so a counter kept there would erase the very history it exists + /// to accumulate. + timeouts: HashMap, } impl Supervisor { @@ -53,6 +97,7 @@ impl Supervisor { identity, proxy, backoff: HashMap::new(), + timeouts: HashMap::new(), } } @@ -64,6 +109,12 @@ impl Supervisor { pub async fn run(mut self, store: &Store, connections: &Connections, oauth: &OAuthFlow) { let start = tokio::time::Instant::now() + self.config.tick_interval; let mut interval = tokio::time::interval_at(start, self.config.tick_interval); + // A cycle walks every install in turn and each probe can take the whole + // probe window, so a tick can outlast its own interval. The default + // behaviour would then fire the missed ticks back to back, re-probing + // servers that were just probed. Delay instead: pace from when the + // cycle finished. + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); tracing::info!( tick_seconds = self.config.tick_interval.as_secs(), @@ -102,26 +153,17 @@ impl Supervisor { if !server.enabled { // The disable path owns tearing the connection down. All that // is left here is to forget any backoff, so re-enabling gets an - // immediate attempt rather than inheriting an old penalty. + // immediate attempt rather than inheriting an old penalty. The + // timeout streak goes with it for the same reason. self.backoff.remove(&server_id); + self.timeouts.remove(&server_id); continue; } - if connections.is_connected(&server_id).await { - if connections - .probe_alive(&server_id, self.config.probe_timeout) - .await - { - self.backoff.remove(&server_id); - continue; - } - - tracing::warn!( - server_id = %server_id, - qualified_name = %server.qualified_name, - "the transport dropped; reconnecting" - ); - connections.disconnect(&server_id).await; + if connections.is_connected(&server_id).await + && self.judge_probe(connections, &server).await == AfterProbe::Keep + { + continue; } if !self @@ -139,6 +181,7 @@ impl Supervisor { { Ok(tools) => { self.backoff.remove(&server_id); + self.timeouts.remove(&server_id); tracing::info!( server_id = %server_id, qualified_name = %server.qualified_name, @@ -161,9 +204,107 @@ 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. + async fn judge_probe( + &mut self, + connections: &Connections, + server: &InstalledServer, + ) -> AfterProbe { + let server_id = server.server_id.clone(); + let outcome = connections + .probe_alive(&server_id, self.config.probe_timeout) + .await; + + match &outcome { + ProbeOutcome::Alive { elapsed } => { + tracing::trace!( + server_id = %server_id, + ?elapsed, + "the liveness probe answered" + ); + self.backoff.remove(&server_id); + self.timeouts.remove(&server_id); + return AfterProbe::Keep; + } + // 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); + } + // 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. + ProbeOutcome::Broken { error, elapsed } => { + tracing::warn!( + server_id = %server_id, + qualified_name = %server.qualified_name, + outcome = outcome.as_str(), + ?elapsed, + "the transport failed its liveness probe; \ + reconnecting: {error}" + ); + self.timeouts.remove(&server_id); + } + // 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. + ProbeOutcome::Missing => { + self.timeouts.remove(&server_id); + } + } + + connections.disconnect(&server_id).await; + AfterProbe::Rebuild + } + /// How many servers currently carry a backoff penalty. #[must_use] pub fn backed_off_count(&self) -> usize { self.backoff.len() } + + /// How many consecutive probe timeouts `server_id` has accumulated. + /// + /// Zero for a server that answered its last probe, that was never probed, + /// or that has just been torn down — the streak is what the next teardown + /// decision is made on, so it is worth being able to read. + #[must_use] + pub fn consecutive_timeouts(&self, server_id: &str) -> u32 { + self.timeouts.get(server_id).copied().unwrap_or(0) + } }