From 651507001c0096757c572dcfc860449c529eb147 Mon Sep 17 00:00:00 2001 From: Kush Bisen Date: Thu, 20 Aug 2026 16:25:08 +0200 Subject: [PATCH] fix: correct historical bounds and segmented storage recovery --- src/api/janus_api/tests.rs | 16 +- src/execution/historical_executor.rs | 62 +++--- src/parsing/janusql_parser/ast.rs | 5 +- src/storage/segmented_storage/background.rs | 45 +---- src/storage/segmented_storage/mod.rs | 31 ++- src/storage/segmented_storage/segment.rs | 179 ++++++++++++++---- .../operators/historical_sliding_window.rs | 22 +-- tests/historical_sliding_window_test.rs | 6 +- tests/historical_window_bounds_test.rs | 19 +- tests/public_spec_behavior_test.rs | 6 +- tests/segmented_storage_error_test.rs | 29 +++ tests/segmented_storage_regression_test.rs | 56 ++++++ 12 files changed, 332 insertions(+), 144 deletions(-) diff --git a/src/api/janus_api/tests.rs b/src/api/janus_api/tests.rs index 884d362..1faed20 100644 --- a/src/api/janus_api/tests.rs +++ b/src/api/janus_api/tests.rs @@ -718,7 +718,7 @@ fn test_sliding_query_defined_baseline_snapshots_change_with_live_evaluation_tim StreamingSegmentedStorage::new(config).expect("Failed to create segmented storage"), ); - for (timestamp, value) in [(86_400_002, "10"), (86_460_000, "20")] { + for (timestamp, value) in [(86_340_002, "10"), (86_400_000, "20")] { storage .write_rdf( timestamp, @@ -730,7 +730,7 @@ fn test_sliding_query_defined_baseline_snapshots_change_with_live_evaluation_tim .expect("Failed to write historical RDF event"); } storage.flush().expect("Failed to flush storage"); - for (timestamp, value) in [(86_460_002, "30"), (86_520_000, "50")] { + for (timestamp, value) in [(86_400_002, "30"), (86_460_000, "50")] { storage .write_rdf( timestamp, @@ -784,14 +784,14 @@ HAVING(AVG(?value) > ?yesterdayAvgValue) let latest_rows = Arc::new(RwLock::new(HashMap::new())); assert_eq!( storage - .query_rdf(86_400_001, 86_460_001) + .query_rdf(86_340_001, 86_400_001) .expect("first historical range should query") .len(), 2 ); assert_eq!( storage - .query_rdf(86_460_001, 86_520_001) + .query_rdf(86_400_001, 86_460_001) .expect("second historical range should query") .len(), 2 @@ -910,8 +910,8 @@ HAVING(AVG(?value) > ?yesterdayAvgValue) let second_snapshot = baseline_registry .get_snapshot("http://example.org/yesterdayBaseline", 172_860_001) .expect("expected snapshot at second evaluation time"); - assert_eq!(first_snapshot.window_start, 86_400_001); - assert_eq!(first_snapshot.window_end, 86_460_001); - assert_eq!(second_snapshot.window_start, 86_460_001); - assert_eq!(second_snapshot.window_end, 86_520_001); + assert_eq!(first_snapshot.window_start, 86_340_001); + assert_eq!(first_snapshot.window_end, 86_400_001); + assert_eq!(second_snapshot.window_start, 86_400_001); + assert_eq!(second_snapshot.window_end, 86_460_001); } diff --git a/src/execution/historical_executor.rs b/src/execution/historical_executor.rs index 3320da7..c6a3f58 100644 --- a/src/execution/historical_executor.rs +++ b/src/execution/historical_executor.rs @@ -19,7 +19,6 @@ use crate::parsing::janusql_parser::WindowDefinition; use crate::querying::oxigraph_adapter::OxigraphAdapter; use crate::storage::segmented_storage::StreamingSegmentedStorage; use crate::stream::operators::historical_fixed_window::HistoricalFixedWindowOperator; -use crate::stream::operators::historical_sliding_window::HistoricalSlidingWindowOperator; use oxigraph::model::{GraphName, NamedNode, Quad}; use rsp_rs::QuadContainer; use std::collections::{HashMap, HashSet}; @@ -166,23 +165,16 @@ impl HistoricalExecutor { window: &WindowDefinition, sparql_query: &'a str, ) -> impl Iterator>, JanusApiError>> + 'a { - let offset = window.offset.unwrap_or(0); - let width = window.width; - let slide = window.slide; - let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_millis() as u64; - let start_time = now.saturating_sub(offset); - SlidingWindowIterator { executor: self, - current_start: start_time, - evaluation_time: now, - width, - slide, + window: window.clone(), + current_evaluation_time: now, + latest_evaluation_time: now, sparql_query: sparql_query.to_string(), } } @@ -397,36 +389,29 @@ impl HistoricalExecutor { &self, window: &WindowDefinition, ) -> Result<(u64, u64), JanusApiError> { - // For fixed windows: use explicit start/end - if let (Some(start), Some(end)) = (window.start, window.end) { - return Ok((start, end)); - } - - // For sliding windows: calculate from offset and width - if let Some(offset) = window.offset { - let now = std::time::SystemTime::now() + let evaluation_time = if window.offset.is_some() { + std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_err(|e| JanusApiError::ExecutionError(format!("System time error: {}", e)))? - .as_millis() as u64; - - let start = now.saturating_sub(offset); - let end = start + window.width; - return Ok((start, end)); - } + .as_millis() as u64 + } else { + window.end.unwrap_or_default() + }; - Err(JanusApiError::ExecutionError( - "Window definition must have either (start, end) or (offset, width)".to_string(), - )) + window.resolve_historical_bounds(evaluation_time).ok_or_else(|| { + JanusApiError::ExecutionError( + "Window definition cannot resolve complete historical bounds".to_string(), + ) + }) } } /// Iterator for sliding windows that queries storage directly struct SlidingWindowIterator<'a> { executor: &'a HistoricalExecutor, - current_start: u64, - evaluation_time: u64, - width: u64, - slide: u64, + window: WindowDefinition, + current_evaluation_time: u64, + latest_evaluation_time: u64, sparql_query: String, } @@ -434,10 +419,10 @@ impl<'a> Iterator for SlidingWindowIterator<'a> { type Item = Result>, JanusApiError>; fn next(&mut self) -> Option { - let window_start = self.current_start; - let window_end = window_start.checked_add(self.width)?; + let (window_start, window_end) = + self.window.resolve_historical_bounds(self.current_evaluation_time)?; - if window_end > self.evaluation_time { + if window_end > self.latest_evaluation_time { return None; } @@ -453,7 +438,8 @@ impl<'a> Iterator for SlidingWindowIterator<'a> { let result = self.executor.execute_sparql_on_events(&events, &self.sparql_query); // Advance window - self.current_start += self.slide; + self.current_evaluation_time = + self.current_evaluation_time.checked_add(self.window.slide)?; Some(result) } @@ -557,7 +543,7 @@ mod tests { .execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }") .collect::>(); - assert_eq!(results.len(), 4); + assert_eq!(results.len(), 6); assert!(results.iter().all(|result| result.is_ok())); } @@ -586,7 +572,7 @@ mod tests { .execute_sliding_windows(&window, "SELECT ?s WHERE { ?s ?p ?o }") .collect::>(); - assert_eq!(results.len(), 2); + assert_eq!(results.len(), 3); assert!(results.iter().all(|result| result.is_ok())); } diff --git a/src/parsing/janusql_parser/ast.rs b/src/parsing/janusql_parser/ast.rs index b597323..db539d9 100644 --- a/src/parsing/janusql_parser/ast.rs +++ b/src/parsing/janusql_parser/ast.rs @@ -74,8 +74,9 @@ impl WindowDefinition { return None; } - let historical_start = evaluation_time.saturating_sub(offset); - let historical_end = historical_start.checked_add(range)?; + // Historical sliding intervals are [T - OFFSET - RANGE, T - OFFSET]. + let historical_end = evaluation_time.checked_sub(offset)?; + let historical_start = historical_end.checked_sub(range)?; Some((historical_start, historical_end)) } } diff --git a/src/storage/segmented_storage/background.rs b/src/storage/segmented_storage/background.rs index 6a7a108..981aae5 100644 --- a/src/storage/segmented_storage/background.rs +++ b/src/storage/segmented_storage/background.rs @@ -77,55 +77,26 @@ impl StreamingSegmentedStorage { events }; - let events_ref = &mut events_to_flush; let flush_result = (|| -> std::io::Result<()> { - let new_segment = Self::write_segment_files(&config, events_ref)?; - - { - let mut segments = segments.write().unwrap(); - segments.push(new_segment); - segments.sort_by_key(|s| s.start_timstamp); - } - + // The dictionary must be durable before a segment referencing its IDs is committed. let dict_path = std::path::Path::new(&config.segment_base_path).join("dictionary.bin"); let dict = dictionary.read().unwrap(); dict.save_to_file(&dict_path)?; + let new_segment = Self::write_segment_files(&config, &mut events_to_flush)?; + + let mut segments = segments.write().unwrap(); + segments.push(new_segment); + segments.sort_by_key(|s| s.start_timstamp); + Ok(()) })(); if let Err(err) = flush_result { - Self::restore_failed_background_flush(&batch_buffer, &events_to_flush); + Self::restore_failed_flush(&batch_buffer, &events_to_flush); return Err(err); } Ok(()) } - - fn restore_failed_background_flush(batch_buffer: &Arc>, events: &[Event]) { - if events.is_empty() { - return; - } - - let mut buffer = batch_buffer.write().unwrap(); - for event in events.iter().rev().cloned() { - buffer.events.push_front(event); - buffer.total_bytes += std::mem::size_of::(); - } - - let restored_oldest = events.first().map(|event| event.timestamp); - let restored_newest = events.last().map(|event| event.timestamp); - - buffer.oldest_timestamp_bound = match (buffer.oldest_timestamp_bound, restored_oldest) { - (Some(existing), Some(restored)) => Some(existing.min(restored)), - (None, restored) => restored, - (existing, None) => existing, - }; - - buffer.newest_timestamp_bound = match (buffer.newest_timestamp_bound, restored_newest) { - (Some(existing), Some(restored)) => Some(existing.max(restored)), - (None, restored) => restored, - (existing, None) => existing, - }; - } } diff --git a/src/storage/segmented_storage/mod.rs b/src/storage/segmented_storage/mod.rs index 49ab2b9..cb6c20c 100644 --- a/src/storage/segmented_storage/mod.rs +++ b/src/storage/segmented_storage/mod.rs @@ -35,14 +35,39 @@ impl StreamingSegmentedStorage { // Load or create dictionary let dict_path = std::path::Path::new(&config.segment_base_path).join("dictionary.bin"); + let has_persisted_segments = std::fs::read_dir(&config.segment_base_path)?.any(|entry| { + entry.ok().is_some_and(|entry| { + entry.file_type().map(|kind| kind.is_file()).unwrap_or(false) + && entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with("segment-") && name.ends_with(".log")) + }) + }); let dictionary = if dict_path.exists() { match Dictionary::load_from_file(&dict_path) { Ok(dict) => dict, Err(e) => { - eprintln!("Warning: Failed to load dictionary: {}, creating new one", e); + if has_persisted_segments { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "Cannot open persisted segment data without a readable dictionary '{}': {e}", + dict_path.display() + ), + )); + } Dictionary::new() } } + } else if has_persisted_segments { + return Err(std::io::Error::new( + std::io::ErrorKind::NotFound, + format!( + "Cannot open persisted segment data because dictionary '{}' is missing", + dict_path.display() + ), + )); } else { Dictionary::new() }; @@ -158,9 +183,7 @@ impl StreamingSegmentedStorage { /// This is useful when you need to ensure data is persisted immediately. pub fn flush(&self) -> std::io::Result<()> { self.ensure_background_flush_healthy()?; - self.flush_batch_buffer_to_segment()?; - self.save_dictionary()?; - Ok(()) + self.flush_batch_buffer_to_segment() } /// Shutdown the storage system gracefully, ensuring all data is flushed to disk. diff --git a/src/storage/segmented_storage/segment.rs b/src/storage/segmented_storage/segment.rs index 653c096..79bda21 100644 --- a/src/storage/segmented_storage/segment.rs +++ b/src/storage/segmented_storage/segment.rs @@ -1,6 +1,7 @@ use std::{ - io::{BufWriter, Seek, Write}, + io::{BufWriter, Read, Seek, SeekFrom, Write}, sync::atomic::{AtomicU64, Ordering}, + sync::{Arc, RwLock}, time::{SystemTime, UNIX_EPOCH}, }; @@ -9,7 +10,7 @@ use crate::{ encoding::{encode_record, RECORD_SIZE}, Event, }, - storage::util::{EnhancedSegmentMetadata, IndexBlock, StreamingConfig}, + storage::util::{BatchBuffer, EnhancedSegmentMetadata, IndexBlock, StreamingConfig}, }; use super::StreamingSegmentedStorage; @@ -43,18 +44,50 @@ impl StreamingSegmentedStorage { events }; - let segment = Self::write_segment_files(&self.config, &mut events_to_flush)?; + // The dictionary must be durable before a segment referencing its IDs is committed. + let flush_result = (|| -> std::io::Result<()> { + self.save_dictionary()?; + let segment = Self::write_segment_files(&self.config, &mut events_to_flush)?; - { let mut segments = self.segments.write().unwrap(); segments.push(segment); segments.sort_by_key(|s| s.start_timstamp); + Ok(()) + })(); + + if let Err(err) = flush_result { + Self::restore_failed_flush(&self.batch_buffer, &events_to_flush); + return Err(err); } - self.save_dictionary()?; Ok(()) } + pub(super) fn restore_failed_flush(batch_buffer: &Arc>, events: &[Event]) { + if events.is_empty() { + return; + } + + let mut buffer = batch_buffer.write().unwrap(); + for event in events.iter().rev().cloned() { + buffer.events.push_front(event); + buffer.total_bytes += std::mem::size_of::(); + } + + let restored_oldest = events.iter().map(|event| event.timestamp).min(); + let restored_newest = events.iter().map(|event| event.timestamp).max(); + buffer.oldest_timestamp_bound = match (buffer.oldest_timestamp_bound, restored_oldest) { + (Some(existing), Some(restored)) => Some(existing.min(restored)), + (None, restored) => restored, + (existing, None) => existing, + }; + buffer.newest_timestamp_bound = match (buffer.newest_timestamp_bound, restored_newest) { + (Some(existing), Some(restored)) => Some(existing.max(restored)), + (None, restored) => restored, + (existing, None) => existing, + }; + } + pub(crate) fn write_segment_files( config: &StreamingConfig, events: &mut [Event], @@ -222,13 +255,25 @@ impl StreamingSegmentedStorage { let index_path = format!("{}/segment-{}.idx", segment_dir, segment_id); if let Ok(_metadata) = fs::metadata(&data_path) { - let (index_directory, start_ts, end_ts, record_count) = - if fs::metadata(&index_path).is_ok() { - Self::load_index_directory_from_file(&index_path) - .unwrap_or_else(|_| (Vec::new(), 0, u64::MAX, 0)) - } else { - (Vec::new(), 0, u64::MAX, 0) - }; + let (start_ts, end_ts, record_count) = + Self::load_segment_log_metadata(&data_path)?; + let mut index_directory = if fs::metadata(&index_path).is_ok() { + Self::load_index_directory_from_file( + &index_path, + self.config.entries_per_index_block, + )? + } else { + Vec::new() + }; + + for block_index in 0..index_directory.len() { + index_directory[block_index].max_timestamp = + if block_index + 1 < index_directory.len() { + index_directory[block_index + 1].min_timestamp + } else { + end_ts + }; + } let segment = EnhancedSegmentMetadata { start_timstamp: start_ts, @@ -256,51 +301,73 @@ impl StreamingSegmentedStorage { Ok(()) } + fn load_segment_log_metadata(data_path: &str) -> std::io::Result<(u64, u64, u64)> { + let mut file = std::fs::File::open(data_path)?; + let byte_len = file.metadata()?.len(); + if byte_len == 0 || byte_len % RECORD_SIZE as u64 != 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("Segment log '{data_path}' is empty or truncated"), + )); + } + + let record_count = byte_len / RECORD_SIZE as u64; + let mut record = [0u8; RECORD_SIZE]; + file.read_exact(&mut record)?; + let (start_timestamp, ..) = crate::core::encoding::decode_record(&record); + file.seek(SeekFrom::Start((record_count - 1) * RECORD_SIZE as u64))?; + file.read_exact(&mut record)?; + let (end_timestamp, ..) = crate::core::encoding::decode_record(&record); + Ok((start_timestamp, end_timestamp, record_count)) + } + pub(super) fn load_index_directory_from_file( index_path: &str, - ) -> std::io::Result<(Vec, u64, u64, u64)> { - use std::io::Read; - + entries_per_index_block: usize, + ) -> std::io::Result> { + const INDEX_ENTRY_SIZE: usize = 16; + if entries_per_index_block == 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "entries_per_index_block must be greater than zero", + )); + } let mut file = std::fs::File::open(index_path)?; let mut buffer = Vec::new(); file.read_to_end(&mut buffer)?; if buffer.is_empty() { - return Ok((Vec::new(), 0, u64::MAX, 0)); + return Ok(Vec::new()); + } + if buffer.len() % INDEX_ENTRY_SIZE != 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("Sparse index '{index_path}' is truncated"), + )); } let mut index_directory = Vec::new(); let mut file_offset = 0u64; - let mut global_min_ts = u64::MAX; - let mut global_max_ts = 0u64; - let mut total_records = 0u64; - - let entries_per_block = 1000; let mut current_block_start = 0; while current_block_start < buffer.len() { - let block_size = - std::cmp::min(entries_per_block * 16, buffer.len() - current_block_start); + let block_size = std::cmp::min( + entries_per_index_block * INDEX_ENTRY_SIZE, + buffer.len() - current_block_start, + ); let block_end = current_block_start + block_size; let block_entries = block_end - current_block_start; - let entry_count = (block_entries / 16) as u32; - - if entry_count == 0 { - break; - } + let entry_count = (block_entries / INDEX_ENTRY_SIZE) as u32; let first_ts = u64::from_le_bytes( buffer[current_block_start..current_block_start + 8].try_into().unwrap(), ); - let last_entry_start = current_block_start + ((entry_count - 1) as usize * 16); + let last_entry_start = + current_block_start + ((entry_count - 1) as usize * INDEX_ENTRY_SIZE); let last_ts = u64::from_le_bytes( buffer[last_entry_start..last_entry_start + 8].try_into().unwrap(), ); - global_min_ts = global_min_ts.min(first_ts); - global_max_ts = global_max_ts.max(last_ts); - total_records += entry_count as u64; - index_directory.push(IndexBlock { min_timestamp: first_ts, max_timestamp: last_ts, @@ -312,6 +379,50 @@ impl StreamingSegmentedStorage { current_block_start = block_end; } - Ok((index_directory, global_min_ts, global_max_ts, total_records)) + Ok(index_directory) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + #[test] + fn recovery_uses_log_metadata_and_configured_index_block_size() { + let temp_dir = TempDir::new().unwrap(); + let config = StreamingConfig { + segment_base_path: temp_dir.path().to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1_000_000, + sparse_interval: 2, + entries_per_index_block: 2, + }; + + { + let storage = StreamingSegmentedStorage::new(config.clone()).unwrap(); + for timestamp in (10..=70).step_by(10) { + storage + .write(Event { timestamp, subject: 0, predicate: 0, object: 0, graph: 0 }) + .unwrap(); + } + storage.flush().unwrap(); + } + + let storage = StreamingSegmentedStorage::new(config).unwrap(); + let segments = storage.segments.read().unwrap(); + assert_eq!(segments.len(), 1); + let segment = &segments[0]; + assert_eq!(segment.record_count, 7); + assert_eq!(segment.start_timstamp, 10); + assert_eq!(segment.end_timestamp, 70); + assert_eq!(segment.index_directory.len(), 2); + assert_eq!(segment.index_directory.last().unwrap().max_timestamp, 70); + drop(segments); + + let tail = storage.query(70, 70).unwrap(); + assert_eq!(tail.len(), 1); + assert_eq!(tail[0].timestamp, 70); } } diff --git a/src/stream/operators/historical_sliding_window.rs b/src/stream/operators/historical_sliding_window.rs index 5a0a898..dcb3e22 100644 --- a/src/stream/operators/historical_sliding_window.rs +++ b/src/stream/operators/historical_sliding_window.rs @@ -8,8 +8,8 @@ use std::rc::Rc; pub struct HistoricalSlidingWindowOperator { storage: Rc, window_def: WindowDefinition, - current_start: u64, - evaluation_time: u64, + current_evaluation_time: u64, + latest_evaluation_time: u64, } impl HistoricalSlidingWindowOperator { @@ -25,16 +25,11 @@ impl HistoricalSlidingWindowOperator { .unwrap() .as_millis() as u64; - // Offset is mandatory for HistoricalSliding windows as per the parser and requirements. - // We subtract it from the query_start to "go back" in time. - let offset = window_def.offset.expect("Offset must be defined for HistoricalSlidingWindow"); - let start_time = now.saturating_sub(offset); - HistoricalSlidingWindowOperator { storage, window_def, - current_start: start_time, - evaluation_time: now, + current_evaluation_time: now, + latest_evaluation_time: now, } } } @@ -43,10 +38,10 @@ impl Iterator for HistoricalSlidingWindowOperator { type Item = Vec; fn next(&mut self) -> Option { - let window_start = self.current_start; - let window_end = window_start.checked_add(self.window_def.width)?; + let (window_start, window_end) = + self.window_def.resolve_historical_bounds(self.current_evaluation_time)?; - if window_end > self.evaluation_time { + if window_end > self.latest_evaluation_time { return None; } @@ -55,7 +50,8 @@ impl Iterator for HistoricalSlidingWindowOperator { match events_result { Ok(events) => { // Advance the window - self.current_start += self.window_def.slide; + self.current_evaluation_time = + self.current_evaluation_time.checked_add(self.window_def.slide)?; Some(events) } Err(e) => { diff --git a/tests/historical_sliding_window_test.rs b/tests/historical_sliding_window_test.rs index a473be3..37c264c 100644 --- a/tests/historical_sliding_window_test.rs +++ b/tests/historical_sliding_window_test.rs @@ -77,7 +77,7 @@ fn test_historical_sliding_window_with_real_iris() { let mut operator = HistoricalSlidingWindowOperator::new(storage.clone(), window_def); - // Window 1: [now-500, now-300] + // Window 1: [now-700, now-500] let w1 = operator.next().unwrap(); assert!(w1.len() >= 2); // At least 2 events (type + value for first sensor) @@ -90,11 +90,11 @@ fn test_historical_sliding_window_with_real_iris() { first_event.timestamp ); - // Window 2: [now-400, now-200] + // Window 2: [now-600, now-400] let w2 = operator.next().unwrap(); assert!(w2.len() >= 2); - // Window 3: [now-300, now-100] + // Window 3: [now-500, now-300] let w3 = operator.next().unwrap(); assert!(w3.len() >= 2); } diff --git a/tests/historical_window_bounds_test.rs b/tests/historical_window_bounds_test.rs index 017c7fc..88eb5ff 100644 --- a/tests/historical_window_bounds_test.rs +++ b/tests/historical_window_bounds_test.rs @@ -31,13 +31,13 @@ fn fixed_window() -> WindowDefinition { #[test] fn resolves_sliding_historical_bounds_for_first_evaluation() { let window = sliding_window(); - assert_eq!(window.resolve_historical_bounds(172_800_000), Some((86_400_000, 86_460_000))); + assert_eq!(window.resolve_historical_bounds(172_800_000), Some((86_340_000, 86_400_000))); } #[test] fn resolves_sliding_historical_bounds_for_next_evaluation() { let window = sliding_window(); - assert_eq!(window.resolve_historical_bounds(172_860_000), Some((86_460_000, 86_520_000))); + assert_eq!(window.resolve_historical_bounds(172_860_000), Some((86_400_000, 86_460_000))); } #[test] @@ -47,6 +47,21 @@ fn sliding_historical_bounds_return_none_when_first_window_would_cross_evaluatio assert_eq!(window.resolve_historical_bounds(172_800_000), None); } +#[test] +fn sliding_historical_bounds_return_none_on_underflow() { + let window = sliding_window(); + assert_eq!(window.resolve_historical_bounds(86_400_000), None); +} + +#[test] +fn sliding_historical_bounds_have_the_configured_width() { + let window = sliding_window(); + let (start, end) = window.resolve_historical_bounds(172_800_000).unwrap(); + assert_eq!(end, 172_800_000 - 86_400_000); + assert_eq!(start, end - 60_000); + assert_eq!(end - start, 60_000); +} + #[test] fn resolves_fixed_historical_bounds_independent_of_evaluation_time() { let window = fixed_window(); diff --git a/tests/public_spec_behavior_test.rs b/tests/public_spec_behavior_test.rs index 8fc6e6f..9bcf5ab 100644 --- a/tests/public_spec_behavior_test.rs +++ b/tests/public_spec_behavior_test.rs @@ -327,7 +327,7 @@ fn spec_hybrid_historical_sliding_query_parses() { } #[test] -fn spec_historical_sliding_bounds_follow_t_minus_offset_plus_range_formula() { +fn spec_historical_sliding_bounds_follow_t_minus_offset_minus_range_formula() { let window = WindowDefinition { window_name: "http://example.org/previousHour".to_string(), source_kind: SourceKind::Log, @@ -341,8 +341,8 @@ fn spec_historical_sliding_bounds_follow_t_minus_offset_plus_range_formula() { }; let evaluation_time = 200_000_000; - let expected_start = evaluation_time - 86_400_000; - let expected_end = expected_start + 3_600_000; + let expected_end = evaluation_time - 86_400_000; + let expected_start = expected_end - 3_600_000; assert_eq!( window.resolve_historical_bounds(evaluation_time), diff --git a/tests/segmented_storage_error_test.rs b/tests/segmented_storage_error_test.rs index bbde812..0a70388 100644 --- a/tests/segmented_storage_error_test.rs +++ b/tests/segmented_storage_error_test.rs @@ -50,3 +50,32 @@ fn test_background_flush_failure_surfaces_as_storage_error() { "unexpected shutdown error: {shutdown_err}" ); } + +#[test] +fn synchronous_flush_does_not_commit_a_segment_when_dictionary_persistence_fails() { + let temp_dir = TempDir::new().expect("failed to create temp dir"); + let storage_dir = temp_dir.path().join("storage"); + let storage = StreamingSegmentedStorage::new(StreamingConfig { + segment_base_path: storage_dir.to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1024 * 1024, + sparse_interval: 1, + entries_per_index_block: 2, + }) + .unwrap(); + + fs::create_dir(storage_dir.join("dictionary.bin")).unwrap(); + storage.write_rdf(1_000, "http://a", "http://b", "value", "http://g").unwrap(); + assert!(storage.flush().is_err()); + assert!(fs::read_dir(&storage_dir).unwrap().all(|entry| !entry + .unwrap() + .file_name() + .to_string_lossy() + .ends_with(".log"))); + assert_eq!(storage.query(0, 2_000).unwrap().len(), 1); + + fs::remove_dir(storage_dir.join("dictionary.bin")).unwrap(); + storage.flush().unwrap(); + assert_eq!(storage.query(0, 2_000).unwrap().len(), 1); +} diff --git a/tests/segmented_storage_regression_test.rs b/tests/segmented_storage_regression_test.rs index a189480..55f707f 100644 --- a/tests/segmented_storage_regression_test.rs +++ b/tests/segmented_storage_regression_test.rs @@ -270,3 +270,59 @@ fn test_shutdown_race_safety() { } } } + +#[test] +fn opening_persisted_segments_without_dictionary_fails() { + let temp_dir = TempDir::new().expect("failed to create temp dir"); + let storage_dir = temp_dir.path().join("storage"); + let config = StreamingConfig { + segment_base_path: storage_dir.to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1_000_000, + sparse_interval: 1, + entries_per_index_block: 2, + }; + + { + let storage = StreamingSegmentedStorage::new(config.clone()).unwrap(); + storage.write_rdf(1_000, "http://a", "http://b", "value", "http://g").unwrap(); + storage.flush().unwrap(); + } + + fs::remove_file(storage_dir.join("dictionary.bin")).unwrap(); + let err = match StreamingSegmentedStorage::new(config) { + Ok(_) => panic!("missing dictionary must fail"), + Err(err) => err, + }; + assert!(err.to_string().contains("persisted segment data")); + assert!(err.to_string().contains("dictionary")); +} + +#[test] +fn opening_persisted_segments_with_corrupt_dictionary_fails() { + let temp_dir = TempDir::new().expect("failed to create temp dir"); + let storage_dir = temp_dir.path().join("storage"); + let config = StreamingConfig { + segment_base_path: storage_dir.to_string_lossy().into_owned(), + max_batch_events: 100, + max_batch_age_seconds: 60, + max_batch_bytes: 1_000_000, + sparse_interval: 1, + entries_per_index_block: 2, + }; + + { + let storage = StreamingSegmentedStorage::new(config.clone()).unwrap(); + storage.write_rdf(1_000, "http://a", "http://b", "value", "http://g").unwrap(); + storage.flush().unwrap(); + } + + fs::write(storage_dir.join("dictionary.bin"), b"not a dictionary").unwrap(); + let err = match StreamingSegmentedStorage::new(config) { + Ok(_) => panic!("corrupt dictionary must fail"), + Err(err) => err, + }; + assert!(err.to_string().contains("persisted segment data")); + assert!(err.to_string().contains("readable dictionary")); +}