fix: correct historical bounds and segmented storage recovery

Author: argahsuknesib

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);
 }

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<Item = Result<Vec<HashMap<String, String>>, 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,47 +389,40 @@ 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,
 }
 
 impl<'a> Iterator for SlidingWindowIterator<'a> {
     type Item = Result<Vec<HashMap<String, String>>, JanusApiError>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        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::<Vec<_>>();
 
-        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::<Vec<_>>();
 
-        assert_eq!(results.len(), 2);
+        assert_eq!(results.len(), 3);
         assert!(results.iter().all(|result| result.is_ok()));
     }
 

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))
             }
         }

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<RwLock<BatchBuffer>>, 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::<Event>();
-        }
-
-        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,
-        };
-    }
 }

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.