From 61c149a5e2ab4e6df6604bd4b2be7f460f317d44 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Fri, 31 Jul 2026 16:31:38 -0600 Subject: [PATCH] Add U5 supervision support --- .../solstone-core-spl/src/link_state_files.rs | 383 ++++++++++++++++++ .../solstone-core-spl/src/posture_gate.rs | 239 +++++++++++ .../solstone-core-spl/src/service_shutdown.rs | 172 ++++++++ .../src/service_transition.rs | 184 +++++++++ 4 files changed, 978 insertions(+) create mode 100644 core/crates/solstone-core-spl/src/link_state_files.rs create mode 100644 core/crates/solstone-core-spl/src/posture_gate.rs create mode 100644 core/crates/solstone-core-spl/src/service_shutdown.rs create mode 100644 core/crates/solstone-core-spl/src/service_transition.rs diff --git a/core/crates/solstone-core-spl/src/link_state_files.rs b/core/crates/solstone-core-spl/src/link_state_files.rs new file mode 100644 index 000000000..8aaef6415 --- /dev/null +++ b/core/crates/solstone-core-spl/src/link_state_files.rs @@ -0,0 +1,383 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +//! Read-only access to the local SPL link identity and service-token files. +//! +//! The owning Python link modules provision these files. This module never +//! creates, updates, or retains their contents after a failed read. + +use std::{fs, io::ErrorKind, path::Path}; + +use serde_json::Value; + +/// The persisted local identity used by the SPL service. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LinkState { + /// The provisioned home instance identifier. + pub instance_id: String, + /// The owner-facing name, or the supplied default when not stored as text. + pub home_label: String, + /// The provisioning lock timestamp when it is a JSON integer. + pub locked_at: Option, +} + +/// The result of loading `link/state.json` without mutating the journal. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum LinkStateRead { + /// A valid identity was read. + Present(LinkState), + /// The identity file has not been provisioned. + Missing, + /// The identity file could not be read. + Unreadable, + /// The identity file was not a valid state object. + Malformed, +} + +/// A service token that intentionally has no formatting implementation. +#[derive(Clone, Eq, PartialEq)] +pub struct LinkServiceToken(String); + +impl LinkServiceToken { + /// Returns the token only to the authenticated relay request builder. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// The result of loading `link/tokens/account.json` without retaining a last-good value. +pub enum LinkServiceTokenRead { + /// A current non-empty token was read. + Present(LinkServiceToken), + /// The token file has not been provisioned. + Missing, + /// The token file could not be read. + Unreadable, + /// The token file was not a JSON object. + Malformed, +} + +enum JsonRead { + Value(Value), + Missing, + Unreadable, + Malformed, +} + +/// Reads `link/state.json` beneath a journal root without creating any path. +pub fn load_link_state(journal_root: &Path, default_label: &str) -> LinkStateRead { + let path = journal_root.join("link").join("state.json"); + let raw = match read_json(&path) { + JsonRead::Value(raw) => raw, + JsonRead::Missing => return LinkStateRead::Missing, + JsonRead::Unreadable => return LinkStateRead::Unreadable, + JsonRead::Malformed => return LinkStateRead::Malformed, + }; + + let Some(object) = raw.as_object() else { + return LinkStateRead::Malformed; + }; + let Some(instance_id) = object + .get("instance_id") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + else { + return LinkStateRead::Malformed; + }; + + let stored_label = object + .get("home_label") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()); + let home_label = match stored_label { + Some(value) => value.to_owned(), + None => default_label.to_owned(), + }; + let locked_at = object.get("locked_at").and_then(Value::as_i64); + + LinkStateRead::Present(LinkState { + instance_id: instance_id.to_owned(), + home_label, + locked_at, + }) +} + +/// Reads `link/tokens/account.json` beneath a journal root without creating any path. +pub fn load_link_service_token(journal_root: &Path) -> LinkServiceTokenRead { + let path = journal_root + .join("link") + .join("tokens") + .join("account.json"); + let raw = match read_json(&path) { + JsonRead::Value(raw) => raw, + JsonRead::Missing => return LinkServiceTokenRead::Missing, + JsonRead::Unreadable => return LinkServiceTokenRead::Unreadable, + JsonRead::Malformed => return LinkServiceTokenRead::Malformed, + }; + + let Some(object) = raw.as_object() else { + return LinkServiceTokenRead::Malformed; + }; + let candidate = object + .get("service_token") + .filter(|value| json_truthy(value)) + .or_else(|| object.get("account_token")); + let token = candidate + .and_then(Value::as_str) + .filter(|value| !value.is_empty()); + + match token { + Some(value) => LinkServiceTokenRead::Present(LinkServiceToken(value.to_owned())), + None => LinkServiceTokenRead::Missing, + } +} + +fn json_truthy(value: &Value) -> bool { + match value { + Value::Null => false, + Value::Bool(value) => *value, + Value::Number(value) => value.as_f64().is_some_and(|value| value != 0.0), + Value::String(value) => !value.is_empty(), + Value::Array(value) => !value.is_empty(), + Value::Object(value) => !value.is_empty(), + } +} + +fn read_json(path: &Path) -> JsonRead { + let text = match fs::read_to_string(path) { + Ok(text) => text, + Err(error) if error.kind() == ErrorKind::NotFound => return JsonRead::Missing, + Err(_) => return JsonRead::Unreadable, + }; + + match serde_json::from_str(&text) { + Ok(value) => JsonRead::Value(value), + Err(_) => JsonRead::Malformed, + } +} + +#[cfg(test)] +mod tests { + use std::{ + error::Error, + fs, + path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, + }; + + use super::{LinkServiceTokenRead, LinkStateRead, load_link_service_token, load_link_state}; + + struct TempJournal { + path: PathBuf, + } + + impl TempJournal { + fn new() -> Result> { + static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(0); + + for _ in 0..100 { + let ordinal = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "solstone-core-spl-link-state-{}-{ordinal}", + std::process::id() + )); + match fs::create_dir(&path) { + Ok(()) => return Ok(Self { path }), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(error) => return Err(error.into()), + } + } + + Err("could not allocate a disposable journal directory".into()) + } + + fn write(&self, relative: &str, contents: &str) -> Result<(), Box> { + let path = self.path.join(relative); + let Some(parent) = path.parent() else { + return Err("test path has no parent".into()); + }; + fs::create_dir_all(parent)?; + fs::write(path, contents)?; + Ok(()) + } + + fn path(&self) -> &Path { + &self.path + } + } + + impl Drop for TempJournal { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } + } + + #[test] + fn existing_state_is_read_without_rewriting_it() -> Result<(), Box> { + let journal = TempJournal::new()?; + let state = r#"{"instance_id":"home-1","home_label":"Study","locked_at":1700000000}"#; + journal.write("link/state.json", state)?; + let state_path = journal.path().join("link/state.json"); + let before = fs::read(&state_path)?; + + let loaded = load_link_state(journal.path(), "solstone"); + + match loaded { + LinkStateRead::Present(value) => { + assert_eq!(value.instance_id, "home-1"); + assert_eq!(value.home_label, "Study"); + assert_eq!(value.locked_at, Some(1_700_000_000)); + } + _ => return Err("valid state was not loaded".into()), + } + assert_eq!(fs::read(state_path)?, before); + Ok(()) + } + + #[test] + fn missing_malformed_and_unreadable_state_remain_distinguishable() -> Result<(), Box> + { + let missing = TempJournal::new()?; + assert!(matches!( + load_link_state(missing.path(), "solstone"), + LinkStateRead::Missing + )); + assert!(!missing.path().join("link").exists()); + + let malformed = TempJournal::new()?; + malformed.write("link/state.json", "not json")?; + assert!(matches!( + load_link_state(malformed.path(), "solstone"), + LinkStateRead::Malformed + )); + + let unreadable = TempJournal::new()?; + fs::create_dir_all(unreadable.path().join("link/state.json"))?; + assert!(matches!( + load_link_state(unreadable.path(), "solstone"), + LinkStateRead::Unreadable + )); + Ok(()) + } + + #[test] + fn state_uses_the_supplied_default_for_empty_or_nontext_labels() -> Result<(), Box> { + let journal = TempJournal::new()?; + + journal.write( + "link/state.json", + r#"{"instance_id":"home-1","home_label":""}"#, + )?; + match load_link_state(journal.path(), "Default Home") { + LinkStateRead::Present(value) => assert_eq!(value.home_label, "Default Home"), + _ => return Err("empty state label was not defaulted".into()), + } + + journal.write( + "link/state.json", + r#"{"instance_id":"home-1","home_label":42}"#, + )?; + match load_link_state(journal.path(), "Default Home") { + LinkStateRead::Present(value) => assert_eq!(value.home_label, "Default Home"), + _ => return Err("nontext state label was not defaulted".into()), + } + Ok(()) + } + + #[test] + fn state_accepts_only_integer_locked_at_values() -> Result<(), Box> { + let journal = TempJournal::new()?; + let cases = [ + ("1700000000", Some(1_700_000_000)), + ("\"1700000000\"", None), + ("true", None), + ("1700000000.5", None), + ]; + + for (locked_at, expected) in cases { + let payload = format!("{{\"instance_id\":\"home-1\",\"locked_at\":{locked_at}}}"); + journal.write("link/state.json", &payload)?; + match load_link_state(journal.path(), "solstone") { + LinkStateRead::Present(value) => assert_eq!(value.locked_at, expected), + _ => return Err("valid state was not loaded".into()), + } + } + Ok(()) + } + + #[test] + fn token_prefers_service_then_legacy_account_without_creating_files() + -> Result<(), Box> { + let journal = TempJournal::new()?; + assert!(matches!( + load_link_service_token(journal.path()), + LinkServiceTokenRead::Missing + )); + assert!(!journal.path().join("link").join("tokens").exists()); + + journal.write( + "tokens/account.json", + r#"{"service_token":"wrong-path-token"}"#, + )?; + assert!(matches!( + load_link_service_token(journal.path()), + LinkServiceTokenRead::Missing + )); + + journal.write( + "link/tokens/account.json", + r#"{"service_token":"current-token","account_token":"legacy-token"}"#, + )?; + match load_link_service_token(journal.path()) { + LinkServiceTokenRead::Present(value) => assert!(value.as_str() == "current-token"), + _ => return Err("current service token was not loaded".into()), + } + + journal.write( + "link/tokens/account.json", + r#"{"service_token":"","account_token":"legacy-token"}"#, + )?; + match load_link_service_token(journal.path()) { + LinkServiceTokenRead::Present(value) => assert!(value.as_str() == "legacy-token"), + _ => return Err("legacy service token was not loaded".into()), + } + + journal.write( + "link/tokens/account.json", + r#"{"service_token":false,"account_token":"legacy-token"}"#, + )?; + match load_link_service_token(journal.path()) { + LinkServiceTokenRead::Present(value) => assert!(value.as_str() == "legacy-token"), + _ => return Err("legacy service token was not loaded".into()), + } + + journal.write( + "link/tokens/account.json", + r#"{"service_token":5,"account_token":"legacy-token"}"#, + )?; + assert!(matches!( + load_link_service_token(journal.path()), + LinkServiceTokenRead::Missing + )); + Ok(()) + } + + #[test] + fn malformed_and_unreadable_tokens_do_not_supply_a_cached_value() -> Result<(), Box> + { + let malformed = TempJournal::new()?; + malformed.write("link/tokens/account.json", "not json")?; + assert!(matches!( + load_link_service_token(malformed.path()), + LinkServiceTokenRead::Malformed + )); + + let unreadable = TempJournal::new()?; + fs::create_dir_all(unreadable.path().join("link/tokens/account.json"))?; + assert!(matches!( + load_link_service_token(unreadable.path()), + LinkServiceTokenRead::Unreadable + )); + Ok(()) + } +} diff --git a/core/crates/solstone-core-spl/src/posture_gate.rs b/core/crates/solstone-core-spl/src/posture_gate.rs new file mode 100644 index 000000000..514651222 --- /dev/null +++ b/core/crates/solstone-core-spl/src/posture_gate.rs @@ -0,0 +1,239 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +/// A cached service token that never exposes its contents through formatting. +#[derive(Clone, Eq, PartialEq)] +pub struct ServiceToken(String); + +impl ServiceToken { + /// Returns the token only for the authenticated request that needs it. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// The latest value obtained from the local posture source. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum PostureInput { + /// A successfully read posture value. + Value(String), + /// The posture source could not be read. + ReadFailed, +} + +/// The latest value obtained from the local service-token source. +#[derive(Clone, Eq, PartialEq)] +pub enum TokenInput { + /// A successfully read token value. + Value(String), + /// The token source could not be read. + ReadFailed, +} + +/// The reason a relay connection is not permitted. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RelayBlocked { + /// No usable posture has been observed, or its value is not exactly `spl`. + PostureNotSpl, + /// Reading the posture source failed. + PostureReadFailed, + /// No non-empty cached service token is available. + TokenMissing, + /// Reading the service-token source failed. + TokenReadFailed, +} + +/// An allowed relay connection with its authenticated service token. +pub struct RelayPermit { + token: ServiceToken, +} + +impl RelayPermit { + /// Returns the service token for the connection authentication header. + pub fn token(&self) -> &ServiceToken { + &self.token + } +} + +/// The current relay connection decision. +pub enum RelayDecision { + /// The relay connection may open. + Allowed(RelayPermit), + /// The relay connection must remain closed. + Blocked(RelayBlocked), +} + +#[derive(Clone, Eq, PartialEq)] +enum PostureState { + Unobserved, + Value(String), + ReadFailed, +} + +enum TokenState { + Empty, + Cached(ServiceToken), + ReadFailed, +} + +/// Pure cache and admission state for the relay WebSocket. +/// +/// Any posture transition clears the cached token before the next decision. +pub struct PostureGate { + posture: PostureState, + token: TokenState, +} + +impl PostureGate { + /// Creates a gate which blocks relay connections until both inputs are read. + pub fn new() -> Self { + Self { + posture: PostureState::Unobserved, + token: TokenState::Empty, + } + } + + /// Records the latest posture value and clears a token cache after a change. + pub fn update_posture(&mut self, input: PostureInput) { + let next = match input { + PostureInput::Value(value) => PostureState::Value(value), + PostureInput::ReadFailed => PostureState::ReadFailed, + }; + + if self.posture != next { + self.token = TokenState::Empty; + } + self.posture = next; + } + + /// Records the latest service-token read result. + pub fn update_token(&mut self, input: TokenInput) { + self.token = match input { + TokenInput::Value(value) if value.is_empty() => TokenState::Empty, + TokenInput::Value(value) => TokenState::Cached(ServiceToken(value)), + TokenInput::ReadFailed => TokenState::ReadFailed, + }; + } + + /// Returns the admission decision for a relay connection attempt. + pub fn decision(&self) -> RelayDecision { + match &self.posture { + PostureState::Value(value) if value == "spl" => match &self.token { + TokenState::Cached(token) => RelayDecision::Allowed(RelayPermit { + token: token.clone(), + }), + TokenState::Empty => RelayDecision::Blocked(RelayBlocked::TokenMissing), + TokenState::ReadFailed => RelayDecision::Blocked(RelayBlocked::TokenReadFailed), + }, + PostureState::ReadFailed => RelayDecision::Blocked(RelayBlocked::PostureReadFailed), + PostureState::Unobserved | PostureState::Value(_) => { + RelayDecision::Blocked(RelayBlocked::PostureNotSpl) + } + } + } +} + +impl Default for PostureGate { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::{PostureGate, PostureInput, RelayBlocked, RelayDecision, TokenInput}; + + fn blocked(gate: &PostureGate) -> Option { + match gate.decision() { + RelayDecision::Allowed(_) => None, + RelayDecision::Blocked(reason) => Some(reason), + } + } + + #[test] + fn unobserved_and_non_spl_postures_block_the_relay() { + let mut gate = PostureGate::new(); + + assert_eq!(blocked(&gate), Some(RelayBlocked::PostureNotSpl)); + + gate.update_posture(PostureInput::Value("home".to_owned())); + gate.update_token(TokenInput::Value("service-token".to_owned())); + assert_eq!(blocked(&gate), Some(RelayBlocked::PostureNotSpl)); + } + + #[test] + fn only_exact_spl_posture_is_eligible() { + let mut gate = PostureGate::new(); + + for posture in ["SPL", "spl ", " spl", "Spl"] { + gate.update_posture(PostureInput::Value(posture.to_owned())); + gate.update_token(TokenInput::Value("service-token".to_owned())); + assert_eq!(blocked(&gate), Some(RelayBlocked::PostureNotSpl)); + } + } + + #[test] + fn exact_spl_with_nonempty_token_is_allowed() { + let mut gate = PostureGate::new(); + gate.update_posture(PostureInput::Value("spl".to_owned())); + gate.update_token(TokenInput::Value("service-token".to_owned())); + + match gate.decision() { + RelayDecision::Allowed(permit) => assert_eq!(permit.token().as_str(), "service-token"), + RelayDecision::Blocked(reason) => assert_eq!(reason, RelayBlocked::TokenMissing), + } + } + + #[test] + fn empty_token_blocks_the_relay() { + let mut gate = PostureGate::new(); + gate.update_posture(PostureInput::Value("spl".to_owned())); + gate.update_token(TokenInput::Value(String::new())); + + assert_eq!(blocked(&gate), Some(RelayBlocked::TokenMissing)); + } + + #[test] + fn posture_read_failure_remains_distinguishable() { + let mut gate = PostureGate::new(); + gate.update_posture(PostureInput::ReadFailed); + gate.update_token(TokenInput::Value("service-token".to_owned())); + + assert_eq!(blocked(&gate), Some(RelayBlocked::PostureReadFailed)); + } + + #[test] + fn token_read_failure_remains_distinguishable() { + let mut gate = PostureGate::new(); + gate.update_posture(PostureInput::Value("spl".to_owned())); + gate.update_token(TokenInput::ReadFailed); + + assert_eq!(blocked(&gate), Some(RelayBlocked::TokenReadFailed)); + } + + #[test] + fn posture_changes_invalidate_the_cached_token() { + let mut gate = PostureGate::new(); + gate.update_posture(PostureInput::Value("spl".to_owned())); + gate.update_token(TokenInput::Value("service-token".to_owned())); + + gate.update_posture(PostureInput::Value("home".to_owned())); + assert_eq!(blocked(&gate), Some(RelayBlocked::PostureNotSpl)); + + gate.update_posture(PostureInput::Value("spl".to_owned())); + assert_eq!(blocked(&gate), Some(RelayBlocked::TokenMissing)); + } + + #[test] + fn unchanged_posture_keeps_the_cached_token() { + let mut gate = PostureGate::new(); + gate.update_posture(PostureInput::Value("spl".to_owned())); + gate.update_token(TokenInput::Value("service-token".to_owned())); + gate.update_posture(PostureInput::Value("spl".to_owned())); + + match gate.decision() { + RelayDecision::Allowed(permit) => assert_eq!(permit.token().as_str(), "service-token"), + RelayDecision::Blocked(reason) => assert_eq!(reason, RelayBlocked::TokenMissing), + } + } +} diff --git a/core/crates/solstone-core-spl/src/service_shutdown.rs b/core/crates/solstone-core-spl/src/service_shutdown.rs new file mode 100644 index 000000000..b2a12cedb --- /dev/null +++ b/core/crates/solstone-core-spl/src/service_shutdown.rs @@ -0,0 +1,172 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +//! Supervised shutdown of the relay client and its listen task. +//! +//! C4 requires that a posture transition away from exact `spl` leaves the +//! process with no WebSocket to the relay. Calling the client `stop()` method +//! cancels tunnels, but does not close the listen WebSocket. The supervisor +//! must therefore cancel and await the separately-running listen task after +//! the client has stopped. +//! +//! Python replacement, quoted verbatim from `solstone/think/spl/service.py`: +//! +//! ```python +//! async def _stop_client(client: RelayClient, run_task: asyncio.Task[None]) -> None: +//! await client.stop() +//! run_task.cancel() +//! with contextlib.suppress(asyncio.CancelledError): +//! await run_task +//! ``` + +use std::{error::Error, future::Future}; + +use tokio::task::{JoinError, JoinHandle}; + +/// A relay client whose tunnel work can be stopped before its listen task ends. +pub trait RelayStop { + /// The error returned when stopping tunnel work fails. + type Error: Error + Send + Sync + 'static; + + /// Stops relay tunnel work without closing the separately-owned listen task. + fn stop(&mut self) -> impl Future> + Send; +} + +/// A failure while stopping a relay client and its listen task. +#[derive(Debug, thiserror::Error)] +pub enum ServiceShutdownError +where + ClientError: Error + 'static, + RunError: Error + 'static, +{ + /// Stopping the relay client's tunnel work failed. + #[error("failed to stop relay client")] + ClientStop(#[source] ClientError), + /// The listen task ended with an unexpected application failure. + #[error("relay listen task failed during shutdown")] + ListenRun(#[source] RunError), + /// The listen task ended abnormally instead of being cancelled. + #[error("relay listen task join failed during shutdown")] + ListenJoin(#[source] JoinError), +} + +/// Stops relay tunnel work, then cancels and awaits the relay listen task. +/// +/// A cancelled listen task is the expected result. A task that finished with +/// an application or join failure before cancellation is returned to the +/// supervisor as a typed error. +pub async fn stop_relay_run( + client: &mut Client, + run_task: JoinHandle>, +) -> Result<(), ServiceShutdownError> +where + Client: RelayStop, + RunError: Error + Send + Sync + 'static, +{ + client + .stop() + .await + .map_err(ServiceShutdownError::ClientStop)?; + + run_task.abort(); + match run_task.await { + Ok(Ok(())) => Ok(()), + Ok(Err(error)) => Err(ServiceShutdownError::ListenRun(error)), + Err(error) if error.is_cancelled() => Ok(()), + Err(error) => Err(ServiceShutdownError::ListenJoin(error)), + } +} + +#[cfg(test)] +mod tests { + use std::{ + convert::Infallible, + future, + sync::{Arc, Mutex}, + }; + + use super::{RelayStop, ServiceShutdownError, stop_relay_run}; + + #[derive(Debug, thiserror::Error)] + #[error("test client stop failure")] + struct TestClientError; + + struct TestClient { + events: Arc>>, + } + + impl RelayStop for TestClient { + type Error = TestClientError; + + async fn stop(&mut self) -> Result<(), Self::Error> { + push_event(&self.events, "stop"); + Ok(()) + } + } + + struct CancellationMarker { + events: Arc>>, + } + + impl Drop for CancellationMarker { + fn drop(&mut self) { + push_event(&self.events, "listen-cancelled"); + } + } + + #[tokio::test] + async fn stops_before_cancelling_and_awaits_the_listen_task() { + let events = Arc::new(Mutex::new(Vec::new())); + let (started_send, started_receive) = tokio::sync::oneshot::channel(); + let task_events = Arc::clone(&events); + let run_task = tokio::spawn(async move { + let _ = started_send.send(()); + let _marker = CancellationMarker { + events: task_events, + }; + future::pending::>().await + }); + assert!(started_receive.await.is_ok()); + + let mut client = TestClient { + events: Arc::clone(&events), + }; + let result = stop_relay_run(&mut client, run_task).await; + + assert!(result.is_ok()); + assert_eq!(read_events(&events), ["stop", "listen-cancelled"]); + } + + #[derive(Debug, thiserror::Error)] + #[error("test listen failure")] + struct TestRunError; + + #[tokio::test] + async fn reports_an_unexpected_listen_task_failure() { + let events = Arc::new(Mutex::new(Vec::new())); + let run_task = tokio::spawn(async { Err::<(), _>(TestRunError) }); + tokio::task::yield_now().await; + let mut client = TestClient { + events: Arc::clone(&events), + }; + + let result = stop_relay_run(&mut client, run_task).await; + + assert!(matches!(result, Err(ServiceShutdownError::ListenRun(_)))); + assert_eq!(read_events(&events), ["stop"]); + } + + fn push_event(events: &Mutex>, event: &'static str) { + match events.lock() { + Ok(mut guard) => guard.push(event), + Err(poisoned) => poisoned.into_inner().push(event), + } + } + + fn read_events(events: &Mutex>) -> Vec<&'static str> { + match events.lock() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } + } +} diff --git a/core/crates/solstone-core-spl/src/service_transition.rs b/core/crates/solstone-core-spl/src/service_transition.rs new file mode 100644 index 000000000..9cf86c3a3 --- /dev/null +++ b/core/crates/solstone-core-spl/src/service_transition.rs @@ -0,0 +1,184 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +/// Whether the supervised SPL client is currently absent or running. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ServiceLifecycle { + /// No client is running. + Idle, + /// A client is running and may be kept alive through a posture read error. + Parked, +} + +/// The current observation of the local SPL posture. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum PostureObservation { + /// The posture source contained exactly `spl`. + Spl, + /// The posture source was read but did not contain exactly `spl`. + NotSpl, + /// The posture source could not be read. + ReadFailed, +} + +impl PostureObservation { + /// Converts an observed posture value without normalizing it. + pub fn from_value(value: &str) -> Self { + if value == "spl" { + Self::Spl + } else { + Self::NotSpl + } + } +} + +/// The current observation of the SPL service-token source. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TokenObservation { + /// A non-empty service token is available to a new client. + Present, + /// No service token is available. + Missing, + /// The service-token source could not be read. + ReadFailed, +} + +/// The supervisor action for one posture/token observation cycle. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ServiceAction { + /// Keep the supervisor idle. + StayIdle, + /// Start a new SPL client. + Start, + /// Keep the existing SPL client running. + StayParked, + /// Stop the existing SPL client and clear its state. + Stop, +} + +/// Decides a lifecycle action without starting, stopping, or inspecting a client. +/// +/// A token gates only a new start. Once parked, the client stays available while +/// the posture remains `spl` or cannot be read; an explicit non-SPL posture is +/// the only observation that stops it. +pub const fn transition( + lifecycle: ServiceLifecycle, + posture: PostureObservation, + token: TokenObservation, +) -> ServiceAction { + match lifecycle { + ServiceLifecycle::Idle => match (posture, token) { + (PostureObservation::Spl, TokenObservation::Present) => ServiceAction::Start, + _ => ServiceAction::StayIdle, + }, + ServiceLifecycle::Parked => match posture { + PostureObservation::NotSpl => ServiceAction::Stop, + PostureObservation::Spl | PostureObservation::ReadFailed => ServiceAction::StayParked, + }, + } +} + +#[cfg(test)] +mod tests { + use super::{ + PostureObservation, ServiceAction, ServiceLifecycle, TokenObservation, transition, + }; + + #[test] + fn only_exact_spl_is_an_eligible_posture() { + assert_eq!( + PostureObservation::from_value("spl"), + PostureObservation::Spl + ); + + for value in ["SPL", "spl ", " spl", "Spl", "home", ""] { + assert_eq!( + PostureObservation::from_value(value), + PostureObservation::NotSpl + ); + } + } + + #[test] + fn idle_transition_table_fails_closed() { + let cases = [ + ( + PostureObservation::Spl, + TokenObservation::Present, + ServiceAction::Start, + ), + ( + PostureObservation::Spl, + TokenObservation::Missing, + ServiceAction::StayIdle, + ), + ( + PostureObservation::Spl, + TokenObservation::ReadFailed, + ServiceAction::StayIdle, + ), + ( + PostureObservation::NotSpl, + TokenObservation::Present, + ServiceAction::StayIdle, + ), + ( + PostureObservation::ReadFailed, + TokenObservation::Present, + ServiceAction::StayIdle, + ), + ]; + + for (posture, token, expected) in cases { + assert_eq!(transition(ServiceLifecycle::Idle, posture, token), expected); + } + } + + #[test] + fn parked_transition_table_only_stops_after_an_explicit_non_spl_posture() { + let cases = [ + ( + PostureObservation::Spl, + TokenObservation::Present, + ServiceAction::StayParked, + ), + ( + PostureObservation::Spl, + TokenObservation::Missing, + ServiceAction::StayParked, + ), + ( + PostureObservation::Spl, + TokenObservation::ReadFailed, + ServiceAction::StayParked, + ), + ( + PostureObservation::ReadFailed, + TokenObservation::Present, + ServiceAction::StayParked, + ), + ( + PostureObservation::NotSpl, + TokenObservation::Present, + ServiceAction::Stop, + ), + ( + PostureObservation::NotSpl, + TokenObservation::Missing, + ServiceAction::Stop, + ), + ( + PostureObservation::NotSpl, + TokenObservation::ReadFailed, + ServiceAction::Stop, + ), + ]; + + for (posture, token, expected) in cases { + assert_eq!( + transition(ServiceLifecycle::Parked, posture, token), + expected + ); + } + } +} -- 2.51.2