Fix unbounded interval sampling, persistence, and Allen relations

src/persistence.rs

@@ -489,6 +489,10 @@ const KEY_DSU_CLUSTER_COUNT: &[u8] = b"dsu_cluster_count";
 const STORAGE_FORMAT_VERSION: u32 = 1;
 const INDEX_FORMAT_VERSION: u32 = 2;
 const TEMPORAL_BUCKET_SECONDS: i64 = 86400;
+const MAX_TEMPORAL_BUCKETS: i64 = 4096;
+// No timestamp divided by TEMPORAL_BUCKET_SECONDS can produce this bucket.
+// Existing day-indexed records remain readable without rebuilding their indexes.
+const WIDE_TEMPORAL_BUCKET: i64 = i64::MIN;
 const DEFAULT_CACHE_CAPACITY: usize = 100_000;
 const DEFAULT_BLOCK_CACHE_MB: u64 = 512;
 const DEFAULT_WRITE_BUFFER_MB: u64 = 128;
@@ -1598,6 +1602,9 @@ impl RecordStore for PersistentStore {
     }
 
     fn get_records_in_interval(&self, interval: crate::temporal::Interval) -> Vec<Record> {
+        if interval.is_empty() {
+            return Vec::new();
+        }
         let cf = match self.db.cf_handle(CF_INDEX_TEMPORAL_BUCKET) {
             Some(cf) => cf,
             None => {
@@ -1606,29 +1613,53 @@ impl RecordStore for PersistentStore {
             }
         };
         let mut candidates = std::collections::HashSet::new();
-        for bucket in buckets_for_interval(interval.start, interval.end) {
-            let prefix = bucket.to_be_bytes();
-            let iter = self
-                .db
-                .iterator_cf(cf, IteratorMode::From(&prefix, Direction::Forward));
-            for entry in iter {
+        let buckets = buckets_for_interval(interval.start, interval.end);
+        if buckets == [WIDE_TEMPORAL_BUCKET] {
+            // A broad query must also find ordinary day-indexed records, including
+            // records written by older versions. Scan existing entries, not days.
+            for entry in self.db.iterator_cf(cf, IteratorMode::Start) {
                 let (key, _) = match entry {
                     Ok(pair) => pair,
                     Err(_) => {
                         self.mark_read_fault();
                         break;
                     }
                 };
-                if !key.starts_with(&prefix) {
-                    break;
-                }
                 if let Some(record_id) = decode_temporal_record_id(&key) {
                     candidates.insert(record_id);
                 } else {
                     self.mark_read_fault();
                     break;
                 }
             }
+        } else {
+            for bucket in buckets
+                .into_iter()
+                .chain(std::iter::once(WIDE_TEMPORAL_BUCKET))
+            {
+                let prefix = bucket.to_be_bytes();
+                let iter = self
+                    .db
+                    .iterator_cf(cf, IteratorMode::From(&prefix, Direction::Forward));
+                for entry in iter {
+                    let (key, _) = match entry {
+                        Ok(pair) => pair,
+                        Err(_) => {
+                            self.mark_read_fault();
+                            break;
+                        }
+                    };
+                    if !key.starts_with(&prefix) {
+                        break;
+                    }
+                    if let Some(record_id) = decode_temporal_record_id(&key) {
+                        candidates.insert(record_id);
+                    } else {
+                        self.mark_read_fault();
+                        break;
+                    }
+                }
+            }
         }
         let mut records = Vec::new();
         for record_id in candidates {
@@ -2462,6 +2493,9 @@ fn buckets_for_interval(start: i64, end: i64) -> Vec<i64> {
     let mut buckets = Vec::new();
     let mut current = start.div_euclid(TEMPORAL_BUCKET_SECONDS);
     let end_bucket = (end - 1).div_euclid(TEMPORAL_BUCKET_SECONDS);
+    if end_bucket - current >= MAX_TEMPORAL_BUCKETS {
+        return vec![WIDE_TEMPORAL_BUCKET];
+    }
     while current <= end_bucket {
         buckets.push(current);
         current += 1;
@@ -3417,6 +3451,27 @@ mod tests {
             .unwrap_or_else(std::sync::PoisonError::into_inner)
     }
 
+    #[test]
+    fn temporal_bucket_expansion_is_bounded() {
+        assert_eq!(buckets_for_interval(0, 0), Vec::<i64>::new());
+        assert_eq!(buckets_for_interval(-1, 1), vec![-1, 0]);
+        assert_eq!(buckets_for_interval(0, TEMPORAL_BUCKET_SECONDS), vec![0]);
+        assert_eq!(
+            buckets_for_interval(0, MAX_TEMPORAL_BUCKETS * TEMPORAL_BUCKET_SECONDS).len(),
+            MAX_TEMPORAL_BUCKETS as usize
+        );
+        assert_eq!(
+            buckets_for_interval(0, MAX_TEMPORAL_BUCKETS * TEMPORAL_BUCKET_SECONDS + 1),
+            vec![WIDE_TEMPORAL_BUCKET]
+        );
+        assert_eq!(
+            buckets_for_interval(i64::MIN, i64::MAX),
+            vec![WIDE_TEMPORAL_BUCKET]
+        );
+        assert_eq!(buckets_for_interval(i64::MIN, i64::MIN + 1).len(), 1);
+        assert_eq!(buckets_for_interval(i64::MAX - 1, i64::MAX).len(), 1);
+    }
+
     #[test]
     fn persistent_store_round_trip() {
         let _guard = lock_persistent_tests();

src/temporal.rs

@@ -90,12 +90,13 @@ impl Interval {
     }
 
     /// Get the duration of this interval in seconds
-    /// Returns None for intervals with infinite endpoints
+    /// Returns None for intervals with infinite endpoints. Finite durations that
+    /// exceed i64::MAX seconds saturate at i64::MAX.
     pub fn duration(&self) -> Option<i64> {
         if self.start == NEG_INF || self.end == POS_INF {
             None
         } else {
-            Some(self.end - self.start)
+            Some(self.end.saturating_sub(self.start))
         }
     }
 
@@ -107,13 +108,14 @@ impl Interval {
     }
 
     /// Calculate the overlap duration between this interval and another.
-    /// Returns 0 if the intervals don't overlap.
+    /// Returns 0 if the intervals don't overlap. Durations exceeding i64::MAX
+    /// seconds saturate at i64::MAX, including overlap across infinite endpoints.
     #[inline]
     pub fn overlap_duration(&self, other: &Interval) -> i64 {
         let overlap_start = self.start.max(other.start);
         let overlap_end = self.end.min(other.end);
         if overlap_start < overlap_end {
-            overlap_end - overlap_start
+            overlap_end.saturating_sub(overlap_start)
         } else {
             0
         }
@@ -164,7 +166,7 @@ impl Ord for Interval {
 /// All relations are mutually exclusive and collectively exhaustive.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum AllenRelation {
-    /// a precedes b: a.end <= b.start
+    /// a precedes b: a.end < b.start
     Precedes,
     /// a meets b: a.end == b.start (adjacent)
     Meets,
@@ -182,13 +184,13 @@ pub enum AllenRelation {
     StartedBy,
     /// a contains b: a.start < b.start && b.end < a.end
     Contains,
-    /// a finished by b: b.start < a.start && b.end == a.end
+    /// a finished by b: a.start < b.start && a.end == b.end
     FinishedBy,
     /// a overlapped by b: b.start < a.start && a.start < b.end < a.end
     OverlappedBy,
     /// a met by b: b.end == a.start
     MetBy,
-    /// a preceded by b: b.end <= a.start
+    /// a preceded by b: b.end < a.start
     PrecededBy,
 }
 
@@ -200,7 +202,7 @@ pub fn allen_relation(a: &Interval, b: &Interval) -> AllenRelation {
         (Ordering::Equal, Ordering::Equal) => Equals,
         (Ordering::Equal, Ordering::Less) => Starts,
         (Ordering::Equal, Ordering::Greater) => StartedBy,
-        (Ordering::Less, Ordering::Equal) => Finishes,
+        (Ordering::Less, Ordering::Equal) => FinishedBy,
         (Ordering::Less, Ordering::Less) => {
             if a.end < b.start {
                 Precedes
@@ -212,7 +214,7 @@ pub fn allen_relation(a: &Interval, b: &Interval) -> AllenRelation {
                 Contains
             }
         }
-        (Ordering::Greater, Ordering::Equal) => FinishedBy,
+        (Ordering::Greater, Ordering::Equal) => Finishes,
         (Ordering::Greater, Ordering::Greater) => {
             if b.end < a.start {
                 PrecededBy
@@ -431,14 +433,65 @@ mod tests {
 
     #[test]
     fn test_allen_relations() {
+        use AllenRelation::*;
+
         let a = Interval::new(100, 200).unwrap();
-        let b = Interval::new(150, 250).unwrap();
-        let c = Interval::new(200, 300).unwrap();
-        let d = Interval::new(50, 100).unwrap();
+        for (start, end, expected, inverse) in [
+            (250, 300, Precedes, PrecededBy),
+            (200, 300, Meets, MetBy),
+            (150, 250, Overlaps, OverlappedBy),
+            (100, 250, Starts, StartedBy),
+            (50, 250, During, Contains),
+            (50, 200, Finishes, FinishedBy),
+            (100, 200, Equals, Equals),
+        ] {
+            let b = Interval::new(start, end).unwrap();
+            assert_eq!(allen_relation(&a, &b), expected);
+            assert_eq!(allen_relation(&b, &a), inverse);
+        }
+    }
 
-        assert_eq!(allen_relation(&a, &b), AllenRelation::Overlaps);
-        assert_eq!(allen_relation(&a, &c), AllenRelation::Meets);
-        assert_eq!(allen_relation(&a, &d), AllenRelation::MetBy); // d meets a, so a is met by d
+    #[test]
+    fn test_overlap_duration_extremes() {
+        let all = Interval::all_time();
+        for (a, b, expected) in [
+            (all, all, i64::MAX),
+            (all, Interval::from_start(-100), i64::MAX),
+            (all, Interval::until_end(100), i64::MAX),
+            (Interval::from_start(100), Interval::until_end(200), 100),
+            (all, Interval::new(-100, 100).unwrap(), 200),
+            (
+                Interval::new(0, 100).unwrap(),
+                Interval::new(50, 150).unwrap(),
+                50,
+            ),
+            (Interval::until_end(100), Interval::from_start(100), 0),
+            (Interval::until_end(100), Interval::from_start(200), 0),
+            (
+                Interval::new(i64::MIN + 1, i64::MAX - 1).unwrap(),
+                all,
+                i64::MAX,
+            ),
+        ] {
+            assert_eq!(a.overlap_duration(&b), expected, "{a} and {b}");
+            assert_eq!(b.overlap_duration(&a), expected, "{b} and {a}");
+        }
+    }
+
+    #[test]
+    fn test_duration_extremes() {
+        for interval in [
+            Interval::all_time(),
+            Interval::from_start(-100),
+            Interval::until_end(100),
+        ] {
+            assert_eq!(interval.duration(), None);
+            assert_eq!(interval.duration_or_zero(), 0);
+        }
+        let wide = Interval::new(i64::MIN + 1, i64::MAX - 1).unwrap();
+        assert_eq!(wide.duration(), Some(i64::MAX));
+        assert_eq!(wide.duration_or_zero(), i64::MAX);
+        assert_eq!(Interval::new(-100, 100).unwrap().duration(), Some(200));
     }
 
     #[test]

tests/distributed_e2e.rs

@@ -158,6 +158,62 @@ fn normalize_remote_query(
     }
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn distributed_ingest_and_query_unbounded_intervals() -> anyhow::Result<()> {
+    let config = support::build_iam_config();
+    let (shard0, shard0_handle) = spawn_shard(0, config.clone()).await?;
+    let (shard1, shard1_handle) = spawn_shard(1, config.clone()).await?;
+    let (router, router_handle) = spawn_router(vec![shard0, shard1], config).await?;
+    let mut client = RouterServiceClient::connect(format!("http://{router}")).await?;
+
+    // Exercise partitioned batch ingestion, then match a later batch against its
+    // persisted unbounded descriptors through the router.
+    for batch in 0..2 {
+        let records = (0..100)
+            .map(|index| {
+                record_input(
+                    index,
+                    "person",
+                    "crm",
+                    &format!("unbounded-{batch}-{index}"),
+                    vec![("email", "unbounded@example.com", i64::MIN, i64::MAX)],
+                )
+            })
+            .collect();
+        let response = client
+            .ingest_records(IngestRecordsRequest {
+                internal_protocol_version: 5,
+                records,
+            })
+            .await?
+            .into_inner();
+        assert_eq!(response.assignments.len(), 100);
+    }
+
+    for (start, end) in [(0, 100), (i64::MIN, i64::MAX)] {
+        let response = client
+            .query_entities(proto::QueryEntitiesRequest {
+                descriptors: vec![proto::QueryDescriptor {
+                    attr: "email".into(),
+                    value: "unbounded@example.com".into(),
+                }],
+                start,
+                end,
+            })
+            .await?
+            .into_inner();
+        assert_eq!(
+            normalize_remote_query(response),
+            ("matches", 1, (start, end))
+        );
+    }
+
+    router_handle.abort();
+    shard0_handle.abort();
+    shard1_handle.abort();
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn distributed_stream_and_query() -> anyhow::Result<()> {
     let config = support::build_iam_config();