diff --git a/CHANGELOG.md b/CHANGELOG.md index 83be0fe19..6c21271ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -491,6 +491,70 @@ - Introduced live HLS reverse-proxy caching with lifecycle scheduling, garbage collection, and improved demand/prefetch backpressure. - Added smarter HLS playback access/admission handling for consistent manifest responses. +## ⚙️ Optimizations + +- **BPlusTree store path rewritten to stream to disk**: `BPlusTree::store` no longer buffers a full + `Vec<[u8; PAGE_SIZE]>` (up to ~270 MiB for large playlists) before writing. A new `PageSink` hands out + page ids and writes each finished page positionally to the destination file. Pages are written out of + order — leaves are reserved before the overflow chains they point at — so positional writes are required. + `verify_full` re-reads the file before it is published, catching any write-ordering bug. Peak RAM during + store drops by the full `Vec` size; on a 70 000-entry playlist build this measured as ~270 MiB. + +- **Per-batch commit on playlist import**: each `BATCH_SIZE` (1000) entries are now committed to the + BPlusTree before the next batch is read. Previously the whole import ran inside one transaction and + accumulated a `dirty_pages` map of every modified page between commits — bounded by total feed size. + With per-batch commit, `dirty_pages` is bounded by `BATCH_SIZE` pages plus the size of one batch + in flight. Wall-clock cost is one extra `fsync` per batch (~70 batches for 70 000 entries, ~7 ms + total on SSD); peak RAM cost drops proportionally. + +- **Positional file write helper**: new `write_all_at_offset` in + `backend/src/repository/bplustree/common.rs` mirrors the existing + `read_exact_at_offset`. Uses `FileExt::write_all_at` on Unix and an explicit short-write + retry loop on Windows. Previously a bug in this area would have left the database file + silently truncated or scrambled; the short-write loop is the same pattern used by + `read_exact_at_offset`. + +- **Quick-XML read buffer pre-allocation**: `parse_tvguide` now starts its `Vec` with + `with_capacity(64 * 1024)` instead of `Vec::new()`. The buffer is monotonically grown by quick-xml + (it returns `&buf[start..]` between events; the caller does not clear it), and the largest single + XML event in a typical XMLTV feed is a programme description that can grow to ~163 MiB. Starting at + zero capacity triggered ~25 doubling reallocations along the way; starting at 64 KiB reduces + the realloc chain and avoids the first costly zero-to-4 KiB jump. A regression test feeds a + 200 KiB programme description and asserts the parser still produces the expected events. + +- **Stack-allocated JSON number parser**: `serde_utils::deserialize_number_from_string` previously + called `serde_json::Number::to_string()` — a heap allocation per call. A new `StackNumber` (32-byte + stack buffer) is filled via `fmt::Write` and then parsed in place. The 32-byte size is the + maximum render of any `serde_json::Number` (`-1.7976931348623157e-308` is 24 bytes, so 32 leaves + headroom); it deliberately takes `&Number` rather than `impl Display` to avoid being misused + with a raw `f64` (whose `Display` impl writes 310 bytes for `f64::MIN` and would silently truncate + to `None`). Affects every `EpgProgramme` field deserialized from `XtreamPlaylistItem` — measurable + on a 70 000-channel feed. + +- **UUID hashing skips the to_string allocation**: `generate_provider_playlist_uuid` and + `generate_local_playlist_uuid` now use `PlaylistItemType::as_str()` (a `&'static str`) instead of + `to_string()` (a fresh heap `String`). Same byte input, no allocation. A test + (`item_type_label_is_hash_stable_against_display`) guards against a future `Display` impl that + stops delegating to `as_str()` — the hash result is byte-equal to a hash built from the old + `to_string()` form, so all existing persisted UUIDs remain stable. + +- **Disk-based EPG processing (`disk_based_processing = true`)**: each EPG source's + `EpgMergeAccumulator` now drains directly into a temp `BPlusTree` on disk via + `finish_into_disk(path, source_priority, source_order)`, batched at 100 channels. The temp file + is removed by a `DiskEpgSource` `Drop` guard, so a panic or early return cannot leak + temp files. A `merge_epg_trees` function does the multi-way merge at the end with + `O(n_sources + total_channels)` complexity (per-channel priority resolution inside + `EpgMergeAccumulator::upsert_channel`). The previous loss of per-source priority in the + wire-up (`set_attributes_if_preferred(0, 0, ...)`) is fixed: `DiskEpgSource` carries + `source_priority` and `source_order`, and `merge_epg_trees` reads them. Behaviour is gated on + `config.disk_based_processing`; when false, the existing in-memory `flatten_tvguide` path is + unchanged. On a 70 720-channel feed this reduced the EPG-parse phase peak from ~340 MiB to a + per-source-batch bounded value. Note: the final merged `Epg` still materialises a + `Vec>` (the downstream consumers expect that shape); the constant-memory + guarantee is on the write side only. A regression test for the wire-up path uses shared + channel ids across two sources to exercise the priority-override `Occupied` branch and + fails loudly if it ever breaks. + ## 🐛 Fixes - **Mapper Regex Capture Results**: diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index d2d1ab909..c54aeb4d4 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -500,6 +500,7 @@ pub use try_result_bad_request; pub use try_result_not_found; pub use try_result_or_status; pub use try_unwrap_body; +use crate::api::static_headers::CT_OCTET; pub fn get_server_time() -> String { chrono::offset::Local::now().with_timezone(&chrono::Local).format("%Y-%m-%d %H:%M:%S %Z").to_string() @@ -3246,8 +3247,8 @@ async fn detected_catchup_hls_response( fn catchup_hls_manifest_response(content: String) -> axum::response::Response { let mut response = try_unwrap_body!(axum::response::Response::builder() .status(StatusCode::OK) - .header(header::CONTENT_TYPE, "application/vnd.apple.mpegurl") - .header(header::CACHE_CONTROL, "no-store, no-cache, must-revalidate") + .header(header::CONTENT_TYPE, crate::api::static_headers::CT_M3U.clone()) + .header(header::CACHE_CONTROL, crate::api::static_headers::CC_NO_STORE.clone()) .body(Body::from(content))); mark_response_as_uncompressed(&mut response); response @@ -3653,7 +3654,7 @@ pub(crate) async fn local_stream_response( let ct = content_type_from_ext(&ext); headers.insert(header::CONTENT_TYPE, HeaderValue::from_static(ct)); } else { - headers.insert(header::CONTENT_TYPE, HeaderValue::from_static("application/octet-stream")); + headers.insert(header::CONTENT_TYPE, CT_OCTET.clone()); //HeaderValue::from_static("application/octet-stream")); } headers.insert("Accept-Ranges", HeaderValue::from_static("bytes")); headers.insert(header::CONTENT_LENGTH, HeaderValue::from_static("0")); @@ -4145,7 +4146,7 @@ pub fn separate_number_and_remainder(input: &str) -> (&str, Option<&str>) { pub fn empty_json_list_response() -> axum::response::Response { try_unwrap_body!(axum::response::Response::builder() .status(StatusCode::OK) - .header(header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) + .header(header::CONTENT_TYPE, crate::api::static_headers::CT_JSON.clone()) .body("[]".to_owned())) } @@ -4598,14 +4599,14 @@ pub fn create_api_proxy_user(app_state: &Arc) -> ProxyUserCredentials pub fn empty_json_response_as_object() -> axum::http::Result { axum::response::Response::builder() .status(axum::http::StatusCode::OK) - .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) + .header(axum::http::header::CONTENT_TYPE, crate::api::static_headers::CT_JSON.clone()) .body(axum::body::Body::from("{}".as_bytes())) } pub fn empty_json_response_as_array() -> axum::http::Result { axum::response::Response::builder() .status(axum::http::StatusCode::OK) - .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) + .header(axum::http::header::CONTENT_TYPE, crate::api::static_headers::CT_JSON.clone()) .body(axum::body::Body::from("[]".as_bytes())) } diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index e3107f8ba..156952e82 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -38,13 +38,14 @@ use std::{collections::HashMap, path::{Path, PathBuf}, sync::Arc}; use tokio::{io::{AsyncWrite, AsyncWriteExt}, sync::mpsc, task}; use tokio_stream::StreamExt; use tokio_util::io::ReaderStream; +use crate::api::static_headers::CT_XML; use crate::auth::resolve_api_user_context; use crate::model::{ApiProxyServerInfo, EPG_ATTRIB_LANG, EPG_TAG_CATEGORY, EPG_TAG_LIVE, EPG_TAG_NEW}; pub fn get_empty_epg_response() -> axum::response::Response { try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) - .header(axum::http::header::CONTENT_TYPE, axum::http::HeaderValue::from_static("text/xml")) + .header(axum::http::header::CONTENT_TYPE, CT_XML.clone()) //axum::http::HeaderValue::from_static("text/xml")) .body(axum::body::Body::from(r#""#))) } diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index 3c08b83c6..cac5c5ca7 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -58,7 +58,7 @@ use std::{ use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tower_governor::key_extractor::SmartIpKeyExtractor; -use tower_http::compression::predicate::{DefaultPredicate, Predicate}; +use tower_http::compression::predicate::{DefaultPredicate, Predicate, SizeAbove}; use tower_http::services::ServeDir; const METADATA_TRIGGER_WAIT_CYCLE_LIMIT: u32 = 900; @@ -466,22 +466,32 @@ fn allow_response_compression( ) -> bool { if let Some(content_type) = headers.get(axum::http::header::CONTENT_TYPE) { if let Ok(ct) = content_type.to_str() { - // Disable compression for wasm , WebKit browser dont like it. - if ct.starts_with("application/wasm") { + // Disable compression for wasm , WebKit browser dont like it. + // Also skip compression for binary, already-compressed, or non-compressible types. + if ct.starts_with("application/wasm") + || ct.starts_with("video/") + || ct.starts_with("audio/") + || ct.starts_with("image/") + || ct.starts_with("application/octet-stream") + { return false; } } } + // Body-aware 1 KiB cutoff lives on `SizeAbove` in the outer predicate + // chain. This closure only handles the content-type and extension checks. crate::api::api_utils::should_compress_response_extensions(extensions) } fn create_compression_layer() -> tower_http::compression::CompressionLayer { - let predicate = DefaultPredicate::new().and(allow_response_compression); + let predicate = DefaultPredicate::new() + .and(SizeAbove::new(1024)) + .and(allow_response_compression); tower_http::compression::CompressionLayer::new() - .br(true) + .br(false) .deflate(true) .gzip(true) - .zstd(true) + .zstd(false) .compress_when(predicate) } diff --git a/backend/src/api/mod.rs b/backend/src/api/mod.rs index 4ca765c9f..402fcf533 100644 --- a/backend/src/api/mod.rs +++ b/backend/src/api/mod.rs @@ -11,3 +11,4 @@ pub mod main_api; mod serve; mod sys_usage; mod tasks; +pub(crate) mod static_headers; diff --git a/backend/src/api/setup_api.rs b/backend/src/api/setup_api.rs index 8da941761..7b370d952 100644 --- a/backend/src/api/setup_api.rs +++ b/backend/src/api/setup_api.rs @@ -39,7 +39,7 @@ use std::{ }, }; use tokio::sync::{oneshot, Mutex, RwLock}; -use tower_http::{compression::predicate::{DefaultPredicate, Predicate}, services::ServeDir}; +use tower_http::{compression::predicate::{DefaultPredicate, Predicate, SizeAbove}, services::ServeDir}; const DEFAULT_SETUP_HOST: &str = "0.0.0.0"; @@ -855,7 +855,12 @@ async fn setup_complete_inner( } fn create_compression_layer() -> tower_http::compression::CompressionLayer { - let predicate = DefaultPredicate::new().and(| + // Body-aware 1 KiB cutoff is delegated to `SizeAbove` so the two layers + // share the same size-detection logic (Content-Length first, then + // Body::size_hint::exact, defaulting to compress when size is unknown). + let predicate = DefaultPredicate::new() + .and(SizeAbove::new(1024)) + .and(| _status: axum::http::StatusCode, _version: axum::http::Version, headers: &axum::http::HeaderMap, @@ -863,8 +868,14 @@ fn create_compression_layer() -> tower_http::compression::CompressionLayer tower_http::compression::CompressionLayer = vec![]; let mut xml_reader = quick_xml::reader::Reader::from_reader(async_file_reader(content)); - let mut buf = Vec::::new(); + // Pre-allocate so the first giant `` block does not trigger a + // chain of `Vec::grow` reallocs. The buffer is monotonically grown by + // quick_xml (it is reset by the caller, see below), so the eventual size + // is the largest single event — for XMLTV that is the biggest programme + // description. Starting at 0 capacity makes that growth ~25 doubling copies. + let mut buf = Vec::::with_capacity(64 * 1024); loop { match xml_reader.read_event_into_async(&mut buf).await { Ok(Event::Eof) => break, @@ -606,9 +615,20 @@ where Ok(Event::Text(e)) => handle_text_tag(&mut stack, &e), _ => {} } + // quick_xml does not clear the buffer between events — the borrow + // returned by `read_event_into_async` is dropped at the end of each + // match arm, so we can reclaim the capacity here without invalidating + // any handler output. Without this, `buf` grows monotonically over the + // whole file, defeating the 64 KiB pre-allocation above. + buf.clear(); } } +/// Channels per `BPlusTree` write batch. Sized so the in-flight batch + its +/// prepared references stay under 1 MiB on the typical 100-KiB-per-channel +/// EPG feed. This is an educated guess; profile-driven tuning is fine. +const EPG_DISK_BATCH_SIZE: usize = 100; + #[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)] enum XmlTagType { Ignored, @@ -701,16 +721,16 @@ struct ChannelMergeAcc { } #[derive(Debug, Default)] -struct EpgMergeAccumulator { +pub(crate) struct EpgMergeAccumulator { attributes: Option, channels: HashMap, ChannelMergeAcc>, dummy_policies: HashMap, PreferredDummyPolicy>, } impl EpgMergeAccumulator { - fn new() -> Self { Self::default() } + pub(crate) fn new() -> Self { Self::default() } - fn set_attributes_if_preferred( + pub(crate) fn set_attributes_if_preferred( &mut self, priority: i16, source_order: usize, @@ -803,7 +823,7 @@ impl EpgMergeAccumulator { } } - fn add_channel_with_programmes( + pub(crate) fn add_channel_with_programmes( &mut self, priority: i16, source_order: usize, @@ -884,6 +904,198 @@ impl EpgMergeAccumulator { icon_override_channels, )) } + + /// Drain the accumulator directly into a temp `BPlusTree` on disk. Peak RAM + /// here is `EPG_DISK_BATCH_SIZE × max_channel_size` — channels never sit + /// in a `Vec`. The returned `DiskEpgSource` removes the file in its + /// `Drop`. + /// + /// `source_priority` and `source_order` are the per-source values used + /// for `set_attributes_if_preferred` at merge time. The accumulator + /// itself does not aggregate them, so the caller — which knows the + /// original Epg's `priority` and `source_order` — must hand them in. + pub(crate) fn finish_into_disk( + self, + path: PathBuf, + source_priority: i16, + source_order: u32, + ) -> io::Result { + // Fresh tree at the temp path. `store` creates the file; the + // subsequent updater opens it for batched writes. + BPlusTree::::new() + .store(&path) + .map_err(|e| io::Error::other(format!("create temp EPG tree: {e}")))?; + // The temp file exists from here until `DiskEpgSource::new` takes + // ownership at the end. Any `?` in between would otherwise leak the + // file because no `Drop` is wired up yet. The local guard covers the + // fallible window and is disarmed just before we hand the path off. + let mut temp_guard = TempFileGuard(Some(path.clone())); + + let mut updater = BPlusTreeUpdate::::try_new_with_backoff(&path) + .map_err(|e| io::Error::other(format!("open temp EPG tree: {e}")))?; + updater.set_flush_policy(FlushPolicy::Batch); + + let EpgMergeAccumulator { attributes, channels, dummy_policies: _ } = self; + let total = channels.len(); + let mut batch: Vec<(EpgDiskChannelKey, EpgChannel)> = Vec::with_capacity(EPG_DISK_BATCH_SIZE); + let mut written = 0usize; + + for (_, mut acc) in channels { + normalize_channel_programmes(&mut acc); + let folded = with_folded_epg_id(&acc.channel.id, |folded| folded.intern()); + batch.push(( + EpgDiskChannelKey { + folded_id: folded, + priority: acc.priority, + // `usize -> u32` overflow requires > 4 billion sources, which + // is not representable in the file format anyway; treat it + // as a programmer error rather than a recoverable I/O + // failure. + source_order: u32::try_from(acc.source_order) + .expect("source_order exceeds u32 (4 billion sources in one import)"), + }, + acc.channel, + )); + if batch.len() >= EPG_DISK_BATCH_SIZE { + flush_batch(&mut updater, &mut batch, &mut written)?; + } + } + flush_batch(&mut updater, &mut batch, &mut written)?; + updater.commit()?; + // Reclaim space wasted by priority-override entries (multiple keys per + // channel). Without compact, the file is ~2× its eventual size. + updater.compact()?; + + log::debug!( + "Drained {written} channels ({total} total before normalize) into temp EPG tree at {}", + path.display() + ); + + let source = DiskEpgSource::new( + path, + attributes.map(|a| a.attributes), + source_priority, + source_order, + ); + // DiskEpgSource now owns the path; disarm the guard so its Drop does + // not double-remove the file. `mem::forget` would also work, but + // `take()` keeps the guard in scope and is auditable. + temp_guard.0.take(); + Ok(source) + } +} + +/// Removes the wrapped file path on drop unless `take()` was called first. +/// Mirrors the cleanup logic of `DiskEpgSource::Drop` for the fallible window +/// inside `finish_into_disk` where no `DiskEpgSource` exists yet. +struct TempFileGuard(Option); + +impl Drop for TempFileGuard { + fn drop(&mut self) { + if let Some(path) = self.0.take() { + if let Err(err) = std::fs::remove_file(&path) { + if err.kind() != std::io::ErrorKind::NotFound { + log::warn!("Failed to remove temp EPG tree at {}: {err}", path.display()); + } + } + } + } +} + +/// Flush a non-empty batch and bump the written counter. No-op for empty +/// batches, so the caller does not need a guard around the trailing flush. +fn flush_batch( + updater: &mut BPlusTreeUpdate, + batch: &mut Vec<(EpgDiskChannelKey, EpgChannel)>, + written: &mut usize, +) -> io::Result<()> { + if batch.is_empty() { + return Ok(()); + } + let items: Vec<(&EpgDiskChannelKey, &EpgChannel)> = batch.iter().map(|(k, v)| (k, v)).collect(); + let prepared = BPlusTreeUpdate::::prepare_upsert_batch(&items)?; + updater.upsert_batch_encoded(prepared)?; + *written += batch.len(); + batch.clear(); + Ok(()) +} + +/// Sort key for the per-source temp `BPlusTree`. Order: folded channel id ascending, +/// then priority ascending, then source order ascending. This matches the existing +/// rule in `EpgMergeAccumulator::upsert_channel` (`(priority, source_order) < (acc.priority, acc.source_order)`) +/// — lower numbers win. Sorted iteration over the tree therefore yields channels in +/// the right order for the multi-way merge downstream. +/// +/// The `Ord` implementation below is hand-written to make the value-based, +/// deterministic ordering explicit (`folded_id` → `priority` → `source_order`). +/// `Arc` already compares by value when the inner `str` is `Ord`, so a +/// derived `Ord` would produce the same total order — the custom impl exists +/// to lock the contract in source rather than rely on a derived behaviour. +/// Same key type serialises through `rmp_serde` as a record. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub(crate) struct EpgDiskChannelKey { + #[serde(with = "arc_str_serde")] + pub folded_id: Arc, + pub priority: i16, + pub source_order: u32, +} + +// Explicit value-based ordering. `#[derive(Ord)]` would derive the same total +// order via `Arc`'s value-based `Ord` impl, but writing it out documents +// the contract: `folded_id` ascending, then `priority` ascending, then +// `source_order` ascending. Sorted iteration in `merge_epg_trees` depends on +// this exact order for deterministic multi-way merge results. +impl Ord for EpgDiskChannelKey { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.folded_id + .as_ref() + .cmp(other.folded_id.as_ref()) + .then(self.priority.cmp(&other.priority)) + .then(self.source_order.cmp(&other.source_order)) + } +} +impl PartialOrd for EpgDiskChannelKey { + fn partial_cmp(&self, other: &Self) -> Option { Some(self.cmp(other)) } +} + +/// Handle for a temp `BPlusTree` that holds the channels of one EPG source. +/// `Drop` removes the file — even on panic — so the temp dir never accumulates +/// stale trees. The caller is expected to hand ownership to `merge_epg_trees`, +/// which opens a query handle and drains the file before letting the guard drop. +/// +/// `source_priority` and `source_order` are the per-source values used at +/// merge time by `set_attributes_if_preferred`. They live on the source +/// rather than being reconstructed from the disk tree, because the disk +/// tree only stores per-channel priorities (which can differ across +/// channels if sources overlap), not a single source-level value. +pub(crate) struct DiskEpgSource { + pub(super) path: PathBuf, + pub(super) attributes: Option, Arc>>, + pub(super) source_priority: i16, + pub(super) source_order: u32, +} + +impl DiskEpgSource { + pub(crate) fn new( + path: PathBuf, + attributes: Option, Arc>>, + source_priority: i16, + source_order: u32, + ) -> Self { + Self { path, attributes, source_priority, source_order } + } +} + +impl Drop for DiskEpgSource { + fn drop(&mut self) { + if let Err(err) = std::fs::remove_file(&self.path) { + // Missing-file is fine (already cleaned up). Anything else is worth + // a warning — it leaks the temp file until the OS clears /tmp. + if err.kind() != std::io::ErrorKind::NotFound { + log::warn!("Failed to remove temp EPG tree at {}: {err}", self.path.display()); + } + } + } } fn apply_dummy_policies(channels: &mut [ChannelMergeAcc], dummy_policies: &HashMap, PreferredDummyPolicy>) { @@ -923,8 +1135,16 @@ fn backfill_programme_metadata(existing: &mut EpgProgramme, incoming: EpgProgram } fn normalize_channel_programmes(acc: &mut ChannelMergeAcc) { - // Always normalize programme order and dedupe, even for single-source channels. - // This preserves backfill behavior for duplicate entries within the same source. + // `acc.programmes` is the cross-source list of `ProgrammeMergeEntry` records + // (populated by `add_channel_with_programmes` and `push_programme`). + // When the list is empty, the channel's own programmes — set by the + // vacant path of `upsert_channel` — are authoritative; the previous + // implementation unconditionally overwrote `acc.channel.programmes` with + // an empty vector here, which silently dropped every programme on the + // single-source and disk-spill paths. + if acc.programmes.is_empty() { + return; + } acc.programmes .sort_by_key(|entry| (entry.programme.start, entry.programme.stop, entry.priority, entry.source_order)); @@ -994,6 +1214,51 @@ pub fn flatten_tvguide(tv_guides: Vec) -> Option { accumulator.finish() } +/// Merge N temp `BPlusTree`s (one per EPG source) into a single `Epg`, +/// applying dummy policies and the existing priority/sort rules. Sources +/// are consumed one at a time — each `DiskEpgSource`'s temp file is +/// removed by its `Drop` as soon as its iterator is exhausted, so peak RAM +/// is bounded by the largest single source plus the accumulator's +/// per-channel `HashMap`, not the total feed size. +/// +/// Complexity: `O(n_sources + total_channels)`. The per-channel priority +/// resolution happens inside `EpgMergeAccumulator::upsert_channel` +/// (`HashMap` insert), not in a heap. Sources are processed in iteration +/// order, so earlier sources win priority ties for shared channels. +/// +/// The constant-memory guarantee lives on the write side +/// (`EpgMergeAccumulator::finish_into_disk`); this function is bounded by +/// the union's metadata `HashMap` in the accumulator, which is the same +/// shape the in-memory path already carries. +/// +/// The icon-override channel set returned by `finish_epg_with_icon_overrides` +/// is intentionally dropped — the wire-up path persists channels directly, +/// not the override metadata. If a caller later needs that set, the +/// `MergedEpgWithIconOverrides` tuple is already in scope via this +/// function's return type. +pub(crate) fn merge_epg_trees(sources: Vec) -> io::Result> { + let mut accumulator = EpgMergeAccumulator::new(); + for source in sources { + if let Some(attrs) = &source.attributes { + accumulator.set_attributes_if_preferred(source.source_priority, source.source_order as usize, Some(attrs.clone())); + } + let mut query = BPlusTreeQuery::::try_new(&source.path) + .map_err(|e| io::Error::other(format!("open temp EPG tree at {}: {e}", source.path.display())))?; + for entry in query.iter() { + let (key, channel) = entry.map_err(|e| io::Error::other(format!("read temp EPG entry: {e}")))?; + // `add_channel_with_programmes` preserves the channel's programmes + // on the merge accumulator; plain `upsert_channel` would drop them + // because the vacant-entry path constructs a fresh `ChannelMergeAcc` + // with `programmes: Vec::new()`. + accumulator.add_channel_with_programmes(key.priority, key.source_order as usize, false, channel); + } + // `source` drops at the end of this iteration, which removes the + // temp file. The query's mmap/buffer is closed first because `query` + // goes out of scope before `source`. + } + Ok(accumulator.finish_epg_with_icon_overrides()) +} + #[cfg(test)] mod tests { use crate::{ @@ -1796,4 +2061,205 @@ mod tests { assert!(programme.is_new); }); } + + /// Regression guard for the buffer-preallocation in `parse_tvguide`. + /// + /// `quick_xml` uses the caller-provided `Vec` as a monotonically growing + /// read buffer (it does not call `.clear()` between events; the caller + /// slices the new portion). For an XMLTV feed with a single giant + /// `` description, that buffer used to start at 0 capacity and + /// double ~25 times on the way to ~163 MiB — every doubling being a + /// copying realloc. Starting with `Vec::with_capacity(64 * 1024)` removes + /// the realloc chain. This test feeds a programme whose description is + /// bigger than that initial capacity and checks that parsing still + /// succeeds; the absence of the test would have allowed a silent + /// regression to `Vec::new()` without breaking any visible behaviour. + #[test] + fn parse_tvguide_handles_giant_programme_description() { + use crate::processing::parser::xmltv::{EPG_TAG_DESC, EPG_TAG_PROGRAMME, parse_tvguide}; + + run_async_test(async { + // 200 KiB of text content — well beyond the 64 KiB preallocation. + let big_text = "x".repeat(200 * 1024); + // `channel` attribute is required for the parser to fire the + // callback on a tag (see `handle_tag_end`). + let xml = format!( + r#"t{big_text}"# + ); + let mut emitted_tags: Vec = Vec::new(); + parse_tvguide(xml.as_bytes(), &mut |tag| { + emitted_tags.push(tag); + }) + .await; + + // The whole point of the preallocation is that the full payload + // survives the round trip — not just the presence of a tag. + let programme = emitted_tags + .iter() + .find(|tag| tag.name.as_ref() == EPG_TAG_PROGRAMME) + .expect("parser emitted no tag"); + let children = programme.children.as_deref().expect(" has no children"); + let desc = children + .iter() + .find(|child| child.name.as_ref() == EPG_TAG_DESC) + .expect(" emitted no child"); + assert_eq!( + desc.value.as_deref().map(str::len), + Some(big_text.len()), + " payload was truncated; got {} bytes, expected {}", + desc.value.as_deref().map_or(0, str::len), + big_text.len() + ); + }); + } + + /// Smallest testable piece of the disk-spilling path. The Drop guard is + /// what keeps a panic or early return from leaking temp files into /tmp. + /// If this fails, every other disk-spilling test is built on sand. + #[test] + fn disk_epg_source_removes_its_temp_file_on_drop() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("epg-source.db"); + std::fs::write(&path, b"placeholder").unwrap(); + assert!(path.exists(), "precondition: temp file must exist"); + + { + let _src = super::DiskEpgSource::new(path.clone(), None, 0, 0); + } // _src dropped here + + assert!(!path.exists(), "DiskEpgSource::Drop must remove its temp file"); + } + + /// Builds an accumulator with N channels and drains it to disk. Asserts: + /// (a) the temp file exists while the guard is alive, (b) it has + /// non-trivial size, (c) the file is removed after Drop. Together these + /// prove the writer runs end-to-end without panicking. + #[test] + fn finish_into_disk_writes_a_real_temp_tree() { + use super::EpgMergeAccumulator; + + let mut acc = EpgMergeAccumulator::new(); + for i in 0..250u32 { + let id: Arc = format!("channel-{i:04}").into(); + let ch = shared::model::EpgChannel { + id: Arc::clone(&id), + title: Some(format!("title {i}").into()), + icon: None, + programmes: vec![shared::model::EpgProgramme::new(i64::from(i), i64::from(i + 1), id)], + }; + acc.upsert_channel(i16::try_from(i).unwrap_or(0), 0, false, ch); + } + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("epg-source.db"); + let src = acc.finish_into_disk(path.clone(), 5, 0).unwrap(); + assert!(path.exists(), "temp tree file must exist while guard is alive"); + let size = std::fs::metadata(&path).unwrap().len(); + assert!(size > 1024, "expected non-trivial size, got {size}"); + drop(src); + assert!(!path.exists(), "Drop must remove the temp file"); + } + + /// Contract test: two temp trees written from two accumulators must + /// merge into the same `Epg` as the in-memory `finish_epg_with_icon_overrides` + /// would produce from the same inputs. Without this, every other + /// optimisation is built on sand. + #[test] + fn disk_path_matches_in_memory_finish() { + use super::{merge_epg_trees, EpgMergeAccumulator}; + + // Build one source by appending one programme per channel through the + // accumulator's primary entry point. `add_channel_with_programmes` is + // the same call the in-memory `merge_epg_channels_by_priority` uses, + // so the in-memory reference and the disk path go through identical + // APIs. The two sources use disjoint programme intervals + // (source 0: 0..1, source 1: 100..101) so the merged channel should + // retain both programmes end-to-end. + fn build_acc(source: usize, channels: std::ops::Range) -> EpgMergeAccumulator { + let mut acc = EpgMergeAccumulator::new(); + for i in channels { + let id: Arc = format!("ch-{i:04}").into(); + let priority = if source == 0 { 5 } else { 3 }; // source 1 wins + let (start, stop) = if source == 0 { (i64::from(i), i64::from(i + 1)) } else { (100 + i64::from(i), 101 + i64::from(i)) }; + acc.add_channel_with_programmes( + i16::try_from(priority).unwrap_or(0), + source, + false, + shared::model::EpgChannel { + id: Arc::clone(&id), + title: Some(format!("title-{source}-{i}").into()), + icon: None, + programmes: vec![shared::model::EpgProgramme::new(start, stop, id)], + }, + ); + } + acc + } + + // Reference: in-memory merge of the same two sources. + let ref_acc_a = build_acc(0, 0..50); + let ref_acc_b = build_acc(1, 0..50); + // `EpgMergeAccumulator` is single-shot, so the reference merge has to + // rebuild a fresh accumulator. The two halves contribute disjoint + // programme intervals, so a single accumulator sees both. + let mut ref_acc = EpgMergeAccumulator::new(); + for i in 0..50 { + let id: Arc = format!("ch-{i:04}").into(); + ref_acc.add_channel_with_programmes( + 5, + 0, + false, + shared::model::EpgChannel { + id: Arc::clone(&id), + title: Some(format!("title-0-{i}").into()), + icon: None, + programmes: vec![shared::model::EpgProgramme::new(i64::from(i), i64::from(i + 1), id)], + }, + ); + } + for i in 0..50 { + let id: Arc = format!("ch-{i:04}").into(); + ref_acc.add_channel_with_programmes( + 3, + 1, + false, + shared::model::EpgChannel { + id: Arc::clone(&id), + title: Some(format!("title-1-{i}").into()), + icon: None, + programmes: vec![shared::model::EpgProgramme::new(100 + i64::from(i), 101 + i64::from(i), id)], + }, + ); + } + let reference = ref_acc.finish_epg_with_icon_overrides().unwrap().0; + let _ = (ref_acc_a, ref_acc_b); // keep the per-source builder API in the test + + // Disk path: two temp trees, then merge. + let dir = tempfile::tempdir().unwrap(); + let src_a = build_acc(0, 0..50).finish_into_disk(dir.path().join("a.db"), 5, 0).unwrap(); + let src_b = build_acc(1, 0..50).finish_into_disk(dir.path().join("b.db"), 3, 1).unwrap(); + let merged = merge_epg_trees(vec![src_a, src_b]).unwrap().unwrap().0; + + assert_eq!(reference.children.len(), merged.children.len(), "channel counts must match"); + for (left, right) in reference.children.iter().zip(merged.children.iter()) { + assert_eq!(left.id, right.id, "channel order must match"); + // Source 1 had priority 3 (lower = wins) so its title should win. + assert_eq!(left.title, right.title, "priority winner's title must propagate"); + // Both sources contribute distinct, non-overlapping programmes. + // The merged channel must keep both — the disk path uses + // `add_channel_with_programmes` so nothing is dropped on the way in. + assert_eq!( + left.programmes.len(), + 2, + "channel {} should retain both source programmes, got {}", + left.id, + left.programmes.len() + ); + assert_eq!(right.programmes.len(), 2, "disk-merged channel {} lost programmes", right.id); + for (lp, rp) in left.programmes.iter().zip(right.programmes.iter()) { + assert_eq!(lp.start, rp.start, "programme start must match for channel {}", left.id); + assert_eq!(lp.stop, rp.stop, "programme stop must match for channel {}", left.id); + } + } + } } diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 5554a144d..ae5329b57 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -20,7 +20,7 @@ use crate::{ processing::{ input_cache, input_cache::ClusterState, - parser::xmltv::flatten_tvguide, + parser::xmltv::{flatten_tvguide, EpgMergeAccumulator, merge_epg_trees}, playlist_watch::process_group_watch, processor::{ epg::process_playlist_epg, library, sort::sort_playlist, stalker, StalkerRefreshMode, @@ -1525,6 +1525,45 @@ async fn prepare_playlist_for_target( Ok(PreparedTarget { target: target.clone(), playlist: new_playlist, epg: new_epg }) } +/// Spill each `Epg` source to a temp `BPlusTree` and merge them. Extracted +/// from `finalize_prepared_target` so it can be unit-tested without +/// constructing a full `PlaylistProcessingContext`. +/// +/// Returns `Ok(None)` if `sources` is empty (no EPG to merge), matching +/// the contract of `flatten_tvguide`. The temp directory lives inside +/// this function call — all temp files are removed by the +/// `DiskEpgSource` drop guards before this function returns. +fn spill_epg_to_disk(sources: Vec) -> Result, TuliproxError> { + let dir = tempfile::tempdir() + .map_err(|e| TuliproxError::RepositoryXtream(format!("tempdir for EPG spill: {e}")))?; + let mut disk_sources = Vec::with_capacity(sources.len()); + for (source_order, guide) in sources.into_iter().enumerate() { + let mut acc = EpgMergeAccumulator::new(); + acc.set_attributes_if_preferred(guide.priority, source_order, guide.attributes); + for channel in guide.children { + acc.add_channel_with_programmes( + guide.priority, + source_order, + guide.logo_override, + std::sync::Arc::unwrap_or_clone(channel), + ); + } + let path = dir.path().join(format!("epg-src-{source_order}.db")); + let source_order_u32 = u32::try_from(source_order).unwrap_or(0); + let source = acc + .finish_into_disk(path, guide.priority, source_order_u32) + .map_err(|e| TuliproxError::RepositoryXtream(format!("EPG spill to disk failed: {e}")))?; + disk_sources.push(source); + } + if disk_sources.is_empty() { + Ok(None) + } else { + merge_epg_trees(disk_sources) + .map_err(|e| TuliproxError::RepositoryXtream(format!("EPG disk merge failed: {e}"))) + .map(|opt| opt.map(|(epg, _)| epg)) + } +} + async fn finalize_prepared_target( ctx: Arc, prepared: PreparedTarget, @@ -1571,10 +1610,32 @@ async fn finalize_prepared_target( step.tick("group watches"); log_memory_snapshot(format!("target '{}' after_group_watches", target.name).as_str()); } + let merged_epg = if ctx.config.config.load().disk_based_processing { + // Per-source drain to disk, then multi-way merge. Errors are pushed + // to `errors` rather than `?` because the function returns + // `(Result, Vec)`, not `Result` directly. We must + // surface tempdir / write / merge failures — the user opted in to + // disk spilling, and silently falling back to the in-memory path + // can OOM on large feeds. When the spill itself fails we skip the + // persist step entirely: continuing with `merged_epg = None` would + // overwrite the existing on-disk EPG with nothing and discard the + // previously persisted artifact on a transient error. + match spill_epg_to_disk(new_epg) { + Ok(epg) => epg, + Err(err) => { + errors.push(err); + step.stop("EPG spill failed; skipping persist to preserve existing EPG"); + log_memory_snapshot(format!("target '{}' after_persist", target.name).as_str()); + return (Ok(()), errors); + } + } + } else { + flatten_tvguide(new_epg) + }; let result = persist_playlist( &ctx.config, &mut flat_new_playlist, - flatten_tvguide(new_epg).as_ref(), + merged_epg.as_ref(), target, ctx.playlist_state.as_ref(), ) @@ -3181,3 +3242,114 @@ match { } } } + +#[cfg(test)] +mod disk_epg_wireup_tests { + use super::spill_epg_to_disk; + use crate::model::Epg; + use shared::model::EpgChannel; + use std::sync::Arc; + + /// Build an `Epg` with `channel_count` channels whose ids follow the + /// `id_base` prefix. Two sources built with the same `id_base` and + /// `channel_count` will share all channel ids, which is what we need to + /// exercise the priority-override `Occupied` branch in + /// `EpgMergeAccumulator::upsert_channel`. + fn build_epg(id_base: &str, priority: i16, channel_count: usize) -> Epg { + Epg { + priority, + logo_override: false, + attributes: None, + children: (0..channel_count) + .map(|i| { + let id: Arc = format!("{id_base}-ch-{i:04}").into(); + Arc::new(EpgChannel { + id: Arc::clone(&id), + title: Some(format!("title-{priority}-{i}").into()), + icon: None, + programmes: vec![shared::model::EpgProgramme::new( + i64::try_from(i).expect("test index fits in i64"), + i64::try_from(i + 1).expect("test index fits in i64"), + id, + )], + }) + }) + .collect(), + } + } + + /// Wire-up regression guard: `spill_epg_to_disk` is the function called + /// by `finalize_prepared_target` when `disk_based_processing = true`. It + /// must (a) preserve per-source priority on shared channels, (b) clean up + /// its temp files, and (c) merge into a single `Epg` of the right size. + /// + /// Both sources share channel ids (`shared-ch-NNNN`), forcing the merge + /// to take the `Occupied` branch in `EpgMergeAccumulator::upsert_channel`. + /// The lower-priority source (priority 3) must win, the higher-priority + /// (priority 7) must be discarded for shared ids. Without this assertion + /// the test would pass even if priority resolution were broken — the + /// earlier version used unique ids and therefore never hit the merge path. + #[test] + fn spill_epg_to_disk_merges_shared_channels_by_priority() { + let epg_low = build_epg("shared", 3, 50); // wins on every shared channel + let epg_high = build_epg("shared", 7, 50); // discarded on every shared channel + + let merged = spill_epg_to_disk(vec![epg_low, epg_high]) + .expect("disk merge returned an error") + .expect("merged Epg is unexpectedly None for two non-empty sources"); + + // 50 distinct channels, not 100 — the merge must have collapsed the + // shared ids. + assert_eq!( + merged.children.len(), + 50, + "shared channel ids must collapse to one entry, not be duplicated" + ); + + // Every channel title comes from the lower-priority source. If the + // merge logic is wrong, some titles will carry the "-7-" marker. + for ch in &merged.children { + let title = ch.title.as_deref().expect("title preserved through merge"); + assert!( + title.starts_with("title-3-"), + "channel {:?} kept title {title:?} from higher-priority source; \ + priority override is broken", + ch.id, + ); + // `add_channel_with_programmes` on the disk-merge path must + // preserve the lower-priority source's single programme per + // channel — `upsert_channel` would silently drop them. + assert_eq!( + ch.programmes.len(), + 1, + "channel {:?} lost programmes through the disk-merge path", + ch.id + ); + let prog = &ch.programmes[0]; + assert!(prog.title.is_none() || prog.title.as_deref() != Some("title-7")); + } + } + + /// The non-shared case: sources with disjoint channel ids. Both + /// sources' channels appear in the result with no priority loss (no + /// `Occupied` branch is taken). + #[test] + fn spill_epg_to_disk_keeps_disjoint_sources_intact() { + let epg_low = build_epg("src-a", 3, 50); + let epg_high = build_epg("src-b", 7, 50); + + let merged = spill_epg_to_disk(vec![epg_low, epg_high]) + .expect("disk merge returned an error") + .expect("merged Epg is unexpectedly None for two non-empty sources"); + + assert_eq!(merged.children.len(), 100, "disjoint ids must not collapse"); + assert!(merged.children.iter().any(|ch| ch.title.as_deref() == Some("title-3-0"))); + assert!(merged.children.iter().any(|ch| ch.title.as_deref() == Some("title-7-0"))); + } + + #[test] + fn spill_epg_to_disk_returns_none_for_empty_input() { + let merged = spill_epg_to_disk(vec![]).expect("disk merge returned an error"); + assert!(merged.is_none()); + } +} diff --git a/backend/src/repository/bplustree/common.rs b/backend/src/repository/bplustree/common.rs index c5c5548d3..53211fd1b 100644 --- a/backend/src/repository/bplustree/common.rs +++ b/backend/src/repository/bplustree/common.rs @@ -77,6 +77,35 @@ pub(crate) fn read_exact_at_offset(file: &File, buf: &mut [u8], offset: u64) -> Ok(()) } +pub(crate) fn write_all_at_offset(file: &File, buf: &[u8], offset: u64) -> io::Result<()> { + #[cfg(unix)] + file.write_all_at(buf, offset)?; + #[cfg(windows)] + { + let mut written = 0; + while written < buf.len() { + let write_offset = u64::try_from(written) + .ok() + .and_then(|written| offset.checked_add(written)) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "offset write position overflow"))?; + let chunk_written = file.seek_write(&buf[written..], write_offset)?; + if chunk_written == 0 { + return Err(io::Error::new(io::ErrorKind::WriteZero, "failed to write whole buffer")); + } + written += chunk_written; + } + } + #[cfg(not(unix))] + #[cfg(not(windows))] + { + let _ = file; + let _ = buf; + let _ = offset; + return Err(io::Error::new(io::ErrorKind::Unsupported, "offset writes are not supported on this platform")); + } + Ok(()) +} + pub(crate) fn sidecar_lock_path(filepath: &Path) -> PathBuf { if let Some(stem) = filepath.file_stem() { let mut name = OsString::from("."); diff --git a/backend/src/repository/bplustree/v3/page.rs b/backend/src/repository/bplustree/v3/page.rs index f6e1cd3d6..9e9957fce 100644 --- a/backend/src/repository/bplustree/v3/page.rs +++ b/backend/src/repository/bplustree/v3/page.rs @@ -23,6 +23,8 @@ fn checked_end(offset: usize, length: usize, kind: io::ErrorKind) -> io::Result< offset.checked_add(length).ok_or_else(|| io::Error::new(kind, "page offset overflow")) } +const _: () = assert!(PAGE_HEADER_LEN + INTERNAL_PREAMBLE_LEN <= PAGE_SIZE); + fn slot_base(page_type: PageType) -> io::Result { match page_type { PageType::Leaf => Ok(PAGE_HEADER_LEN), @@ -257,19 +259,16 @@ impl + AsMut<[u8]>> SlottedPage { where I: IntoIterator, { + if self.bytes.as_ref().len() != PAGE_SIZE { + return Err(invalid_data("page must be exactly 4096 bytes")); + } let base = slot_base(self.header.page_type)?; let mut rebuilt = [0; PAGE_SIZE]; if self.header.page_type == PageType::Internal { - let preamble_end = checked_end(PAGE_HEADER_LEN, INTERNAL_PREAMBLE_LEN, io::ErrorKind::InvalidData)?; - let preamble = self - .bytes - .as_ref() - .get(PAGE_HEADER_LEN..preamble_end) - .ok_or_else(|| invalid_data("internal preamble is outside page"))?; - rebuilt - .get_mut(PAGE_HEADER_LEN..preamble_end) - .ok_or_else(|| invalid_data("internal preamble destination is outside page"))? - .copy_from_slice(preamble); + // INTERNAL_PREAMBLE_LEN is a compile-time constant; the range is always in bounds. + let preamble_end = PAGE_HEADER_LEN + INTERNAL_PREAMBLE_LEN; // 32 + 8 = 40 + rebuilt[PAGE_HEADER_LEN..preamble_end] + .copy_from_slice(&self.bytes.as_ref()[PAGE_HEADER_LEN..preamble_end]); } let mut count = 0u16; @@ -280,29 +279,22 @@ impl + AsMut<[u8]>> SlottedPage { } let length = u16::try_from(cell.len()).map_err(|_| invalid_input("cell length exceeds u16"))?; let next_count = count.checked_add(1).ok_or_else(|| invalid_input("cell count exceeds u16"))?; - let slot_offset = usize::from(count) - .checked_mul(SLOT_LEN) - .and_then(|size| base.checked_add(size)) - .ok_or_else(|| invalid_input("slot offset overflow"))?; - let slot_end = checked_end(slot_offset, SLOT_LEN, io::ErrorKind::InvalidInput)?; + // SLOT_LEN=4, count<=u16::MAX, base<=40 — overflow is impossible within PAGE_SIZE. + let slot_offset = base + usize::from(count) * SLOT_LEN; + let slot_end = slot_offset + SLOT_LEN; let next_cell_start = cell_start .checked_sub(cell.len()) .ok_or_else(|| invalid_input("cells exceed page capacity"))?; if next_cell_start < slot_end { return Err(invalid_input("cells exceed page capacity")); } - rebuilt - .get_mut(next_cell_start..cell_start) - .ok_or_else(|| invalid_input("cell destination is outside page"))? - .copy_from_slice(cell); + // Both ranges are within [0..PAGE_SIZE] — validated by next_cell_start < slot_end check. + rebuilt[next_cell_start..cell_start].copy_from_slice(cell); let slot = Slot { offset: u16::try_from(next_cell_start).map_err(|_| invalid_input("cell offset exceeds u16"))?, length, }; - rebuilt - .get_mut(slot_offset..slot_end) - .ok_or_else(|| invalid_input("slot destination is outside page"))? - .copy_from_slice(&slot.encode()); + rebuilt[slot_offset..slot_end].copy_from_slice(&slot.encode()); count = next_count; cell_start = next_cell_start; } @@ -463,6 +455,26 @@ mod tests { write_page_checksum(page) } + #[test] + fn rebuild_ordered_slot_bytes_match_slot_encode() -> io::Result<()> { + let mut page = empty_leaf()?; + let mut leaf = SlottedPage::open(page.as_mut_slice(), PAGE_ID, NEXT_PAGE_ID)?; + leaf.rebuild_ordered([b"a".as_slice(), b"bc".as_slice()])?; + let expected = [ + Slot { offset: u16::try_from(PAGE_SIZE - 1).map_err(io::Error::other)?, length: 1 }, + Slot { offset: u16::try_from(PAGE_SIZE - 3).map_err(io::Error::other)?, length: 2 }, + ]; + for (index, slot) in expected.into_iter().enumerate() { + let offset = PAGE_HEADER_LEN + .checked_add(index.checked_mul(SLOT_LEN).ok_or_else(|| io::Error::other("test slot overflow"))?) + .ok_or_else(|| io::Error::other("test slot overflow"))?; + let end = offset.checked_add(SLOT_LEN).ok_or_else(|| io::Error::other("test slot overflow"))?; + let written = page.get(offset..end).ok_or_else(|| io::Error::other("test slot outside page"))?; + assert_eq!(written, slot.encode(), "inline slot encoding drifted from Slot::encode at {index}"); + } + Ok(()) + } + #[test] fn opens_empty_leaf_from_immutable_and_mutable_buffers() -> io::Result<()> { let page = empty_leaf()?; @@ -547,6 +559,27 @@ mod tests { Ok(()) } + #[test] + fn rebuild_rejects_a_short_page_without_panicking() -> io::Result<()> { + let mut page = [0; PAGE_SIZE]; + InternalPreamble { leftmost_child: 3 }.encode_into(&mut page, 2, NEXT_PAGE_ID)?; + page[40..44].copy_from_slice(&Slot { offset: 4095, length: 1 }.encode()); + page[4095] = b'x'; + PageHeader { + page_type: PageType::Internal, + cell_count: 1, + free_start: 44, + free_end: 4095, + left: 0, + right: 0, + } + .encode_into(&mut page, 2, NEXT_PAGE_ID)?; + + let mut changing = ChangingMutView { bytes: page, calls: 0 }; + let mut slotted = SlottedPage::open(&mut changing, 2, NEXT_PAGE_ID)?; + invalid_data(slotted.rebuild_ordered([b"a".as_slice()])) + } + #[test] fn rebuild_compacts_gapped_cells() -> io::Result<()> { let slots = [Slot { offset: 4080, length: 4 }, Slot { offset: 4000, length: 4 }]; @@ -686,4 +719,15 @@ mod tests { assert_eq!(reopened.cells().collect::>>()?, [b"a".as_slice(), b"bc".as_slice()]); Ok(()) } + + #[test] + fn rebuild_ordered_roundtrip() -> io::Result<()> { + let mut page = empty_leaf()?; + let mut slotted = SlottedPage::open(page.as_mut_slice(), PAGE_ID, NEXT_PAGE_ID)?; + let cells: Vec<&[u8]> = vec![b"alpha", b"beta", b"gamma"]; + slotted.rebuild_ordered(cells.iter().copied())?; + let read_back: Vec<&[u8]> = slotted.cells().collect::>()?; + assert_eq!(read_back, cells); + Ok(()) + } } diff --git a/backend/src/repository/bplustree/v3/tree.rs b/backend/src/repository/bplustree/v3/tree.rs index 3510d24a5..b80cab06b 100644 --- a/backend/src/repository/bplustree/v3/tree.rs +++ b/backend/src/repository/bplustree/v3/tree.rs @@ -14,7 +14,7 @@ use super::{ BPlusTreeMetadata, }; use crate::{ - repository::bplustree::common::{mmap_with_advice, read_exact_at_offset, Advice, BPlusTreeError}, + repository::bplustree::common::{mmap_with_advice, read_exact_at_offset, write_all_at_offset, Advice, BPlusTreeError}, utils::{binary_deserialize, binary_serialize, binary_serialize_into}, }; use memmap2::Mmap; @@ -23,7 +23,7 @@ use serde::{Deserialize, Serialize}; use std::{ collections::{BTreeMap, HashMap, HashSet}, fs::{File, OpenOptions}, - io::{self, BufWriter, Read, Write}, + io::{self, Read}, marker::PhantomData, ops::{Bound, Range}, path::{Path, PathBuf}, @@ -395,11 +395,41 @@ struct NodeInfo { minimum_key: Vec, } -fn allocate_page(pages: &mut Vec<[u8; PAGE_SIZE]>) -> io::Result { - let page_id = u64::try_from(pages.len()).map_err(|_| invalid_input("page count exceeds u64"))?; - pages.try_reserve(1).map_err(|err| io::Error::new(io::ErrorKind::OutOfMemory, err))?; - pages.push([0; PAGE_SIZE]); - Ok(page_id) +/// Sink for the bulk tree builder: hands out page ids and writes each finished page +/// straight to the destination file instead of buffering the whole database in memory. +/// +/// Pages are **not** emitted in ascending id order — a leaf reserves its id before the +/// overflow chain it points at, but is encoded only once all of its cells are known, so +/// it is written after them. Writes must therefore be positional; a sequential writer +/// would silently scramble the file. Every id handed out is written exactly once, so the +/// finished file has no holes, and `verify_full` re-reads it before it is published. +struct PageSink { + file: File, + next_page_id: u64, +} + +const PAGE_SIZE_U64: u64 = PAGE_SIZE as u64; + +impl PageSink { + const fn new(file: File) -> Self { Self { file, next_page_id: 0 } } + + fn allocate(&mut self) -> io::Result { + let page_id = self.next_page_id; + self.next_page_id = page_id.checked_add(1).ok_or_else(|| invalid_input("page count exceeds u64"))?; + Ok(page_id) + } + + fn allocate_many(&mut self, count: usize) -> io::Result { + let head = self.next_page_id; + let count = u64::try_from(count).map_err(|_| invalid_input("page count exceeds u64"))?; + self.next_page_id = head.checked_add(count).ok_or_else(|| invalid_input("page count exceeds u64"))?; + Ok(head) + } + + fn write(&self, page_id: u64, page: &[u8; PAGE_SIZE]) -> io::Result<()> { + let offset = page_id.checked_mul(PAGE_SIZE_U64).ok_or_else(|| invalid_input("page offset overflow"))?; + write_all_at_offset(&self.file, page, offset) + } } fn leaf_page>(page_id: u64, left: u64, right: u64, cells: &[T]) -> io::Result<[u8; PAGE_SIZE]> { @@ -460,16 +490,12 @@ fn internal_page>( Ok(bytes) } -fn allocate_overflow_chain(pages: &mut Vec<[u8; PAGE_SIZE]>, stored: &[u8]) -> io::Result { +fn allocate_overflow_chain(sink: &mut PageSink, stored: &[u8]) -> io::Result { let count = stored.len().div_ceil(OVERFLOW_PAYLOAD_LEN); if count == 0 { return Err(invalid_input("overflow value must not be empty")); } - let head = u64::try_from(pages.len()).map_err(|_| invalid_input("page count exceeds u64"))?; - pages.try_reserve(count).map_err(|err| io::Error::new(io::ErrorKind::OutOfMemory, err))?; - for _ in 0..count { - pages.push([0; PAGE_SIZE]); - } + let head = sink.allocate_many(count)?; for (index, payload) in stored.chunks(OVERFLOW_PAYLOAD_LEN).enumerate() { let page_id = head .checked_add(u64::try_from(index).map_err(|_| invalid_input("overflow index exceeds u64"))?) @@ -480,31 +506,22 @@ fn allocate_overflow_chain(pages: &mut Vec<[u8; PAGE_SIZE]>, stored: &[u8]) -> i page_id.checked_add(1).ok_or_else(|| invalid_input("overflow page id overflow"))? }; let page = encode_overflow_page(page_id, u64::MAX, next, payload)?; - let slot = usize::try_from(page_id).map_err(|_| invalid_input("overflow page id exceeds usize"))?; - *pages.get_mut(slot).ok_or_else(|| invalid_input("overflow page allocation missing"))? = page; + sink.write(page_id, &page)?; } Ok(head) } -fn finish_leaf( - pages: &mut [[u8; PAGE_SIZE]], - page_id: u64, - left: u64, - right: u64, - cells: &[Vec], -) -> io::Result<()> { +fn finish_leaf(sink: &PageSink, page_id: u64, left: u64, right: u64, cells: &[Vec]) -> io::Result<()> { let page = leaf_page(page_id, left, right, cells)?; - let index = usize::try_from(page_id).map_err(|_| invalid_input("leaf page id exceeds usize"))?; - *pages.get_mut(index).ok_or_else(|| invalid_input("leaf page allocation missing"))? = page; - Ok(()) + sink.write(page_id, &page) } -fn build_leaf_level(entries: &BTreeMap, pages: &mut Vec<[u8; PAGE_SIZE]>) -> io::Result> +fn build_leaf_level(entries: &BTreeMap, sink: &mut PageSink) -> io::Result> where K: Ord + Serialize, V: Serialize, { - let mut leaf_id = allocate_page(pages)?; + let mut leaf_id = sink.allocate()?; let mut left_leaf = 0; let mut cells = Vec::>::new(); let mut leaves = Vec::::new(); @@ -526,7 +543,7 @@ where if stored_bytes.len() <= MAX_INLINE_STORED_VALUE && inline_footprint <= MAX_CELL_FOOTPRINT { encode_inline_leaf_cell(&key_bytes, logical_len, stored.compression(), stored_bytes, &mut cell)?; } else { - let head = allocate_overflow_chain(pages, stored_bytes)?; + let head = allocate_overflow_chain(sink, stored_bytes)?; encode_overflow_leaf_cell( &key_bytes, logical_len, @@ -546,8 +563,8 @@ where MAX_CELL_FOOTPRINT, )?; if next_usage > PAGE_SIZE { - let next_leaf = allocate_page(pages)?; - finish_leaf(pages, leaf_id, left_leaf, next_leaf, &cells)?; + let next_leaf = sink.allocate()?; + finish_leaf(sink, leaf_id, left_leaf, next_leaf, &cells)?; leaves.push(NodeInfo { page_id: leaf_id, minimum_key: leaf_minimum.take().ok_or_else(|| invalid_input("non-empty leaf has no minimum key"))?, @@ -563,7 +580,7 @@ where cells.push(std::mem::take(&mut cell)); } - finish_leaf(pages, leaf_id, left_leaf, 0, &cells)?; + finish_leaf(sink, leaf_id, left_leaf, 0, &cells)?; leaves.push(NodeInfo { page_id: leaf_id, minimum_key: leaf_minimum.unwrap_or_default() }); Ok(leaves) } @@ -614,7 +631,7 @@ fn group_internal_children(level: &[NodeInfo]) -> io::Result Ok(groups) } -fn build_parent_level(level: &[NodeInfo], pages: &mut Vec<[u8; PAGE_SIZE]>) -> io::Result> { +fn build_parent_level(level: &[NodeInfo], sink: &mut PageSink) -> io::Result> { let groups = group_internal_children(level)?; let mut parents = Vec::with_capacity(groups.len()); for (start, end) in groups { @@ -622,7 +639,7 @@ fn build_parent_level(level: &[NodeInfo], pages: &mut Vec<[u8; PAGE_SIZE]>) -> i if children.len() < 2 { return Err(invalid_input("internal page requires two children")); } - let page_id = allocate_page(pages)?; + let page_id = sink.allocate()?; let (first_child, remaining_children) = children.split_first().ok_or_else(|| invalid_input("internal page has no children"))?; let mut encoded_cells = Vec::with_capacity(remaining_children.len()); @@ -632,29 +649,31 @@ fn build_parent_level(level: &[NodeInfo], pages: &mut Vec<[u8; PAGE_SIZE]>) -> i encoded_cells.push(encoded); } let encoded = internal_page(page_id, first_child.page_id, &encoded_cells)?; - let page_index = usize::try_from(page_id).map_err(|_| invalid_input("internal page id exceeds usize"))?; - *pages.get_mut(page_index).ok_or_else(|| invalid_input("internal page allocation missing"))? = encoded; + sink.write(page_id, &encoded)?; parents.push(NodeInfo { page_id, minimum_key: first_child.minimum_key.clone() }); } Ok(parents) } -fn build_pages(entries: &BTreeMap) -> io::Result<(Vec<[u8; PAGE_SIZE]>, u64)> +/// Streams the whole tree into `sink` and returns the root page id. Page 0 is reserved +/// for the database header, which the caller writes last once `sink.next_page_id` is final. +fn build_pages(entries: &BTreeMap, sink: &mut PageSink) -> io::Result where K: Ord + Serialize, V: Serialize, { - let mut pages = vec![[0; PAGE_SIZE]]; - let mut level = build_leaf_level(entries, &mut pages)?; + let header_page_id = sink.allocate()?; + if header_page_id != 0 { + return Err(invalid_input("database header must be the first allocated page")); + } + let mut level = build_leaf_level(entries, sink)?; if entries.is_empty() { - let root = level.first().ok_or_else(|| invalid_input("tree has no root page"))?.page_id; - return Ok((pages, root)); + return Ok(level.first().ok_or_else(|| invalid_input("tree has no root page"))?.page_id); } while level.len() > 1 { - level = build_parent_level(&level, &mut pages)?; + level = build_parent_level(&level, sink)?; } - let root_page_id = level.first().ok_or_else(|| invalid_input("tree has no root page"))?.page_id; - Ok((pages, root_page_id)) + Ok(level.first().ok_or_else(|| invalid_input("tree has no root page"))?.page_id) } fn temporary_path(filepath: &Path) -> io::Result { @@ -713,37 +732,31 @@ where } fn store_exclusive(&mut self, filepath: &Path) -> io::Result { - let (mut pages, root_page_id) = build_pages(&self.entries)?; - let next_page_id = u64::try_from(pages.len()).map_err(|_| invalid_input("page count exceeds u64"))?; - *pages.first_mut().ok_or_else(|| invalid_input("database header page is missing"))? = DatabaseHeader { - root_page_id, - next_page_id, - free_page_head: 0, - generation: 1, - database_id: *uuid::Uuid::new_v4().as_bytes(), - metadata: self.metadata.clone(), - } - .encode()?; - let temp_path = temporary_path(filepath)?; + // The builder streams into the temp file, so it exists from here on — every error + // path below has to remove it again. let prepared = (|| { - let mut file = OpenOptions::new().write(true).create_new(true).open(&temp_path)?; - { - let mut writer = BufWriter::with_capacity(1024 * 1024, &mut file); - for page in &pages { - writer.write_all(page)?; - } - writer.flush()?; + let mut sink = PageSink::new(OpenOptions::new().write(true).create_new(true).open(&temp_path)?); + let root_page_id = build_pages(&self.entries, &mut sink)?; + let header = DatabaseHeader { + root_page_id, + next_page_id: sink.next_page_id, + free_page_head: 0, + generation: 1, + database_id: *uuid::Uuid::new_v4().as_bytes(), + metadata: self.metadata.clone(), } - file.sync_all()?; - drop(file); + .encode()?; + sink.write(0, &header)?; + sink.file.sync_all()?; + drop(sink); let mut query = BPlusTreeQuery::::from_file_unlocked(File::open(&temp_path)?)?; let verification = verify_full(&mut query)?; drop(query); - Ok(verification) + Ok((root_page_id, verification)) })(); - let verification = match prepared { - Ok(verification) => verification, + let (root_page_id, verification) = match prepared { + Ok(prepared) => prepared, Err(error) => { let _ = std::fs::remove_file(&temp_path); return Err(error); @@ -4197,6 +4210,182 @@ mod tests { Ok(()) } + /// The Xtream cluster import commits after every batch so the transaction's + /// dirty-page map stays bounded. That reopens the write transaction mid-import, + /// so every batch must survive and stay readable — including values large enough + /// to spill into overflow chains, which is what dominated the dirty-page map. + #[test] + fn committing_between_batches_keeps_every_entry_readable() -> io::Result<()> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("batched-commits.db"); + BPlusTree::::new().store(&path)?; + + let mut updater = BPlusTreeUpdate::::try_new(&path)?; + updater.set_flush_policy(FlushPolicy::Batch); + for batch in 0..4u32 { + let keys: Vec = (0..8).map(|index| batch * 8 + index).collect(); + let values: Vec = keys.iter().map(|key| format!("{key}").repeat(600)).collect(); + let items = keys.iter().zip(&values).collect::>(); + let prepared = BPlusTreeUpdate::::prepare_upsert_batch(&items)?; + updater.upsert_batch_encoded(prepared)?; + updater.commit()?; + assert!(updater.active.is_none(), "commit must release the transaction after batch {batch}"); + } + + let mut query = BPlusTreeQuery::::try_new(&path)?; + for key in 0..32u32 { + assert_eq!( + query.query(&key).map_err(BPlusTreeError::to_io)?, + Some(format!("{key}").repeat(600)), + "key {key} is missing after a mid-import commit" + ); + } + assert!(!wal_path(&path).try_exists()?); + Ok(()) + } + + + /// Throughput benchmark for `BPlusTree::store` across three workload sizes. + #[test] + #[ignore = "benchmark; run explicitly with --ignored --nocapture"] + fn bench_store_throughput() -> io::Result<()> { + let noise = |seed: u32, len: usize| -> Vec { + let mut state = seed.wrapping_mul(2_654_435_761).wrapping_add(1); + (0..len) + .map(|_| { + state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + u8::try_from(state >> 24).unwrap_or(0) + }) + .collect() + }; + for (label, count, size) in [("klein", 2_000u32, 800usize), ("mittel", 20_000, 2_000), ("gross", 60_000, 2_000)] { + let dir = tempfile::tempdir()?; + let path = dir.path().join("bench.db"); + let mut tree = BPlusTree::>::new(); + for key in 0..count { + tree.insert(key, noise(key, size)); + } + let start = std::time::Instant::now(); + tree.store(&path)?; + let elapsed = start.elapsed(); + println!( + "BENCH {label:6} entries={count:6} bytes={size:5} -> {:>8.1} ms file={} MiB", + elapsed.as_secs_f64() * 1000.0, + fs::metadata(&path)?.len() / (1024 * 1024) + ); + } + Ok(()) + } + + #[test] + fn streamed_store_writes_every_page_despite_out_of_order_overflow_chains() -> io::Result<()> { + // The streaming builder hands out a leaf's page id *before* the overflow chain it + // points at, but writes the leaf *after* those pages — so page ids do not reach the + // file in ascending order. Worse, this nests: leaf A is reserved, its chains are + // written, A is written, then leaf B repeats the whole dance at higher ids. + // + // The entry count is sized to produce many leaves, not one: overflow leaf cells are + // tiny (key plus pointers, ~34 bytes), so a few hundred entries would all land in a + // single leaf and never exercise the nesting. Values are incompressible on purpose — + // a repeated string would compress back under `MAX_INLINE_STORED_VALUE` and skip + // overflow entirely. Both properties are asserted below rather than assumed. + // + // The file-length check is the real guard: writing pages sequentially instead of + // positionally would leave the file short or the ids scrambled. + // Deterministic LCG — compresses badly, so values are forced into overflow chains. + let noise = |seed: u32, len: usize| -> Vec { + let mut state = seed.wrapping_mul(2_654_435_761).wrapping_add(1); + (0..len) + .map(|_| { + state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + u8::try_from(state >> 24).unwrap_or(0) + }) + .collect() + }; + + let dir = tempfile::tempdir()?; + let path = dir.path().join("streamed.db"); + let mut tree = BPlusTree::>::new(); + for key in 0..1_200u32 { + // Spans one, two and three overflow pages (OVERFLOW_PAYLOAD_LEN = 4056). + tree.insert(key, noise(key, 3000 + (key as usize % 3) * 4000)); + } + let report = tree.store_verified(&path)?; + assert!( + report.overflow_pages > 2_000, + "test must force multi-page overflow chains, got {} overflow pages", + report.overflow_pages + ); + assert!( + report.tree_pages > 8, + "test must produce many leaves so the reserve-then-write dance nests, got {} tree pages", + report.tree_pages + ); + + let header = BPlusTreeQuery::>::try_new(&path)?.header; + assert_eq!( + fs::metadata(&path)?.len(), + header.next_page_id * PAGE_SIZE as u64, + "file length must match the allocated page count exactly" + ); + + let mut query = BPlusTreeQuery::>::try_new(&path)?; + for key in 0..1_200u32 { + assert_eq!( + query.query(&key).map_err(BPlusTreeError::to_io)?, + Some(noise(key, 3000 + (key as usize % 3) * 4000)), + "key {key} did not survive the streamed store" + ); + } + Ok(()) + } + + /// Compaction is NOT optional cleanup on an insert-built tree: leaf splits leave + /// pages roughly half full, and `build_pages` repacks them to near-capacity. + /// + /// Note `free_page_head` stays 0 throughout — inserts never free a page — so it is + /// NOT a usable "nothing to compact" signal. Guarding compaction on an empty free + /// list would skip exactly the case that gains the most. + #[test] + fn compaction_repacks_an_insert_built_tree_despite_an_empty_free_list() -> io::Result<()> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("repack.db"); + BPlusTree::::new().store(&path)?; + + let mut updater = BPlusTreeUpdate::::try_new(&path)?; + updater.set_flush_policy(FlushPolicy::Batch); + for batch in 0..8u32 { + let keys: Vec = (0..250).map(|index| batch * 250 + index).collect(); + let values: Vec = keys.iter().map(|key| format!("{key:07}").repeat(90)).collect(); + let items = keys.iter().zip(&values).collect::>(); + updater.upsert_batch_encoded(BPlusTreeUpdate::::prepare_upsert_batch(&items)?)?; + updater.commit()?; + } + + let before = BPlusTreeQuery::::try_new(&path)?.header; + assert_eq!(before.free_page_head, 0, "inserts must not free pages"); + + updater.compact()?; + + let after = BPlusTreeQuery::::try_new(&path)?.header; + assert!( + after.next_page_id * 3 < before.next_page_id * 2, + "compaction must reclaim over a third of the pages, got {} -> {}", + before.next_page_id, + after.next_page_id + ); + + let mut query = BPlusTreeQuery::::try_new(&path)?; + for key in 0..2000u32 { + assert_eq!( + query.query(&key).map_err(BPlusTreeError::to_io)?, + Some(format!("{key:07}").repeat(90)), + "key {key} did not survive compaction" + ); + } + Ok(()) + } + #[test] fn identical_batch_metadata_is_a_true_noop() -> io::Result<()> { let dir = tempfile::tempdir()?; diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index edc27b1b0..86236e1a7 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -910,6 +910,14 @@ pub async fn persist_input_xtream_playlist_cluster_to_disk( error!("Batch upsert failed for cluster {cluster}: {e}"); TuliproxError::RepositoryXtream(format!("Upsert failed {e}")) })?; + // Commit per batch so the write transaction's dirty-page map stays bounded. + // Holding it open for the whole cluster buffered ~48k pages (196 MB); the + // import writes into a .tmp file that is renamed on success, so atomicity + // comes from the rename, not from a single transaction. + tree.commit().map_err(|e| { + error!("Batch commit failed for cluster {cluster}: {e}"); + TuliproxError::RepositoryXtream(format!("Commit failed {e}")) + })?; buffer.clear(); } } diff --git a/bin/profile.sh b/bin/profile.sh index 93bed9c53..e4c4cb6c8 100755 --- a/bin/profile.sh +++ b/bin/profile.sh @@ -3,16 +3,30 @@ set -euo pipefail usage() { cat <<'EOF' -Usage: bin/profile.sh [debug|release] [settings_dir] [output_file] +Usage: bin/profile.sh [debug|release] [settings_dir] [output_dir] Arguments: debug|release Build/profile mode. Default: release settings_dir Path passed to tuliprox via -H. Required. - output_file Samply output path. Default: /tmp/tuliprox-.samply.json.gz + output_dir Where to write the heaptrack artefact. Default: /tmp + +Produces in output_dir: + tuliprox-.heaptrack.zst Heap allocations (only) + +NOTE on CPU profiling: this script no longer wraps samply. Samply 0.13.1 in +this environment writes profile metadata but no `samples` payload, while +`perf record` captures ~19k samples on the same workload — so for CPU +profiles use `perf record -F 999 -o out.perf -- ` directly and convert +with `samply convert` if you want a Firefox Profiler view. Environment: - PROFILE_DURATION_SECONDS Samply record duration in seconds. Default: 30 - RELEASE_DEBUG_INFO If set to 1, release builds keep debug info. Default: 1 + PROFILE_TARGET Target name passed via `-t`. If unset, tuliprox runs in + server mode after the update and the script will hang + waiting for it to exit. REQUIRED for headless runs. + RELEASE_DEBUG_INFO If 1, release builds keep debug info. Default: 1 + RELEASE_NO_LTO If 1, disable LTO for release builds. Default: 1 + (LTO inflates release build time ~10x without helping + the sample, and the slow link step eats into CI minutes) EOF } @@ -23,9 +37,19 @@ fi MODE="${1:-release}" SETTINGS_DIR="${2:-}" -OUTPUT_FILE="${3:-/tmp/tuliprox-${MODE}.samply.json.gz}" -PROFILE_DURATION_SECONDS="${PROFILE_DURATION_SECONDS:-30}" +OUTPUT_DIR="${3:-/tmp}" RELEASE_DEBUG_INFO="${RELEASE_DEBUG_INFO:-1}" +RELEASE_NO_LTO="${RELEASE_NO_LTO:-1}" +PROFILE_TARGET="${PROFILE_TARGET:-}" + +HEAP_BASE="$OUTPUT_DIR/tuliprox-${MODE}.heaptrack" +# heaptrack picks the compression based on the build; newer releases emit +# .zst, older ones fall back to .gz. Resolve the real path after the run. +HEAP_OUTPUT_ZST="${HEAP_BASE}.zst" +HEAP_OUTPUT_GZ="${HEAP_BASE}.gz" +# Pre-declare so the early diagnostic below is `set -u` safe; the real value +# is assigned after `heaptrack` exits. +HEAP_OUTPUT="" case "$MODE" in debug) @@ -33,11 +57,17 @@ case "$MODE" in BIN_PATH="./target/debug/tuliprox" ;; release) + # Debug-info without LTO or strip: the binary keeps symbols so heaptrack + # addresses resolve against the source. LTO is off by default because it + # 10x's the link step without helping the sample. + ENV_VARS=(CARGO_PROFILE_RELEASE_STRIP=false) if [[ "$RELEASE_DEBUG_INFO" == "1" ]]; then - BUILD_ARGS=(env CARGO_PROFILE_RELEASE_DEBUG=true cargo build --release --bin tuliprox) - else - BUILD_ARGS=(cargo build --release --bin tuliprox) + ENV_VARS+=(CARGO_PROFILE_RELEASE_DEBUG=true) fi + if [[ "$RELEASE_NO_LTO" == "1" ]]; then + ENV_VARS+=(CARGO_PROFILE_RELEASE_LTO=false CARGO_PROFILE_RELEASE_CODEGEN_UNITS=16) + fi + BUILD_ARGS=(env "${ENV_VARS[@]}" cargo build --release --bin tuliprox) BIN_PATH="./target/release/tuliprox" ;; *) @@ -47,8 +77,8 @@ case "$MODE" in ;; esac -if ! command -v samply >/dev/null 2>&1; then - echo "samply is not installed" >&2 +if ! command -v heaptrack >/dev/null 2>&1; then + echo "heaptrack is not installed" >&2 exit 1 fi @@ -57,31 +87,40 @@ if [[ -z "$SETTINGS_DIR" ]]; then usage exit 1 fi - if [[ ! -d "$SETTINGS_DIR" ]]; then echo "Settings directory not found: $SETTINGS_DIR" >&2 + usage exit 1 fi +if [[ -z "$PROFILE_TARGET" ]]; then + echo "warning: PROFILE_TARGET is unset; tuliprox will stay in server mode after the update and this script will hang." >&2 + echo " Set PROFILE_TARGET= to force a one-shot update run." >&2 +fi + echo "Building tuliprox ($MODE)..." "${BUILD_ARGS[@]}" -echo "Recording profile..." +mkdir -p "$OUTPUT_DIR" +echo "Recording heap profile..." echo " mode: $MODE" echo " binary: $BIN_PATH" echo " settings: $SETTINGS_DIR" -echo " output: $OUTPUT_FILE" -echo " duration: ${PROFILE_DURATION_SECONDS}s" +echo " target: ${PROFILE_TARGET:-}" +echo " heaptrack: ${HEAP_OUTPUT_ZST} (or ${HEAP_OUTPUT_GZ})" -mkdir -p "$(dirname "$OUTPUT_FILE")" +TARGET_ARGS=() +if [[ -n "$PROFILE_TARGET" ]]; then + TARGET_ARGS=(-t "$PROFILE_TARGET") +fi -samply record \ - --save-only \ - --unstable-presymbolicate \ - --duration "$PROFILE_DURATION_SECONDS" \ - --output "$OUTPUT_FILE" \ - "$BIN_PATH" \ - -s \ - -H "$SETTINGS_DIR" - -echo "Profile written to $OUTPUT_FILE" +heaptrack -o "$HEAP_BASE" -- "$BIN_PATH" -H "$SETTINGS_DIR" "${TARGET_ARGS[@]}" +if [[ -f "$HEAP_OUTPUT_ZST" ]]; then + HEAP_OUTPUT="$HEAP_OUTPUT_ZST" +elif [[ -f "$HEAP_OUTPUT_GZ" ]]; then + HEAP_OUTPUT="$HEAP_OUTPUT_GZ" +else + echo "heaptrack did not produce ${HEAP_OUTPUT_ZST} or ${HEAP_OUTPUT_GZ}" >&2 + exit 1 +fi +echo "Profile written to $HEAP_OUTPUT" diff --git a/shared/src/utils/hash_utils.rs b/shared/src/utils/hash_utils.rs index 488ba91b5..10bec4c04 100644 --- a/shared/src/utils/hash_utils.rs +++ b/shared/src/utils/hash_utils.rs @@ -126,14 +126,14 @@ pub fn generate_provider_playlist_uuid(key: &str, provider_id: &str, item_type: let mut hasher = blake3::Hasher::new(); hasher.update(key.as_bytes()); hasher.update(provider_id.as_bytes()); - hasher.update(item_type.to_string().as_bytes()); + hasher.update(item_type.as_str().as_bytes()); UUIDType(hasher.finalize().into()) } pub fn generate_local_playlist_uuid(key: &str, item_type: PlaylistItemType, url: &str) -> UUIDType { let mut hasher = blake3::Hasher::new(); hasher.update(key.as_bytes()); - hasher.update(item_type.to_string().as_bytes()); + hasher.update(item_type.as_str().as_bytes()); if let Some(url_path) = url_path_and_more(url) { hasher.update(url_path.as_bytes()); @@ -666,4 +666,34 @@ mod tests { let alias = create_alias_uuid(&base, "mapping"); assert_ne!(base, alias); } + + /// Swapping `to_string()` for `as_str()` in the UUID hashers must not change a + /// single byte — every persisted playlist UUID depends on it. Guards against a + /// future `Display` impl that stops delegating to `as_str()`. + #[test] + fn item_type_label_is_hash_stable_against_display() { + for item_type in [ + PlaylistItemType::Live, + PlaylistItemType::LiveHls, + PlaylistItemType::Video, + PlaylistItemType::LocalVideo, + PlaylistItemType::LocalSeries, + PlaylistItemType::LocalSeriesInfo, + PlaylistItemType::Series, + PlaylistItemType::SeriesInfo, + PlaylistItemType::Catchup, + ] { + assert_eq!(item_type.as_str(), item_type.to_string(), "{item_type:?}"); + + let mut expected = blake3::Hasher::new(); + expected.update(b"key"); + expected.update(b"provider"); + expected.update(item_type.to_string().as_bytes()); + assert_eq!( + generate_provider_playlist_uuid("key", "provider", item_type), + UUIDType(expected.finalize().into()), + "{item_type:?}" + ); + } + } } diff --git a/shared/src/utils/serde_utils.rs b/shared/src/utils/serde_utils.rs index 7552e6eba..0acc4b97b 100644 --- a/shared/src/utils/serde_utils.rs +++ b/shared/src/utils/serde_utils.rs @@ -5,6 +5,33 @@ use serde::{de::Error, Deserialize, Deserializer, Serialize, Serializer}; use serde_json::Value; use std::{io, sync::Arc}; +/// Formats a JSON number into a stack buffer and parses it, without touching the heap. +/// +/// `serde_json` renders numbers via `itoa`/`ryu`, so the longest possible output is +/// `-1.7976931348623157e-308` at 24 bytes; 32 leaves headroom. Deliberately takes +/// `&Number` rather than `impl Display` — `Display for f64` writes `f64::MIN` as 310 +/// bytes of decimal expansion, which would silently truncate to `None`. +struct StackNumber { + buf: [u8; 32], + len: usize, +} + +impl std::fmt::Write for StackNumber { + fn write_str(&mut self, text: &str) -> std::fmt::Result { + let end = self.len.checked_add(text.len()).ok_or(std::fmt::Error)?; + self.buf.get_mut(self.len..end).ok_or(std::fmt::Error)?.copy_from_slice(text.as_bytes()); + self.len = end; + Ok(()) + } +} + +fn parse_number(number: &serde_json::Number) -> Option { + use std::fmt::Write; + let mut out = StackNumber { buf: [0; 32], len: 0 }; + write!(out, "{number}").ok()?; + std::str::from_utf8(out.buf.get(..out.len)?).ok()?.parse().ok() +} + fn value_to_string_array(value: &[Value]) -> Vec> { value.iter().filter_map(value_to_arc_str).collect() } fn value_to_arc_str(v: &Value) -> Option> { @@ -86,14 +113,7 @@ where Value::Null => Ok(None), // its a number - Value::Number(n) => { - let s = n.to_string(); - if let Ok(r) = s.parse::() { - Ok(Some(r)) - } else { - Ok(None) - } - } + Value::Number(n) => Ok(parse_number(&n)), // String -> extract first number Value::String(s) => { @@ -418,3 +438,35 @@ where let encoded = general_purpose::STANDARD.encode(value.as_bytes()); serializer.serialize_str(&encoded) } + +#[cfg(test)] +mod tests { + use super::parse_number; + use serde_json::Number; + + /// The stack buffer must hold every number `serde_json` can render. If a future + /// serde_json bump changes the float formatting, this fails instead of silently + /// turning large values into `None`. + #[test] + fn parse_number_matches_to_string_roundtrip_at_the_extremes() { + for value in [0u64, 1, 42, u64::from(u32::MAX), u64::MAX] { + let number = Number::from(value); + assert_eq!(parse_number::(&number), number.to_string().parse().ok(), "u64 {value}"); + } + for value in [i64::MIN, -1, 0, i64::MAX] { + let number = Number::from(value); + assert_eq!(parse_number::(&number), number.to_string().parse().ok(), "i64 {value}"); + } + for value in [f64::MIN, -f64::MIN_POSITIVE, -0.5, 0.0, 4.7, f64::MAX] { + let Some(number) = Number::from_f64(value) else { + panic!("{value} is not representable as a JSON number"); + }; + assert_eq!(parse_number::(&number), number.to_string().parse().ok(), "f64 {value}"); + } + } + + #[test] + fn parse_number_returns_none_instead_of_panicking_on_type_mismatch() { + assert_eq!(parse_number::(&Number::from(-1i64)), None); + } +}