diff --git a/CLAUDE.md b/CLAUDE.md index 340adc5..7ab3ecc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -246,6 +246,45 @@ UltraLog embeds an MCP (Model Context Protocol) HTTP server so Claude Desktop ca **Load-bearing contract:** the IPC server wakes the GUI via a repaint callback (`request_repaint()`), not a polling timer — see `IpcServer::start_with_repaint` in `src/ipc/server.rs` and its wiring in `UltraLogApp::new`. Incoming commands are drained in `UltraLogApp::process_ipc_commands`, which caps processing at **10 commands per frame** to avoid blocking the UI thread; if more are queued, it requests another repaint to continue next frame. +**Response size contract (load-bearing):** every MCP tool result travels as a single +Server-Sent Event, and streamable-HTTP MCP clients cap one SSE event at **1 MiB** +(`DEFAULT_MAX_EVENT_SIZE_BYTES` in the reference client). An event over that cap is +discarded inside the client's SSE decoder — the caller receives no value *and* no error, +so the tool call simply never returns, even past the IPC layer's own 30s timeout. That +is issue #80's cousin for the MCP path: issue #88, where `evaluate_formula` worked up to +~22,000 rows and hung on everything larger. + +Two defenses keep that from recurring, and both must stay in place: + +1. **Sample budget** — `UltraLogApp::limit_samples` (`src/ipc/handler.rs`) reduces any + per-record series to `DEFAULT_MAX_POINTS` (2000), clamped to `MAX_POINTS_LIMIT` + (10,000), using the chart's LTTB so peaks and dropouts survive. `GetChannelData` and + `EvaluateFormula` carry an optional `max_points`; responses report `total_samples` and + `downsampled` so a caller can tell what it got. `FindPeaks` is capped at `MAX_PEAKS` + (500) because peak count scales with channel noise rather than with anything the + caller asked for; it *selects* by prominence but *returns* chronologically, and + reports `total_peaks`/`truncated` — a bare truncated list is indistinguishable from a + complete one, so a caller counting events would read exactly 500 and believe it. +2. **Byte guard** — `UltraLogMcpServer::json_result` (`src/mcp/server.rs`) serializes + compactly (never `to_string_pretty`: one array element per line roughly doubles the + payload for no benefit) and refuses anything over `MAX_RESPONSE_BYTES` (512 KiB) with + an error naming `max_points` and the time range. An honest error beats silence. + +**Corollary:** anything needing exact aggregates must read the *full* series, not a +downsampled response. `UltraLogApp::channel_series` is that accessor, and +`handle_get_channel_stats`, `handle_find_peaks` and `handle_correlate_channels` all go +through it. `handle_evaluate_formula` does not use it — it has already evaluated its own +series — but it holds the same invariant by calling `compute_stats` *before* +`limit_samples`. Either way the rule is the same: routing an aggregate through a +downsampled series would silently compute statistics over 2000 samples instead of 178,000. + +`channel_series` also enforces that `times` and `values` are the same length +(`require_aligned`). `Log::get_channel_data` is a `filter_map` that drops a row missing +the column, and an analysis-derived `cached_data` is only as long as the algorithm made +it, so a ragged log yields a pair that is *misaligned*, not merely short — and +`downsample_lttb` indexes `values` off `times.len()`, which panics on the GUI thread. +`src/ui/chart.rs` refuses to plot that case; the data API refuses to serve it. + ### UI Modules (src/ui/) UI rendering is split into focused modules that implement methods on `UltraLogApp`. The current layout is a VS Code-style activity bar + side panel; a couple of pre-activity-bar modules remain in the tree but are superseded (noted below): diff --git a/Cargo.lock b/Cargo.lock index ba0178f..808b0ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4578,7 +4578,7 @@ checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" [[package]] name = "ultralog" -version = "2.14.0" +version = "2.14.1" dependencies = [ "anyhow", "arboard", diff --git a/Cargo.toml b/Cargo.toml index cd3b42f..a22e105 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ultralog" -version = "2.14.0" +version = "2.14.1" edition = "2024" # egui/eframe 0.36 is the binding constraint on the minimum supported Rust # version; edition 2024 itself only needs 1.85. diff --git a/README.md b/README.md index 60c4d19..d1953d9 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ A high-performance, cross-platform ECU log viewer written in Rust. ![CI](https://github.com/ClassicMiniDIY/UltraLog/actions/workflows/ci.yml/badge.svg) ![License](https://img.shields.io/badge/license-AGPL--3.0-blue.svg) -![Version](https://img.shields.io/badge/version-2.14.0-green.svg) +![Version](https://img.shields.io/badge/version-2.14.1-green.svg) --- diff --git a/docs/index.html b/docs/index.html index b344939..8cd0fc8 100644 --- a/docs/index.html +++ b/docs/index.html @@ -112,8 +112,8 @@ "applicationCategory": "UtilitiesApplication", "applicationSubCategory": "Automotive Software", "operatingSystem": ["Windows 10", "Windows 11", "macOS", "Linux"], - "softwareVersion": "2.14.0", - "releaseNotes": "https://github.com/ClassicMiniDIY/UltraLog/releases/tag/v2.14.0", + "softwareVersion": "2.14.1", + "releaseNotes": "https://github.com/ClassicMiniDIY/UltraLog/releases/tag/v2.14.1", "downloadUrl": "https://github.com/ClassicMiniDIY/UltraLog/releases/latest", "installUrl": "https://github.com/ClassicMiniDIY/UltraLog/releases/latest", "screenshot": [ @@ -1427,7 +1427,7 @@

