use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::Mutex;
use tokio::task::AbortHandle;
use super::pull_support::{PullProbe, PullSupport};
use super::respawn::RespawnBackoff;
use super::restart::RestartGeneration;
use super::save_support::SaveSupport;
use crate::config::ServerId;
use crate::error::ServerSpawnFailure;
use crate::lsp::{ConnectionId, LspClient, LspServer};
#[derive(Debug)]
#[cfg_attr(
test,
allow(
clippy::large_enum_variant,
reason = "the test-only ClientOnly variant holds just a client"
)
)]
pub(super) enum Backend {
Process(LspServer),
#[cfg(test)]
ClientOnly(LspClient),
#[cfg(test)]
Split {
client: LspClient,
server: LspServer,
},
}
impl Backend {
pub(super) const fn client(&self) -> &LspClient {
match self {
Self::Process(server) => server.client(),
#[cfg(test)]
Self::ClientOnly(client) | Self::Split { client, .. } => client,
}
}
#[allow(
clippy::unnecessary_wraps,
reason = "the test-only ClientOnly variant has no server"
)]
pub(super) const fn server(&self) -> Option<&LspServer> {
match self {
Self::Process(server) => Some(server),
#[cfg(test)]
Self::ClientOnly(_) => None,
#[cfg(test)]
Self::Split { server, .. } => Some(server),
}
}
#[allow(
clippy::unnecessary_wraps,
reason = "the test-only ClientOnly variant has no server"
)]
pub(super) const fn server_mut(&mut self) -> Option<&mut LspServer> {
match self {
Self::Process(server) => Some(server),
#[cfg(test)]
Self::ClientOnly(_) => None,
#[cfg(test)]
Self::Split { server, .. } => Some(server),
}
}
}
#[derive(Debug)]
#[allow(
clippy::large_enum_variant,
reason = "one value per configured server; boxing the backend buys nothing"
)]
pub(super) enum ServerStatus {
Expected {
prior_failure: Option<ServerSpawnFailure>,
},
Running(Backend),
Restarting,
Failed(ServerSpawnFailure),
Stopped(LspClient),
}
#[derive(Debug, Default)]
pub(super) struct RestartState {
pub(super) generation: RestartGeneration,
pub(super) last_attempt: Option<Instant>,
}
#[derive(Debug)]
pub(super) struct ServerSlot {
status: ServerStatus,
pub(super) respawn_lock: Arc<Mutex<()>>,
pub(super) backoff: Option<RespawnBackoff>,
pub(super) restart: RestartState,
pub(super) notification_task: Option<AbortHandle>,
pull_probe: Option<(ConnectionId, PullProbe)>,
}
impl ServerStatus {
const fn awaits_registration(&self) -> bool {
matches!(self, Self::Expected { .. } | Self::Restarting)
}
}
impl ServerSlot {
fn new(status: ServerStatus) -> Self {
Self {
status,
respawn_lock: Arc::new(Mutex::new(())),
backoff: None,
restart: RestartState::default(),
notification_task: None,
pull_probe: None,
}
}
}
#[derive(Debug, Default)]
pub(super) struct Servers(HashMap<ServerId, ServerSlot>);
impl Servers {
pub(super) fn get(&self, id: &ServerId) -> Option<&ServerSlot> {
self.0.get(id)
}
pub(super) fn get_mut(&mut self, id: &ServerId) -> Option<&mut ServerSlot> {
self.0.get_mut(id)
}
pub(super) fn pull_probe(&self, id: &ServerId) -> PullProbe {
let Some(slot) = self.0.get(id) else {
return PullProbe::default();
};
match (slot.pull_probe, self.client_ref(id)) {
(Some((conn, probe)), Some(client)) if conn == client.connection_id() => probe,
_ => PullProbe::default(),
}
}
pub(super) fn update_pull_probe(
&mut self,
id: &ServerId,
conn: ConnectionId,
next: impl FnOnce(PullProbe) -> PullProbe,
) -> Option<PullProbe> {
if self.client_ref(id)?.connection_id() != conn {
return None;
}
let current = self.pull_probe(id);
let probe = next(current);
self.0.get_mut(id)?.pull_probe = Some((conn, probe));
Some(probe)
}
pub(super) fn ids(&self) -> impl Iterator<Item = &ServerId> {
self.0.keys()
}
pub(super) fn client(&self, id: &ServerId) -> Option<LspClient> {
self.client_ref(id).cloned()
}
pub(super) fn client_ref(&self, id: &ServerId) -> Option<&LspClient> {
match &self.0.get(id)?.status {
ServerStatus::Running(backend) => Some(backend.client()),
ServerStatus::Stopped(client) => Some(client),
ServerStatus::Expected { .. } | ServerStatus::Restarting | ServerStatus::Failed(_) => {
None
}
}
}
pub(super) fn clients(&self) -> impl Iterator<Item = &LspClient> {
self.0.values().filter_map(|slot| match &slot.status {
ServerStatus::Running(backend) => Some(backend.client()),
ServerStatus::Stopped(client) => Some(client),
ServerStatus::Expected { .. } | ServerStatus::Restarting | ServerStatus::Failed(_) => {
None
}
})
}
pub(super) fn pull_support(&self, id: &ServerId) -> PullSupport {
let advertised = self
.server(id)
.is_some_and(|server| server.capabilities().diagnostic_provider.is_some());
PullSupport::of(advertised, self.pull_probe(id))
}
pub(super) fn save_support(&self, id: &ServerId) -> SaveSupport {
SaveSupport::of(
self.server(id)
.and_then(|server| server.capabilities().text_document_sync.as_ref()),
)
}
pub(super) fn server(&self, id: &ServerId) -> Option<&LspServer> {
match &self.0.get(id)?.status {
ServerStatus::Running(backend) => backend.server(),
_ => None,
}
}
pub(super) fn server_mut(&mut self, id: &ServerId) -> Option<&mut LspServer> {
match &mut self.0.get_mut(id)?.status {
ServerStatus::Running(backend) => backend.server_mut(),
_ => None,
}
}
pub(super) fn running_servers(&self) -> impl Iterator<Item = (&ServerId, &LspServer)> {
self.0.iter().filter_map(|(id, slot)| match &slot.status {
ServerStatus::Running(backend) => backend.server().map(|server| (id, server)),
_ => None,
})
}
pub(super) fn is_expected(&self, id: &ServerId) -> bool {
self.0
.get(id)
.is_some_and(|slot| slot.status.awaits_registration())
}
pub(super) fn any_expected(&self) -> bool {
self.0
.values()
.any(|slot| slot.status.awaits_registration())
}
pub(super) fn expected(&self) -> HashSet<ServerId> {
self.0
.iter()
.filter(|(_, slot)| slot.status.awaits_registration())
.map(|(id, _)| id.clone())
.collect()
}
pub(super) fn failure(&self, id: &ServerId) -> Option<&ServerSpawnFailure> {
match &self.0.get(id)?.status {
ServerStatus::Failed(failure)
| ServerStatus::Expected {
prior_failure: Some(failure),
} => Some(failure),
_ => None,
}
}
pub(super) fn failures(&self) -> Vec<ServerSpawnFailure> {
let mut failures: Vec<ServerSpawnFailure> = self
.0
.keys()
.filter_map(|id| self.failure(id).cloned())
.collect();
failures.sort_by(|a, b| a.server_id.as_str().cmp(b.server_id.as_str()));
failures
}
pub(super) fn set_expected(&mut self, set: &HashSet<ServerId>) {
for id in set {
match self.0.get_mut(id) {
None => {
self.0.insert(
id.clone(),
ServerSlot::new(ServerStatus::Expected {
prior_failure: None,
}),
);
}
Some(slot) => {
if let ServerStatus::Failed(failure) = &slot.status {
slot.status = ServerStatus::Expected {
prior_failure: Some(failure.clone()),
};
}
}
}
}
self.clear_expected_except(set);
}
pub(super) fn clear_expected(&mut self) {
self.clear_expected_except(&HashSet::new());
}
fn clear_expected_except(&mut self, keep: &HashSet<ServerId>) {
self.0.retain(|id, slot| {
if keep.contains(id) {
return true;
}
match &mut slot.status {
ServerStatus::Expected {
prior_failure: Some(failure),
} => {
slot.status = ServerStatus::Failed(failure.clone());
true
}
ServerStatus::Expected {
prior_failure: None,
} => false,
_ => true,
}
});
}
pub(super) fn announce_failure(&mut self, failure: &ServerSpawnFailure) {
if let Some(slot) = self.0.get_mut(&failure.server_id)
&& matches!(slot.status, ServerStatus::Expected { .. })
{
slot.status = ServerStatus::Expected {
prior_failure: Some(failure.clone()),
};
}
}
pub(super) fn record_failure(&mut self, failure: &ServerSpawnFailure) -> bool {
match self.0.get_mut(&failure.server_id) {
None => {
self.0.insert(
failure.server_id.clone(),
ServerSlot::new(ServerStatus::Failed(failure.clone())),
);
false
}
Some(slot) => match &slot.status {
ServerStatus::Expected { .. } => {
slot.status = ServerStatus::Failed(failure.clone());
true
}
ServerStatus::Failed(_) => {
slot.status = ServerStatus::Failed(failure.clone());
false
}
ServerStatus::Running(_) | ServerStatus::Restarting | ServerStatus::Stopped(_) => {
tracing::debug!(
id = %failure.server_id,
"startup failure ignored: the server is registered"
);
false
}
},
}
}
pub(super) fn fail_unsettled(&mut self, configs: impl IntoIterator<Item = ServerSpawnFailure>) {
for failure in configs {
let settled = matches!(
self.0.get(&failure.server_id).map(|slot| &slot.status),
Some(
ServerStatus::Running(_)
| ServerStatus::Restarting
| ServerStatus::Stopped(_)
| ServerStatus::Failed(_)
| ServerStatus::Expected {
prior_failure: Some(_)
}
)
);
if !settled {
self.record_failure(&failure);
}
}
}
pub(super) fn register(&mut self, id: ServerId, backend: Backend) -> Registered {
if let Some(slot) = self.0.get_mut(&id) {
let was_expected = matches!(slot.status, ServerStatus::Expected { .. });
let displaced =
match std::mem::replace(&mut slot.status, ServerStatus::Running(backend)) {
ServerStatus::Running(previous) => Some(previous),
_ => None,
};
return Registered {
was_expected,
displaced,
};
}
self.0
.insert(id, ServerSlot::new(ServerStatus::Running(backend)));
Registered {
was_expected: false,
displaced: None,
}
}
pub(super) fn take_for_restart(&mut self, id: &ServerId) -> Option<Backend> {
let slot = self.0.get_mut(id)?;
if !matches!(slot.status, ServerStatus::Running(_)) {
return None;
}
match std::mem::replace(&mut slot.status, ServerStatus::Restarting) {
ServerStatus::Running(backend) => Some(backend),
other => {
slot.status = other;
None
}
}
}
pub(super) fn restore(&mut self, id: &ServerId, backend: Backend) -> Option<Backend> {
match self.0.get_mut(id) {
Some(slot) if matches!(slot.status, ServerStatus::Restarting) => {
slot.status = ServerStatus::Running(backend);
None
}
None => {
self.0
.insert(id.clone(), ServerSlot::new(ServerStatus::Running(backend)));
None
}
Some(_) => Some(backend),
}
}
pub(super) fn remove_server(&mut self, id: &ServerId) -> Option<LspServer> {
let slot = self.0.get_mut(id)?;
if !matches!(slot.status, ServerStatus::Running(_)) {
return None;
}
match std::mem::replace(
&mut slot.status,
ServerStatus::Expected {
prior_failure: None,
},
) {
ServerStatus::Running(Backend::Process(server)) => {
slot.status = ServerStatus::Stopped(server.client().clone());
Some(server)
}
#[cfg(test)]
ServerStatus::Running(Backend::Split { client, server }) => {
slot.status = ServerStatus::Stopped(client);
Some(server)
}
#[cfg(test)]
ServerStatus::Running(backend @ Backend::ClientOnly(_)) => {
slot.status = ServerStatus::Running(backend);
None
}
other => {
slot.status = other;
None
}
}
}
pub(super) fn drain_servers(&mut self) -> Vec<(ServerId, LspServer)> {
let ids: Vec<ServerId> = self.0.keys().cloned().collect();
let mut drained = Vec::new();
for id in ids {
if let Some(server) = self.remove_server(&id) {
drained.push((id, server));
}
}
drained
}
pub(super) fn settlements(&self) -> HashMap<ServerId, crate::config::ServerSettlement> {
use crate::config::ServerSettlement;
self.0
.iter()
.map(|(id, slot)| {
let settlement = match &slot.status {
ServerStatus::Running(_) | ServerStatus::Stopped(_) => {
ServerSettlement::Registered
}
ServerStatus::Failed(_)
| ServerStatus::Expected {
prior_failure: Some(_),
} => ServerSettlement::Failed,
ServerStatus::Expected {
prior_failure: None,
}
| ServerStatus::Restarting => ServerSettlement::Pending,
};
(id.clone(), settlement)
})
.collect()
}
pub(super) fn respawn_lock(&self, id: &ServerId) -> Arc<Mutex<()>> {
self.0.get(id).map_or_else(
|| Arc::new(Mutex::new(())),
|slot| Arc::clone(&slot.respawn_lock),
)
}
pub(super) fn slot_for_swap(
&mut self,
id: &ServerId,
backend: Backend,
) -> Result<Option<Backend>, Box<Backend>> {
if let Some(slot) = self.0.get_mut(id) {
if matches!(slot.status, ServerStatus::Stopped(_)) {
return Err(Box::new(backend));
}
return Ok(
match std::mem::replace(&mut slot.status, ServerStatus::Running(backend)) {
ServerStatus::Running(old) => Some(old),
_ => None,
},
);
}
self.0
.insert(id.clone(), ServerSlot::new(ServerStatus::Running(backend)));
Ok(None)
}
#[cfg(test)]
pub(super) fn register_test_client(&mut self, id: ServerId, client: LspClient) {
let backend = match self.0.get_mut(&id).map(|slot| {
std::mem::replace(
&mut slot.status,
ServerStatus::Expected {
prior_failure: None,
},
)
}) {
Some(ServerStatus::Running(
Backend::Process(server) | Backend::Split { server, .. },
)) => Backend::Split { client, server },
_ => Backend::ClientOnly(client),
};
self.register(id, backend);
}
#[cfg(test)]
pub(super) fn register_test_server(&mut self, id: ServerId, server: LspServer) {
let backend = match self.0.get_mut(&id).map(|slot| {
std::mem::replace(
&mut slot.status,
ServerStatus::Expected {
prior_failure: None,
},
)
}) {
Some(ServerStatus::Running(
Backend::ClientOnly(client) | Backend::Split { client, .. },
)) => Backend::Split { client, server },
_ => Backend::Process(server),
};
self.register(id, backend);
}
}
#[derive(Debug)]
pub(super) struct Registered {
pub(super) was_expected: bool,
pub(super) displaced: Option<Backend>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(super) enum Phase {
#[default]
Idle,
Settling,
Settled,
InitPanicked,
ShuttingDown,
}
impl Phase {
pub(super) const fn begin_startup(&mut self) {
if matches!(self, Self::Idle | Self::Settled) {
*self = Self::Settling;
}
}
pub(super) fn finish_startup(&mut self) {
if *self == Self::Settling {
*self = Self::Settled;
}
}
pub(super) fn init_panicked(&mut self) {
if *self != Self::ShuttingDown {
*self = Self::InitPanicked;
}
}
pub(super) const fn begin_shutdown(&mut self) {
*self = Self::ShuttingDown;
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::error::StartupFailure;
#[test]
fn test_phase_shutdown_is_terminal_and_init_panic_survives_settling() {
let mut phase = Phase::default();
phase.begin_startup();
phase.init_panicked();
phase.finish_startup();
assert_eq!(phase, Phase::InitPanicked);
phase.begin_shutdown();
phase.init_panicked();
phase.begin_startup();
assert_eq!(phase, Phase::ShuttingDown);
}
#[test]
fn test_phase_startup_guard_drop_only_leaves_settling() {
let mut phase = Phase::Idle;
phase.finish_startup();
assert_eq!(phase, Phase::Idle);
phase.begin_startup();
phase.finish_startup();
assert_eq!(phase, Phase::Settled);
}
fn failure(id: &str) -> ServerSpawnFailure {
ServerSpawnFailure {
server_id: ServerId::new(id).unwrap(),
language_id: crate::config::LanguageId::from_static("rust"),
command: crate::config::ServerCommand::from_static("x"),
reason: StartupFailure::InitTaskPanicked,
}
}
#[test]
fn test_failed_then_expected_then_cleared_keeps_the_failure() {
let mut servers = Servers::default();
let id = ServerId::from_static("rust");
servers.record_failure(&failure("rust"));
servers.set_expected(&HashSet::from([id.clone()]));
assert!(servers.is_expected(&id));
assert!(servers.failure(&id).is_some());
servers.clear_expected();
assert!(!servers.is_expected(&id));
assert!(servers.failure(&id).is_some());
}
#[test]
fn test_failure_of_an_expected_server_settles_it_as_failed() {
let mut servers = Servers::default();
let id = ServerId::from_static("rust");
servers.set_expected(&HashSet::from([id.clone()]));
assert!(servers.record_failure(&failure("rust")));
assert!(servers.failure(&id).is_some());
assert!(!servers.is_expected(&id));
assert_eq!(servers.failures().len(), 1);
}
fn running_backend() -> Backend {
Backend::Process(LspServer::new_for_test(
lsp_types::ServerCapabilities::default(),
))
}
#[tokio::test]
async fn test_failure_on_a_running_or_stopped_slot_is_ignored() {
let mut servers = Servers::default();
let id = ServerId::from_static("rust");
servers.register(id.clone(), running_backend());
assert!(!servers.record_failure(&failure("rust")));
assert!(servers.failure(&id).is_none());
assert!(servers.server(&id).is_some());
let stopped = servers.remove_server(&id);
assert!(stopped.is_some());
assert!(!servers.record_failure(&failure("rust")));
assert!(servers.failure(&id).is_none());
}
#[tokio::test]
async fn test_stopped_slot_keeps_the_dead_client_and_is_never_revived_by_a_swap() {
let mut servers = Servers::default();
let id = ServerId::from_static("rust");
servers.register(id.clone(), running_backend());
drop(servers.remove_server(&id));
assert!(servers.client(&id).is_some());
assert!(servers.server(&id).is_none());
assert!(servers.slot_for_swap(&id, running_backend()).is_err());
assert!(servers.server(&id).is_none());
}
#[tokio::test]
async fn test_swap_replaces_the_backend_in_one_assignment_and_returns_the_old_one() {
let mut servers = Servers::default();
let id = ServerId::from_static("rust");
servers.register(id.clone(), running_backend());
let old = servers.slot_for_swap(&id, running_backend()).unwrap();
assert!(old.is_some());
let server = servers.server(&id).unwrap();
assert!(std::ptr::eq(
std::ptr::from_ref(server.client()),
std::ptr::from_ref(server.client())
));
assert!(servers.client(&id).is_some());
}
#[tokio::test]
async fn test_pull_probe_is_bound_to_the_running_connection() {
let mut servers = Servers::default();
let id = ServerId::from_static("rust");
servers.register(id.clone(), running_backend());
assert_eq!(servers.pull_probe(&id), PullProbe::Untried);
let first = servers.client(&id).unwrap().connection_id();
servers.update_pull_probe(&id, first, |_| PullProbe::Refused);
let held = servers.take_for_restart(&id).unwrap();
assert!(servers.restore(&id, held).is_none());
assert_eq!(servers.pull_probe(&id), PullProbe::Refused);
drop(servers.slot_for_swap(&id, running_backend()).unwrap());
assert_eq!(servers.pull_probe(&id), PullProbe::Untried);
let second = servers.client(&id).unwrap().connection_id();
assert_ne!(first, second);
assert_eq!(
servers.update_pull_probe(&id, first, |_| PullProbe::Refused),
None
);
assert_eq!(servers.pull_probe(&id), PullProbe::Untried);
servers.update_pull_probe(&id, second, |_| PullProbe::Answered);
drop(servers.register(id.clone(), running_backend()));
assert_eq!(servers.pull_probe(&id), PullProbe::Untried);
}
#[tokio::test]
async fn test_take_and_restore_round_trip_and_refuse_a_settled_slot() {
let mut servers = Servers::default();
let id = ServerId::from_static("rust");
servers.register(id.clone(), running_backend());
let held = servers.take_for_restart(&id).unwrap();
assert!(servers.is_expected(&id));
assert!(servers.restore(&id, held).is_none());
assert!(servers.server(&id).is_some());
let held = servers.take_for_restart(&id).unwrap();
servers.register(id.clone(), running_backend());
assert!(servers.restore(&id, held).is_some());
}
#[test]
fn test_clear_expected_keeps_failed_and_drops_plain_expected() {
let mut servers = Servers::default();
let (failed, plain) = (
ServerId::from_static("failed"),
ServerId::from_static("plain"),
);
servers.record_failure(&failure("failed"));
servers.set_expected(&HashSet::from([failed.clone(), plain.clone()]));
servers.clear_expected();
assert!(servers.failure(&failed).is_some());
assert!(servers.get(&plain).is_none());
}
#[test]
fn test_set_expected_drops_expectation_of_servers_outside_the_set() {
let mut servers = Servers::default();
let (a, b) = (ServerId::from_static("a"), ServerId::from_static("b"));
servers.set_expected(&HashSet::from([a.clone(), b.clone()]));
servers.set_expected(&HashSet::from([a.clone()]));
assert!(servers.is_expected(&a));
assert!(!servers.is_expected(&b));
assert!(servers.get(&b).is_none());
}
}