diff --git a/core/crates/solstone-core-spl/src/admission.rs b/core/crates/solstone-core-spl/src/admission.rs new file mode 100644 index 000000000..2e2705519 --- /dev/null +++ b/core/crates/solstone-core-spl/src/admission.rs @@ -0,0 +1,212 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +use std::collections::HashMap; +use std::sync::{Mutex, MutexGuard}; + +const DEFAULT_GLOBAL_CEILING: usize = 32; +const DEFAULT_SENDER_CEILING: usize = 4; + +#[derive(Debug, Default)] +struct AdmissionState { + global_count: usize, + sender_counts: HashMap, + saturated_count: u64, +} + +/// Synchronous, in-memory admission limits for concurrent blob transfers. +/// +/// Each method holds the mutex only while it updates or reads a counter. Callers +/// acquire before transfer work and release afterward, so the gate can be shared +/// across tunnel tasks without a lock spanning the transfer itself. +#[derive(Debug)] +pub struct BlobAdmissionGate { + global_ceiling: usize, + sender_ceiling: usize, + state: Mutex, +} + +impl BlobAdmissionGate { + /// Creates a gate with independent global and per-sender ceilings. + pub fn new(global_ceiling: usize, sender_ceiling: usize) -> Self { + Self { + global_ceiling, + sender_ceiling, + state: Mutex::new(AdmissionState::default()), + } + } + + /// Acquires one global slot without waiting, or records a refusal. + pub fn try_acquire_global(&self) -> bool { + let mut state = self.lock_state(); + if state.global_count < self.global_ceiling { + state.global_count += 1; + true + } else { + state.saturated_count = state.saturated_count.saturating_add(1); + false + } + } + + /// Releases one global slot. Releasing an empty gate leaves it empty. + pub fn release_global(&self) { + let mut state = self.lock_state(); + state.global_count = state.global_count.saturating_sub(1); + } + + /// Acquires one sender slot without waiting, or records a refusal. + pub fn try_acquire_sender(&self, fp: &str) -> bool { + let mut state = self.lock_state(); + let count = sender_count(&state.sender_counts, fp); + if count < self.sender_ceiling { + state.sender_counts.insert(fp.to_owned(), count + 1); + true + } else { + state.saturated_count = state.saturated_count.saturating_add(1); + false + } + } + + /// Releases one sender slot and removes the sender entry when it reaches zero. + pub fn release_sender(&self, fp: &str) { + let mut state = self.lock_state(); + match state.sender_counts.get(fp).copied() { + Some(0) | None => {} + Some(1) => { + state.sender_counts.remove(fp); + } + Some(count) => { + state.sender_counts.insert(fp.to_owned(), count - 1); + } + } + } + + /// Returns the number of held global slots. + pub fn global_count(&self) -> usize { + self.lock_state().global_count + } + + /// Returns the number of held slots for one sender. + pub fn sender_count(&self, fp: &str) -> usize { + sender_count(&self.lock_state().sender_counts, fp) + } + + /// Returns the number of senders currently holding at least one slot. + pub fn active_senders(&self) -> usize { + self.lock_state().sender_counts.len() + } + + /// Returns the cumulative number of global or per-sender refusals. + pub fn saturated_count(&self) -> u64 { + self.lock_state().saturated_count + } + + fn lock_state(&self) -> MutexGuard<'_, AdmissionState> { + match self.state.lock() { + Ok(state) => state, + Err(poisoned) => poisoned.into_inner(), + } + } +} + +impl Default for BlobAdmissionGate { + fn default() -> Self { + Self::new(DEFAULT_GLOBAL_CEILING, DEFAULT_SENDER_CEILING) + } +} + +fn sender_count(sender_counts: &HashMap, fp: &str) -> usize { + match sender_counts.get(fp) { + Some(count) => *count, + None => 0, + } +} + +#[cfg(test)] +mod tests { + use super::BlobAdmissionGate; + + #[test] + fn default_ceilings_are_32_global_and_4_per_sender() { + let gate = BlobAdmissionGate::default(); + + for _ in 0..32 { + assert!(gate.try_acquire_global()); + } + assert!(!gate.try_acquire_global()); + + for _ in 0..4 { + assert!(gate.try_acquire_sender("sender")); + } + assert!(!gate.try_acquire_sender("sender")); + } + + #[test] + fn global_and_sender_ceilings_refuse_independently() { + let gate = BlobAdmissionGate::new(1, 1); + + assert!(gate.try_acquire_global()); + assert!(!gate.try_acquire_global()); + assert_eq!(gate.saturated_count(), 1); + + assert!(gate.try_acquire_sender("first")); + assert!(!gate.try_acquire_sender("first")); + assert!(gate.try_acquire_sender("second")); + assert_eq!(gate.saturated_count(), 2); + + gate.release_global(); + assert!(gate.try_acquire_global()); + assert_eq!(gate.saturated_count(), 2); + } + + #[test] + fn refusal_count_is_monotonic_and_only_refusals_change_it() { + let gate = BlobAdmissionGate::new(1, 1); + + assert!(gate.try_acquire_global()); + assert_eq!(gate.saturated_count(), 0); + assert!(!gate.try_acquire_global()); + assert_eq!(gate.saturated_count(), 1); + gate.release_global(); + assert_eq!(gate.saturated_count(), 1); + assert!(gate.try_acquire_global()); + assert_eq!(gate.saturated_count(), 1); + + assert!(gate.try_acquire_sender("sender")); + assert!(!gate.try_acquire_sender("sender")); + assert_eq!(gate.saturated_count(), 2); + gate.release_sender("sender"); + assert_eq!(gate.saturated_count(), 2); + } + + #[test] + fn releases_do_not_underflow_or_leak_sender_entries() { + let gate = BlobAdmissionGate::new(2, 2); + + gate.release_global(); + gate.release_sender("missing"); + assert_eq!(gate.global_count(), 0); + assert_eq!(gate.sender_count("missing"), 0); + assert_eq!(gate.active_senders(), 0); + + assert!(gate.try_acquire_sender("sender")); + assert!(gate.try_acquire_sender("sender")); + assert_eq!(gate.active_senders(), 1); + gate.release_sender("sender"); + assert_eq!(gate.sender_count("sender"), 1); + assert_eq!(gate.active_senders(), 1); + gate.release_sender("sender"); + gate.release_sender("sender"); + assert_eq!(gate.sender_count("sender"), 0); + assert_eq!(gate.active_senders(), 0); + } + + #[test] + fn refused_new_sender_does_not_create_an_active_entry() { + let gate = BlobAdmissionGate::new(1, 0); + + assert!(!gate.try_acquire_sender("sender")); + assert_eq!(gate.sender_count("sender"), 0); + assert_eq!(gate.active_senders(), 0); + } +} diff --git a/core/crates/solstone-core-spl/src/health.rs b/core/crates/solstone-core-spl/src/health.rs new file mode 100644 index 000000000..014450c48 --- /dev/null +++ b/core/crates/solstone-core-spl/src/health.rs @@ -0,0 +1,50 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +pub const REASON_HOME_MISSING_MOBILE: &str = "home_missing_mobile"; +pub const REASON_SERVICE_TOKEN_REJECTED: &str = "service_token_rejected"; +pub const REASON_RELAY_TUNNEL_REJECTED: &str = "relay_tunnel_rejected"; +pub const REASON_RELAY_TUNNEL_UNREACHABLE: &str = "relay_tunnel_unreachable"; +pub const REASON_LOCAL_PRIVATE_LISTENER_UNREACHABLE: &str = "local_private_listener_unreachable"; +pub const REASON_RELAY_ADMISSION_SATURATED: &str = "relay_admission_saturated"; + +pub const OFFLINE_TUNNEL_REASONS: [&str; 2] = [ + REASON_SERVICE_TOKEN_REJECTED, + REASON_LOCAL_PRIVATE_LISTENER_UNREACHABLE, +]; +pub const LINK_HEALTH_EVENT: &str = "health"; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn health_vocabulary_matches_the_owner_visible_contract() { + assert_eq!( + [ + REASON_HOME_MISSING_MOBILE, + REASON_SERVICE_TOKEN_REJECTED, + REASON_RELAY_TUNNEL_REJECTED, + REASON_RELAY_TUNNEL_UNREACHABLE, + REASON_LOCAL_PRIVATE_LISTENER_UNREACHABLE, + REASON_RELAY_ADMISSION_SATURATED, + ], + [ + "home_missing_mobile", + "service_token_rejected", + "relay_tunnel_rejected", + "relay_tunnel_unreachable", + "local_private_listener_unreachable", + "relay_admission_saturated", + ] + ); + assert_eq!( + OFFLINE_TUNNEL_REASONS, + [ + "service_token_rejected", + "local_private_listener_unreachable", + ] + ); + assert_eq!(LINK_HEALTH_EVENT, "health"); + } +} diff --git a/core/crates/solstone-core-spl/src/ws_buffer.rs b/core/crates/solstone-core-spl/src/ws_buffer.rs new file mode 100644 index 000000000..28938a2ae --- /dev/null +++ b/core/crates/solstone-core-spl/src/ws_buffer.rs @@ -0,0 +1,326 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +use std::future::Future; +use std::time::Duration; + +use bytes::{Bytes, BytesMut}; +use thiserror::Error; +use tokio::time::Instant; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct WsClosed; + +/// The narrow WebSocket seam shared by tunnel and blob receivers. +pub trait WsByteSource { + fn next_message(&mut self) -> impl Future, WsClosed>> + Send; +} + +#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)] +pub enum WsBufferError { + #[error("websocket read exceeded its absolute deadline")] + ReadTimeout, + #[error("websocket read did not make the required progress")] + ProgressTimeout, + #[error("websocket closed before the requested bytes arrived")] + Closed, +} + +pub struct BufferedWsReader { + source: S, + buffer: BytesMut, +} + +impl BufferedWsReader { + pub fn new(source: S) -> Self { + Self { + source, + buffer: BytesMut::new(), + } + } + + /// Return the next `n` bytes without consuming them. + /// + /// WebSocket message boundaries are not byte boundaries, so this may read + /// across several binary frames before returning. + pub async fn peek(&mut self, n: usize) -> Result { + self.fill(n).await?; + let bytes = self.buffer.get(..n).ok_or(WsBufferError::Closed)?; + Ok(Bytes::copy_from_slice(bytes)) + } + + /// Consume and return exactly `n` bytes, spanning binary frames as needed. + pub async fn read_exactly(&mut self, n: usize) -> Result { + self.fill(n).await?; + Ok(self.buffer.split_to(n).freeze()) + } + + pub async fn peek_bounded( + &mut self, + n: usize, + deadline: Duration, + ) -> Result { + self.fill_bounded(n, deadline, None).await?; + let bytes = self.buffer.get(..n).ok_or(WsBufferError::Closed)?; + Ok(Bytes::copy_from_slice(bytes)) + } + + pub async fn read_exactly_bounded( + &mut self, + n: usize, + deadline: Duration, + ) -> Result { + self.fill_bounded(n, deadline, None).await?; + Ok(self.buffer.split_to(n).freeze()) + } + + pub async fn read_exactly_progress( + &mut self, + n: usize, + deadline: Duration, + window: Duration, + min_bytes_per_window: usize, + ) -> Result { + self.fill_bounded(n, deadline, Some((window, min_bytes_per_window))) + .await?; + Ok(self.buffer.split_to(n).freeze()) + } + + /// Return all residue already received from the WebSocket and clear it. + /// + /// This preserves bytes exactly for loopback replay; it does not wait for a + /// further frame. + pub fn drain_buffer(&mut self) -> Bytes { + self.buffer.split().freeze() + } + + async fn fill(&mut self, n: usize) -> Result<(), WsBufferError> { + while self.buffer.len() < n { + self.receive().await?; + } + Ok(()) + } + + async fn fill_bounded( + &mut self, + n: usize, + deadline: Duration, + progress: Option<(Duration, usize)>, + ) -> Result<(), WsBufferError> { + let progress = progress.filter(|(_, minimum)| *minimum > 0); + let started = Instant::now(); + let absolute_deadline = started.checked_add(deadline); + let mut window_started = started; + let mut window_bytes = 0; + + while self.buffer.len() < n { + let now = Instant::now(); + if deadline_has_elapsed(absolute_deadline, now) { + return Err(WsBufferError::ReadTimeout); + } + let timeout_at = match progress { + Some((window, minimum)) => { + if let Some(window_ended) = elapsed_window_end(window_started, window, now) { + if window_bytes < minimum { + return Err(WsBufferError::ProgressTimeout); + } + window_started = window_ended; + window_bytes = 0; + } + earliest_deadline(absolute_deadline, window_started.checked_add(window)) + } + None => absolute_deadline, + }; + + let frame = match wait_for_frame(&mut self.source, timeout_at).await { + Ok(Ok(Some(frame))) => frame, + Ok(Ok(None)) | Ok(Err(_)) => return Err(WsBufferError::Closed), + Err(_) => { + let now = Instant::now(); + if deadline_has_elapsed(absolute_deadline, now) { + return Err(WsBufferError::ReadTimeout); + } + if let Some((window, minimum)) = progress + && let Some(window_ended) = elapsed_window_end(window_started, window, now) + { + if window_bytes < minimum { + return Err(WsBufferError::ProgressTimeout); + } + window_started = window_ended; + window_bytes = 0; + continue; + } + return Err(WsBufferError::ReadTimeout); + } + }; + window_bytes += frame.len(); + self.buffer.extend_from_slice(&frame); + } + Ok(()) + } + + async fn receive(&mut self) -> Result<(), WsBufferError> { + match self.source.next_message().await { + Ok(Some(frame)) => { + let is_empty = frame.is_empty(); + self.buffer.extend_from_slice(&frame); + if is_empty { + tokio::task::yield_now().await; + } + Ok(()) + } + Ok(None) | Err(_) => Err(WsBufferError::Closed), + } + } +} + +fn deadline_has_elapsed(deadline: Option, now: Instant) -> bool { + deadline.is_some_and(|deadline| now >= deadline) +} + +fn elapsed_window_end(started: Instant, window: Duration, now: Instant) -> Option { + started.checked_add(window).filter(|ends| now >= *ends) +} + +fn earliest_deadline(left: Option, right: Option) -> Option { + match (left, right) { + (Some(left), Some(right)) => Some(left.min(right)), + (Some(deadline), None) | (None, Some(deadline)) => Some(deadline), + (None, None) => None, + } +} + +async fn wait_for_frame( + source: &mut S, + deadline: Option, +) -> Result, WsClosed>, tokio::time::error::Elapsed> { + match deadline { + Some(deadline) => tokio::time::timeout_at(deadline, source.next_message()).await, + None => Ok(source.next_message().await), + } +} + +#[cfg(test)] +mod tests { + use std::collections::VecDeque; + + use super::*; + + struct Frames(VecDeque); + + impl Frames { + fn new(frames: &[&[u8]]) -> Self { + Self( + frames + .iter() + .map(|frame| Bytes::copy_from_slice(frame)) + .collect(), + ) + } + } + + impl WsByteSource for Frames { + fn next_message(&mut self) -> impl Future, WsClosed>> + Send { + std::future::ready(Ok(self.0.pop_front())) + } + } + + #[tokio::test] + async fn split_sbo1_header_peeks_then_reads_the_same_wire_bytes() { + const HEADER: &[u8] = b"SBO1\x00\x00\x00\x10"; + + let mut reader = + BufferedWsReader::new(Frames::new(&[b"SB", b"O1\x00", b"\x00\x00\x10body"])); + + assert_eq!( + reader.peek(HEADER.len()).await, + Ok(Bytes::from_static(HEADER)) + ); + assert_eq!( + reader.read_exactly(HEADER.len()).await, + Ok(Bytes::from_static(HEADER)) + ); + assert_eq!( + reader.read_exactly(4).await, + Ok(Bytes::from_static(b"body")) + ); + } + + #[tokio::test] + async fn consumed_tls_prefix_and_residue_replay_the_exact_client_hello() { + const CLIENT_HELLO: &[u8] = &[ + 0x16, 0x03, 0x01, 0x00, 0x14, 0x01, 0x00, 0x00, 0x10, 0x03, 0x03, 0x42, 0x42, 0x42, + 0x42, 0x42, 0x42, 0x42, 0x42, 0x42, 0x42, 0x42, 0x42, 0x42, 0x42, + ]; + let mut reader = + BufferedWsReader::new(Frames::new(&[&CLIENT_HELLO[..5], &CLIENT_HELLO[5..]])); + + let consumed = reader.read_exactly(8).await; + let residue = reader.drain_buffer(); + assert_eq!(consumed, Ok(Bytes::copy_from_slice(&CLIENT_HELLO[..8]))); + let mut replay = consumed.map_or_else(|_| Vec::new(), |bytes| bytes.to_vec()); + replay.extend_from_slice(&residue); + + assert_eq!(replay, CLIENT_HELLO); + assert_eq!(reader.drain_buffer(), Bytes::new()); + } + + #[tokio::test] + async fn closed_stream_is_not_a_short_success() { + let mut reader = BufferedWsReader::new(Frames::new(&[b"ab"])); + assert_eq!(reader.read_exactly(3).await, Err(WsBufferError::Closed)); + } + + #[tokio::test] + async fn absolute_deadline_times_out_when_no_frame_arrives() { + struct Stalled; + impl WsByteSource for Stalled { + fn next_message( + &mut self, + ) -> impl Future, WsClosed>> + Send { + std::future::pending() + } + } + let mut reader = BufferedWsReader::new(Stalled); + assert_eq!( + reader.read_exactly_bounded(1, Duration::ZERO).await, + Err(WsBufferError::ReadTimeout) + ); + } + + #[tokio::test] + async fn progress_read_spans_frames_when_each_window_has_enough_bytes() { + let mut reader = BufferedWsReader::new(Frames::new(&[b"hel", b"lo"])); + + assert_eq!( + reader + .read_exactly_progress(5, Duration::from_secs(1), Duration::from_secs(1), 3,) + .await, + Ok(Bytes::from_static(b"hello")) + ); + } + + #[tokio::test] + async fn elapsed_progress_window_reports_progress_timeout_without_sleeping() { + let mut reader = BufferedWsReader::new(Frames::new(&[])); + + assert_eq!( + reader + .read_exactly_progress(1, Duration::from_secs(1), Duration::ZERO, 1) + .await, + Err(WsBufferError::ProgressTimeout) + ); + } + + #[tokio::test] + async fn zero_progress_threshold_keeps_the_bounded_read_usable() { + let mut reader = BufferedWsReader::new(Frames::new(&[b"x"])); + + assert_eq!( + reader + .read_exactly_progress(1, Duration::from_secs(1), Duration::ZERO, 0) + .await, + Ok(Bytes::from_static(b"x")) + ); + } +}