Unlock Your Performanc
New - v2.14.0 + v2.14.1 Open Source diff --git a/docs/sitemap.xml b/docs/sitemap.xml index 8e61d58..1e65b0f 100644 --- a/docs/sitemap.xml +++ b/docs/sitemap.xml @@ -3,7 +3,7 @@ xmlns:image="http://www.google.com/schemas/sitemap-image/1.1"> https://ultralog.co/ - 2026-08-26 + 2026-09-15 weekly 1.0 diff --git a/src/ipc/commands.rs b/src/ipc/commands.rs index f83128c..3a1f70f 100644 --- a/src/ipc/commands.rs +++ b/src/ipc/commands.rs @@ -2,6 +2,37 @@ use serde::{Deserialize, Serialize}; +/// Default number of samples returned by the data-bearing commands +/// (`GetChannelData`, `EvaluateFormula`) when the caller does not ask for a +/// specific count. +/// +/// This matches the chart's own LTTB budget in `src/ui/chart.rs`: 2000 points +/// is enough to see every feature of a trace and costs ~55 KB of compact JSON. +pub const DEFAULT_MAX_POINTS: usize = 2000; + +/// Hard ceiling on the caller-supplied `max_points`. +/// +/// Streamable-HTTP MCP clients drop any single SSE event larger than 1 MiB +/// (`DEFAULT_MAX_EVENT_SIZE_BYTES` in the reference client), and the drop is +/// silent: the caller never receives a result *or* an error, it just hangs +/// (issue #88). 10,000 samples is ~275 KB of compact JSON, which leaves ample +/// headroom under that limit for the surrounding envelope. +pub const MAX_POINTS_LIMIT: usize = 10_000; + +/// Maximum number of peaks returned by `FindPeaks`. +/// +/// Peak counts grow with the noise in a channel, not with anything the caller +/// controls, so this is capped for the same reason as `MAX_POINTS_LIMIT`. The +/// most prominent peaks are kept. +pub const MAX_PEAKS: usize = 500; + +/// Clamp a caller-supplied sample budget into the supported range. +pub fn resolve_max_points(requested: Option) -> usize { + requested + .unwrap_or(DEFAULT_MAX_POINTS) + .clamp(1, MAX_POINTS_LIMIT) +} + /// Commands that can be sent from the MCP server to the GUI #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", content = "payload")] @@ -27,6 +58,10 @@ pub enum IpcCommand { channel_name: String, /// Optional time range (start, end) in seconds time_range: Option<(f64, f64)>, + /// Maximum number of samples to return. Defaults to + /// [`DEFAULT_MAX_POINTS`] and is clamped to [`MAX_POINTS_LIMIT`]. + #[serde(default)] + max_points: Option, }, /// Get statistics for a channel @@ -72,6 +107,10 @@ pub enum IpcCommand { formula: String, /// Optional time range time_range: Option<(f64, f64)>, + /// Maximum number of samples to return. Defaults to + /// [`DEFAULT_MAX_POINTS`] and is clamped to [`MAX_POINTS_LIMIT`]. + #[serde(default)] + max_points: Option, }, /// Set the visible time range on the chart @@ -148,17 +187,34 @@ pub enum ResponseData { /// List of channels Channels(Vec), - /// Channel time series data - ChannelData { times: Vec, values: Vec }, + /// Channel time series data. + /// + /// `times`/`values` may be downsampled; `total_samples` is always the + /// number of records the series was drawn from. + ChannelData { + times: Vec, + values: Vec, + #[serde(default)] + total_samples: usize, + #[serde(default)] + downsampled: bool, + }, /// Channel statistics Stats(ChannelStats), - /// Formula evaluation result + /// Formula evaluation result. + /// + /// `times`/`values` may be downsampled; `stats` is always computed over + /// the full series, and `total_samples` is its length. FormulaResult { times: Vec, values: Vec, stats: ChannelStats, + #[serde(default)] + total_samples: usize, + #[serde(default)] + downsampled: bool, }, /// Values at cursor position @@ -167,8 +223,17 @@ pub enum ResponseData { /// List of computed channel templates ComputedChannels(Vec), - /// Peak detection results - Peaks(Vec), + /// Peak detection results. + /// + /// `peaks` is capped at [`MAX_PEAKS`]; `total_peaks` is how many the + /// detector actually found, and `truncated` says whether the cap bit. + Peaks { + peaks: Vec, + #[serde(default)] + total_peaks: usize, + #[serde(default)] + truncated: bool, + }, /// Correlation result Correlation { @@ -350,6 +415,7 @@ mod tests { file_id: "0".to_string(), channel_name: "RPM".to_string(), time_range: Some((10.0, 20.0)), + max_points: Some(500), }; let json = serde_json::to_string(&cmd).unwrap(); let parsed: IpcCommand = serde_json::from_str(&json).unwrap(); @@ -357,11 +423,13 @@ mod tests { file_id, channel_name, time_range, + max_points, } = parsed { assert_eq!(file_id, "0"); assert_eq!(channel_name, "RPM"); assert_eq!(time_range, Some((10.0, 20.0))); + assert_eq!(max_points, Some(500)); } else { panic!("Expected GetChannelData command"); } @@ -373,6 +441,7 @@ mod tests { file_id: "0".to_string(), channel_name: "Boost".to_string(), time_range: None, + max_points: None, }; let json = serde_json::to_string(&cmd).unwrap(); let parsed: IpcCommand = serde_json::from_str(&json).unwrap(); @@ -380,11 +449,13 @@ mod tests { file_id, channel_name, time_range, + max_points, } = parsed { assert_eq!(file_id, "0"); assert_eq!(channel_name, "Boost"); assert!(time_range.is_none()); + assert!(max_points.is_none()); } else { panic!("Expected GetChannelData command"); } @@ -488,12 +559,22 @@ mod tests { let resp = IpcResponse::ok_with_data(ResponseData::ChannelData { times: vec![0.0, 0.1, 0.2, 0.3], values: vec![1000.0, 1500.0, 2000.0, 2500.0], + total_samples: 4, + downsampled: false, }); let json = serde_json::to_string(&resp).unwrap(); let parsed: IpcResponse = serde_json::from_str(&json).unwrap(); - if let IpcResponse::Ok(Some(ResponseData::ChannelData { times, values })) = parsed { + if let IpcResponse::Ok(Some(ResponseData::ChannelData { + times, + values, + total_samples, + downsampled, + })) = parsed + { assert_eq!(times, vec![0.0, 0.1, 0.2, 0.3]); assert_eq!(values, vec![1000.0, 1500.0, 2000.0, 2500.0]); + assert_eq!(total_samples, 4); + assert!(!downsampled); } else { panic!("Expected ChannelData response"); } @@ -596,18 +677,61 @@ mod tests { prominence: 800.0, }, ]; - let resp = IpcResponse::ok_with_data(ResponseData::Peaks(peaks)); + let resp = IpcResponse::ok_with_data(ResponseData::Peaks { + peaks, + total_peaks: 2, + truncated: false, + }); let json = serde_json::to_string(&resp).unwrap(); let parsed: IpcResponse = serde_json::from_str(&json).unwrap(); - if let IpcResponse::Ok(Some(ResponseData::Peaks(p))) = parsed { + if let IpcResponse::Ok(Some(ResponseData::Peaks { + peaks: p, + total_peaks, + truncated, + })) = parsed + { assert_eq!(p.len(), 2); assert_eq!(p[0].time, 10.5); assert_eq!(p[1].value, 7500.0); + assert_eq!(total_peaks, 2); + assert!(!truncated); } else { panic!("Expected Peaks response"); } } + #[test] + fn test_truncated_peaks_response_reports_the_true_total() { + // A truncated list must not be indistinguishable from a complete one: + // a caller counting events off `peaks.len()` alone would read exactly + // MAX_PEAKS and believe the channel had no more. + let peaks: Vec = (0..MAX_PEAKS) + .map(|i| Peak { + time: i as f64, + value: 1000.0 + i as f64, + prominence: 50.0, + }) + .collect(); + let resp = IpcResponse::ok_with_data(ResponseData::Peaks { + peaks, + total_peaks: 41_337, + truncated: true, + }); + let json = serde_json::to_string(&resp).unwrap(); + let parsed: IpcResponse = serde_json::from_str(&json).unwrap(); + let IpcResponse::Ok(Some(ResponseData::Peaks { + peaks, + total_peaks, + truncated, + })) = parsed + else { + panic!("Expected Peaks response"); + }; + assert_eq!(peaks.len(), MAX_PEAKS); + assert_eq!(total_peaks, 41_337); + assert!(truncated); + } + // ======================================================================== // JSON Format Compatibility Tests // ======================================================================== @@ -641,20 +765,33 @@ mod tests { #[test] fn test_command_can_be_parsed_from_external_json() { - // Test parsing JSON that might come from an external MCP client + // Test parsing JSON that might come from an external MCP client. + // `max_points` is absent here on purpose: it is `#[serde(default)]`, so + // a payload written before the sample budget existed still parses. let json = r#"{"type":"GetChannelData","payload":{"file_id":"0","channel_name":"RPM","time_range":[0.0,10.0]}}"#; let cmd: IpcCommand = serde_json::from_str(json).unwrap(); if let IpcCommand::GetChannelData { file_id, channel_name, time_range, + max_points, } = cmd { assert_eq!(file_id, "0"); assert_eq!(channel_name, "RPM"); assert_eq!(time_range, Some((0.0, 10.0))); + assert!(max_points.is_none(), "Omitted max_points must default"); } else { panic!("Expected GetChannelData command"); } } + + #[test] + fn test_resolve_max_points_clamps_into_supported_range() { + assert_eq!(resolve_max_points(None), DEFAULT_MAX_POINTS); + assert_eq!(resolve_max_points(Some(750)), 750); + assert_eq!(resolve_max_points(Some(0)), 1, "Zero would return no data"); + assert_eq!(resolve_max_points(Some(usize::MAX)), MAX_POINTS_LIMIT); + assert_eq!(resolve_max_points(Some(MAX_POINTS_LIMIT)), MAX_POINTS_LIMIT); + } } diff --git a/src/ipc/handler.rs b/src/ipc/handler.rs index 8b8c5d9..eed3497 100644 --- a/src/ipc/handler.rs +++ b/src/ipc/handler.rs @@ -29,7 +29,8 @@ impl UltraLogApp { file_id, channel_name, time_range, - } => self.handle_get_channel_data(&file_id, &channel_name, time_range), + max_points, + } => self.handle_get_channel_data(&file_id, &channel_name, time_range, max_points), IpcCommand::GetChannelStats { file_id, @@ -66,7 +67,8 @@ impl UltraLogApp { file_id, formula, time_range, - } => self.handle_evaluate_formula(&file_id, &formula, time_range), + max_points, + } => self.handle_evaluate_formula(&file_id, &formula, time_range, max_points), IpcCommand::SetTimeRange { start, end } => self.handle_set_time_range(start, end), @@ -240,15 +242,20 @@ impl UltraLogApp { IpcResponse::ok_with_data(ResponseData::Channels(channels)) } - fn handle_get_channel_data( + /// Resolve a channel (raw or computed) to its full, un-downsampled series. + /// + /// Callers that need exact aggregates - stats, peaks, correlation - must go + /// through this rather than [`Self::handle_get_channel_data`], whose + /// response is downsampled for transport. + fn channel_series( &self, file_id: &str, channel_name: &str, time_range: Option<(f64, f64)>, - ) -> IpcResponse { + ) -> Result<(Vec, Vec), String> { let file_idx = match file_id.parse::() { Ok(idx) if idx < self.files.len() => idx, - _ => return IpcResponse::error(format!("Invalid file ID: {}", file_id)), + _ => return Err(format!("Invalid file ID: {}", file_id)), }; let file = &self.files[file_idx]; @@ -260,55 +267,136 @@ impl UltraLogApp { .iter() .position(|c| c.name().eq_ignore_ascii_case(channel_name)); - let (times, values) = if let Some(idx) = channel_idx { + if let Some(idx) = channel_idx { let all_times = file.log.get_times_as_f64().to_vec(); let all_values = file.log.get_channel_data(idx); - self.filter_by_time_range(all_times, all_values, time_range) - } else { - // Check computed channels - if let Some(computed) = self.file_computed_channels.get(&file_idx) { - if let Some(c) = computed - .iter() - .find(|c| c.name().eq_ignore_ascii_case(channel_name)) - { - if let Some(data) = &c.cached_data { - let all_times = file.log.get_times_as_f64().to_vec(); - self.filter_by_time_range(all_times, data.clone(), time_range) - } else { - return IpcResponse::error("Computed channel not evaluated yet"); - } - } else { - return IpcResponse::error(format!("Channel not found: {}", channel_name)); - } - } else { - return IpcResponse::error(format!("Channel not found: {}", channel_name)); - } + Self::require_aligned(channel_name, &all_times, &all_values)?; + return Ok(self.filter_by_time_range(all_times, all_values, time_range)); + } + + // Check computed channels + let Some(computed) = self.file_computed_channels.get(&file_idx) else { + return Err(format!("Channel not found: {}", channel_name)); + }; + let Some(c) = computed + .iter() + .find(|c| c.name().eq_ignore_ascii_case(channel_name)) + else { + return Err(format!("Channel not found: {}", channel_name)); + }; + let Some(data) = &c.cached_data else { + return Err("Computed channel not evaluated yet".to_string()); }; - IpcResponse::ok_with_data(ResponseData::ChannelData { times, values }) + let all_times = file.log.get_times_as_f64().to_vec(); + Self::require_aligned(channel_name, &all_times, data)?; + Ok(self.filter_by_time_range(all_times, data.clone(), time_range)) } - fn handle_get_channel_stats( + /// Reject a times/values pair whose lengths disagree. + /// + /// `Log::get_channel_data` is a `filter_map` that drops a row missing the + /// column entirely, and an analysis-derived `cached_data` is only as long as + /// the algorithm made it, so a ragged log can yield fewer values than times. + /// Such a pair is not merely short, it is misaligned from the first dropped + /// row onward - and `filter_by_time_range`'s `zip` would quietly paper over + /// the mismatch while `downsample_lttb` indexes `values` off `times.len()` + /// and panics on the GUI thread. + /// + /// `src/ui/chart.rs` refuses to plot this case for the same reason; the data + /// API refuses to serve it rather than return numbers attributed to the + /// wrong timestamps. + pub fn require_aligned( + channel_name: &str, + times: &[f64], + values: &[f64], + ) -> Result<(), String> { + if times.len() != values.len() { + return Err(format!( + "Channel '{}' has {} values for {} timestamps; the log rows are ragged, \ + so samples cannot be matched to times", + channel_name, + values.len(), + times.len() + )); + } + Ok(()) + } + + fn handle_get_channel_data( &self, file_id: &str, channel_name: &str, time_range: Option<(f64, f64)>, + max_points: Option, ) -> IpcResponse { - // First get the data - let data_response = self.handle_get_channel_data(file_id, channel_name, time_range); + let (times, values) = match self.channel_series(file_id, channel_name, time_range) { + Ok(series) => series, + Err(message) => return IpcResponse::error(message), + }; - match data_response { - IpcResponse::Ok(Some(ResponseData::ChannelData { times, values })) => { - if values.is_empty() { - return IpcResponse::error("No data in range"); - } + let total_samples = times.len(); + let (times, values, downsampled) = Self::limit_samples(times, values, max_points); - let stats = self.compute_stats(×, &values); - IpcResponse::ok_with_data(ResponseData::Stats(stats)) - } - IpcResponse::Error { message } => IpcResponse::error(message), - _ => IpcResponse::error("Unexpected response"), + IpcResponse::ok_with_data(ResponseData::ChannelData { + times, + values, + total_samples, + downsampled, + }) + } + + /// Reduce a series to at most `max_points` samples for transport. + /// + /// An unbounded per-record array is not deliverable over MCP: streamable-HTTP + /// clients silently discard any SSE event above 1 MiB, leaving the caller + /// hanging with neither a result nor an error (issue #88). LTTB is used + /// rather than plain striding so peaks and dropouts survive the reduction. + /// + /// Returns the reduced series plus whether any reduction actually happened. + pub fn limit_samples( + times: Vec, + values: Vec, + max_points: Option, + ) -> (Vec, Vec, bool) { + let budget = resolve_max_points(max_points); + if times.len() <= budget { + return (times, values, false); + } + + // LTTB needs at least 3 buckets; below that, keep evenly spaced samples. + if budget < 3 { + let step = times.len().div_ceil(budget); + let t = times.iter().step_by(step).copied().take(budget).collect(); + let v = values.iter().step_by(step).copied().take(budget).collect(); + return (t, v, true); + } + + let points = Self::downsample_lttb(×, &values, budget); + let t = points.iter().map(|p| p[0]).collect(); + let v = points.iter().map(|p| p[1]).collect(); + (t, v, true) + } + + fn handle_get_channel_stats( + &self, + file_id: &str, + channel_name: &str, + time_range: Option<(f64, f64)>, + ) -> IpcResponse { + // Stats must be exact, so they run over the full series rather than the + // downsampled payload `handle_get_channel_data` returns. + let (times, values) = match self.channel_series(file_id, channel_name, time_range) { + Ok(series) => series, + Err(message) => return IpcResponse::error(message), + }; + + if values.is_empty() { + return IpcResponse::error("No data in range"); } + + let stats = self.compute_stats(×, &values); + IpcResponse::ok_with_data(ResponseData::Stats(stats)) } fn handle_select_channel(&mut self, file_id: &str, channel_name: &str) -> IpcResponse { @@ -517,6 +605,7 @@ impl UltraLogApp { file_id: &str, formula: &str, time_range: Option<(f64, f64)>, + max_points: Option, ) -> IpcResponse { let file_idx = match file_id.parse::() { Ok(idx) if idx < self.files.len() => idx, @@ -555,14 +644,27 @@ impl UltraLogApp { }; let all_times = file.log.get_times_as_f64().to_vec(); + // Same alignment precondition as `channel_series`: the evaluator emits + // one value per record, so a disagreement here means the log itself is + // ragged and `limit_samples` would index past the end of `values`. + if let Err(e) = Self::require_aligned(formula, &all_times, &all_values) { + return IpcResponse::error(e); + } let (times, values) = self.filter_by_time_range(all_times, all_values, time_range); + // Stats are computed before downsampling so they describe every record + // in range, not just the samples that fit in the response. let stats = self.compute_stats(×, &values); + let total_samples = times.len(); + let (times, values, downsampled) = Self::limit_samples(times, values, max_points); + IpcResponse::ok_with_data(ResponseData::FormulaResult { times, values, stats, + total_samples, + downsampled, }) } @@ -635,16 +737,43 @@ impl UltraLogApp { channel_name: &str, min_prominence: Option, ) -> IpcResponse { - let data_response = self.handle_get_channel_data(file_id, channel_name, None); + let (times, values) = match self.channel_series(file_id, channel_name, None) { + Ok(series) => series, + Err(message) => return IpcResponse::error(message), + }; - match data_response { - IpcResponse::Ok(Some(ResponseData::ChannelData { times, values })) => { - let peaks = self.find_peaks_in_data(×, &values, min_prominence.unwrap_or(0.1)); - IpcResponse::ok_with_data(ResponseData::Peaks(peaks)) - } - IpcResponse::Error { message } => IpcResponse::error(message), - _ => IpcResponse::error("Unexpected response"), + let mut peaks = self.find_peaks_in_data(×, &values, min_prominence.unwrap_or(0.1)); + + // Peak count scales with channel noise, not with anything the caller + // asked for, so a noisy channel can otherwise produce a payload too + // large to deliver (issue #88). Select the most prominent, then restore + // chronological order so the series still reads as a timeline. + // + // `total_peaks` is reported alongside because a bare truncated list is + // indistinguishable from a complete one: a caller asked "how many boost + // spikes?" would read exactly MAX_PEAKS off a channel with thousands and + // believe it. + let total_peaks = peaks.len(); + let truncated = total_peaks > MAX_PEAKS; + if truncated { + peaks.sort_by(|a, b| { + b.prominence + .partial_cmp(&a.prominence) + .unwrap_or(std::cmp::Ordering::Equal) + }); + peaks.truncate(MAX_PEAKS); + peaks.sort_by(|a, b| { + a.time + .partial_cmp(&b.time) + .unwrap_or(std::cmp::Ordering::Equal) + }); } + + IpcResponse::ok_with_data(ResponseData::Peaks { + peaks, + total_peaks, + truncated, + }) } fn handle_correlate_channels( @@ -653,31 +782,26 @@ impl UltraLogApp { channel_a: &str, channel_b: &str, ) -> IpcResponse { - let data_a = self.handle_get_channel_data(file_id, channel_a, None); - let data_b = self.handle_get_channel_data(file_id, channel_b, None); - - match (data_a, data_b) { - ( - IpcResponse::Ok(Some(ResponseData::ChannelData { values: a, .. })), - IpcResponse::Ok(Some(ResponseData::ChannelData { values: b, .. })), - ) => { - if a.len() != b.len() || a.is_empty() { - return IpcResponse::error("Channels have different lengths or are empty"); - } - - let coefficient = self.compute_correlation(&a, &b); - let interpretation = self.interpret_correlation(coefficient); + let (_, a) = match self.channel_series(file_id, channel_a, None) { + Ok(series) => series, + Err(message) => return IpcResponse::error(message), + }; + let (_, b) = match self.channel_series(file_id, channel_b, None) { + Ok(series) => series, + Err(message) => return IpcResponse::error(message), + }; - IpcResponse::ok_with_data(ResponseData::Correlation { - coefficient, - interpretation, - }) - } - (IpcResponse::Error { message }, _) | (_, IpcResponse::Error { message }) => { - IpcResponse::error(message) - } - _ => IpcResponse::error("Unexpected response"), + if a.len() != b.len() || a.is_empty() { + return IpcResponse::error("Channels have different lengths or are empty"); } + + let coefficient = self.compute_correlation(&a, &b); + let interpretation = self.interpret_correlation(coefficient); + + IpcResponse::ok_with_data(ResponseData::Correlation { + coefficient, + interpretation, + }) } fn handle_show_scatter_plot( diff --git a/src/mcp/server.rs b/src/mcp/server.rs index ea151ac..e841bd7 100644 --- a/src/mcp/server.rs +++ b/src/mcp/server.rs @@ -23,7 +23,29 @@ use tokio::sync::oneshot; use super::client::GuiClient; use crate::ipc::DEFAULT_IPC_PORT; -use crate::ipc::commands::{IpcCommand, IpcResponse, ResponseData}; +use crate::ipc::commands::{ + DEFAULT_MAX_POINTS, IpcCommand, IpcResponse, MAX_PEAKS, MAX_POINTS_LIMIT, ResponseData, +}; + +/// Maximum size of a single tool-result payload, in bytes. +/// +/// Streamable-HTTP MCP clients cap a single SSE event at 1 MiB +/// (`DEFAULT_MAX_EVENT_SIZE_BYTES` in the reference client). An event above the +/// cap is discarded by the client's SSE decoder without surfacing anything to +/// the caller, so an oversized tool result reads as a hang: no value, no error, +/// not even after the IPC layer's own 30s timeout would have fired (issue #88). +/// +/// Payloads are bounded well before this point by the sample budget in +/// `src/ipc/handler.rs`; this is the backstop that turns any remaining +/// oversized response into an actionable error instead of silence. +pub const MAX_RESPONSE_BYTES: usize = 512 * 1024; + +// The sample budget is quoted as a literal in the `#[schemars(description)]` +// attributes below, which cannot interpolate constants. Fail the build if the +// constants move so the tool schemas can never advertise stale numbers. +const _: () = assert!(DEFAULT_MAX_POINTS == 2000, "update the tool descriptions"); +const _: () = assert!(MAX_POINTS_LIMIT == 10_000, "update the tool descriptions"); +const _: () = assert!(MAX_PEAKS == 500, "update the find_peaks description"); /// Default port for the MCP HTTP server /// Port 52453 = 5-2-4-5-3, a nod to the 1-2-4-5-3 firing order of legendary inline-5 engines @@ -160,6 +182,30 @@ impl UltraLogMcpServer { .map_err(Self::mcp_error) } + /// Serialize a tool result compactly and refuse to emit anything the + /// transport would silently drop. + /// + /// Compact rather than pretty: pretty-printing a numeric array puts one + /// value per line, which roughly doubles the payload for no benefit to the + /// caller. + pub fn json_result( + value: &T, + ) -> Result { + let text = serde_json::to_string(value) + .map_err(|e| Self::mcp_error(format!("Failed to serialize response: {}", e)))?; + + if text.len() > MAX_RESPONSE_BYTES { + return Err(Self::mcp_error(format!( + "Response is {} bytes, over the {} byte limit this transport can deliver. \ + Narrow the time range (start_time/end_time) or lower max_points.", + text.len(), + MAX_RESPONSE_BYTES + ))); + } + + Ok(CallToolResult::success(vec![ContentBlock::text(text)])) + } + fn mcp_error(message: impl Into) -> McpError { McpError { code: ErrorCode(-32603), @@ -211,6 +257,11 @@ pub struct ChannelDataRequest { #[schemars(description = "Optional end time in seconds")] #[serde(default)] pub end_time: Option, + #[schemars( + description = "Maximum samples to return (default 2000, max 10000). Longer series are downsampled with LTTB, which preserves peaks and dropouts. Use start_time/end_time for full resolution over a narrower window." + )] + #[serde(default)] + pub max_points: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -240,6 +291,11 @@ pub struct EvaluateFormulaRequest { #[schemars(description = "Optional end time in seconds")] #[serde(default)] pub end_time: Option, + #[schemars( + description = "Maximum samples to return (default 2000, max 10000). Longer series are downsampled with LTTB; the returned stats are always computed over every record in range, not just the returned samples." + )] + #[serde(default)] + pub max_points: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -317,11 +373,7 @@ impl UltraLogMcpServer { Parameters(_): Parameters, ) -> Result { match self.send_command_async(IpcCommand::GetState).await? { - IpcResponse::Ok(Some(ResponseData::State(state))) => { - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&state).unwrap_or_default(), - )])) - } + IpcResponse::Ok(Some(ResponseData::State(state))) => Self::json_result(&state), IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), } @@ -338,11 +390,7 @@ impl UltraLogMcpServer { .send_command_async(IpcCommand::LoadFile { path: req.path }) .await? { - IpcResponse::Ok(Some(ResponseData::FileLoaded(info))) => { - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&info).unwrap_or_default(), - )])) - } + IpcResponse::Ok(Some(ResponseData::FileLoaded(info))) => Self::json_result(&info), IpcResponse::Ok(Some(ResponseData::Ack)) => { Ok(CallToolResult::success(vec![ContentBlock::text( "File is being loaded. Use get_state to check when ready.", @@ -384,18 +432,14 @@ impl UltraLogMcpServer { }) .await? { - IpcResponse::Ok(Some(ResponseData::Channels(channels))) => { - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&channels).unwrap_or_default(), - )])) - } + IpcResponse::Ok(Some(ResponseData::Channels(channels))) => Self::json_result(&channels), IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), } } #[tool( - description = "Get time series data for a specific channel. Optionally filter by time range." + description = "Get time series data for a specific channel. Optionally filter by time range. Returns at most max_points samples (default 2000); longer series are downsampled with LTTB and the response reports total_samples and downsampled so you can tell." )] async fn get_channel_data( &self, @@ -411,19 +455,22 @@ impl UltraLogMcpServer { file_id: req.file_id, channel_name: req.channel_name, time_range, + max_points: req.max_points, }) .await? { - IpcResponse::Ok(Some(ResponseData::ChannelData { times, values })) => { - let result = serde_json::json!({ - "sample_count": times.len(), - "times": times, - "values": values - }); - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&result).unwrap_or_default(), - )])) - } + IpcResponse::Ok(Some(ResponseData::ChannelData { + times, + values, + total_samples, + downsampled, + })) => Self::json_result(&serde_json::json!({ + "sample_count": times.len(), + "total_samples": total_samples, + "downsampled": downsampled, + "times": times, + "values": values + })), IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), } @@ -447,11 +494,7 @@ impl UltraLogMcpServer { }) .await? { - IpcResponse::Ok(Some(ResponseData::Stats(stats))) => { - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&stats).unwrap_or_default(), - )])) - } + IpcResponse::Ok(Some(ResponseData::Stats(stats))) => Self::json_result(&stats), IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), } @@ -564,9 +607,7 @@ impl UltraLogMcpServer { .await? { IpcResponse::Ok(Some(ResponseData::ComputedChannels(channels))) => { - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&channels).unwrap_or_default(), - )])) + Self::json_result(&channels) } IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), @@ -574,7 +615,7 @@ impl UltraLogMcpServer { } #[tool( - description = "Evaluate a mathematical formula against the log data without creating a permanent channel. Returns the computed values and statistics." + description = "Evaluate a mathematical formula against the log data without creating a permanent channel. Returns the computed values and statistics. Returns at most max_points samples (default 2000); longer series are downsampled with LTTB, but the statistics always cover every record in range." )] async fn evaluate_formula( &self, @@ -590,6 +631,7 @@ impl UltraLogMcpServer { file_id: req.file_id, formula: req.formula, time_range, + max_points: req.max_points, }) .await? { @@ -597,17 +639,16 @@ impl UltraLogMcpServer { times, values, stats, - })) => { - let result = serde_json::json!({ - "sample_count": times.len(), - "stats": stats, - "times": times, - "values": values - }); - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&result).unwrap_or_default(), - )])) - } + total_samples, + downsampled, + })) => Self::json_result(&serde_json::json!({ + "sample_count": times.len(), + "total_samples": total_samples, + "downsampled": downsampled, + "stats": stats, + "times": times, + "values": values + })), IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), } @@ -705,18 +746,14 @@ impl UltraLogMcpServer { }) .await? { - IpcResponse::Ok(Some(ResponseData::CursorValues(values))) => { - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&values).unwrap_or_default(), - )])) - } + IpcResponse::Ok(Some(ResponseData::CursorValues(values))) => Self::json_result(&values), IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), } } #[tool( - description = "Find peaks (local maxima) in a channel. Useful for finding acceleration events, boost spikes, etc." + description = "Find peaks (local maxima) in a channel. Useful for finding acceleration events, boost spikes, etc. At most 500 peaks are returned, selected by prominence and listed chronologically; check total_peaks and truncated before treating peak_count as an event count, and raise min_prominence to cut noise." )] async fn find_peaks( &self, @@ -730,11 +767,16 @@ impl UltraLogMcpServer { }) .await? { - IpcResponse::Ok(Some(ResponseData::Peaks(peaks))) => { - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&peaks).unwrap_or_default(), - )])) - } + IpcResponse::Ok(Some(ResponseData::Peaks { + peaks, + total_peaks, + truncated, + })) => Self::json_result(&serde_json::json!({ + "peak_count": peaks.len(), + "total_peaks": total_peaks, + "truncated": truncated, + "peaks": peaks, + })), IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), } @@ -763,9 +805,7 @@ impl UltraLogMcpServer { "coefficient": coefficient, "interpretation": interpretation }); - Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&result).unwrap_or_default(), - )])) + Self::json_result(&result) } IpcResponse::Error { message } => Err(Self::mcp_error(message)), _ => Err(Self::mcp_error("Unexpected response")), diff --git a/tests/core/mcp_tests.rs b/tests/core/mcp_tests.rs index 0ee54e5..8682945 100644 --- a/tests/core/mcp_tests.rs +++ b/tests/core/mcp_tests.rs @@ -12,9 +12,13 @@ use std::io::{BufRead, BufReader, Write}; use std::net::TcpStream; use std::time::Duration; +use ultralog::app::UltraLogApp; use ultralog::ipc::IpcServer; -use ultralog::ipc::commands::{IpcCommand, IpcResponse, ResponseData}; +use ultralog::ipc::commands::{ + ChannelStats, DEFAULT_MAX_POINTS, IpcCommand, IpcResponse, MAX_POINTS_LIMIT, ResponseData, +}; use ultralog::mcp::UltraLogMcpServer; +use ultralog::mcp::server::MAX_RESPONSE_BYTES; use rmcp::ServerHandler; @@ -384,3 +388,298 @@ fn test_mcp_server_handle_url() { assert_eq!(handle.port(), mcp_port); assert_eq!(handle.url(), format!("http://127.0.0.1:{}/mcp", mcp_port)); } + +// ============================================================================ +// Response Size Budget Tests (issue #88) +// ============================================================================ +// +// Streamable-HTTP MCP clients cap a single SSE event at 1 MiB and drop anything +// larger inside their SSE decoder, so an oversized tool result surfaces to the +// caller as neither a value nor an error - the call simply never returns. +// `evaluate_formula` and `get_channel_data` used to serialize one entry per log +// record, so any log past ~22,000 rows crossed that cap and hung. These tests +// pin the two defenses: a sample budget on the data itself, and a hard byte +// guard on the serialized payload. + +/// The SSE event size limit that motivated the budget, in bytes. +const SSE_EVENT_LIMIT: usize = 1024 * 1024; + +#[test] +fn test_limit_samples_defaults_to_budget() { + let n = 178_000; + let times: Vec = (0..n).map(|i| i as f64 * 0.01).collect(); + let values: Vec = (0..n).map(|i| (i as f64).sin()).collect(); + + let (t, v, downsampled) = UltraLogApp::limit_samples(times, values, None); + + assert!(downsampled, "A 178k-record series must report downsampling"); + assert_eq!(t.len(), DEFAULT_MAX_POINTS); + assert_eq!(v.len(), DEFAULT_MAX_POINTS); +} + +#[test] +fn test_limit_samples_leaves_short_series_untouched() { + let times: Vec = (0..500).map(|i| i as f64).collect(); + let values: Vec = (0..500).map(|i| i as f64 * 2.0).collect(); + + let (t, v, downsampled) = UltraLogApp::limit_samples(times.clone(), values.clone(), None); + + assert!( + !downsampled, + "A series under budget must not be downsampled" + ); + assert_eq!(t, times); + assert_eq!(v, values); +} + +#[test] +fn test_limit_samples_clamps_request_to_ceiling() { + let n = 200_000; + let times: Vec = (0..n).map(|i| i as f64 * 0.01).collect(); + let values: Vec = (0..n).map(|i| (i as f64).cos()).collect(); + + let (t, _, downsampled) = UltraLogApp::limit_samples(times, values, Some(usize::MAX)); + + assert!(downsampled); + assert_eq!( + t.len(), + MAX_POINTS_LIMIT, + "An unbounded max_points must clamp to the ceiling, not honour the request" + ); +} + +#[test] +fn test_limit_samples_preserves_endpoints() { + let n = 50_000; + let times: Vec = (0..n).map(|i| i as f64 * 0.1).collect(); + let values: Vec = (0..n).map(|i| i as f64).collect(); + let (first_t, last_t) = (times[0], times[n - 1]); + let (first_v, last_v) = (values[0], values[n - 1]); + + let (t, v, _) = UltraLogApp::limit_samples(times, values, Some(1000)); + + assert_eq!(t.first().copied(), Some(first_t)); + assert_eq!(t.last().copied(), Some(last_t)); + assert_eq!(v.first().copied(), Some(first_v)); + assert_eq!(v.last().copied(), Some(last_v)); +} + +#[test] +fn test_limit_samples_handles_tiny_budgets() { + let times: Vec = (0..10_000).map(|i| i as f64).collect(); + let values: Vec = (0..10_000).map(|i| i as f64).collect(); + + for budget in [1usize, 2, 3] { + let (t, v, downsampled) = + UltraLogApp::limit_samples(times.clone(), values.clone(), Some(budget)); + assert!(downsampled); + assert_eq!( + t.len(), + budget, + "budget {} should be honoured exactly", + budget + ); + assert_eq!(v.len(), budget); + } +} + +#[test] +fn test_max_budget_payload_fits_under_sse_event_limit() { + // Worst case a caller can ask for: the ceiling, with wide values that + // serialize to long decimal expansions. + let times: Vec = (0..MAX_POINTS_LIMIT) + .map(|i| i as f64 * 0.123_456_789_012) + .collect(); + let values: Vec = (0..MAX_POINTS_LIMIT) + .map(|i| (i as f64).sin() * -123_456.789_012_345) + .collect(); + + let payload = serde_json::json!({ + "sample_count": times.len(), + "total_samples": 500_000, + "downsampled": true, + "stats": { + "min": -1.0, "max": 1.0, "mean": 0.5, "std_dev": 0.1, + "median": 0.5, "count": 500_000, "min_time": 0.0, "max_time": 1.0 + }, + "times": times, + "values": values, + }); + let encoded = serde_json::to_string(&payload).unwrap(); + + assert!( + encoded.len() <= MAX_RESPONSE_BYTES, + "Worst-case payload is {} bytes, over the {} byte guard", + encoded.len(), + MAX_RESPONSE_BYTES + ); + assert!( + encoded.len() < SSE_EVENT_LIMIT, + "Worst-case payload is {} bytes, at or over the {} byte SSE event limit", + encoded.len(), + SSE_EVENT_LIMIT + ); + const { + assert!( + MAX_RESPONSE_BYTES < SSE_EVENT_LIMIT, + "The guard must sit below the limit it is protecting against" + ) + }; +} + +#[test] +fn test_json_result_rejects_oversized_payload() { + let oversized = serde_json::json!({ "values": vec![1.234_567_890_123_f64; 200_000] }); + let encoded_len = serde_json::to_string(&oversized).unwrap().len(); + assert!( + encoded_len > MAX_RESPONSE_BYTES, + "Fixture must actually exceed the guard (was {} bytes)", + encoded_len + ); + + let err = UltraLogMcpServer::json_result(&oversized) + .expect_err("An oversized payload must be refused, not emitted"); + + // The caller has to be told what to do about it, since the transport would + // otherwise drop the event with no diagnostic at all. + assert!( + err.message.contains("max_points") && err.message.contains("time range"), + "Error should name the knobs that fix it, got: {}", + err.message + ); +} + +#[test] +fn test_json_result_accepts_budgeted_payload() { + let times: Vec = (0..DEFAULT_MAX_POINTS).map(|i| i as f64 * 0.01).collect(); + let payload = serde_json::json!({ "sample_count": times.len(), "times": times }); + + let result = UltraLogMcpServer::json_result(&payload).expect("Budgeted payload must be sent"); + assert_eq!(result.is_error, Some(false)); +} + +#[test] +fn test_evaluate_formula_response_roundtrips_at_full_scale() { + // End-to-end over the real IPC transport: a 178,000-record log (the size + // reported in issue #88) must come back bounded and fast. + let port = find_available_port(); + let server = IpcServer::start_on_port(port).expect("Failed to start server"); + std::thread::sleep(Duration::from_millis(200)); + + std::thread::spawn(move || { + let deadline = std::time::Instant::now() + Duration::from_secs(20); + while std::time::Instant::now() < deadline { + if let Some((command, response_tx)) = server.poll_command() { + let IpcCommand::EvaluateFormula { max_points, .. } = command else { + let _ = response_tx.send(IpcResponse::error("Unexpected command")); + continue; + }; + let n = 178_000; + let times: Vec = (0..n).map(|i| i as f64 * 0.01).collect(); + let values: Vec = (0..n).map(|i| (i as f64).sin() * 1234.5678).collect(); + let total_samples = times.len(); + let (times, values, downsampled) = + UltraLogApp::limit_samples(times, values, max_points); + let _ = response_tx.send(IpcResponse::ok_with_data(ResponseData::FormulaResult { + times, + values, + stats: ChannelStats { + min: -1234.5678, + max: 1234.5678, + mean: 0.0, + std_dev: 1.0, + median: 0.0, + count: total_samples, + min_time: 0.0, + max_time: 1779.99, + }, + total_samples, + downsampled, + })); + return; + } + std::thread::sleep(Duration::from_millis(5)); + } + }); + + let client = ultralog::mcp::client::GuiClient::with_port(port); + let response = client + .send_command(IpcCommand::EvaluateFormula { + file_id: "0".to_string(), + formula: "RPM * 2".to_string(), + time_range: None, + max_points: None, + }) + .expect("Full-scale evaluate_formula must return a response"); + + let IpcResponse::Ok(Some(ResponseData::FormulaResult { + times, + values, + stats, + total_samples, + downsampled, + })) = response + else { + panic!("Expected FormulaResult, got {:?}", response); + }; + + assert_eq!(total_samples, 178_000, "The true record count must survive"); + assert_eq!(stats.count, 178_000, "Stats must describe every record"); + assert!(downsampled); + assert_eq!(times.len(), DEFAULT_MAX_POINTS); + assert_eq!(values.len(), DEFAULT_MAX_POINTS); + + let encoded = serde_json::to_string(&serde_json::json!({ + "sample_count": times.len(), + "total_samples": total_samples, + "downsampled": downsampled, + "stats": stats, + "times": times, + "values": values, + })) + .unwrap(); + assert!( + encoded.len() < SSE_EVENT_LIMIT, + "Full-scale response is {} bytes, which the transport would drop", + encoded.len() + ); +} + +#[test] +fn test_require_aligned_rejects_ragged_series() { + // `Log::get_channel_data` is a `filter_map` that drops a row missing the + // column, so a ragged log yields fewer values than times - misaligned, not + // merely short. Feeding that pair to LTTB indexes `values` off `times.len()` + // and panics on the GUI thread, so it has to be refused up front. + let times: Vec = (0..100).map(|i| i as f64).collect(); + let values: Vec = (0..97).map(|i| i as f64).collect(); + + let err = UltraLogApp::require_aligned("MAP", ×, &values) + .expect_err("A ragged series must be refused"); + assert!( + err.contains("MAP") && err.contains("97") && err.contains("100"), + "Error should name the channel and both counts, got: {}", + err + ); + + UltraLogApp::require_aligned("MAP", ×, ×).expect("Aligned series must pass"); + UltraLogApp::require_aligned("MAP", &[], &[]).expect("Empty series must pass"); +} + +#[test] +fn test_limit_samples_never_panics_on_aligned_series() { + // Guards the LTTB call itself: the budget path must hold for a range of + // lengths straddling DEFAULT_MAX_POINTS, not just the big ones. + for n in [0usize, 1, 2, 3, 4, 1999, 2000, 2001, 5000] { + let times: Vec = (0..n).map(|i| i as f64 * 0.01).collect(); + let values: Vec = (0..n).map(|i| (i as f64).sin()).collect(); + let (t, v, _) = UltraLogApp::limit_samples(times, values, None); + assert_eq!(t.len(), v.len(), "n={} produced a misaligned result", n); + assert_eq!( + t.len(), + n.min(DEFAULT_MAX_POINTS), + "n={} exceeded budget", + n + ); + } +}