diff --git a/crates/hypercolor-daemon/src/api/ws/cache.rs b/crates/hypercolor-daemon/src/api/ws/cache.rs index dc79d1527..eca85bf09 100644 --- a/crates/hypercolor-daemon/src/api/ws/cache.rs +++ b/crates/hypercolor-daemon/src/api/ws/cache.rs @@ -26,13 +26,12 @@ use hypercolor_leptos_ext::ws::{ZONE_PREVIEW_FRAME_HEADER_LEN, ZONE_PREVIEW_FRAM use hypercolor_types::canvas::PublishedSurfaceStorageIdentity; use hypercolor_types::scene::{SceneId, ZoneId}; +use hypercolor_leptos_ext::ws::registry::{CanvasFormat, FrameFormat}; + use super::preview_encode::{ PreviewJpegEncoder, PreviewRawEncoder, encode_canvas_jpeg_payload_scaled_stateless, }; -use super::protocol::{ - ActiveFramesConfig, CanvasFormat, FrameFormat, FrameZoneSelection, - validate_preview_surface_bytes, -}; +use super::protocol::{ActiveFramesConfig, FrameZoneSelection, validate_preview_surface_bytes}; use crate::api::AppState; use crate::display_frames::DisplayFrameSnapshot; diff --git a/crates/hypercolor-daemon/src/api/ws/interactive_preview_relay.rs b/crates/hypercolor-daemon/src/api/ws/interactive_preview_relay.rs index 58316f635..df665859f 100644 --- a/crates/hypercolor-daemon/src/api/ws/interactive_preview_relay.rs +++ b/crates/hypercolor-daemon/src/api/ws/interactive_preview_relay.rs @@ -13,8 +13,9 @@ use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::warn; +use hypercolor_leptos_ext::ws::registry::CanvasFormat; + use super::preview_encode::{PreviewJpegEncoder, PreviewRawEncoder}; -use super::protocol::CanvasFormat; use super::relays::PreviewOutboundSender; use crate::interactive_preview::InteractivePreviewFrame; use crate::preview_runtime::PreviewPixelFormat; diff --git a/crates/hypercolor-daemon/src/api/ws/mod.rs b/crates/hypercolor-daemon/src/api/ws/mod.rs index 52beb7c09..7bfbb0d5c 100644 --- a/crates/hypercolor-daemon/src/api/ws/mod.rs +++ b/crates/hypercolor-daemon/src/api/ws/mod.rs @@ -13,6 +13,7 @@ mod preview_scale; mod protocol; mod relays; mod session; +mod topics; #[cfg(test)] mod tests; diff --git a/crates/hypercolor-daemon/src/api/ws/preview_encode.rs b/crates/hypercolor-daemon/src/api/ws/preview_encode.rs index baded8ade..847bf0af0 100644 --- a/crates/hypercolor-daemon/src/api/ws/preview_encode.rs +++ b/crates/hypercolor-daemon/src/api/ws/preview_encode.rs @@ -11,10 +11,10 @@ use hypercolor_leptos_ext::ws::{ }; use hypercolor_types::canvas::{linear_to_srgb_u8, srgb_u8_to_linear}; +use hypercolor_leptos_ext::ws::registry::CanvasFormat; + use super::preview_scale::{PreviewScaleFormat, PreviewScaleWorkspace}; -use super::protocol::{ - CanvasFormat, validate_preview_surface_bytes, validate_preview_surface_resource, -}; +use super::protocol::{validate_preview_surface_bytes, validate_preview_surface_resource}; const PREVIEW_JPEG_QUALITY: u8 = 80; const PREVIEW_JPEG_SUBSAMP: TurboJpegSubsamp = TurboJpegSubsamp::Sub2x2; diff --git a/crates/hypercolor-daemon/src/api/ws/protocol.rs b/crates/hypercolor-daemon/src/api/ws/protocol.rs index 9b1cc3477..4435299cf 100644 --- a/crates/hypercolor-daemon/src/api/ws/protocol.rs +++ b/crates/hypercolor-daemon/src/api/ws/protocol.rs @@ -3,15 +3,19 @@ //! These types describe the wire format on `/api/v1/ws`. Everything here is data — //! no network I/O, no caches, no runtime state. -use std::collections::HashSet; +use std::collections::{BTreeMap, HashSet}; use std::fmt; use std::hash::DefaultHasher; use std::hash::{Hash, Hasher}; -use serde::de::{self, IgnoredAny, SeqAccess, Visitor}; +use serde::de::{self, IgnoredAny, MapAccess, SeqAccess, Visitor}; use serde::{Deserialize, Deserializer, Serialize}; use serde_json::json; +use hypercolor_leptos_ext::ws::registry::{ + CanvasConfig, CanvasFormat, FramesConfig, TopicId, TopicSet, +}; +use hypercolor_leptos_ext::ws::topic::{PatchError, SubscriptionTable}; use hypercolor_leptos_ext::ws::{ DEFAULT_PREVIEW_MAX_DECODED_PUBLICATION_BYTES, INTERACTIVE_PREVIEW_ID_MAX_BYTES, PreviewTransportCapability, @@ -25,384 +29,289 @@ use crate::device_metrics::DeviceMetricsSnapshot; // ── Subscription Types ─────────────────────────────────────────────────── -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub(super) enum WsChannel { - Frames, - Spectrum, - Events, - FrameEvents, - Canvas, - ScreenCanvas, - ScreenZones, - WebViewportCanvas, - ZonePreview, - Metrics, - DeviceMetrics, - Sensors, - DisplayPreview, - InputEvents, -} - -impl WsChannel { - pub(super) const SUPPORTED: [Self; 14] = [ - Self::Frames, - Self::Spectrum, - Self::Events, - Self::FrameEvents, - Self::Canvas, - Self::ScreenCanvas, - Self::ScreenZones, - Self::WebViewportCanvas, - Self::ZonePreview, - Self::Metrics, - Self::DeviceMetrics, - Self::Sensors, - Self::DisplayPreview, - Self::InputEvents, - ]; - - pub(super) const fn as_str(self) -> &'static str { - match self { - Self::Frames => "frames", - Self::Spectrum => "spectrum", - Self::Events => "events", - Self::FrameEvents => "frame_events", - Self::Canvas => "canvas", - Self::ScreenCanvas => "screen_canvas", - Self::ScreenZones => "screen_zones", - Self::WebViewportCanvas => "web_viewport_canvas", - Self::ZonePreview => "zone_preview", - Self::Metrics => "metrics", - Self::DeviceMetrics => "device_metrics", - Self::Sensors => "sensors", - Self::DisplayPreview => "display_preview", - Self::InputEvents => "input_events", - } - } - - pub(super) fn parse(raw: &str) -> Option { - match raw { - "frames" => Some(Self::Frames), - "spectrum" => Some(Self::Spectrum), - "events" => Some(Self::Events), - "frame_events" => Some(Self::FrameEvents), - "canvas" => Some(Self::Canvas), - "screen_canvas" => Some(Self::ScreenCanvas), - "screen_zones" => Some(Self::ScreenZones), - "web_viewport_canvas" => Some(Self::WebViewportCanvas), - "zone_preview" => Some(Self::ZonePreview), - "metrics" => Some(Self::Metrics), - "device_metrics" => Some(Self::DeviceMetrics), - "sensors" => Some(Self::Sensors), - "display_preview" => Some(Self::DisplayPreview), - "input_events" => Some(Self::InputEvents), - _ => None, - } - } - - pub(super) fn is_supported(self) -> bool { - Self::SUPPORTED.contains(&self) - } +/// One validated wire selector: the topic a client named plus the +/// canonical key its key type parsed. Every topic is unkeyed today, so +/// `key` is always `None`; routing it through the vtable anyway keeps +/// the boundary — not the caller — in charge of what a key looks like. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct TopicSelection { + pub(super) topic: TopicId, + pub(super) key: Option, +} - pub(super) const fn requires_control_subscription(self) -> bool { - matches!( - self, - Self::ScreenCanvas | Self::ScreenZones | Self::InputEvents - ) - } +/// One connection's live subscriptions. +/// +/// Membership and per-subscription config are two views of one fact, so +/// they move together: [`SubscriptionState::admit`] is the only place a +/// topic joins the set, and it materializes that topic's default config +/// in the same step. Every client-visible change goes through +/// [`SubscriptionState::subscribe`] or +/// [`SubscriptionState::unsubscribe`], which build a whole replacement +/// state the caller swaps in only after the runtime accepts it. +/// +/// Config outlives membership on purpose: unsubscribing drops the topic +/// from the set, and its config moves aside into [`DormantConfigs`] so a +/// client that re-subscribes gets its own settings back rather than the +/// defaults. The live table only ever holds live subscriptions, which is +/// what its own contract promises and what `any_for` has to keep meaning. +#[derive(Debug, Clone)] +pub(super) struct SubscriptionState { + topics: TopicSet, + live: SubscriptionTable, + dormant: DormantConfigs, +} - const fn bit(self) -> u16 { - match self { - Self::Frames => 1 << 0, - Self::Spectrum => 1 << 1, - Self::Events => 1 << 2, - Self::FrameEvents => 1 << 3, - Self::Canvas => 1 << 4, - Self::ScreenCanvas => 1 << 5, - Self::ScreenZones => 1 << 12, - Self::WebViewportCanvas => 1 << 6, - Self::ZonePreview => 1 << 7, - Self::Metrics => 1 << 8, - Self::DeviceMetrics => 1 << 9, - Self::Sensors => 1 << 10, - Self::DisplayPreview => 1 << 11, - Self::InputEvents => 1 << 13, - } +impl Default for SubscriptionState { + /// A fresh connection starts subscribed to `events` and nothing else. + fn default() -> Self { + let mut state = Self { + topics: TopicSet::EMPTY, + live: SubscriptionTable::default(), + dormant: DormantConfigs::default(), + }; + state.admit(TopicId::Events, None); + state } } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -pub(super) struct ChannelSet(u16); - -impl ChannelSet { - pub(super) const fn contains(self, channel: WsChannel) -> bool { - self.0 & channel.bit() != 0 +/// Config a client set for a topic it is not currently subscribed to. +/// +/// Kept apart from the live subscription table on purpose: that table +/// means "subscribed", and a config that outlives its subscription would +/// make it lie. Keyed the same way, so the two halves stay swappable as +/// keyed topics arrive. +#[derive(Debug, Clone, Default)] +struct DormantConfigs(BTreeMap<(u32, Option), serde_json::Value>); + +impl DormantConfigs { + fn get(&self, bit: u32, key: Option<&str>) -> Option<&serde_json::Value> { + self.0.get(&(bit, key.map(str::to_owned))) } - pub(super) fn insert(&mut self, channel: WsChannel) { - self.0 |= channel.bit(); + fn insert(&mut self, bit: u32, key: Option, config: serde_json::Value) { + self.0.insert((bit, key), config); } - pub(super) fn remove(&mut self, channel: WsChannel) { - self.0 &= !channel.bit(); + fn take(&mut self, bit: u32, key: Option<&str>) -> Option { + self.0.remove(&(bit, key.map(str::to_owned))) } +} - pub(super) fn iter(self) -> impl Iterator { - WsChannel::SUPPORTED - .into_iter() - .filter(move |channel| self.contains(*channel)) +impl SubscriptionState { + pub(super) const fn topics(&self) -> TopicSet { + self.topics } - pub(super) fn from_channels(channels: &[WsChannel]) -> Self { - let mut set = Self::default(); - for channel in channels { - set.insert(*channel); - } - set + pub(super) const fn contains(&self, topic: TopicId) -> bool { + self.topics.contains(topic) } -} -#[derive(Debug, Clone)] -pub(super) struct SubscriptionState { - pub(super) channels: ChannelSet, - pub(super) config: ChannelConfig, -} - -impl Default for SubscriptionState { - fn default() -> Self { - let mut channels = ChannelSet::default(); - channels.insert(WsChannel::Events); - Self { - channels, - config: ChannelConfig::default(), + /// This topic's config, live or dormant, or its default when the + /// client has never configured it. Dormant counts because the engine + /// reads across subscriptions: a `screen_zones`-only client still + /// borrows the `screen_canvas` cadence. + pub(super) fn config_of(&self, topic: TopicId) -> C + where + C: serde::de::DeserializeOwned + Default, + { + match self.stored_config(topic.bit(), None) { + // Borrowed, not cloned: relays re-read config on every frame + // they pace. + Some(stored) => C::deserialize(stored) + .expect("stored topic config round-trips through its own config type"), + None => C::default(), } } -} -#[derive(Debug, Clone, Serialize, Default)] -pub(super) struct ChannelConfig { - pub(super) frames: FramesConfig, - pub(super) spectrum: SpectrumConfig, - pub(super) canvas: CanvasConfig, - pub(super) screen_canvas: CanvasConfig, - pub(super) web_viewport_canvas: CanvasConfig, - pub(super) zone_preview: CanvasConfig, - pub(super) metrics: MetricsConfig, - pub(super) device_metrics: MetricsConfig, - pub(super) display_preview: DisplayPreviewConfig, -} - -impl ChannelConfig { - pub(super) fn apply_patch(&mut self, patch: ChannelConfigPatch) -> Result<(), WsProtocolError> { - let mut next = self.clone(); - next.apply_patch_inner(patch)?; - *self = next; - Ok(()) + fn stored_config(&self, bit: u32, key: Option<&str>) -> Option<&serde_json::Value> { + self.live + .config(bit, key) + .or_else(|| self.dormant.get(bit, key)) } - fn apply_patch_inner(&mut self, patch: ChannelConfigPatch) -> Result<(), WsProtocolError> { - if let Some(frames) = patch.frames { - if let Some(fps) = frames.fps { - validate_range(fps, 1, 60, "config.frames.fps", "expected 1..=60")?; - self.frames.fps = fps; - } - if let Some(format) = frames.format { - self.frames.format = format; - } - if let Some(zones) = frames.zones { - if zones.is_empty() { - return Err(WsProtocolError::invalid_config( - "config.frames.zones", - "must not be empty", - )); - } - self.frames.zones = zones; + /// The config stanza the subscribe acknowledgment echoes: every + /// live subscription that has config, in declaration order. + pub(super) fn config_projection(&self) -> serde_json::Value { + let mut map = serde_json::Map::new(); + for topic in self.topics.iter() { + for (_key, config) in self.live.entries_for(topic.bit()) { + map.insert(topic.as_str().to_owned(), config.clone()); } } + serde_json::Value::Object(map) + } - if let Some(spectrum) = patch.spectrum { - if let Some(fps) = spectrum.fps { - validate_range(fps, 1, 60, "config.spectrum.fps", "expected 1..=60")?; - self.spectrum.fps = fps; - } - if let Some(bins) = spectrum.bins { - if ![8, 16, 32, 64, 128].contains(&bins) { - return Err(WsProtocolError::invalid_config( - "config.spectrum.bins", - "expected one of [8, 16, 32, 64, 128]", - )); - } - self.spectrum.bins = bins; - } + /// Build the state a subscribe request would produce. + /// + /// The whole request is one transaction: every selector joins, every + /// config stanza applies, and every runtime admission runs against a + /// candidate copy. Any failure returns the error with the live state + /// untouched, so a request that names four topics and mis-configures + /// the fourth changes nothing. + pub(super) fn subscribe( + &self, + selections: &[TopicSelection], + patch: Option<&serde_json::Map>, + ) -> Result { + let mut next = self.clone(); + for selection in selections { + next.admit(selection.topic, selection.key.clone()); } - if let Some(canvas) = patch.canvas { - if let Some(fps) = canvas.fps { - validate_range(fps, 1, 60, "config.canvas.fps", "expected 1..=60")?; - self.canvas.fps = fps; - } - if let Some(format) = canvas.format { - self.canvas.format = format; - } - if let Some(width) = canvas.width { - self.canvas.width = width; - } - if let Some(height) = canvas.height { - self.canvas.height = height; + if let Some(patch) = patch { + // Declaration order, so a request carrying two bad stanzas + // always reports the same one. + for topic in TopicId::ALL.iter().copied() { + let Some(stanza) = patch.get(topic.as_str()) else { + continue; + }; + // A null stanza on a topic that takes config has always + // meant "no patch" on this wire, and clients still send + // it. Configless topics keep going to the vtable, which + // refuses null on apply. + if stanza.is_null() && topic.vtable().configurable { + continue; + } + next.apply_patch(topic, stanza)?; } - validate_passive_preview_shape(&self.canvas, "config.canvas")?; } - if let Some(screen_canvas) = patch.screen_canvas { - if let Some(fps) = screen_canvas.fps { - validate_range(fps, 1, 60, "config.screen_canvas.fps", "expected 1..=60")?; - self.screen_canvas.fps = fps; - } - if let Some(format) = screen_canvas.format { - self.screen_canvas.format = format; - } - if let Some(width) = screen_canvas.width { - self.screen_canvas.width = width; - } - if let Some(height) = screen_canvas.height { - self.screen_canvas.height = height; - } - validate_passive_preview_shape(&self.screen_canvas, "config.screen_canvas")?; - } + Ok(next) + } - if let Some(web_viewport_canvas) = patch.web_viewport_canvas { - if let Some(fps) = web_viewport_canvas.fps { - validate_range( - fps, - 1, - 60, - "config.web_viewport_canvas.fps", - "expected 1..=60", - )?; - self.web_viewport_canvas.fps = fps; - } - if let Some(format) = web_viewport_canvas.format { - self.web_viewport_canvas.format = format; - } - if let Some(width) = web_viewport_canvas.width { - self.web_viewport_canvas.width = width; - } - if let Some(height) = web_viewport_canvas.height { - self.web_viewport_canvas.height = height; - } - validate_passive_preview_shape( - &self.web_viewport_canvas, - "config.web_viewport_canvas", - )?; + /// Build the state an unsubscribe request would produce. Stored + /// config moves aside rather than dying, so a later re-subscribe + /// reinstates it. + pub(super) fn unsubscribe(&self, selections: &[TopicSelection]) -> Self { + let mut next = self.clone(); + for selection in selections { + next.retire(selection.topic); } + next + } - if let Some(zone_preview) = patch.zone_preview { - if let Some(fps) = zone_preview.fps { - validate_range(fps, 1, 60, "config.zone_preview.fps", "expected 1..=60")?; - self.zone_preview.fps = fps; - } - if let Some(format) = zone_preview.format { - self.zone_preview.format = format; - } - if let Some(width) = zone_preview.width { - self.zone_preview.width = width; - } - if let Some(height) = zone_preview.height { - self.zone_preview.height = height; - } - validate_passive_preview_shape(&self.zone_preview, "config.zone_preview")?; + /// The single write path for joining: the set gains the topic and + /// the live table gains its config in the same step. + fn admit(&mut self, topic: TopicId, key: Option) { + self.topics.insert(topic); + if !topic.vtable().configurable { + return; } - - if let Some(metrics) = patch.metrics - && let Some(interval_ms) = metrics.interval_ms - { - validate_range( - interval_ms, - 100, - 10_000, - "config.metrics.interval_ms", - "expected 100..=10000", - )?; - self.metrics.interval_ms = interval_ms; + let bit = topic.bit(); + if self.live.config(bit, key.as_deref()).is_some() { + return; } + let config = self + .dormant + .take(bit, key.as_deref()) + .unwrap_or_else(|| (topic.vtable().default_config_json)()); + self.live.insert(bit, key, config); + } - if let Some(device_metrics) = patch.device_metrics - && let Some(interval_ms) = device_metrics.interval_ms - { - validate_range( - interval_ms, - 100, - 10_000, - "config.device_metrics.interval_ms", - "expected 100..=10000", - )?; - self.device_metrics.interval_ms = interval_ms; + /// The single write path for leaving: the set loses the topic and + /// its live config moves to the dormant cache in the same step. + fn retire(&mut self, topic: TopicId) { + self.topics.remove(topic); + let bit = topic.bit(); + let carried: Vec<(Option, serde_json::Value)> = self + .live + .entries_for(bit) + .map(|(key, config)| (key.map(str::to_owned), config.clone())) + .collect(); + for (key, config) in carried { + self.live.remove(bit, key.as_deref()); + self.dormant.insert(bit, key, config); } + } - if let Some(display_preview) = patch.display_preview { - // Double-Option: outer `Some` means the client sent the key; - // inner `None` explicitly clears the target (disabling the - // relay). Trim non-empty strings so accidental whitespace - // doesn't sneak a subscription through with no real device. - if let Some(device_id) = display_preview.device_id { - match device_id { - Some(id) => { - let trimmed = id.trim(); - if trimmed.is_empty() { - return Err(WsProtocolError::invalid_config( - "config.display_preview.device_id", - "must be non-empty when provided", - )); - } - self.display_preview.device_id = Some(trimmed.to_owned()); - } - None => self.display_preview.device_id = None, - } - } - if let Some(fps) = display_preview.fps { - validate_range(fps, 1, 30, "config.display_preview.fps", "expected 1..=30")?; - self.display_preview.fps = fps; - } + fn apply_patch( + &mut self, + topic: TopicId, + stanza: &serde_json::Value, + ) -> Result<(), WsProtocolError> { + let bit = topic.bit(); + let current = self + .stored_config(bit, None) + .cloned() + .unwrap_or_else(|| (topic.vtable().default_config_json)()); + let next = (topic.vtable().apply_patch_json)(¤t, stanza) + .map_err(|error| config_patch_error(topic, &error))?; + super::topics::admit_config(topic, &next)?; + // Membership decides which half owns the result, so configuring + // a topic the request never named cannot fake a subscription. + if self.topics.contains(topic) { + // For an unkeyed topic a set bit implies a live entry, so + // this insert only ever replaces. Keyed topics (3.2c) must + // patch per admitted key, or a patch would mint a live + // entry for a key that never subscribed. + debug_assert!( + self.live.config(bit, None).is_some(), + "patch target must already be live for its key" + ); + self.live.insert(bit, None, next); + } else { + self.dormant.insert(bit, None, next); } - Ok(()) } +} - pub(super) fn filtered_json(&self, channels: ChannelSet) -> serde_json::Value { - let mut map = serde_json::Map::new(); +#[cfg(test)] +impl SubscriptionState { + /// Whether the live table still means what its name says: a topic + /// that takes config has a live entry exactly when it is subscribed. + pub(super) fn live_table_agrees_with_membership(&self) -> bool { + TopicId::ALL.iter().copied().all(|topic| { + let live = self.live.any_for(topic.bit()); + live == (self.topics.contains(topic) && topic.vtable().configurable) + }) + } - for channel in channels.iter() { - let value = match channel { - WsChannel::Frames => serde_json::to_value(&self.frames), - WsChannel::Spectrum => serde_json::to_value(&self.spectrum), - WsChannel::Canvas => serde_json::to_value(&self.canvas), - WsChannel::ScreenCanvas => serde_json::to_value(&self.screen_canvas), - WsChannel::WebViewportCanvas => serde_json::to_value(&self.web_viewport_canvas), - WsChannel::ZonePreview => serde_json::to_value(&self.zone_preview), - WsChannel::Metrics => serde_json::to_value(&self.metrics), - WsChannel::DeviceMetrics => serde_json::to_value(&self.device_metrics), - WsChannel::DisplayPreview => serde_json::to_value(&self.display_preview), - WsChannel::Events - | WsChannel::FrameEvents - | WsChannel::Sensors - | WsChannel::ScreenZones - | WsChannel::InputEvents => continue, - }; - - if let Ok(json_value) = value { - map.insert(channel.as_str().to_owned(), json_value); - } - } + /// Whether this topic's config is parked for a later re-subscribe. + pub(super) fn has_dormant_config(&self, topic: TopicId) -> bool { + self.dormant.get(topic.bit(), None).is_some() + } - serde_json::Value::Object(map) + /// Drive one subscribe request the way the wire drives it: channel + /// names in, the same parse, transaction, and admission out. + pub(super) fn subscribed( + &self, + channels: &[&str], + config: serde_json::Value, + ) -> Result { + let names: Vec = channels.iter().map(|name| (*name).to_owned()).collect(); + let selections = parse_channels(&names)?; + self.subscribe(&selections, config.as_object()) + } + + /// Drive one unsubscribe request the same way. + pub(super) fn unsubscribed(&self, channels: &[&str]) -> Self { + let names: Vec = channels.iter().map(|name| (*name).to_owned()).collect(); + let selections = parse_channels(&names).expect("test channel names parse"); + self.unsubscribe(&selections) } } -#[derive(Debug, Clone, Serialize)] -pub(super) struct FramesConfig { - pub(super) fps: u32, - pub(super) format: FrameFormat, - pub(super) zones: Vec, +/// Project a rejected patch onto the wire's error vocabulary. +/// +/// A configless topic refuses config in two phases — a stanza with +/// fields fails to deserialize, an explicit `null` fails to apply — and +/// both are the same client mistake, so both get the same response. +/// Field-level rejections name the field under its topic; whole-value +/// rejections name the topic alone. +fn config_patch_error(topic: TopicId, error: &PatchError) -> WsProtocolError { + if !topic.vtable().configurable { + return WsProtocolError::invalid_config( + format!("config.{}", topic.as_str()), + "topic accepts no config", + ); + } + + let field = match error.field { + "config" | "patch" => format!("config.{}", topic.as_str()), + field => format!("config.{}.{field}", topic.as_str()), + }; + WsProtocolError::invalid_config(field, error.reason.clone()) } #[derive(Debug, Clone)] @@ -461,50 +370,9 @@ impl ActiveFramesConfig { } } -impl Default for FramesConfig { - fn default() -> Self { - Self { - fps: 30, - format: FrameFormat::Binary, - zones: vec!["all".to_owned()], - } - } -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct SpectrumConfig { - pub(super) fps: u32, - pub(super) bins: u16, -} - -impl Default for SpectrumConfig { - fn default() -> Self { - Self { fps: 30, bins: 64 } - } -} - -#[derive(Debug, Clone, Serialize)] -pub(super) struct CanvasConfig { - pub(super) fps: u32, - pub(super) format: CanvasFormat, - pub(super) width: u32, - pub(super) height: u32, -} - -impl Default for CanvasConfig { - fn default() -> Self { - Self { - fps: 15, - format: CanvasFormat::Rgb, - width: 0, - height: 0, - } - } -} - -fn validate_passive_preview_shape( +pub(super) fn validate_passive_preview_shape( config: &CanvasConfig, - field: &'static str, + field: impl Into, ) -> Result<(), WsProtocolError> { if config.width == 0 || config.height == 0 { return Ok(()); @@ -516,52 +384,6 @@ fn validate_passive_preview_shape( }) } -#[derive(Debug, Clone, Serialize)] -pub(super) struct MetricsConfig { - pub(super) interval_ms: u32, -} - -impl Default for MetricsConfig { - fn default() -> Self { - Self { interval_ms: 1000 } - } -} - -/// Configuration for the per-display preview channel. `device_id` is -/// `None` until the client sends its first subscribe with a target — -/// once set, the relay task follows that device's JPEG frame watch and -/// streams every new frame out as a binary `0x07` payload. -#[derive(Debug, Clone, Serialize)] -pub(super) struct DisplayPreviewConfig { - #[serde(skip_serializing_if = "Option::is_none")] - pub(super) device_id: Option, - pub(super) fps: u32, -} - -impl Default for DisplayPreviewConfig { - fn default() -> Self { - Self { - device_id: None, - fps: 15, - } - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub(super) enum FrameFormat { - Binary, - Json, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub(super) enum CanvasFormat { - Rgb, - Rgba, - Jpeg, -} - #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub(super) enum InteractivePreviewTarget { @@ -595,10 +417,14 @@ pub(super) const MAX_INPUT_WHEEL_DELTA: i32 = 120 * 100; #[serde(tag = "type", rename_all = "snake_case")] pub(super) enum ClientMessage { /// Subscribe to one or more channels. + /// + /// `config` stays a raw object here: each stanza is validated by the + /// topic that owns it, through the registry vtable, so this message + /// does not need a field per topic. Subscribe { channels: Vec, #[serde(default)] - config: Option, + config: Option, #[serde(default)] preview_transport: Option, }, @@ -658,6 +484,59 @@ pub(super) enum ClientMessage { }, } +/// The `config` object of a subscribe: one stanza per channel, keyed by +/// wire name, with each stanza left as raw JSON for the topic that owns +/// it to validate. +/// +/// Naming a channel twice is refused rather than resolved. A client that +/// sends two stanzas for one channel does not agree with itself about +/// which config should win, and silently keeping the last one hides that +/// from the only party who can fix it. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub(super) struct ConfigStanzas(serde_json::Map); + +impl ConfigStanzas { + pub(super) const fn stanzas(&self) -> &serde_json::Map { + &self.0 + } +} + +impl<'de> Deserialize<'de> for ConfigStanzas { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct StanzaVisitor; + + impl<'de> Visitor<'de> for StanzaVisitor { + type Value = serde_json::Map; + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("an object of per-channel config stanzas") + } + + fn visit_map(self, mut access: A) -> Result + where + A: MapAccess<'de>, + { + let mut stanzas = serde_json::Map::new(); + while let Some(channel) = access.next_key::()? { + let stanza = access.next_value::()?; + if stanzas.contains_key(&channel) { + return Err(de::Error::custom(format_args!( + "duplicate config stanza `{channel}`" + ))); + } + stanzas.insert(channel, stanza); + } + Ok(stanzas) + } + } + + deserializer.deserialize_map(StanzaVisitor).map(Self) + } +} + /// Wire form of one injected input edge from a browser preview. #[derive(Debug, Clone, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] @@ -960,105 +839,6 @@ where } } -#[derive(Debug, Deserialize, Default)] -pub(super) struct ChannelConfigPatch { - #[serde(default)] - pub(super) frames: Option, - #[serde(default)] - pub(super) spectrum: Option, - #[serde(default)] - pub(super) canvas: Option, - #[serde(default)] - pub(super) screen_canvas: Option, - #[serde(default)] - pub(super) web_viewport_canvas: Option, - #[serde(default)] - pub(super) zone_preview: Option, - #[serde(default)] - pub(super) metrics: Option, - #[serde(default)] - pub(super) device_metrics: Option, - #[serde(default)] - pub(super) display_preview: Option, -} - -#[derive(Debug, Deserialize)] -pub(super) struct FramesConfigPatch { - #[serde(default)] - pub(super) fps: Option, - #[serde(default)] - pub(super) format: Option, - #[serde(default)] - pub(super) zones: Option>, -} - -#[derive(Debug, Deserialize)] -pub(super) struct SpectrumConfigPatch { - #[serde(default)] - pub(super) fps: Option, - #[serde(default)] - pub(super) bins: Option, -} - -#[derive(Debug, Deserialize)] -pub(super) struct CanvasConfigPatch { - #[serde(default)] - pub(super) fps: Option, - #[serde(default)] - pub(super) format: Option, - #[serde(default)] - pub(super) width: Option, - #[serde(default)] - pub(super) height: Option, -} - -#[derive(Debug, Deserialize)] -pub(super) struct MetricsConfigPatch { - #[serde(default)] - pub(super) interval_ms: Option, -} - -/// Patch for `DisplayPreviewConfig`. `device_id` uses a double-Option so -/// clients can distinguish "leave as-is" (`device_id: undefined`) from -/// "clear the target" (`device_id: null`). Setting the outer `Some(None)` -/// detaches the relay and stops emitting frames. -/// -/// The custom `deserialize_with` is required because plain -/// `Option>` with serde's default behavior collapses -/// `null` and missing-key to the same `None` — losing the tri-state we -/// need for "clear". -#[derive(Debug, Deserialize)] -pub(super) struct DisplayPreviewConfigPatch { - #[serde( - default, - deserialize_with = "deserialize_double_option_string", - skip_serializing_if = "Option::is_none" - )] - #[allow( - clippy::option_option, - reason = "the patch protocol needs distinct states for missing, null, and string values" - )] - pub(super) device_id: Option>, - #[serde(default)] - pub(super) fps: Option, -} - -/// Deserialize a double-Option so `null` maps to `Some(None)` (explicit -/// clear) and a missing key keeps the outer `None` (via `#[serde(default)]`). -/// Without this helper serde's default collapses both into `None`. -#[allow( - clippy::option_option, - reason = "serde needs the tri-state shape to preserve missing-vs-null during patch application" -)] -fn deserialize_double_option_string<'de, D>( - deserializer: D, -) -> Result>, D::Error> -where - D: serde::Deserializer<'de>, -{ - Option::::deserialize(deserializer).map(Some) -} - /// Server-to-client acknowledgment messages. #[derive(Debug, Serialize)] #[serde(tag = "type", rename_all = "snake_case")] @@ -1618,20 +1398,23 @@ impl WsProtocolError { } } - pub(super) fn invalid_config(field: &'static str, message: &'static str) -> Self { + pub(super) fn invalid_config(field: impl Into, reason: impl Into) -> Self { + let field = field.into(); + let reason = reason.into(); Self { code: "invalid_config", - message: format!("Invalid configuration for {field}: {message}"), - details: Some(json!({"field": field, "reason": message})), + message: format!("Invalid configuration for {field}: {reason}"), + details: Some(json!({"field": field, "reason": reason})), } } pub(super) fn invalid_config_resource( - field: &'static str, + field: impl Into, width: u32, height: u32, reason: String, ) -> Self { + let field = field.into(); Self { code: "invalid_config", message: format!("Invalid configuration for {field}: {reason}"), @@ -1645,14 +1428,6 @@ impl WsProtocolError { } } - pub(super) fn unsupported_channel(channel: &str) -> Self { - Self { - code: "unsupported_channel", - message: format!("Channel '{channel}' is not supported by this server"), - details: Some(json!({"channel": channel})), - } - } - pub(super) fn into_message(self) -> ServerMessage { ServerMessage::Error { code: self.code.to_owned(), @@ -1675,20 +1450,8 @@ pub(super) fn frame_selection_hash(selected: &[String]) -> u64 { hasher.finish() } -pub(super) fn validate_range( - value: u32, - min: u32, - max: u32, - field: &'static str, - message: &'static str, -) -> Result<(), WsProtocolError> { - if !(min..=max).contains(&value) { - return Err(WsProtocolError::invalid_config(field, message)); - } - Ok(()) -} - -pub(super) fn parse_channels(channels: &[String]) -> Result, WsProtocolError> { +/// Parse the wire's `channels` array into validated selectors. +pub(super) fn parse_channels(channels: &[String]) -> Result, WsProtocolError> { if channels.is_empty() { return Err(WsProtocolError::invalid_request( "channels must contain at least one channel", @@ -1697,37 +1460,43 @@ pub(super) fn parse_channels(channels: &[String]) -> Result, WsPr let mut parsed = Vec::with_capacity(channels.len()); for channel in channels { - let parsed_channel = WsChannel::parse(channel).ok_or_else(|| { + let topic = TopicId::parse(channel).ok_or_else(|| { WsProtocolError::invalid_request(format!("Unknown channel '{channel}'")) })?; - - if !parsed_channel.is_supported() { - return Err(WsProtocolError::unsupported_channel(channel)); - } - - parsed.push(parsed_channel); + // The key the topic's own key type accepts, canonicalized — the + // table stores what the boundary validated, never raw client text. + let key = (topic.vtable().validate_key)(None).map_err(|error| { + WsProtocolError::invalid_request(format!( + "Invalid key for channel '{channel}': {error}" + )) + })?; + parsed.push(TopicSelection { topic, key }); } Ok(parsed) } -pub(super) fn sorted_channel_names(channels: ChannelSet) -> Vec { - let mut names: Vec = channels +pub(super) fn sorted_channel_names(topics: TopicSet) -> Vec { + let mut names: Vec = topics .iter() - .map(|channel| channel.as_str().to_owned()) + .map(|topic| topic.as_str().to_owned()) .collect(); names.sort(); names } -pub(super) fn unique_sorted_channel_names(channels: &[WsChannel]) -> Vec { - sorted_channel_names(ChannelSet::from_channels(channels)) +pub(super) fn unique_sorted_channel_names(selections: &[TopicSelection]) -> Vec { + let mut topics = TopicSet::EMPTY; + for selection in selections { + topics.insert(selection.topic); + } + sorted_channel_names(topics) } pub(super) fn ws_capabilities() -> Vec { - let mut capabilities: Vec = WsChannel::SUPPORTED + let mut capabilities: Vec = TopicId::ALL .iter() - .map(|channel| channel.as_str().to_owned()) + .map(|topic| topic.as_str().to_owned()) .collect(); capabilities.push("commands".to_owned()); capabilities.push("canvas_format_jpeg".to_owned()); @@ -1794,13 +1563,13 @@ pub(super) fn to_snake_case(input: &str) -> String { pub(super) fn should_relay_event( event: &hypercolor_types::event::HypercolorEvent, - channels: ChannelSet, + topics: TopicSet, ) -> bool { if matches!( event, hypercolor_types::event::HypercolorEvent::FrameRendered { .. } ) { - return channels.contains(WsChannel::FrameEvents); + return topics.contains(TopicId::FrameEvents); } // Host input events carry keystroke data and never ride the default @@ -1810,8 +1579,8 @@ pub(super) fn should_relay_event( event, hypercolor_types::event::HypercolorEvent::InputEventReceived { .. } ) { - return channels.contains(WsChannel::InputEvents); + return topics.contains(TopicId::InputEvents); } - channels.contains(WsChannel::Events) + topics.contains(TopicId::Events) } diff --git a/crates/hypercolor-daemon/src/api/ws/relays.rs b/crates/hypercolor-daemon/src/api/ws/relays.rs index e36f929ab..da99fc03c 100644 --- a/crates/hypercolor-daemon/src/api/ws/relays.rs +++ b/crates/hypercolor-daemon/src/api/ws/relays.rs @@ -17,6 +17,10 @@ use hypercolor_core::bus::EventTimestamp; use hypercolor_core::device::usb_actor_metrics_snapshot; use hypercolor_core::engine::RenderLoopState; use hypercolor_core::input::BrowserInputPublicationId; +use hypercolor_leptos_ext::ws::registry::{ + CanvasConfig, CanvasFormat, DisplayPreviewConfig, FramesConfig, MetricsConfig, SpectrumConfig, + TopicId, +}; use hypercolor_leptos_ext::ws::{ InteractivePreviewFrame as WireInteractivePreviewFrame, PREVIEW_CHUNK_FIXED_HEADER_LEN, PreviewCancelFrame, PreviewChunkFrame, PreviewFrame as WirePreviewFrame, PreviewFrameChannel, @@ -40,11 +44,11 @@ use super::cache::{ try_encode_cached_canvas_preview_binary, try_encode_cached_zone_preview_binary_scaled, }; use super::protocol::{ - ActiveFramesConfig, CanvasConfig, MetricsCopies, MetricsDevices, MetricsDisplayLane, - MetricsDisplayOutput, MetricsEffectHealth, MetricsFps, MetricsFrameTime, MetricsMemory, - MetricsPacing, MetricsPayload, MetricsPreview, MetricsPreviewDemand, MetricsRenderSurfaces, - MetricsStages, MetricsTimeline, MetricsWebsocket, ServerMessage, SpectrumConfig, - SubscriptionState, WsChannel, event_message_parts, should_relay_event, + ActiveFramesConfig, MetricsCopies, MetricsDevices, MetricsDisplayLane, MetricsDisplayOutput, + MetricsEffectHealth, MetricsFps, MetricsFrameTime, MetricsMemory, MetricsPacing, + MetricsPayload, MetricsPreview, MetricsPreviewDemand, MetricsRenderSurfaces, MetricsStages, + MetricsTimeline, MetricsWebsocket, ServerMessage, SubscriptionState, event_message_parts, + should_relay_event, }; use crate::api::AppState; use crate::interactive_preview::PreviewResourceLease; @@ -437,6 +441,21 @@ pub(super) fn preview_outbound_channel_with_limits( } impl PreviewOutboundSender { + /// The capability this sender would agree on with `peer`, without + /// agreeing to it. Staging the answer lets a subscribe reject on + /// something else entirely with the transport still untouched. + pub(super) fn project_negotiation( + &self, + peer: PreviewTransportCapability, + ) -> Result { + let state = self + .shared + .state + .lock() + .unwrap_or_else(PoisonError::into_inner); + Self::negotiated_capability(&state, peer) + } + pub(super) fn negotiate_transport( &self, peer: PreviewTransportCapability, @@ -446,6 +465,21 @@ impl PreviewOutboundSender { .state .lock() .unwrap_or_else(PoisonError::into_inner); + // Checked and applied under one lock: either the whole + // capability swap lands or the transport is left as it was. + let negotiated = Self::negotiated_capability(&state, peer)?; + state.capability = negotiated; + state.limits = PreviewOutboundLimits { + max_publication_bytes: negotiated.max_encoded_publication_bytes, + max_connection_bytes: negotiated.max_connection_bytes, + }; + Ok(negotiated) + } + + fn negotiated_capability( + state: &PreviewOutboundState, + peer: PreviewTransportCapability, + ) -> Result { if !state.current.is_empty() || !state.queued.is_empty() || !state.in_flight.is_empty() @@ -463,13 +497,7 @@ impl PreviewOutboundSender { .min(state.limits.max_connection_bytes), ..supported }; - let negotiated = local.negotiated_with(peer); - state.capability = negotiated; - state.limits = PreviewOutboundLimits { - max_publication_bytes: negotiated.max_encoded_publication_bytes, - max_connection_bytes: negotiated.max_connection_bytes, - }; - Ok(negotiated) + Ok(local.negotiated_with(peer)) } pub(super) fn publish( @@ -687,7 +715,7 @@ impl PreviewOutboundSender { Ok(true) } - pub(super) fn cancel_channel(&self, channel: WsChannel) -> Result { + pub(super) fn cancel_topic(&self, topic: TopicId) -> Result { let mut state = self .shared .state @@ -703,7 +731,7 @@ impl PreviewOutboundSender { state .current .keys() - .filter(|stream| preview_stream_matches_channel(stream, channel)) + .filter(|stream| preview_stream_matches_topic(stream, topic)) .cloned(), ); let additional = streams @@ -742,22 +770,22 @@ pub(super) enum PreviewOutboundItem { Cancellation(PreviewCancelFrame), } -const fn preview_stream_matches_channel(stream: &PreviewStreamId, channel: WsChannel) -> bool { - match (stream, channel) { - (PreviewStreamId::Passive(frame_channel), WsChannel::Canvas) => { +const fn preview_stream_matches_topic(stream: &PreviewStreamId, topic: TopicId) -> bool { + match (stream, topic) { + (PreviewStreamId::Passive(frame_channel), TopicId::Canvas) => { matches!(frame_channel, PreviewFrameChannel::Canvas) } - (PreviewStreamId::Passive(frame_channel), WsChannel::ScreenCanvas) => { + (PreviewStreamId::Passive(frame_channel), TopicId::ScreenCanvas) => { matches!(frame_channel, PreviewFrameChannel::ScreenCanvas) } - (PreviewStreamId::Passive(frame_channel), WsChannel::WebViewportCanvas) => { + (PreviewStreamId::Passive(frame_channel), TopicId::WebViewportCanvas) => { matches!(frame_channel, PreviewFrameChannel::WebViewportCanvas) } - (PreviewStreamId::Passive(frame_channel), WsChannel::DisplayPreview) => { + (PreviewStreamId::Passive(frame_channel), TopicId::DisplayPreview) => { matches!(frame_channel, PreviewFrameChannel::DisplayPreview) } - (PreviewStreamId::ScreenZones, WsChannel::ScreenZones) - | (PreviewStreamId::Zone { .. }, WsChannel::ZonePreview) => true, + (PreviewStreamId::ScreenZones, TopicId::ScreenZones) + | (PreviewStreamId::Zone { .. }, TopicId::ZonePreview) => true, _ => false, } } @@ -999,16 +1027,23 @@ impl PreviewCursorQueue { } } - pub(super) fn set_max_streams( - &mut self, - max_streams: usize, + /// Whether the queue's live cursors fit inside `capability`. + pub(super) fn check_capability( + &self, + capability: PreviewTransportCapability, ) -> Result<(), PreviewOutboundError> { - if self.cursors.len() > max_streams { - return Err(PreviewOutboundError::StreamBudgetExceeded { - maximum: max_streams, + if capability.version == PreviewTransportVersion::V1 { + if self.cursors.len() > capability.max_streams { + return Err(PreviewOutboundError::StreamBudgetExceeded { + maximum: capability.max_streams, + }); + } + } else if self.state_bytes > capability.max_cursor_state_bytes { + return Err(PreviewOutboundError::CursorStateBudgetExceeded { + maximum: capability.max_cursor_state_bytes, + actual: self.state_bytes, }); } - self.max_streams = max_streams; Ok(()) } @@ -1016,14 +1051,7 @@ impl PreviewCursorQueue { &mut self, capability: PreviewTransportCapability, ) -> Result<(), PreviewOutboundError> { - if capability.version == PreviewTransportVersion::V1 { - self.set_max_streams(capability.max_streams)?; - } else if self.state_bytes > capability.max_cursor_state_bytes { - return Err(PreviewOutboundError::CursorStateBudgetExceeded { - maximum: capability.max_cursor_state_bytes, - actual: self.state_bytes, - }); - } + self.check_capability(capability)?; self.max_streams = capability.max_streams; self.max_state_bytes = capability.max_cursor_state_bytes; self.version = capability.version; @@ -1359,7 +1387,7 @@ pub(super) async fn relay_events( Ok(timestamped) => { let should_relay = { let subs = subscriptions.borrow(); - should_relay_event(×tamped.event, subs.channels) + should_relay_event(×tamped.event, subs.topics()) }; if !should_relay { continue; @@ -1381,7 +1409,7 @@ pub(super) async fn relay_events( } Err(broadcast::error::RecvError::Lagged(n)) => { warn!("WebSocket consumer lagged by {n} events"); - if subscriptions.borrow().channels.contains(WsChannel::Events) { + if subscriptions.borrow().contains(TopicId::Events) { let msg = ServerMessage::Event { event: "resync_required".to_owned(), timestamp: EventTimestamp::now().to_string(), @@ -1419,8 +1447,10 @@ pub(super) async fn relay_frames( if active_frame_config.is_none() { active_frame_config = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::Frames) { - Some(ActiveFramesConfig::new(subs.config.frames.clone())) + if subs.contains(TopicId::Frames) { + Some(ActiveFramesConfig::new( + subs.config_of::(TopicId::Frames), + )) } else { None } @@ -1506,8 +1536,8 @@ pub(super) async fn relay_spectrum( if active_spectrum_config.is_none() { active_spectrum_config = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::Spectrum) { - Some(subs.config.spectrum.clone()) + if subs.contains(TopicId::Spectrum) { + Some(subs.config_of::(TopicId::Spectrum)) } else { None } @@ -1591,8 +1621,8 @@ pub(super) async fn relay_canvas( if active_canvas_config.is_none() { active_canvas_config = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::Canvas) { - Some(subs.config.canvas.clone()) + if subs.contains(TopicId::Canvas) { + Some(subs.config_of::(TopicId::Canvas)) } else { None } @@ -1727,8 +1757,8 @@ pub(super) async fn relay_screen_canvas( if active_canvas_config.is_none() { active_canvas_config = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::ScreenCanvas) { - Some(subs.config.screen_canvas.clone()) + if subs.contains(TopicId::ScreenCanvas) { + Some(subs.config_of::(TopicId::ScreenCanvas)) } else { None } @@ -1838,10 +1868,7 @@ pub(super) async fn relay_screen_zones( let mut zones_rx = None::>; loop { - let subscribed = subscriptions - .borrow() - .channels - .contains(WsChannel::ScreenZones); + let subscribed = subscriptions.borrow().contains(TopicId::ScreenZones); if subscribed && zones_rx.is_none() { let mut receiver = preview_runtime.screen_zones_receiver(); receiver.mark_changed(); @@ -1938,8 +1965,8 @@ pub(super) async fn relay_web_viewport_canvas( if active_canvas_config.is_none() { active_canvas_config = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::WebViewportCanvas) { - Some(subs.config.web_viewport_canvas.clone()) + if subs.contains(TopicId::WebViewportCanvas) { + Some(subs.config_of::(TopicId::WebViewportCanvas)) } else { None } @@ -2053,8 +2080,8 @@ pub(super) async fn relay_zone_preview( if active_canvas_config.is_none() { active_canvas_config = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::ZonePreview) { - Some(subs.config.zone_preview.clone()) + if subs.contains(TopicId::ZonePreview) { + Some(subs.config_of::(TopicId::ZonePreview)) } else { None } @@ -2216,13 +2243,13 @@ pub(super) async fn relay_display_preview( // display worker config changes also close and recreate the sender. let desired = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::DisplayPreview) { - subs.config - .display_preview + if subs.contains(TopicId::DisplayPreview) { + let config = subs.config_of::(TopicId::DisplayPreview); + config .device_id .as_ref() .and_then(|raw| DeviceId::from_str(raw).ok()) - .map(|id| (id, subs.config.display_preview.fps.max(1))) + .map(|id| (id, config.fps.max(1))) } else { None } @@ -2347,9 +2374,9 @@ fn preview_stream_demand(config: &CanvasConfig) -> PreviewStreamDemand { PreviewStreamDemand { fps: config.fps, format: match config.format { - super::protocol::CanvasFormat::Rgb => PreviewPixelFormat::Rgb, - super::protocol::CanvasFormat::Rgba => PreviewPixelFormat::Rgba, - super::protocol::CanvasFormat::Jpeg => PreviewPixelFormat::Jpeg, + CanvasFormat::Rgb => PreviewPixelFormat::Rgb, + CanvasFormat::Rgba => PreviewPixelFormat::Rgba, + CanvasFormat::Jpeg => PreviewPixelFormat::Jpeg, }, width: config.width, height: config.height, @@ -2382,8 +2409,11 @@ pub(super) async fn relay_metrics( if active_interval_ms.is_none() { active_interval_ms = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::Metrics) { - Some(subs.config.metrics.interval_ms) + if subs.contains(TopicId::Metrics) { + Some( + subs.config_of::(TopicId::Metrics) + .interval_ms, + ) } else { None } @@ -2411,7 +2441,7 @@ pub(super) async fn relay_metrics( let still_subscribed = { let subs = subscriptions.borrow(); - subs.channels.contains(WsChannel::Metrics) + subs.contains(TopicId::Metrics) }; if !still_subscribed { continue; @@ -2447,8 +2477,11 @@ pub(super) async fn relay_device_metrics( if active_interval_ms.is_none() { active_interval_ms = { let subs = subscriptions.borrow(); - if subs.channels.contains(WsChannel::DeviceMetrics) { - Some(subs.config.device_metrics.interval_ms) + if subs.contains(TopicId::DeviceMetrics) { + Some( + subs.config_of::(TopicId::DeviceMetrics) + .interval_ms, + ) } else { None } @@ -2476,7 +2509,7 @@ pub(super) async fn relay_device_metrics( let still_subscribed = { let subs = subscriptions.borrow(); - subs.channels.contains(WsChannel::DeviceMetrics) + subs.contains(TopicId::DeviceMetrics) }; if !still_subscribed { continue; @@ -2499,7 +2532,7 @@ pub(super) async fn relay_sensors( let mut sent_current_snapshot = false; loop { - if !subscriptions.borrow().channels.contains(WsChannel::Sensors) { + if !subscriptions.borrow().contains(TopicId::Sensors) { sent_current_snapshot = false; if subscriptions.changed().await.is_err() { break; @@ -2541,7 +2574,7 @@ pub(super) async fn relay_sensors( continue; } - if subscriptions.borrow().channels.contains(WsChannel::Sensors) { + if subscriptions.borrow().contains(TopicId::Sensors) { let snapshot = Arc::clone(&rx.borrow_and_update()); enqueue_sensor_snapshot(&json_tx, snapshot.as_ref()); } diff --git a/crates/hypercolor-daemon/src/api/ws/session.rs b/crates/hypercolor-daemon/src/api/ws/session.rs index 21316f7e1..3dbb6cc25 100644 --- a/crates/hypercolor-daemon/src/api/ws/session.rs +++ b/crates/hypercolor-daemon/src/api/ws/session.rs @@ -16,6 +16,7 @@ use axum::http::{HeaderMap, HeaderValue, header}; use axum::response::{IntoResponse, Response}; use hypercolor_leptos_ext::axum::upgrade_handler; use hypercolor_leptos_ext::ws::PreviewTransportCapability; +use hypercolor_leptos_ext::ws::registry::{CanvasConfig, CanvasFormat, SpectrumConfig, TopicId}; use serde::Serialize; use serde_json::json; use tokio::sync::watch; @@ -44,18 +45,17 @@ use super::cache::{ use super::command::dispatch_command; use super::interactive_preview_relay::spawn_interactive_preview_relay; use super::protocol::{ - BrowserInputEdgeWire, ClientMessage, HelloFps, HelloState, InteractivePreviewConfig, - MAX_WS_MESSAGE_BYTES, NameRef, SceneRef, ServerMessage, SubscriptionState, WsChannel, - WsProtocolError, parse_channels, sorted_channel_names, unique_sorted_channel_names, - validate_interactive_preview_shape, ws_capabilities, + BrowserInputEdgeWire, ClientMessage, ConfigStanzas, HelloFps, HelloState, + InteractivePreviewConfig, MAX_WS_MESSAGE_BYTES, NameRef, SceneRef, ServerMessage, + SubscriptionState, TopicSelection, WsProtocolError, parse_channels, sorted_channel_names, + unique_sorted_channel_names, validate_interactive_preview_shape, ws_capabilities, }; use super::relays::{ PreviewCursorQueue, PreviewOutboundItem, PreviewOutboundSender, PreviewSendCursor, WS_PREVIEW_CHUNK_SENT_COUNT, WS_PREVIEW_PUBLICATION_SENT_COUNT, preview_outbound_channel, - publish_subscriptions, relay_canvas, relay_device_metrics, relay_display_preview, relay_events, - relay_frames, relay_metrics, relay_screen_canvas, relay_screen_zones, relay_sensors, - relay_spectrum, relay_web_viewport_canvas, relay_zone_preview, + publish_subscriptions, }; +use super::topics::{RelayContext, spawn_relays}; use crate::api::AppState; use crate::api::effects::active_primary_effect; use crate::api::layouts::validate_layout_sampling_radii; @@ -266,15 +266,13 @@ async fn handle_socket( server: state.server_identity.clone(), state: build_hello_state(&state).await, capabilities: ws_capabilities(), - subscriptions: sorted_channel_names(subscriptions.channels), + subscriptions: sorted_channel_names(subscriptions.topics()), } }; if send_json(&mut socket, &hello).await.is_err() { return; } - // Subscribe to the event bus and watch channels. - let event_rx = state.event_bus.subscribe_all(); // JSON and small binary telemetry stay count-bounded. Preview surfaces use // a keyed, byte-accounted latest-value router below. let (json_tx, mut json_rx) = tokio::sync::mpsc::channel::(WS_BUFFER_SIZE); @@ -287,72 +285,16 @@ async fn handle_socket( preview_tx.clone(), ); - // Spawn event relay tasks — each watches immutable subscription snapshots. - let relay_handle = tokio::spawn(relay_events( - event_rx, - json_tx.clone(), - subscriptions_rx.clone(), - )); - let frame_relay_handle = tokio::spawn(relay_frames( - Arc::clone(&state), - json_tx.clone(), - binary_tx.clone(), - subscriptions_rx.clone(), - )); - let spectrum_relay_handle = tokio::spawn(relay_spectrum( - Arc::clone(&state), - json_tx.clone(), - binary_tx.clone(), - subscriptions_rx.clone(), - )); - let canvas_power_rx = state.power_state.subscribe(); - let canvas_relay_handle = tokio::spawn(relay_canvas( - Arc::clone(&state.preview_runtime), - canvas_power_rx, - preview_tx.clone(), - subscriptions_rx.clone(), - )); - let screen_canvas_relay_handle = tokio::spawn(relay_screen_canvas( - Arc::clone(&state.preview_runtime), - preview_tx.clone(), - subscriptions_rx.clone(), - )); - let screen_zones_relay_handle = tokio::spawn(relay_screen_zones( - Arc::clone(&state.preview_runtime), - subscriptions_rx.clone(), - preview_tx.clone(), - )); - let web_viewport_canvas_relay_handle = tokio::spawn(relay_web_viewport_canvas( - Arc::clone(&state.preview_runtime), - preview_tx.clone(), - subscriptions_rx.clone(), - )); - let zone_preview_relay_handle = tokio::spawn(relay_zone_preview( - Arc::clone(&state.preview_runtime), - preview_tx.clone(), - subscriptions_rx.clone(), - )); - let display_preview_relay_handle = tokio::spawn(relay_display_preview( - Arc::clone(&state), - Arc::clone(&state.display_frames), - preview_tx.clone(), - subscriptions_rx.clone(), - )); - let metrics_relay_handle = tokio::spawn(relay_metrics( - Arc::clone(&state), - json_tx.clone(), - subscriptions_rx.clone(), - )); - let device_metrics_relay_handle = tokio::spawn(relay_device_metrics( - Arc::clone(&state), - json_tx.clone(), - subscriptions_rx.clone(), - )); - let sensors_relay_handle = tokio::spawn(relay_sensors( - Arc::clone(&state), - json_tx.clone(), - subscriptions_rx.clone(), - )); + // Spawn every registered relay task — each watches immutable + // subscription snapshots, and the registry decides which topics a + // task serves. + let relay_handles = spawn_relays(&RelayContext { + state: Arc::clone(&state), + json_tx: json_tx.clone(), + binary_tx: binary_tx.clone(), + preview_tx: preview_tx.clone(), + subscriptions: subscriptions_rx.clone(), + }); let mut ping_interval = tokio::time::interval(WS_PING_INTERVAL); ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -550,21 +492,7 @@ async fn handle_socket( } } - abort_and_join_relays([ - relay_handle, - frame_relay_handle, - spectrum_relay_handle, - canvas_relay_handle, - screen_canvas_relay_handle, - screen_zones_relay_handle, - web_viewport_canvas_relay_handle, - display_preview_relay_handle, - zone_preview_relay_handle, - metrics_relay_handle, - device_metrics_relay_handle, - sensors_relay_handle, - ]) - .await; + abort_and_join_relays(relay_handles).await; browser_previews.shutdown().await; while let Some(cursor) = preview_cursors.pop_next() { preview_rx.complete(cursor.publication()); @@ -584,7 +512,7 @@ async fn wait_for_shutdown(shutdown: Option<&CancellationToken>) { } } -async fn abort_and_join_relays(handles: [JoinHandle<()>; N]) { +async fn abort_and_join_relays(handles: Vec>) { for handle in &handles { handle.abort(); } @@ -630,34 +558,38 @@ impl WsInputDemandLeases { } } - pub(super) fn synchronize( - &mut self, + /// Work out what engine demand a subscription state implies, + /// without registering any of it. Every rejection this projection + /// can raise happens before a single lease moves, so a subscribe + /// that fails here leaves the engine exactly as it was. + pub(super) fn project( + &self, subscriptions: &SubscriptionState, - ) -> Result<(), WsProtocolError> { - let screen_active = subscriptions.channels.contains(WsChannel::ScreenCanvas) - || subscriptions.channels.contains(WsChannel::ScreenZones); + ) -> Result { + let screen_canvas = subscriptions.config_of::(TopicId::ScreenCanvas); + let screen_active = subscriptions.contains(TopicId::ScreenCanvas) + || subscriptions.contains(TopicId::ScreenZones); let (screen_demand, screen_requested_extent) = if screen_active { - let requested_hz = - NonZeroU32::new(subscriptions.config.screen_canvas.fps).ok_or_else(|| { - WsProtocolError::invalid_config( - "config.screen_canvas.fps", - "expected a non-zero cadence", - ) - })?; + let requested_hz = NonZeroU32::new(screen_canvas.fps).ok_or_else(|| { + WsProtocolError::invalid_config( + "config.screen_canvas.fps", + "expected a non-zero cadence", + ) + })?; let mut branches = Vec::with_capacity(2); let mut requested_extent = None; - if subscriptions.channels.contains(WsChannel::ScreenCanvas) { + if subscriptions.contains(TopicId::ScreenCanvas) { let output = resolve_canvas_output_size( self.screen_base_extent.width(), self.screen_base_extent.height(), - subscriptions.config.screen_canvas.width, - subscriptions.config.screen_canvas.height, + screen_canvas.width, + screen_canvas.height, ) .map_err(|error| { WsProtocolError::invalid_config_resource( "config.screen_canvas", - subscriptions.config.screen_canvas.width, - subscriptions.config.screen_canvas.height, + screen_canvas.width, + screen_canvas.height, error.to_string(), ) })?; @@ -671,8 +603,8 @@ impl WsInputDemandLeases { ) })?; let extent_request = ScreenExtentRequest::bounded( - NonZeroU32::new(subscriptions.config.screen_canvas.width), - NonZeroU32::new(subscriptions.config.screen_canvas.height), + NonZeroU32::new(screen_canvas.width), + NonZeroU32::new(screen_canvas.height), ScreenUpscalePolicy::Never, ); branches.push(screen_branch_demand( @@ -683,7 +615,7 @@ impl WsInputDemandLeases { )); requested_extent = Some(canvas_extent); } - if subscriptions.channels.contains(WsChannel::ScreenZones) { + if subscriptions.contains(TopicId::ScreenZones) { let extent_request = ScreenExtentRequest::bounded( NonZeroU32::new(self.screen_base_extent.width()), NonZeroU32::new(self.screen_base_extent.height()), @@ -703,43 +635,60 @@ impl WsInputDemandLeases { let requested_extent = requested_extent.expect("an active screen subscription has an extent"); ( - InputPublicationDemand::default().with_screen_branches(branches), + Some(InputPublicationDemand::default().with_screen_branches(branches)), Some(requested_extent), ) } else { - (InputPublicationDemand::default(), None) + (None, None) }; - Self::synchronize_domain( - &self.demands, - &mut self.spectrum, - subscriptions.channels.contains(WsChannel::Spectrum), - InputPublicationDemand::default().with_source( - hypercolor_core::input::SourceKind::Audio, - subscriptions.config.spectrum.fps, - ), - ); - Self::synchronize_domain( - &self.demands, - &mut self.screen, - screen_active, - screen_demand, - ); - Self::synchronize_domain( - &self.demands, - &mut self.interaction, - subscriptions.channels.contains(WsChannel::InputEvents), - InputPublicationDemand::default().with_source( - hypercolor_core::input::SourceKind::Interaction, - self.interaction_hz, - ), - ); + // Only the tests read the resolved extent back; production + // reads it through the registered branches instead. + #[cfg(not(test))] + let _ = screen_requested_extent; + + Ok(ProjectedInputDemand { + spectrum: subscriptions.contains(TopicId::Spectrum).then(|| { + InputPublicationDemand::default().with_source( + hypercolor_core::input::SourceKind::Audio, + subscriptions + .config_of::(TopicId::Spectrum) + .fps, + ) + }), + screen: screen_demand, + interaction: subscriptions.contains(TopicId::InputEvents).then(|| { + InputPublicationDemand::default().with_source( + hypercolor_core::input::SourceKind::Interaction, + self.interaction_hz, + ) + }), + #[cfg(test)] + screen_requested_extent, + }) + } + + /// Register a projection. Infallible by construction: everything + /// that could refuse already did. + pub(super) fn commit(&mut self, projected: ProjectedInputDemand) { + Self::synchronize_domain(&self.demands, &mut self.spectrum, projected.spectrum); + Self::synchronize_domain(&self.demands, &mut self.screen, projected.screen); + Self::synchronize_domain(&self.demands, &mut self.interaction, projected.interaction); #[cfg(test)] { - self.screen_requested_extent = screen_requested_extent; + self.screen_requested_extent = projected.screen_requested_extent; } - #[cfg(not(test))] - let _ = screen_requested_extent; + } + + /// Project and commit in one step. Production splits the two so a + /// later rejection cannot strand a half-applied lease. + #[cfg(test)] + pub(super) fn synchronize( + &mut self, + subscriptions: &SubscriptionState, + ) -> Result<(), WsProtocolError> { + let projected = self.project(subscriptions)?; + self.commit(projected); Ok(()) } @@ -751,21 +700,30 @@ impl WsInputDemandLeases { fn synchronize_domain( demands: &InputPublicationDemandHandle, registration: &mut Option, - active: bool, - demand: InputPublicationDemand, + demand: Option, ) { - match (registration.as_ref(), active) { - (Some(registration), true) => registration.update(demand), - (None, true) => { + match (registration.as_ref(), demand) { + (Some(registration), Some(demand)) => registration.update(demand), + (None, Some(demand)) => { *registration = Some(demands.register(InputPublicationConsumer::PassiveStream, demand)); } - (Some(_), false) => *registration = None, - (None, false) => {} + (Some(_), None) => *registration = None, + (None, None) => {} } } } +/// The engine demand one subscription state implies, computed but not +/// yet registered. +pub(super) struct ProjectedInputDemand { + spectrum: Option, + screen: Option, + interaction: Option, + #[cfg(test)] + screen_requested_extent: Option, +} + fn screen_branch_demand( kind: ScreenPublicationKind, extent: ScreenExtentRequest, @@ -1086,9 +1044,9 @@ const fn runtime_preview_spec(config: InteractivePreviewConfig) -> RuntimeIntera width: config.width, height: config.height, format: match config.format { - super::protocol::CanvasFormat::Rgb => PreviewPixelFormat::Rgb, - super::protocol::CanvasFormat::Rgba => PreviewPixelFormat::Rgba, - super::protocol::CanvasFormat::Jpeg => PreviewPixelFormat::Jpeg, + CanvasFormat::Rgb => PreviewPixelFormat::Rgb, + CanvasFormat::Rgba => PreviewPixelFormat::Rgba, + CanvasFormat::Jpeg => PreviewPixelFormat::Jpeg, }, } } @@ -1128,17 +1086,17 @@ fn authoritative_claim_error(preview_id: &str, error: AuthoritativeClaimError) - pub(super) fn authorize_subscription_channels( auth_context: RequestAuthContext, - channels: &[WsChannel], + selections: &[TopicSelection], ) -> Result<(), WsProtocolError> { if auth_context.can_control() { return Ok(()); } - let restricted_channels: Vec<&'static str> = channels + let restricted_channels: Vec<&'static str> = selections .iter() - .copied() - .filter(|channel| channel.requires_control_subscription()) - .map(WsChannel::as_str) + .map(|selection| selection.topic) + .filter(|topic| topic.requires_control()) + .map(TopicId::as_str) .collect(); if restricted_channels.is_empty() { @@ -1151,18 +1109,49 @@ pub(super) fn authorize_subscription_channels( } } -pub(super) fn negotiate_preview_transport( +/// A transport capability both ends have agreed on but neither has +/// adopted yet. +pub(super) struct StagedPreviewTransport { + peer: PreviewTransportCapability, + negotiated: PreviewTransportCapability, +} + +/// Work out the capability this subscribe would settle on, touching +/// nothing. A subscribe that goes on to fail its demand projection must +/// leave the connection speaking exactly the transport it spoke before. +pub(super) fn stage_preview_transport( encoded_capability: &str, preview_outbound: &PreviewOutboundSender, - preview_cursors: &mut PreviewCursorQueue, - preview_capability: &mut PreviewTransportCapability, -) -> Result { + preview_cursors: &PreviewCursorQueue, +) -> Result { let peer = PreviewTransportCapability::decode(encoded_capability).map_err(|error| { WsProtocolError::invalid_request(format!("Invalid preview_transport capability: {error}")) })?; let negotiated = preview_outbound - .negotiate_transport(peer) + .project_negotiation(peer) + .map_err(|error| WsProtocolError::invalid_request(error.to_string()))?; + preview_cursors + .check_capability(negotiated) + .map_err(|error| WsProtocolError::invalid_request(error.to_string()))?; + Ok(StagedPreviewTransport { peer, negotiated }) +} + +/// Adopt a staged capability. The sender re-checks and swaps under one +/// lock, and the cursor queue already proved it fits, so this either +/// lands whole or refuses without changing anything. +pub(super) fn commit_preview_transport( + staged: StagedPreviewTransport, + preview_outbound: &PreviewOutboundSender, + preview_cursors: &mut PreviewCursorQueue, + preview_capability: &mut PreviewTransportCapability, +) -> Result { + let negotiated = preview_outbound + .negotiate_transport(staged.peer) .map_err(|error| WsProtocolError::invalid_request(error.to_string()))?; + debug_assert_eq!( + negotiated, staged.negotiated, + "committing a staged transport must settle on what staging projected" + ); preview_cursors .set_capability(negotiated) .map_err(|error| WsProtocolError::invalid_request(error.to_string()))?; @@ -1170,6 +1159,22 @@ pub(super) fn negotiate_preview_transport( Ok(negotiated) } +#[cfg(test)] +pub(super) fn negotiate_preview_transport( + encoded_capability: &str, + preview_outbound: &PreviewOutboundSender, + preview_cursors: &mut PreviewCursorQueue, + preview_capability: &mut PreviewTransportCapability, +) -> Result { + let staged = stage_preview_transport(encoded_capability, preview_outbound, preview_cursors)?; + commit_preview_transport( + staged, + preview_outbound, + preview_cursors, + preview_capability, + ) +} + /// Process a client subscription/unsubscription message. async fn handle_client_message( text: &str, @@ -1204,7 +1209,11 @@ async fn handle_client_message( config, preview_transport, } => { - let parsed_channels = match parse_channels(&channels) { + // Validate the whole request first. Every step below builds + // a candidate and refuses on its own terms, so a request + // that names four topics and mis-configures the fourth + // leaves the connection exactly as it was. + let selections = match parse_channels(&channels) { Ok(parsed) => parsed, Err(error) => { let _ = send_json(socket, &error.into_message()).await; @@ -1212,25 +1221,47 @@ async fn handle_client_message( } }; - if let Err(error) = authorize_subscription_channels(auth_context, &parsed_channels) { + if let Err(error) = authorize_subscription_channels(auth_context, &selections) { let _ = send_json(socket, &error.into_message()).await; return; } - let mut next_subscriptions = subscriptions.clone(); - if let Some(config_patch) = config - && let Err(error) = next_subscriptions.config.apply_patch(config_patch) + let next_subscriptions = match subscriptions + .subscribe(&selections, config.as_ref().map(ConfigStanzas::stanzas)) { - let _ = send_json(socket, &error.into_message()).await; - return; - } + Ok(next) => next, + Err(error) => { + let _ = send_json(socket, &error.into_message()).await; + return; + } + }; - for channel in &parsed_channels { - next_subscriptions.channels.insert(*channel); - } - if let Some(encoded_capability) = preview_transport - && let Err(error) = negotiate_preview_transport( - &encoded_capability, + let staged_transport = match preview_transport + .as_deref() + .map(|encoded| stage_preview_transport(encoded, preview_outbound, preview_cursors)) + .transpose() + { + Ok(staged) => staged, + Err(error) => { + let _ = send_json(socket, &error.into_message()).await; + return; + } + }; + + let projected_demand = match input_demand_leases.project(&next_subscriptions) { + Ok(projected) => projected, + Err(error) => { + let _ = send_json(socket, &error.into_message()).await; + return; + } + }; + + // Commit phase. The transport goes first because adopting it + // is the only step that can still refuse, and it refuses + // without having changed anything. + if let Some(staged) = staged_transport + && let Err(error) = commit_preview_transport( + staged, preview_outbound, preview_cursors, preview_capability, @@ -1239,22 +1270,19 @@ async fn handle_client_message( let _ = send_json(socket, &error.into_message()).await; return; } - if let Err(error) = input_demand_leases.synchronize(&next_subscriptions) { - let _ = send_json(socket, &error.into_message()).await; - return; - } + input_demand_leases.commit(projected_demand); *subscriptions = next_subscriptions; let ack = ServerMessage::Subscribed { - channels: unique_sorted_channel_names(&parsed_channels), - config: subscriptions.config.filtered_json(subscriptions.channels), + channels: unique_sorted_channel_names(&selections), + config: subscriptions.config_projection(), preview_transport: preview_capability.encode(), }; publish_subscriptions(subscriptions_tx, subscriptions); let _ = send_json(socket, &ack).await; } ClientMessage::Unsubscribe { channels } => { - let parsed_channels = match parse_channels(&channels) { + let selections = match parse_channels(&channels) { Ok(parsed) => parsed, Err(error) => { let _ = send_json(socket, &error.into_message()).await; @@ -1262,24 +1290,25 @@ async fn handle_client_message( } }; - let mut next_subscriptions = subscriptions.clone(); - for channel in &parsed_channels { - next_subscriptions.channels.remove(*channel); - } - if let Err(error) = input_demand_leases.synchronize(&next_subscriptions) { - let _ = send_json(socket, &error.into_message()).await; - return; - } + let next_subscriptions = subscriptions.unsubscribe(&selections); + let projected_demand = match input_demand_leases.project(&next_subscriptions) { + Ok(projected) => projected, + Err(error) => { + let _ = send_json(socket, &error.into_message()).await; + return; + } + }; + input_demand_leases.commit(projected_demand); *subscriptions = next_subscriptions; - for channel in &parsed_channels { - if let Err(error) = preview_outbound.cancel_channel(*channel) { - warn!(%error, channel = channel.as_str(), "Failed to cancel unsubscribed preview channel"); + for selection in &selections { + if let Err(error) = preview_outbound.cancel_topic(selection.topic) { + warn!(%error, channel = selection.topic.as_str(), "Failed to cancel unsubscribed preview channel"); } } - let remaining = sorted_channel_names(subscriptions.channels); + let remaining = sorted_channel_names(subscriptions.topics()); let ack = ServerMessage::Unsubscribed { - channels: unique_sorted_channel_names(&parsed_channels), + channels: unique_sorted_channel_names(&selections), remaining, }; publish_subscriptions(subscriptions_tx, subscriptions); diff --git a/crates/hypercolor-daemon/src/api/ws/tests.rs b/crates/hypercolor-daemon/src/api/ws/tests.rs index 3a4bd50dc..be2c86df9 100644 --- a/crates/hypercolor-daemon/src/api/ws/tests.rs +++ b/crates/hypercolor-daemon/src/api/ws/tests.rs @@ -18,6 +18,9 @@ use hypercolor_core::input::{ SourceKind, SourceSessionSlot, SourceStatusHandle, SourceStatusReporter, }; use hypercolor_core::scene::SceneManager; +use hypercolor_leptos_ext::ws::registry::{ + CanvasFormat, FrameFormat, FramesConfig, TopicId, TopicSet, +}; use hypercolor_leptos_ext::ws::{ InteractivePreviewFrame as WireInteractivePreviewFrame, PREVIEW_CHUNK_FRAME_TAG, PREVIEW_MIN_MESSAGE_BYTES, PreviewChunkFrame, PreviewFrame as WirePreviewFrame, @@ -61,13 +64,13 @@ use super::preview_encode::{ encode_canvas_jpeg_payload_scaled_stateless, }; use super::protocol::{ - ActiveFramesConfig, BrowserInputEdgeWire, CanvasFormat, ChannelConfig, ChannelConfigPatch, - ChannelSet, ClientMessage, FrameFormat, FrameZoneSelection, FramesConfig, InputButtonStateWire, - InteractivePreviewConfig, InteractivePreviewTarget, MAX_INPUT_INJECT_EVENTS, - MAX_INPUT_NAME_BYTES, MAX_INPUT_WHEEL_DELTA, MAX_PREVIEW_PUBLICATION_BYTES, ServerMessage, - SubscriptionState, WsChannel, deserialize_finite_coordinate, event_message_parts, - parse_channels, should_relay_event, to_snake_case, unique_sorted_channel_names, - validate_interactive_preview_id, validate_interactive_preview_shape, ws_capabilities, + ActiveFramesConfig, BrowserInputEdgeWire, ClientMessage, FrameZoneSelection, + InputButtonStateWire, InteractivePreviewConfig, InteractivePreviewTarget, + MAX_INPUT_INJECT_EVENTS, MAX_INPUT_NAME_BYTES, MAX_INPUT_WHEEL_DELTA, + MAX_PREVIEW_PUBLICATION_BYTES, ServerMessage, SubscriptionState, TopicSelection, + deserialize_finite_coordinate, event_message_parts, parse_channels, should_relay_event, + to_snake_case, unique_sorted_channel_names, validate_interactive_preview_id, + validate_interactive_preview_shape, ws_capabilities, }; use super::relays::{ PreviewCursorQueue, PreviewOutboundError, PreviewOutboundItem, PreviewOutboundLimits, @@ -79,7 +82,8 @@ use super::relays::{ }; use super::session::{ BrowserPreviewSession, WsInputDemandLeases, authorize_subscription_channels, - negotiate_preview_transport, validated_zone_layout_preview, + commit_preview_transport, negotiate_preview_transport, stage_preview_transport, + validated_zone_layout_preview, }; use crate::api::AppState; use crate::api::security::{RequestAuthContext, SecurityState}; @@ -99,6 +103,28 @@ use crate::preview_runtime::{ use crate::render_thread::{InputPublicationConsumer, InputPublicationDemandHandle}; use crate::startup::input_status_events::InputStatusEventPublisher; +/// Selectors for the authorization tests, in the shape the wire parse +/// produces. +fn selections(topics: &[TopicId]) -> Vec { + topics + .iter() + .map(|topic| TopicSelection { + topic: *topic, + key: None, + }) + .collect() +} + +/// Membership set for the routing tests, built the way the registry +/// builds one. +fn topic_set(topics: &[TopicId]) -> TopicSet { + let mut set = TopicSet::EMPTY; + for topic in topics { + set.insert(*topic); + } + set +} + #[test] fn websocket_input_demand_leases_follow_subscription_lifetime() { let demands = InputPublicationDemandHandle::new(); @@ -114,9 +140,12 @@ fn websocket_input_demand_leases_follow_subscription_lifetime() { 0 ); - subscriptions.channels.insert(WsChannel::ScreenCanvas); - subscriptions.config.screen_canvas.width = 5_120; - subscriptions.config.screen_canvas.height = 0; + subscriptions = subscriptions + .subscribed( + &["screen_canvas"], + serde_json::json!({"screen_canvas": {"width": 5_120, "height": 0}}), + ) + .expect("screen canvas subscribe applies"); leases .synchronize(&subscriptions) .expect("partial-axis screen demand synchronizes"); @@ -135,18 +164,31 @@ fn websocket_input_demand_leases_follow_subscription_lifetime() { }; assert_eq!(canvas_bounds.max_width().map(NonZeroU32::get), Some(5_120)); assert_eq!(canvas_bounds.max_height(), None); + // A refused cadence never reaches the config store, and the lease + // it would have moved stays exactly where it was. let canvas_revision = demands.revision(); - subscriptions.config.screen_canvas.fps = 0; - assert!(leases.synchronize(&subscriptions).is_err()); + let refused = subscriptions + .subscribed( + &["screen_canvas"], + serde_json::json!({"screen_canvas": {"fps": 0}}), + ) + .expect_err("a zero cadence is refused before it can be stored"); + assert_eq!(refused.code, "invalid_config"); + leases + .synchronize(&subscriptions) + .expect("the live subscription still synchronizes"); assert_eq!(demands.revision(), canvas_revision); assert_eq!(demands.screen_branches(), canvas_only); - subscriptions.config.screen_canvas.fps = 15; - subscriptions.channels.insert(WsChannel::Spectrum); - subscriptions.config.spectrum.fps = 24; - subscriptions.config.screen_canvas.height = 720; - subscriptions.channels.insert(WsChannel::ScreenZones); - subscriptions.channels.insert(WsChannel::InputEvents); + subscriptions = subscriptions + .subscribed( + &["spectrum", "screen_zones", "input_events"], + serde_json::json!({ + "spectrum": {"fps": 24}, + "screen_canvas": {"height": 720} + }), + ) + .expect("mixed subscribe applies"); leases .synchronize(&subscriptions) .expect("wide screen demand synchronizes"); @@ -179,8 +221,10 @@ fn websocket_input_demand_leases_follow_subscription_lifetime() { )); assert_eq!(demands.requested_hz(SourceKind::Interaction), 60); - subscriptions.config.spectrum.fps = 48; - subscriptions.channels.remove(WsChannel::ScreenCanvas); + subscriptions = subscriptions + .subscribed(&["spectrum"], serde_json::json!({"spectrum": {"fps": 48}})) + .expect("spectrum cadence patch applies") + .unsubscribed(&["screen_canvas"]); leases .synchronize(&subscriptions) .expect("screen zone demand synchronizes"); @@ -194,8 +238,7 @@ fn websocket_input_demand_leases_follow_subscription_lifetime() { ScreenPublicationKind::Zones { .. } )); - subscriptions.channels.remove(WsChannel::ScreenZones); - subscriptions.channels.remove(WsChannel::InputEvents); + subscriptions = subscriptions.unsubscribed(&["screen_zones", "input_events"]); leases .synchronize(&subscriptions) .expect("removed screen demand synchronizes"); @@ -1331,9 +1374,12 @@ async fn relay_metrics_wakes_when_subscription_changes() { let relay_handle = tokio::spawn(relay_metrics(Arc::clone(&state), json_tx, subscriptions_rx)); - let mut subscriptions = initial_subscriptions; - subscriptions.channels.insert(WsChannel::Metrics); - subscriptions.config.metrics.interval_ms = 100; + let subscriptions = initial_subscriptions + .subscribed( + &["metrics"], + serde_json::json!({"metrics": {"interval_ms": 100}}), + ) + .expect("metrics subscribe applies"); publish_subscriptions(&subscriptions_tx, &subscriptions); let message = tokio::time::timeout(std::time::Duration::from_millis(250), json_rx.recv()) @@ -1401,9 +1447,12 @@ async fn relay_device_metrics_wakes_when_subscription_changes() { subscriptions_rx, )); - let mut subscriptions = initial_subscriptions; - subscriptions.channels.insert(WsChannel::DeviceMetrics); - subscriptions.config.device_metrics.interval_ms = 100; + let subscriptions = initial_subscriptions + .subscribed( + &["device_metrics"], + serde_json::json!({"device_metrics": {"interval_ms": 100}}), + ) + .expect("device metrics subscribe applies"); publish_subscriptions(&subscriptions_tx, &subscriptions); let message = tokio::time::timeout(std::time::Duration::from_millis(250), json_rx.recv()) @@ -1437,8 +1486,9 @@ async fn relay_sensors_streams_latest_snapshot_from_watch() { let relay_handle = tokio::spawn(relay_sensors(Arc::clone(&state), json_tx, subscriptions_rx)); - let mut subscriptions = initial_subscriptions; - subscriptions.channels.insert(WsChannel::Sensors); + let subscriptions = initial_subscriptions + .subscribed(&["sensors"], serde_json::Value::Null) + .expect("sensors subscribe applies"); publish_subscriptions(&subscriptions_tx, &subscriptions); let message = tokio::time::timeout(std::time::Duration::from_millis(250), json_rx.recv()) @@ -1486,8 +1536,9 @@ async fn relay_frames_wakes_when_subscription_changes() { )); assert_eq!(state.event_bus.frame_receiver_count(), 0); - let mut subscriptions = initial_subscriptions; - subscriptions.channels.insert(WsChannel::Frames); + let mut subscriptions = initial_subscriptions + .subscribed(&["frames"], serde_json::Value::Null) + .expect("frames subscribe applies"); publish_subscriptions(&subscriptions_tx, &subscriptions); let payload = tokio::time::timeout(std::time::Duration::from_millis(250), binary_rx.recv()) @@ -1497,7 +1548,7 @@ async fn relay_frames_wakes_when_subscription_changes() { assert_eq!(payload.first().copied(), Some(0x01)); assert_eq!(state.event_bus.frame_receiver_count(), 1); - subscriptions.channels.remove(WsChannel::Frames); + subscriptions = subscriptions.unsubscribed(&["frames"]); publish_subscriptions(&subscriptions_tx, &subscriptions); tokio::time::timeout(std::time::Duration::from_millis(250), async { loop { @@ -1533,8 +1584,9 @@ async fn relay_spectrum_subscribes_lazily() { )); assert_eq!(state.event_bus.spectrum_receiver_count(), 0); - let mut subscriptions = initial_subscriptions; - subscriptions.channels.insert(WsChannel::Spectrum); + let mut subscriptions = initial_subscriptions + .subscribed(&["spectrum"], serde_json::Value::Null) + .expect("spectrum subscribe applies"); publish_subscriptions(&subscriptions_tx, &subscriptions); let payload = tokio::time::timeout(std::time::Duration::from_millis(250), binary_rx.recv()) @@ -1544,7 +1596,7 @@ async fn relay_spectrum_subscribes_lazily() { assert_eq!(payload.first().copied(), Some(0x02)); assert_eq!(state.event_bus.spectrum_receiver_count(), 1); - subscriptions.channels.remove(WsChannel::Spectrum); + subscriptions = subscriptions.unsubscribed(&["spectrum"]); publish_subscriptions(&subscriptions_tx, &subscriptions); tokio::time::timeout(std::time::Duration::from_millis(250), async { loop { @@ -2092,10 +2144,12 @@ async fn relay_display_preview_reattaches_after_frame_stream_reopens() { .normalized(); let device_id = state.device_registry.add(config.device_info()).await; let display_frames = Arc::new(RwLock::new(DisplayFrameRuntime::new())); - let mut subscriptions = SubscriptionState::default(); - subscriptions.channels.insert(WsChannel::DisplayPreview); - subscriptions.config.display_preview.device_id = Some(device_id.to_string()); - subscriptions.config.display_preview.fps = 30; + let subscriptions = SubscriptionState::default() + .subscribed( + &["display_preview"], + serde_json::json!({"display_preview": {"device_id": device_id.to_string(), "fps": 30}}), + ) + .expect("display preview subscribe applies"); let (_subscriptions_tx, subscriptions_rx) = watch::channel(subscriptions); let (preview_tx, preview_rx) = preview_outbound_channel(); @@ -2140,9 +2194,12 @@ async fn relay_display_preview_does_not_subscribe_unknown_device() { let state = Arc::new(AppState::new()); let display_frames = Arc::new(RwLock::new(DisplayFrameRuntime::new())); let unknown_device_id = DeviceId::new(); - let mut subscriptions = SubscriptionState::default(); - subscriptions.channels.insert(WsChannel::DisplayPreview); - subscriptions.config.display_preview.device_id = Some(unknown_device_id.to_string()); + let subscriptions = SubscriptionState::default() + .subscribed( + &["display_preview"], + serde_json::json!({"display_preview": {"device_id": unknown_device_id.to_string()}}), + ) + .expect("display preview subscribe applies"); let (_subscriptions_tx, subscriptions_rx) = watch::channel(subscriptions); let (preview_tx, preview_rx) = preview_outbound_channel(); @@ -2179,19 +2236,24 @@ fn parse_channels_accepts_supported_channel() { "device_metrics".to_owned(), ]; let parsed = parse_channels(&channels).expect("events should parse"); + let topics: Vec = parsed.iter().map(|selection| selection.topic).collect(); assert_eq!( - parsed, + topics, vec![ - WsChannel::Events, - WsChannel::Frames, - WsChannel::Spectrum, - WsChannel::Canvas, - WsChannel::ScreenCanvas, - WsChannel::FrameEvents, - WsChannel::Metrics, - WsChannel::DeviceMetrics, + TopicId::Events, + TopicId::Frames, + TopicId::Spectrum, + TopicId::Canvas, + TopicId::ScreenCanvas, + TopicId::FrameEvents, + TopicId::Metrics, + TopicId::DeviceMetrics, ] ); + assert!( + parsed.iter().all(|selection| selection.key.is_none()), + "every topic is unkeyed on today's wire" + ); } #[test] @@ -2203,12 +2265,12 @@ fn parse_channels_rejects_unknown_channel() { #[test] fn read_only_auth_rejects_private_capture_subscriptions() { - let channels = [ - WsChannel::Events, - WsChannel::ScreenCanvas, - WsChannel::ScreenZones, - WsChannel::InputEvents, - ]; + let channels = selections(&[ + TopicId::Events, + TopicId::ScreenCanvas, + TopicId::ScreenZones, + TopicId::InputEvents, + ]); let error = authorize_subscription_channels(RequestAuthContext::read_only(), &channels) .expect_err("read-only clients must not subscribe to capture-demand channels"); @@ -2224,12 +2286,12 @@ fn read_only_auth_rejects_private_capture_subscriptions() { #[test] fn read_only_auth_allows_non_capture_preview_subscriptions() { - let channels = [ - WsChannel::Events, - WsChannel::Metrics, - WsChannel::Canvas, - WsChannel::WebViewportCanvas, - ]; + let channels = selections(&[ + TopicId::Events, + TopicId::Metrics, + TopicId::Canvas, + TopicId::WebViewportCanvas, + ]); authorize_subscription_channels(RequestAuthContext::read_only(), &channels) .expect("read-only clients may subscribe to non-capture channels"); @@ -2237,11 +2299,11 @@ fn read_only_auth_allows_non_capture_preview_subscriptions() { #[test] fn control_auth_allows_private_capture_subscriptions() { - let channels = [ - WsChannel::ScreenCanvas, - WsChannel::ScreenZones, - WsChannel::InputEvents, - ]; + let channels = selections(&[ + TopicId::ScreenCanvas, + TopicId::ScreenZones, + TopicId::InputEvents, + ]); authorize_subscription_channels(RequestAuthContext::control(), &channels) .expect("control clients may subscribe to capture preview channels"); @@ -2327,22 +2389,28 @@ async fn zone_layout_preview_rejects_invalid_sampling_radii() { #[test] fn channel_config_apply_patch_supports_all_channels() { - let mut config = ChannelConfig::default(); - let patch: ChannelConfigPatch = serde_json::from_value(serde_json::json!({ - "frames": {"fps": 30, "format": "binary"}, - "spectrum": {"fps": 20, "bins": 32}, - "canvas": {"fps": 60, "format": "jpeg", "width": 320, "height": 0}, - "screen_canvas": {"fps": 24, "format": "jpeg", "width": 480, "height": 270}, - "metrics": {"interval_ms": 500}, - "device_metrics": {"interval_ms": 250} - })) - .expect("valid json patch"); - - config - .apply_patch(patch) + let state = SubscriptionState::default() + .subscribed( + &[ + "frames", + "spectrum", + "canvas", + "screen_canvas", + "metrics", + "device_metrics", + ], + serde_json::json!({ + "frames": {"fps": 30, "format": "binary"}, + "spectrum": {"fps": 20, "bins": 32}, + "canvas": {"fps": 60, "format": "jpeg", "width": 320, "height": 0}, + "screen_canvas": {"fps": 24, "format": "jpeg", "width": 480, "height": 270}, + "metrics": {"interval_ms": 500}, + "device_metrics": {"interval_ms": 250} + }), + ) .expect("full channel config patch should be accepted"); - let json = serde_json::to_value(config).expect("config serializes"); + let json = state.config_projection(); assert_eq!(json["canvas"]["fps"], 60); assert_eq!(json["canvas"]["format"], "jpeg"); assert_eq!(json["canvas"]["width"], 320); @@ -2357,48 +2425,63 @@ fn channel_config_apply_patch_supports_all_channels() { #[test] fn channel_config_admits_wide_shapes_and_preserves_auto_dimensions() { - let mut config = ChannelConfig::default(); - let patch: ChannelConfigPatch = serde_json::from_value(serde_json::json!({ - "canvas": {"width": 100_000, "height": 1_000}, - "screen_canvas": {"width": u32::MAX, "height": 0} - })) - .expect("wide preview patch"); - - config.apply_patch(patch).expect("wide shapes are admitted"); + let state = SubscriptionState::default() + .subscribed( + &["canvas", "screen_canvas"], + serde_json::json!({ + "canvas": {"width": 100_000, "height": 1_000}, + "screen_canvas": {"width": u32::MAX, "height": 0} + }), + ) + .expect("wide shapes are admitted"); - assert_eq!( - (config.canvas.width, config.canvas.height), - (100_000, 1_000) - ); - assert_eq!(config.screen_canvas.width, u32::MAX); - assert_eq!(config.screen_canvas.height, 0); + let json = state.config_projection(); + assert_eq!(json["canvas"]["width"], 100_000); + assert_eq!(json["canvas"]["height"], 1_000); + assert_eq!(json["screen_canvas"]["width"], u32::MAX); + assert_eq!(json["screen_canvas"]["height"], 0); } #[test] fn channel_config_rejects_over_budget_shape_transactionally() { - let mut config = ChannelConfig::default(); - let patch: ChannelConfigPatch = serde_json::from_value(serde_json::json!({ - "canvas": {"fps": 60}, - "zone_preview": {"width": 32_768, "height": 4_097} - })) - .expect("over-budget preview patch"); - - let error = config - .apply_patch(patch) + let live = SubscriptionState::default() + .subscribed(&["canvas", "zone_preview"], serde_json::Value::Null) + .expect("bare subscribe applies"); + + let error = live + .subscribed( + &["canvas"], + serde_json::json!({ + "canvas": {"fps": 60}, + "zone_preview": {"width": 32_768, "height": 4_097} + }), + ) .expect_err("over-budget shape is rejected"); assert_eq!(error.code, "invalid_config"); - assert_eq!(config.canvas.fps, 15); - assert_eq!( - (config.zone_preview.width, config.zone_preview.height), - (0, 0) - ); + // The valid stanza in the same request did not land either. + let json = live.config_projection(); + assert_eq!(json["canvas"]["fps"], 15); + assert_eq!(json["zone_preview"]["width"], 0); + assert_eq!(json["zone_preview"]["height"], 0); } #[test] fn channel_config_defaults_are_stable() { - let config = ChannelConfig::default(); - let json = serde_json::to_value(config).expect("config serializes"); + let json = SubscriptionState::default() + .subscribed( + &[ + "frames", + "spectrum", + "canvas", + "screen_canvas", + "metrics", + "device_metrics", + ], + serde_json::Value::Null, + ) + .expect("bare subscribe applies") + .config_projection(); assert_eq!(json["frames"]["fps"], 30); assert_eq!(json["frames"]["format"], "binary"); @@ -2413,10 +2496,231 @@ fn channel_config_defaults_are_stable() { assert_eq!(json["device_metrics"]["interval_ms"], 1000); } +#[test] +fn config_for_a_configless_topic_is_refused_the_same_way_in_both_phases() { + // A stanza with fields fails while deserializing the patch; an + // explicit null deserializes and fails on apply. Both are the same + // client mistake, so the client sees one answer. + for stanza in [serde_json::Value::Null, serde_json::json!({"fps": 10})] { + let error = SubscriptionState::default() + .subscribed(&["sensors"], serde_json::json!({"sensors": stanza})) + .expect_err("sensors takes no config"); + + assert_eq!(error.code, "invalid_config"); + assert_eq!( + error.details, + Some(serde_json::json!({ + "field": "config.sensors", + "reason": "topic accepts no config" + })) + ); + } +} + +#[test] +fn a_repeated_config_stanza_is_refused_rather_than_resolved() { + // Two stanzas for one channel means the client does not agree with + // itself about which config wins. Taking the last one silently would + // hide that from the only party who can fix it. + let repeated = r#"{"type":"subscribe","channels":["metrics"],"config":{"metrics":{"interval_ms":100},"metrics":{"interval_ms":900}}}"#; + let error = serde_json::from_str::(repeated) + .expect_err("a repeated config stanza must not resolve to the last one"); + assert!( + error + .to_string() + .contains("duplicate config stanza `metrics`"), + "expected a duplicate-stanza rejection, got: {error}" + ); + + // One stanza for the same channel still parses and carries its value. + let single = + r#"{"type":"subscribe","channels":["metrics"],"config":{"metrics":{"interval_ms":900}}}"#; + let message: ClientMessage = + serde_json::from_str(single).expect("a single stanza per channel parses"); + let ClientMessage::Subscribe { config, .. } = message else { + panic!("expected a subscribe"); + }; + let config = config.expect("the stanza survives parsing"); + assert_eq!(config.stanzas()["metrics"]["interval_ms"], 900); +} + +#[test] +fn an_absent_or_null_config_object_is_no_config_at_all() { + for raw in [ + r#"{"type":"subscribe","channels":["metrics"]}"#, + r#"{"type":"subscribe","channels":["metrics"],"config":null}"#, + ] { + let message: ClientMessage = + serde_json::from_str(raw).expect("subscribe without config parses"); + let ClientMessage::Subscribe { config, .. } = message else { + panic!("expected a subscribe"); + }; + assert!(config.is_none(), "{raw}"); + } +} + +#[test] +fn a_null_stanza_leaves_a_configurable_topic_alone() { + let state = SubscriptionState::default() + .subscribed( + &["metrics"], + serde_json::json!({"metrics": {"interval_ms": 250}}), + ) + .expect("metrics subscribe applies") + .subscribed(&["metrics"], serde_json::json!({"metrics": null})) + .expect("a null stanza is not a patch"); + + assert_eq!(state.config_projection()["metrics"]["interval_ms"], 250); +} + +#[test] +fn config_for_an_unrecognized_channel_is_ignored() { + let state = SubscriptionState::default() + .subscribed(&["metrics"], serde_json::json!({"lasers": {"fps": 1}})) + .expect("a stanza for no known topic is not a subscribe failure"); + + let config = state.config_projection(); + assert!(config.get("lasers").is_none()); + assert_eq!(config["metrics"]["interval_ms"], 1000); +} + +#[test] +fn unsubscribing_keeps_the_config_a_resubscribe_reinstates() { + let configured = SubscriptionState::default() + .subscribed( + &["metrics"], + serde_json::json!({"metrics": {"interval_ms": 250}}), + ) + .expect("metrics subscribe applies"); + assert!(configured.live_table_agrees_with_membership()); + assert!(!configured.has_dormant_config(TopicId::Metrics)); + + // Unsubscribing parks the config rather than dropping it, and the + // live table stops claiming a topic nobody is subscribed to. + let dropped = configured.unsubscribed(&["metrics"]); + assert!(!dropped.contains(TopicId::Metrics)); + assert!(dropped.config_projection().get("metrics").is_none()); + assert!(dropped.live_table_agrees_with_membership()); + assert!(dropped.has_dormant_config(TopicId::Metrics)); + + let restored = dropped + .subscribed(&["metrics"], serde_json::Value::Null) + .expect("resubscribe applies"); + assert_eq!( + restored.config_projection()["metrics"]["interval_ms"], + 250, + "a resubscribe reinstates the client's own cadence, not the default" + ); + assert!(restored.live_table_agrees_with_membership()); + assert!( + !restored.has_dormant_config(TopicId::Metrics), + "a reinstated config moves back rather than being copied" + ); +} + +#[test] +fn the_live_table_never_claims_an_unsubscribed_topic() { + // Every shape that writes config: subscribe, patch-without-subscribe, + // unsubscribe, resubscribe. The live table has to agree with + // membership after each one, because that is what any_for promises. + let mut state = SubscriptionState::default(); + assert!(state.live_table_agrees_with_membership()); + + state = state + .subscribed( + &["frames", "canvas"], + serde_json::json!({"frames": {"fps": 12}, "display_preview": {"fps": 9}}), + ) + .expect("subscribe with an unrelated stanza applies"); + assert!( + state.live_table_agrees_with_membership(), + "configuring an unsubscribed topic must not fake a subscription" + ); + assert!(state.has_dormant_config(TopicId::DisplayPreview)); + + state = state.unsubscribed(&["frames"]); + assert!(state.live_table_agrees_with_membership()); + + state = state + .subscribed(&["frames", "display_preview"], serde_json::Value::Null) + .expect("resubscribe applies"); + assert!(state.live_table_agrees_with_membership()); + assert_eq!(state.config_projection()["frames"]["fps"], 12); + assert_eq!(state.config_projection()["display_preview"]["fps"], 9); +} + +#[test] +fn config_lands_for_a_topic_the_request_does_not_subscribe() { + let state = SubscriptionState::default() + .subscribed(&["events"], serde_json::json!({"frames": {"fps": 12}})) + .expect("configuring an unsubscribed topic is allowed"); + + assert!( + state.config_projection().get("frames").is_none(), + "an unsubscribed topic is not echoed" + ); + assert!(state.has_dormant_config(TopicId::Frames)); + assert!(state.live_table_agrees_with_membership()); + assert_eq!( + state + .subscribed(&["frames"], serde_json::Value::Null) + .expect("frames subscribe applies") + .config_projection()["frames"]["fps"], + 12 + ); +} + +#[test] +fn staging_a_preview_transport_does_not_adopt_it() { + let peer = PreviewTransportCapability { + max_encoded_publication_bytes: 1024, + max_connection_bytes: 2048, + max_message_bytes: 256, + ..PreviewTransportCapability::default().legacy_v1() + }; + let (sender, receiver) = preview_outbound_channel(); + let mut capability = PreviewTransportCapability::default(); + let mut cursors = PreviewCursorQueue::new(capability.max_streams); + + let staged = + stage_preview_transport(&peer.encode(), &sender, &cursors).expect("capability stages"); + assert_eq!( + capability, + PreviewTransportCapability::default(), + "staging must not adopt the peer's capability" + ); + + // The peer's byte budget is not in force yet, so a publication that + // only the server's own budget admits still goes through. + let frame = preview_test_frame(PreviewFrameChannel::Canvas, 1, 4096); + assert!( + frame.len() > peer.max_encoded_publication_bytes, + "the fixture frame must exceed the peer's budget to be a real test" + ); + sender + .publish( + PreviewStreamId::Passive(PreviewFrameChannel::Canvas), + frame, + None, + ) + .expect("the staged budget is not in force"); + try_receive_preview_publication(&receiver).expect("publication under the old budget"); + + // Adopting it now refuses, because the transport is busy — and it + // refuses without having changed anything. + let error = commit_preview_transport(staged, &sender, &mut cursors, &mut capability) + .expect_err("an active transport cannot renegotiate"); + assert_eq!(error.code, "invalid_request"); + assert_eq!(capability, PreviewTransportCapability::default()); +} + #[test] fn unique_channel_names_are_sorted() { - let names = - unique_sorted_channel_names(&[WsChannel::Events, WsChannel::Events, WsChannel::Events]); + let names = unique_sorted_channel_names(&selections(&[ + TopicId::Events, + TopicId::Events, + TopicId::Events, + ])); assert_eq!(names, vec!["events"]); } @@ -2564,7 +2868,7 @@ fn event_message_parts_exposes_input_status_as_a_dedicated_safe_event() { #[test] fn frame_rendered_events_require_frame_events_even_with_metrics() { - let channels = ChannelSet::from_channels(&[WsChannel::Events, WsChannel::Metrics]); + let channels = topic_set(&[TopicId::Events, TopicId::Metrics]); let event = HypercolorEvent::FrameRendered { frame_number: 7, timing: FrameTiming { @@ -2583,7 +2887,7 @@ fn frame_rendered_events_require_frame_events_even_with_metrics() { #[test] fn frame_rendered_events_require_frame_events_even_with_device_metrics() { - let channels = ChannelSet::from_channels(&[WsChannel::Events, WsChannel::DeviceMetrics]); + let channels = topic_set(&[TopicId::Events, TopicId::DeviceMetrics]); let event = HypercolorEvent::FrameRendered { frame_number: 7, timing: FrameTiming { @@ -2602,7 +2906,7 @@ fn frame_rendered_events_require_frame_events_even_with_device_metrics() { #[test] fn frame_rendered_events_are_suppressed_for_event_only_clients() { - let channels = ChannelSet::from_channels(&[WsChannel::Events]); + let channels = topic_set(&[TopicId::Events]); let event = HypercolorEvent::FrameRendered { frame_number: 7, timing: FrameTiming { @@ -2621,7 +2925,7 @@ fn frame_rendered_events_are_suppressed_for_event_only_clients() { #[test] fn frame_rendered_events_pass_through_for_frame_event_clients() { - let channels = ChannelSet::from_channels(&[WsChannel::FrameEvents]); + let channels = topic_set(&[TopicId::FrameEvents]); let event = HypercolorEvent::FrameRendered { frame_number: 7, timing: FrameTiming { @@ -2673,10 +2977,9 @@ fn input_event_websocket_payload_conforms_to_shared_timed_schema() { async fn input_event_relay_preserves_equal_timestamps_and_sequence_gaps() { let bus = HypercolorBus::new(); let event_rx = bus.subscribe_all(); - let subscriptions = SubscriptionState { - channels: ChannelSet::from_channels(&[WsChannel::InputEvents]), - ..SubscriptionState::default() - }; + let subscriptions = SubscriptionState::default() + .subscribed(&["input_events"], serde_json::Value::Null) + .expect("input events subscribe applies"); let (_subscriptions_tx, subscriptions_rx) = watch::channel(subscriptions); let (json_tx, mut json_rx) = tokio::sync::mpsc::channel::(4); let relay_handle = tokio::spawn(relay_events(event_rx, json_tx, subscriptions_rx)); @@ -2716,10 +3019,7 @@ async fn lagged_event_relay_emits_reliable_resync_hint() { for _ in 0..300 { bus.publish(HypercolorEvent::Paused); } - let subscriptions = SubscriptionState { - channels: ChannelSet::from_channels(&[WsChannel::Events]), - ..SubscriptionState::default() - }; + let subscriptions = SubscriptionState::default(); let (_subscriptions_tx, subscriptions_rx) = watch::channel(subscriptions); let (json_tx, mut json_rx) = tokio::sync::mpsc::channel::(1); let relay_handle = tokio::spawn(relay_events(event_rx, json_tx, subscriptions_rx)); @@ -2914,31 +3214,28 @@ async fn input_status_publisher_rebuilds_watchers_after_graph_change() { #[test] fn input_events_never_relay_on_the_default_events_channel() { - let channels = ChannelSet::from_channels(&[WsChannel::Events]); + let channels = topic_set(&[TopicId::Events]); assert!(!should_relay_event(&sample_input_event(), channels)); } #[test] fn input_events_relay_only_on_the_input_events_channel() { - let channels = ChannelSet::from_channels(&[WsChannel::InputEvents]); + let channels = topic_set(&[TopicId::InputEvents]); assert!(should_relay_event(&sample_input_event(), channels)); } #[test] fn input_events_channel_requires_control_subscription() { - assert!(WsChannel::InputEvents.requires_control_subscription()); - assert_eq!( - WsChannel::parse("input_events"), - Some(WsChannel::InputEvents) - ); - assert_eq!(WsChannel::InputEvents.as_str(), "input_events"); + assert!(TopicId::InputEvents.requires_control()); + assert_eq!(TopicId::parse("input_events"), Some(TopicId::InputEvents)); + assert_eq!(TopicId::InputEvents.as_str(), "input_events"); } #[test] fn default_subscription_excludes_input_events() { - let default_channels = SubscriptionState::default().channels; - assert!(default_channels.contains(WsChannel::Events)); - assert!(!default_channels.contains(WsChannel::InputEvents)); + let initial = SubscriptionState::default(); + assert!(initial.contains(TopicId::Events)); + assert!(!initial.contains(TopicId::InputEvents)); } #[test] @@ -3715,9 +4012,9 @@ fn websocket_manifest_matches_protocol_constants() { .to_owned() }) .collect::>(); - let protocol_channels = WsChannel::SUPPORTED + let protocol_channels = TopicId::ALL .iter() - .map(|channel| channel.as_str().to_owned()) + .map(|topic| topic.as_str().to_owned()) .collect::>(); assert_eq!(manifest_channels, protocol_channels); @@ -3867,113 +4164,85 @@ fn websocket_manifest_matches_protocol_constants() { } #[test] -fn display_preview_patch_tri_state_distinguishes_missing_null_and_value() { +fn display_preview_patch_applies_tri_state_at_the_boundary() { // Three JSON shapes the client can send: - // - key absent → device_id stays `None` (leave as-is) - // - `null` → device_id becomes `Some(None)` (explicit clear) - // - a string → device_id becomes `Some(Some(...))` (set target) - // Without the custom deserializer, `null` and "missing" collapse to - // the same `None`, losing the explicit-clear path. - - let absent: ChannelConfigPatch = - serde_json::from_value(serde_json::json!({ "display_preview": { "fps": 10 } })) - .expect("fps-only patch should deserialize"); - let absent_display = absent.display_preview.expect("display_preview present"); - assert!(absent_display.device_id.is_none(), "missing key → None"); - - let null_value: ChannelConfigPatch = - serde_json::from_value(serde_json::json!({ "display_preview": { "device_id": null } })) - .expect("null device_id should deserialize"); - let null_display = null_value.display_preview.expect("display_preview present"); - assert_eq!( - null_display.device_id, - Some(None), - "null key → Some(None) (explicit clear)" - ); - - let set_value: ChannelConfigPatch = serde_json::from_value( - serde_json::json!({ "display_preview": { "device_id": "device-abc" } }), - ) - .expect("string device_id should deserialize"); - let set_display = set_value.display_preview.expect("display_preview present"); + // - key absent → the target stays as it was + // - `null` → the target is cleared + // - a string → the target is set + // The patch type's own coverage lives in the leptos-ext registry + // suite; this is the same tri-state seen through a subscribe. + + let targeted = SubscriptionState::default() + .subscribed( + &["display_preview"], + serde_json::json!({"display_preview": {"device_id": "device-abc", "fps": 20}}), + ) + .expect("set applied"); assert_eq!( - set_display.device_id, - Some(Some("device-abc".to_owned())), - "string value → Some(Some(value))" + targeted.config_projection()["display_preview"], + serde_json::json!({"device_id": "device-abc", "fps": 20}) ); -} -#[test] -fn display_preview_patch_applies_tri_state_to_config() { - let mut config = ChannelConfig::default(); - - // Start with a set target. - let set_patch: ChannelConfigPatch = serde_json::from_value(serde_json::json!({ - "display_preview": { "device_id": "device-abc", "fps": 20 } - })) - .expect("valid set patch"); - config.apply_patch(set_patch).expect("set applied"); + let retargeted = targeted + .subscribed( + &["display_preview"], + serde_json::json!({"display_preview": {"fps": 15}}), + ) + .expect("fps-only applied"); assert_eq!( - config.display_preview.device_id.as_deref(), - Some("device-abc") + retargeted.config_projection()["display_preview"], + serde_json::json!({"device_id": "device-abc", "fps": 15}) ); - assert_eq!(config.display_preview.fps, 20); - // Missing key leaves device_id as-is but updates fps. - let leave_patch: ChannelConfigPatch = - serde_json::from_value(serde_json::json!({ "display_preview": { "fps": 15 } })) - .expect("valid fps-only patch"); - config.apply_patch(leave_patch).expect("fps-only applied"); + let cleared = retargeted + .subscribed( + &["display_preview"], + serde_json::json!({"display_preview": {"device_id": null}}), + ) + .expect("clear applied"); assert_eq!( - config.display_preview.device_id.as_deref(), - Some("device-abc") + cleared.config_projection()["display_preview"], + serde_json::json!({"fps": 15}) ); - assert_eq!(config.display_preview.fps, 15); - - // null explicitly clears the target. - let clear_patch: ChannelConfigPatch = serde_json::from_value(serde_json::json!({ - "display_preview": { "device_id": null } - })) - .expect("valid clear patch"); - config.apply_patch(clear_patch).expect("clear applied"); - assert!(config.display_preview.device_id.is_none()); - assert_eq!(config.display_preview.fps, 15); } #[test] fn display_preview_patch_rejects_empty_device_id_string() { - let mut config = ChannelConfig::default(); - let bad: ChannelConfigPatch = - serde_json::from_value(serde_json::json!({ "display_preview": { "device_id": " " } })) - .expect("empty whitespace still deserializes"); - let err = config - .apply_patch(bad) + let error = SubscriptionState::default() + .subscribed( + &["display_preview"], + serde_json::json!({"display_preview": {"device_id": " "}}), + ) .expect_err("empty-string device_id should be rejected"); - let message = format!("{err:?}"); - assert!( - message.contains("device_id") || message.contains("non-empty"), - "expected device_id validation error, got: {message}" + + assert_eq!(error.code, "invalid_config"); + assert_eq!( + error.details, + Some(serde_json::json!({ + "field": "config.display_preview.device_id", + "reason": "must be non-empty when provided" + })) ); } #[test] fn display_preview_patch_fps_must_be_in_range() { - let mut config = ChannelConfig::default(); - let too_high: ChannelConfigPatch = serde_json::from_value(serde_json::json!({ - "display_preview": { "fps": 120 } - })) - .expect("high fps deserializes"); - config - .apply_patch(too_high) - .expect_err("fps above 30 should be rejected"); - - let too_low: ChannelConfigPatch = serde_json::from_value(serde_json::json!({ - "display_preview": { "fps": 0 } - })) - .expect("zero fps deserializes"); - config - .apply_patch(too_low) - .expect_err("fps of 0 should be rejected"); + for fps in [0, 120] { + let error = SubscriptionState::default() + .subscribed( + &["display_preview"], + serde_json::json!({"display_preview": {"fps": fps}}), + ) + .expect_err("out-of-range cadence should be rejected"); + assert_eq!(error.code, "invalid_config"); + assert_eq!( + error.details, + Some(serde_json::json!({ + "field": "config.display_preview.fps", + "reason": "expected 1..=30" + })) + ); + } } #[tokio::test] diff --git a/crates/hypercolor-daemon/src/api/ws/topics.rs b/crates/hypercolor-daemon/src/api/ws/topics.rs new file mode 100644 index 000000000..25f313062 --- /dev/null +++ b/crates/hypercolor-daemon/src/api/ws/topics.rs @@ -0,0 +1,256 @@ +//! Daemon-side runtime facts for each WS topic. +//! +//! The registry in `hypercolor-leptos-ext` owns what the wire agrees on: +//! names, keys, configs, patches, tags, gating. What a topic costs to +//! serve is local knowledge, and it lives here keyed by [`TopicId`]: +//! which relay task feeds it, and whether the daemon can afford the +//! surface a config asks for. +//! +//! Relays are registered per task, not per topic. Three topics share the +//! event relay because one bus subscription routes all three, and +//! `zone_preview` fans out to every live scene zone inside its own +//! relay. A table keyed by topic would have to invent a task per entry +//! and lose both. + +use std::sync::Arc; + +use axum::body::Bytes; +use axum::extract::ws::Utf8Bytes; +use hypercolor_leptos_ext::ws::registry::{CanvasConfig, TopicId}; +use serde::Deserialize; +use tokio::sync::{mpsc, watch}; +use tokio::task::JoinHandle; +use tracing::trace; + +use super::protocol::{SubscriptionState, WsProtocolError, validate_passive_preview_shape}; +use super::relays::{ + PreviewOutboundSender, relay_canvas, relay_device_metrics, relay_display_preview, relay_events, + relay_frames, relay_metrics, relay_screen_canvas, relay_screen_zones, relay_sensors, + relay_spectrum, relay_web_viewport_canvas, relay_zone_preview, +}; +use crate::api::AppState; + +/// Everything a relay task needs to attach itself to one connection. +pub(super) struct RelayContext { + pub(super) state: Arc, + pub(super) json_tx: mpsc::Sender, + pub(super) binary_tx: mpsc::Sender, + pub(super) preview_tx: PreviewOutboundSender, + pub(super) subscriptions: watch::Receiver, +} + +/// One relay task and the topics it serves. +struct RelayRegistration { + /// Topics this task feeds. More than one means the task routes. + topics: &'static [TopicId], + spawn: fn(&RelayContext) -> JoinHandle<()>, +} + +/// Every relay a connection spawns, once each. +static RELAYS: &[RelayRegistration] = &[ + RelayRegistration { + // One bus subscription carries all three, and the relay routes by + // event kind rather than opening three broadcast receivers. + topics: &[TopicId::Events, TopicId::FrameEvents, TopicId::InputEvents], + spawn: |context| { + tokio::spawn(relay_events( + context.state.event_bus.subscribe_all(), + context.json_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::Frames], + spawn: |context| { + tokio::spawn(relay_frames( + Arc::clone(&context.state), + context.json_tx.clone(), + context.binary_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::Spectrum], + spawn: |context| { + tokio::spawn(relay_spectrum( + Arc::clone(&context.state), + context.json_tx.clone(), + context.binary_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::Canvas], + spawn: |context| { + tokio::spawn(relay_canvas( + Arc::clone(&context.state.preview_runtime), + context.state.power_state.subscribe(), + context.preview_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::ScreenCanvas], + spawn: |context| { + tokio::spawn(relay_screen_canvas( + Arc::clone(&context.state.preview_runtime), + context.preview_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::ScreenZones], + spawn: |context| { + tokio::spawn(relay_screen_zones( + Arc::clone(&context.state.preview_runtime), + context.subscriptions.clone(), + context.preview_tx.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::WebViewportCanvas], + spawn: |context| { + tokio::spawn(relay_web_viewport_canvas( + Arc::clone(&context.state.preview_runtime), + context.preview_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + // Fans out to one stream per live scene zone inside the relay. + topics: &[TopicId::ZonePreview], + spawn: |context| { + tokio::spawn(relay_zone_preview( + Arc::clone(&context.state.preview_runtime), + context.preview_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::DisplayPreview], + spawn: |context| { + tokio::spawn(relay_display_preview( + Arc::clone(&context.state), + Arc::clone(&context.state.display_frames), + context.preview_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::Metrics], + spawn: |context| { + tokio::spawn(relay_metrics( + Arc::clone(&context.state), + context.json_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::DeviceMetrics], + spawn: |context| { + tokio::spawn(relay_device_metrics( + Arc::clone(&context.state), + context.json_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, + RelayRegistration { + topics: &[TopicId::Sensors], + spawn: |context| { + tokio::spawn(relay_sensors( + Arc::clone(&context.state), + context.json_tx.clone(), + context.subscriptions.clone(), + )) + }, + }, +]; + +/// Spawn every relay this connection needs. +pub(super) fn spawn_relays(context: &RelayContext) -> Vec> { + RELAYS + .iter() + .map(|registration| { + trace!(topics = ?registration.topics, "Spawning WebSocket relay"); + (registration.spawn)(context) + }) + .collect() +} + +/// Admit a candidate config against the daemon's runtime budgets. +/// +/// Runs immediately after the topic's patch applies, so a request with +/// two bad stanzas reports the earlier one, exactly as a single +/// hand-written validation pass did. +pub(super) fn admit_config( + topic: TopicId, + config: &serde_json::Value, +) -> Result<(), WsProtocolError> { + match topic { + TopicId::Canvas + | TopicId::ScreenCanvas + | TopicId::WebViewportCanvas + | TopicId::ZonePreview => { + let canvas = CanvasConfig::deserialize(config) + .expect("a validated canvas config deserializes into its own type"); + validate_passive_preview_shape(&canvas, format!("config.{}", topic.as_str())) + } + TopicId::Frames + | TopicId::Spectrum + | TopicId::Events + | TopicId::FrameEvents + | TopicId::ScreenZones + | TopicId::Metrics + | TopicId::DeviceMetrics + | TopicId::Sensors + | TopicId::DisplayPreview + | TopicId::InputEvents => Ok(()), + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeSet; + + use hypercolor_leptos_ext::ws::registry::TopicId; + + use super::RELAYS; + + #[test] + fn every_topic_is_served_by_exactly_one_relay() { + let mut served = BTreeSet::new(); + for registration in RELAYS { + for topic in registration.topics { + assert!( + served.insert(*topic), + "{} is claimed by two relays", + topic.as_str() + ); + } + } + + let missing: Vec<&str> = TopicId::ALL + .iter() + .filter(|topic| !served.contains(topic)) + .map(|topic| topic.as_str()) + .collect(); + assert!(missing.is_empty(), "topics with no relay: {missing:?}"); + } + + #[test] + fn relays_are_fewer_than_topics_because_the_event_relay_routes_three() { + assert_eq!(RELAYS.len(), 12); + assert_eq!(TopicId::COUNT, 14); + } +} diff --git a/crates/hypercolor-leptos-ext/src/ws/mod.rs b/crates/hypercolor-leptos-ext/src/ws/mod.rs index 15c3b3ca9..19b31b02e 100644 --- a/crates/hypercolor-leptos-ext/src/ws/mod.rs +++ b/crates/hypercolor-leptos-ext/src/ws/mod.rs @@ -1,6 +1,7 @@ mod backoff; mod input_event; mod preview; +pub mod registry; mod spectrum; pub mod topic; pub mod transport; diff --git a/crates/hypercolor-leptos-ext/src/ws/registry.rs b/crates/hypercolor-leptos-ext/src/ws/registry.rs new file mode 100644 index 000000000..faf0b751f --- /dev/null +++ b/crates/hypercolor-leptos-ext/src/ws/registry.rs @@ -0,0 +1,445 @@ +//! The live WebSocket topic topology (Spec 76 §5). +//! +//! Every subscribable topic on `/api/v1/ws` is declared once here, with +//! its wire name, key shape, config, patch, owned binary tags, and +//! control-tier gate. The daemon reads this registry instead of keeping +//! its own parallel channel enum, bitset, and config struct, so a wire +//! fact has exactly one home. +//! +//! What lives here is what the wire agrees on. Which bus lane feeds a +//! topic, which relay task serves it, and what engine demand a +//! subscription implies are runtime facts, and they stay in the daemon's +//! own table keyed by [`TopicId`]. +//! +//! # Tag ownership +//! +//! A topic owns the binary tags only its own codec writes, and the +//! macro asserts at compile time that no two topics claim the same +//! byte. Three tags are deliberately unowned and listed in +//! [`SHARED_TRANSPORT_TAGS`]: the wide passive preview frame, the +//! preview chunk, and the preview cancellation are transport envelopes +//! that four passive preview topics share, so no single topic can claim +//! them. Interactive preview is not a subscribable topic at all — it is +//! a session protocol keyed by preview id — so its two tags sit in the +//! same reserved list. + +use serde::{Deserialize, Serialize}; + +use super::topic::{NoPatch, PatchError, TopicPatch}; +use crate::define_ws_topics; + +/// Frame delivery encoding for the `frames` topic. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum FrameFormat { + /// Packed binary LED frames. + Binary, + /// JSON frame payloads. + Json, +} + +/// Pixel encoding for the passive preview canvas topics. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CanvasFormat { + /// Three bytes per pixel. + Rgb, + /// Four bytes per pixel. + Rgba, + /// JPEG-compressed frames. + Jpeg, +} + +/// Per-subscription configuration for the `frames` topic. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FramesConfig { + /// Delivery cadence in frames per second. + pub fps: u32, + /// Frame encoding. + pub format: FrameFormat, + /// Zone ids to deliver; `["all"]` selects every zone. + pub zones: Vec, +} + +impl Default for FramesConfig { + fn default() -> Self { + Self { + fps: 30, + format: FrameFormat::Binary, + zones: vec!["all".to_owned()], + } + } +} + +/// Patch for [`FramesConfig`]. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FramesConfigPatch { + /// Replacement cadence. + #[serde(default)] + pub fps: Option, + /// Replacement encoding. + #[serde(default)] + pub format: Option, + /// Replacement zone selection. + #[serde(default)] + pub zones: Option>, +} + +impl TopicPatch for FramesConfigPatch { + fn apply(&self, config: &mut FramesConfig) -> Result<(), PatchError> { + if let Some(fps) = self.fps { + validate_range(fps, 1, 60, "fps", "expected 1..=60")?; + config.fps = fps; + } + if let Some(format) = self.format { + config.format = format; + } + if let Some(zones) = self.zones.clone() { + if zones.is_empty() { + return Err(PatchError::new("zones", "must not be empty")); + } + config.zones = zones; + } + Ok(()) + } +} + +/// Per-subscription configuration for the `spectrum` topic. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SpectrumConfig { + /// Delivery cadence in frames per second. + pub fps: u32, + /// FFT bin count. + pub bins: u16, +} + +impl Default for SpectrumConfig { + fn default() -> Self { + Self { fps: 30, bins: 64 } + } +} + +/// Patch for [`SpectrumConfig`]. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SpectrumConfigPatch { + /// Replacement cadence. + #[serde(default)] + pub fps: Option, + /// Replacement bin count. + #[serde(default)] + pub bins: Option, +} + +impl TopicPatch for SpectrumConfigPatch { + fn apply(&self, config: &mut SpectrumConfig) -> Result<(), PatchError> { + if let Some(fps) = self.fps { + validate_range(fps, 1, 60, "fps", "expected 1..=60")?; + config.fps = fps; + } + if let Some(bins) = self.bins { + if ![8, 16, 32, 64, 128].contains(&bins) { + return Err(PatchError::new( + "bins", + "expected one of [8, 16, 32, 64, 128]", + )); + } + config.bins = bins; + } + Ok(()) + } +} + +/// Per-subscription configuration shared by the passive preview canvas +/// topics. `width` and `height` of zero mean "server picks", which is +/// why neither carries an upper bound here: the admissible surface size +/// is a runtime resource question the daemon answers. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CanvasConfig { + /// Delivery cadence in frames per second. + pub fps: u32, + /// Pixel encoding. + pub format: CanvasFormat, + /// Requested width, or zero for the server default. + pub width: u32, + /// Requested height, or zero for the server default. + pub height: u32, +} + +impl Default for CanvasConfig { + fn default() -> Self { + Self { + fps: 15, + format: CanvasFormat::Rgb, + width: 0, + height: 0, + } + } +} + +/// Patch for [`CanvasConfig`]. +#[derive(Debug, Clone, Copy, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CanvasConfigPatch { + /// Replacement cadence. + #[serde(default)] + pub fps: Option, + /// Replacement pixel encoding. + #[serde(default)] + pub format: Option, + /// Replacement width. + #[serde(default)] + pub width: Option, + /// Replacement height. + #[serde(default)] + pub height: Option, +} + +impl TopicPatch for CanvasConfigPatch { + fn apply(&self, config: &mut CanvasConfig) -> Result<(), PatchError> { + if let Some(fps) = self.fps { + validate_range(fps, 1, 60, "fps", "expected 1..=60")?; + config.fps = fps; + } + if let Some(format) = self.format { + config.format = format; + } + if let Some(width) = self.width { + config.width = width; + } + if let Some(height) = self.height { + config.height = height; + } + Ok(()) + } +} + +/// Per-subscription configuration for the periodic telemetry topics. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MetricsConfig { + /// Snapshot period in milliseconds. + pub interval_ms: u32, +} + +impl Default for MetricsConfig { + fn default() -> Self { + Self { interval_ms: 1000 } + } +} + +/// Patch for [`MetricsConfig`]. +#[derive(Debug, Clone, Copy, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MetricsConfigPatch { + /// Replacement snapshot period. + #[serde(default)] + pub interval_ms: Option, +} + +impl TopicPatch for MetricsConfigPatch { + fn apply(&self, config: &mut MetricsConfig) -> Result<(), PatchError> { + if let Some(interval_ms) = self.interval_ms { + validate_range( + interval_ms, + 100, + 10_000, + "interval_ms", + "expected 100..=10000", + )?; + config.interval_ms = interval_ms; + } + Ok(()) + } +} + +/// Per-subscription configuration for the `display_preview` topic. +/// `device_id` stays `None` until a client names a target; clearing it +/// detaches the relay from the device's frame stream. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DisplayPreviewConfig { + /// Target device, or `None` while the subscription is detached. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub device_id: Option, + /// Delivery cadence in frames per second. + pub fps: u32, +} + +impl Default for DisplayPreviewConfig { + fn default() -> Self { + Self { + device_id: None, + fps: 15, + } + } +} + +/// Patch for [`DisplayPreviewConfig`]. `device_id` is a double-`Option` +/// because the three client intents are distinct on the wire: an absent +/// key leaves the target alone, `null` clears it, and a string sets it. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DisplayPreviewConfigPatch { + /// Tri-state target update. + #[serde(default, deserialize_with = "deserialize_double_option_string")] + #[allow( + clippy::option_option, + reason = "the patch protocol needs distinct states for missing, null, and string values" + )] + pub device_id: Option>, + /// Replacement cadence. + #[serde(default)] + pub fps: Option, +} + +impl TopicPatch for DisplayPreviewConfigPatch { + fn apply(&self, config: &mut DisplayPreviewConfig) -> Result<(), PatchError> { + if let Some(device_id) = self.device_id.clone() { + match device_id { + Some(id) => { + // Trim so accidental whitespace cannot sneak a + // subscription through with no real device behind it. + let trimmed = id.trim(); + if trimmed.is_empty() { + return Err(PatchError::new( + "device_id", + "must be non-empty when provided", + )); + } + config.device_id = Some(trimmed.to_owned()); + } + None => config.device_id = None, + } + } + if let Some(fps) = self.fps { + validate_range(fps, 1, 30, "fps", "expected 1..=30")?; + config.fps = fps; + } + Ok(()) + } +} + +fn validate_range( + value: u32, + min: u32, + max: u32, + field: &'static str, + reason: &'static str, +) -> Result<(), PatchError> { + if (min..=max).contains(&value) { + Ok(()) + } else { + Err(PatchError::new(field, reason)) + } +} + +/// Deserialize a double-`Option` so `null` maps to `Some(None)` (explicit +/// clear) and a missing key keeps the outer `None` through +/// `#[serde(default)]`. Serde's own behavior collapses both into `None`. +#[allow( + clippy::option_option, + reason = "serde needs the tri-state shape to preserve missing-vs-null during patch application" +)] +fn deserialize_double_option_string<'de, D>( + deserializer: D, +) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + Option::::deserialize(deserializer).map(Some) +} + +define_ws_topics! { + registry TopicId; + + topic Frames => "frames" { + key: unkeyed, config: FramesConfig, patch: FramesConfigPatch, + tags: [0x01], control: false, + } + topic Spectrum => "spectrum" { + key: unkeyed, config: SpectrumConfig, patch: SpectrumConfigPatch, + tags: [0x02], control: false, + } + topic Events => "events" { + key: unkeyed, config: (), patch: NoPatch, + tags: [], control: false, + } + topic FrameEvents => "frame_events" { + key: unkeyed, config: (), patch: NoPatch, + tags: [], control: false, + } + topic Canvas => "canvas" { + key: unkeyed, config: CanvasConfig, patch: CanvasConfigPatch, + tags: [0x03], control: false, + } + topic ScreenCanvas => "screen_canvas" { + key: unkeyed, config: CanvasConfig, patch: CanvasConfigPatch, + tags: [0x05], control: true, + } + topic ScreenZones => "screen_zones" { + key: unkeyed, config: (), patch: NoPatch, + tags: [0x09, 0x0e, 0x11], control: true, + } + topic WebViewportCanvas => "web_viewport_canvas" { + key: unkeyed, config: CanvasConfig, patch: CanvasConfigPatch, + tags: [0x06], control: false, + } + topic ZonePreview => "zone_preview" { + key: unkeyed, config: CanvasConfig, patch: CanvasConfigPatch, + tags: [0x08, 0x0c], control: false, + } + topic Metrics => "metrics" { + key: unkeyed, config: MetricsConfig, patch: MetricsConfigPatch, + tags: [], control: false, + } + topic DeviceMetrics => "device_metrics" { + key: unkeyed, config: MetricsConfig, patch: MetricsConfigPatch, + tags: [], control: false, + } + topic Sensors => "sensors" { + key: unkeyed, config: (), patch: NoPatch, + tags: [], control: false, + } + topic DisplayPreview => "display_preview" { + key: unkeyed, config: DisplayPreviewConfig, patch: DisplayPreviewConfigPatch, + tags: [0x07], control: false, + } + topic InputEvents => "input_events" { + key: unkeyed, config: (), patch: NoPatch, + tags: [], control: true, + } +} + +/// Binary tags no single topic owns. +/// +/// `0x0b` is the wide form of the passive preview frame, which four +/// topics publish; `0x0f` and `0x10` are the chunk and cancellation +/// envelopes every preview stream rides. `0x0a` and `0x0d` belong to +/// interactive preview, a keyed session protocol rather than a +/// subscribable topic. +pub const SHARED_TRANSPORT_TAGS: &[u8] = &[0x0a, 0x0b, 0x0d, 0x0f, 0x10]; + +// The shared tags are as exclusive as the owned ones: a topic quietly +// claiming a transport envelope byte would collide on the wire without +// the per-topic assertion ever noticing. +const _: () = { + assert!( + super::topic::tags_disjoint(&[ + SHARED_TRANSPORT_TAGS, + ::OWNED_TAGS, + ::OWNED_TAGS, + ::OWNED_TAGS, + ::OWNED_TAGS, + ::OWNED_TAGS, + ::OWNED_TAGS, + ::OWNED_TAGS, + ::OWNED_TAGS, + ]), + "a topic claims a shared preview transport tag" + ); +}; diff --git a/crates/hypercolor-leptos-ext/tests/ws_registry_tests.rs b/crates/hypercolor-leptos-ext/tests/ws_registry_tests.rs new file mode 100644 index 000000000..9b5eed0b1 --- /dev/null +++ b/crates/hypercolor-leptos-ext/tests/ws_registry_tests.rs @@ -0,0 +1,365 @@ +//! Contract coverage for the live WS topic topology (Spec 76 §5). +//! +//! Every assertion here pins something a client can observe: the wire +//! names and their order, the config JSON a fresh subscription echoes, +//! which topics take config at all, which need the control tier, and +//! which binary tags each topic owns. The tag-disjointness assertions +//! are compile-time, so this file merely COMPILING proves them. +#![cfg(feature = "ws-core")] + +use std::collections::BTreeSet; + +use hypercolor_leptos_ext::ws::registry::{ + CanvasConfig, CanvasConfigPatch, DisplayPreviewConfig, DisplayPreviewConfigPatch, FramesConfig, + FramesConfigPatch, MetricsConfig, MetricsConfigPatch, SHARED_TRANSPORT_TAGS, SpectrumConfig, + SpectrumConfigPatch, TopicId, +}; +use hypercolor_leptos_ext::ws::topic::{TopicPatch, apply_patch_transactionally}; +use serde_json::json; + +/// Wire names in declaration order. The daemon's capability list and its +/// protocol manifest are both this sequence, so a reorder is a wire change. +const WIRE_NAMES: [&str; 14] = [ + "frames", + "spectrum", + "events", + "frame_events", + "canvas", + "screen_canvas", + "screen_zones", + "web_viewport_canvas", + "zone_preview", + "metrics", + "device_metrics", + "sensors", + "display_preview", + "input_events", +]; + +#[test] +fn topic_wire_names_are_frozen_in_declaration_order() { + let names: Vec<&str> = TopicId::ALL.iter().map(|topic| topic.as_str()).collect(); + assert_eq!(names, WIRE_NAMES); + assert_eq!(TopicId::COUNT, WIRE_NAMES.len()); +} + +#[test] +fn every_wire_name_round_trips_through_parse() { + for topic in TopicId::ALL.iter().copied() { + assert_eq!(TopicId::parse(topic.as_str()), Some(topic)); + } + assert_eq!(TopicId::parse("lasers"), None); + assert_eq!(TopicId::parse("interactive_preview"), None); +} + +#[test] +fn membership_bits_are_unique_across_the_registry() { + let bits: BTreeSet = TopicId::ALL.iter().map(|topic| topic.bit()).collect(); + assert_eq!(bits.len(), TopicId::COUNT); +} + +#[test] +fn control_tier_gates_screen_capture_and_host_input() { + let gated: Vec<&str> = TopicId::ALL + .iter() + .filter(|topic| topic.requires_control()) + .map(|topic| topic.as_str()) + .collect(); + assert_eq!(gated, vec!["screen_canvas", "screen_zones", "input_events"]); +} + +#[test] +fn configless_topics_carry_no_config_stanza() { + let configless: Vec<&str> = TopicId::ALL + .iter() + .filter(|topic| !topic.vtable().configurable) + .map(|topic| topic.as_str()) + .collect(); + assert_eq!( + configless, + vec![ + "events", + "frame_events", + "screen_zones", + "sensors", + "input_events" + ] + ); +} + +#[test] +fn default_configs_are_the_frozen_wire_defaults() { + for (topic, expected) in [ + ( + TopicId::Frames, + json!({"fps": 30, "format": "binary", "zones": ["all"]}), + ), + (TopicId::Spectrum, json!({"fps": 30, "bins": 64})), + ( + TopicId::Canvas, + json!({"fps": 15, "format": "rgb", "width": 0, "height": 0}), + ), + ( + TopicId::ScreenCanvas, + json!({"fps": 15, "format": "rgb", "width": 0, "height": 0}), + ), + ( + TopicId::WebViewportCanvas, + json!({"fps": 15, "format": "rgb", "width": 0, "height": 0}), + ), + ( + TopicId::ZonePreview, + json!({"fps": 15, "format": "rgb", "width": 0, "height": 0}), + ), + (TopicId::Metrics, json!({"interval_ms": 1000})), + (TopicId::DeviceMetrics, json!({"interval_ms": 1000})), + (TopicId::DisplayPreview, json!({"fps": 15})), + ] { + assert_eq!( + (topic.vtable().default_config_json)(), + expected, + "{} default config", + topic.as_str() + ); + } +} + +#[test] +fn owned_and_shared_tags_cover_the_whole_binary_space() { + let mut owned: Vec = TopicId::ALL + .iter() + .flat_map(|topic| topic.vtable().owned_tags.iter().copied()) + .collect(); + owned.sort_unstable(); + assert_eq!( + owned, + vec![ + 0x01, 0x02, 0x03, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0c, 0x0e, 0x11 + ] + ); + + let mut all: Vec = owned; + all.extend_from_slice(SHARED_TRANSPORT_TAGS); + all.sort_unstable(); + assert_eq!( + all, + vec![ + 0x01, 0x02, 0x03, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, + 0x10, 0x11 + ], + "0x04 stays unassigned; every other tag has exactly one home" + ); +} + +#[test] +fn tag_owners_match_their_codecs() { + assert_eq!(TopicId::Frames.vtable().owned_tags, [0x01]); + assert_eq!(TopicId::Spectrum.vtable().owned_tags, [0x02]); + assert_eq!(TopicId::Canvas.vtable().owned_tags, [0x03]); + assert_eq!(TopicId::ScreenCanvas.vtable().owned_tags, [0x05]); + assert_eq!(TopicId::WebViewportCanvas.vtable().owned_tags, [0x06]); + assert_eq!(TopicId::DisplayPreview.vtable().owned_tags, [0x07]); + assert_eq!(TopicId::ZonePreview.vtable().owned_tags, [0x08, 0x0c]); + assert_eq!(TopicId::ScreenZones.vtable().owned_tags, [0x09, 0x0e, 0x11]); + assert_eq!(TopicId::Metrics.vtable().owned_tags, [] as [u8; 0]); +} + +#[test] +fn unkeyed_topics_reject_a_wire_key() { + for topic in TopicId::ALL.iter().copied() { + assert!(!topic.vtable().keyed, "{} is unkeyed", topic.as_str()); + assert_eq!((topic.vtable().validate_key)(None), Ok(None)); + assert!((topic.vtable().validate_key)(Some("anything")).is_err()); + } +} + +#[test] +fn configless_topics_refuse_config_in_both_phases() { + let vtable = TopicId::Sensors.vtable(); + + // Non-null config fails while deserializing the patch. + let deserialize_error = (vtable.apply_patch_json)(&json!(null), &json!({"fps": 10})) + .expect_err("a configless topic takes no config object"); + assert_eq!(deserialize_error.field, "patch"); + + // Explicit null deserializes, then fails on apply. + let apply_error = (vtable.apply_patch_json)(&json!(null), &json!(null)) + .expect_err("a configless topic cannot apply a patch"); + assert_eq!(apply_error.field, "config"); + assert_eq!(apply_error.reason, "topic accepts no config"); +} + +#[test] +fn unknown_config_fields_are_rejected_rather_than_dropped() { + let stale = json!({"fps": 30, "format": "binary", "zones": ["all"], "gone": true}); + let error = (TopicId::Frames.vtable().apply_patch_json)(&stale, &json!({"fps": 10})) + .expect_err("a stale config field must fail loudly"); + assert_eq!(error.field, "config"); +} + +#[test] +fn unknown_patch_fields_are_rejected_rather_than_silently_ignored() { + let error = (TopicId::Canvas.vtable().apply_patch_json)( + &(TopicId::Canvas.vtable().default_config_json)(), + &json!({"fpz": 30}), + ) + .expect_err("a typo'd patch field must not succeed as a no-op"); + assert_eq!(error.field, "patch"); +} + +#[test] +fn frames_patch_validates_cadence_and_zone_selection() { + let config = FramesConfig::default(); + + for fps in [0_u32, 61] { + let patch = FramesConfigPatch { + fps: Some(fps), + ..FramesConfigPatch::default() + }; + let error = apply_patch_transactionally(&config, &patch).expect_err("fps out of range"); + assert_eq!(error.field, "fps"); + assert_eq!(error.reason, "expected 1..=60"); + } + + let empty = FramesConfigPatch { + zones: Some(Vec::new()), + ..FramesConfigPatch::default() + }; + let error = apply_patch_transactionally(&config, &empty).expect_err("empty zones"); + assert_eq!(error.field, "zones"); + assert_eq!(error.reason, "must not be empty"); + + let accepted = FramesConfigPatch { + fps: Some(60), + zones: Some(vec!["desk".to_owned()]), + ..FramesConfigPatch::default() + }; + let next = apply_patch_transactionally(&config, &accepted).expect("in-range patch applies"); + assert_eq!(next.fps, 60); + assert_eq!(next.zones, vec!["desk".to_owned()]); +} + +#[test] +fn spectrum_patch_admits_only_the_declared_bin_counts() { + let config = SpectrumConfig::default(); + for bins in [8_u16, 16, 32, 64, 128] { + let patch = SpectrumConfigPatch { + bins: Some(bins), + ..SpectrumConfigPatch::default() + }; + assert_eq!( + apply_patch_transactionally(&config, &patch) + .expect("declared bin count") + .bins, + bins + ); + } + + let patch = SpectrumConfigPatch { + bins: Some(48), + ..SpectrumConfigPatch::default() + }; + let error = apply_patch_transactionally(&config, &patch).expect_err("undeclared bin count"); + assert_eq!(error.field, "bins"); + assert_eq!(error.reason, "expected one of [8, 16, 32, 64, 128]"); +} + +#[test] +fn metrics_patch_bounds_the_snapshot_period() { + let config = MetricsConfig::default(); + for interval_ms in [99_u32, 10_001] { + let patch = MetricsConfigPatch { + interval_ms: Some(interval_ms), + }; + let error = apply_patch_transactionally(&config, &patch).expect_err("interval bound"); + assert_eq!(error.field, "interval_ms"); + assert_eq!(error.reason, "expected 100..=10000"); + } +} + +#[test] +fn canvas_patch_leaves_dimensions_unbounded_for_the_daemon_to_admit() { + let config = CanvasConfig::default(); + let patch = CanvasConfigPatch { + width: Some(u32::MAX), + height: Some(0), + ..CanvasConfigPatch::default() + }; + let next = apply_patch_transactionally(&config, &patch).expect("wide shapes deserialize"); + assert_eq!((next.width, next.height), (u32::MAX, 0)); +} + +#[test] +fn a_failing_field_leaves_the_whole_config_untouched() { + let config = CanvasConfig::default(); + let patch: CanvasConfigPatch = + serde_json::from_value(json!({"width": 640, "fps": 240})).expect("patch parses"); + + let error = apply_patch_transactionally(&config, &patch).expect_err("fps out of range"); + assert_eq!(error.field, "fps"); + assert_eq!(config, CanvasConfig::default()); +} + +#[test] +fn display_preview_target_is_a_tri_state() { + let absent: DisplayPreviewConfigPatch = + serde_json::from_value(json!({"fps": 10})).expect("fps-only patch parses"); + assert!(absent.device_id.is_none(), "missing key leaves the target"); + + let cleared: DisplayPreviewConfigPatch = + serde_json::from_value(json!({"device_id": null})).expect("null patch parses"); + assert_eq!(cleared.device_id, Some(None), "null clears the target"); + + let set: DisplayPreviewConfigPatch = + serde_json::from_value(json!({"device_id": "device-abc"})).expect("value patch parses"); + assert_eq!(set.device_id, Some(Some("device-abc".to_owned()))); + + let mut config = DisplayPreviewConfig::default(); + set.apply(&mut config).expect("target applies"); + assert_eq!(config.device_id.as_deref(), Some("device-abc")); + absent.apply(&mut config).expect("cadence applies"); + assert_eq!(config.device_id.as_deref(), Some("device-abc")); + assert_eq!(config.fps, 10); + cleared.apply(&mut config).expect("clear applies"); + assert!(config.device_id.is_none()); + assert_eq!(config.fps, 10); +} + +#[test] +fn display_preview_target_must_name_a_real_device() { + let patch: DisplayPreviewConfigPatch = + serde_json::from_value(json!({"device_id": " "})).expect("whitespace parses"); + let error = apply_patch_transactionally(&DisplayPreviewConfig::default(), &patch) + .expect_err("whitespace is not a device"); + assert_eq!(error.field, "device_id"); + assert_eq!(error.reason, "must be non-empty when provided"); + + let trimmed: DisplayPreviewConfigPatch = + serde_json::from_value(json!({"device_id": " device-abc "})).expect("padded parses"); + let config = apply_patch_transactionally(&DisplayPreviewConfig::default(), &trimmed) + .expect("padding is trimmed"); + assert_eq!(config.device_id.as_deref(), Some("device-abc")); +} + +#[test] +fn display_preview_cadence_stops_at_thirty() { + for fps in [0_u32, 31] { + let patch = DisplayPreviewConfigPatch { + fps: Some(fps), + ..DisplayPreviewConfigPatch::default() + }; + let error = apply_patch_transactionally(&DisplayPreviewConfig::default(), &patch) + .expect_err("cadence bound"); + assert_eq!(error.field, "fps"); + assert_eq!(error.reason, "expected 1..=30"); + } +} + +#[test] +fn a_detached_display_preview_omits_its_target_from_the_wire() { + let config = DisplayPreviewConfig::default(); + assert_eq!( + serde_json::to_value(&config).expect("config serializes"), + json!({"fps": 15}) + ); +} diff --git a/docs/content/api/websocket.md b/docs/content/api/websocket.md index 9764b88fa..f0e48287c 100644 --- a/docs/content/api/websocket.md +++ b/docs/content/api/websocket.md @@ -508,22 +508,30 @@ value, or a forbidden control-tier subscription. ```json { "type": "error", - "code": "unsupported_channel", - "message": "Channel 'bogus' is not supported by this server", - "details": { "channel": "bogus" } + "code": "invalid_config", + "message": "Invalid configuration for config.frames.fps: expected 1..=60", + "details": { "field": "config.frames.fps", "reason": "expected 1..=60" } } ``` -Error codes you may see: `invalid_request` (bad JSON or empty channel list), -`invalid_config` (out-of-range or invalid config value, with `details.field` -and `details.reason`), `unsupported_channel`, and `forbidden` (a control-tier +Error codes you may see: `invalid_request` (bad JSON, an empty channel list, or +an unknown channel name), `invalid_config` (an invalid config stanza, with +`details.field` and `details.reason`), and `forbidden` (a control-tier subscription or mutation attempted without a control key). ## Channel configuration Each configurable channel carries parameters that control throughput and format. -Send them in the `config` field of a `subscribe` message. Out-of-range values -are rejected with an `invalid_config` error and the channel is left unchanged. +Send them in the `config` field of a `subscribe` message. A rejected stanza +fails the whole request with an `invalid_config` error, and every channel named +in it is left exactly as it was. + +Each stanza is validated by the channel that owns it, so four shapes are +refused rather than ignored: a value outside the documented range, a field the +channel does not define, a stanza sent for a channel that takes no config +(`events`, `frame_events`, `screen_zones`, `sensors`, `input_events`), and the +same channel named twice in one `config` object. A `null` stanza on a +configurable channel means "leave this channel alone". ### frames config diff --git a/docs/development/CLIENT_GENERATION.md b/docs/development/CLIENT_GENERATION.md index 51414ddb3..850ffae00 100644 --- a/docs/development/CLIENT_GENERATION.md +++ b/docs/development/CLIENT_GENERATION.md @@ -40,8 +40,8 @@ protocol/websocket-v1.json It records channel names, advertised capabilities, binary frame tags, preview pixel formats, and subscription config bounds. The daemon has a regression test -that compares the manifest with `WsChannel::SUPPORTED`, `ws_capabilities()`, -and the binary tag constants. +that compares the manifest with `TopicId::ALL`, `ws_capabilities()`, and the +binary tag constants. Python generates protocol constants from the manifest: