1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
//! Temporal index for efficient time-based queries in `EpisodicMemory`.
//!
//! Replaces the naive O(N) scan with an indexed structure that provides:
//! - O(log N) range queries by timestamp
//! - O(1) insertion and deletion
//! - Efficient retrieval of recent events
// Reason: u64 to usize casts are for deserialization counts. Data is created/loaded
// on the same architecture, and counts represent actual serialized entry counts.
// These values are validated against buffer bounds before use.
#![allow(clippy::cast_possible_truncation)]
use parking_lot::RwLock;
use std::collections::BTreeMap;
/// Entry in the temporal index.
#[derive(Debug, Clone)]
pub struct TemporalEntry {
/// Event ID.
pub id: u64,
/// Event timestamp (Unix seconds or milliseconds).
pub timestamp: i64,
}
/// Temporal index using a B-tree for efficient range queries.
///
/// Maintains a sorted index of timestamps to event IDs, allowing
/// efficient retrieval of events within time ranges.
pub struct TemporalIndex {
/// B-tree mapping timestamp to list of event IDs.
/// Multiple events can have the same timestamp.
by_timestamp: RwLock<BTreeMap<i64, Vec<u64>>>,
/// Reverse mapping from ID to timestamp for efficient removal.
id_to_timestamp: RwLock<std::collections::HashMap<u64, i64>>,
}
impl Default for TemporalIndex {
fn default() -> Self {
Self::new()
}
}
impl TemporalIndex {
/// Creates a new empty temporal index.
#[must_use]
pub fn new() -> Self {
Self {
by_timestamp: RwLock::new(BTreeMap::new()),
id_to_timestamp: RwLock::new(std::collections::HashMap::new()),
}
}
/// Inserts an entry into the index.
///
/// If the ID already exists, it will be updated with the new timestamp.
// Both guards span the whole update deliberately: `id_to_timestamp` and
// `by_timestamp` are two halves of one index, and releasing either between
// the two writes exposes a window where a reader sees an id in one map and
// not the other.
#[expect(clippy::significant_drop_tightening)]
pub fn insert(&self, id: u64, timestamp: i64) {
let mut id_to_ts = self.id_to_timestamp.write();
let mut by_ts = self.by_timestamp.write();
if let Some(&old_ts) = id_to_ts.get(&id) {
if old_ts == timestamp {
return;
}
if let Some(ids) = by_ts.get_mut(&old_ts) {
ids.retain(|&x| x != id);
if ids.is_empty() {
by_ts.remove(&old_ts);
}
}
}
id_to_ts.insert(id, timestamp);
by_ts.entry(timestamp).or_default().push(id);
}
/// Removes an entry from the index.
pub fn remove(&self, id: u64) {
let mut id_to_ts = self.id_to_timestamp.write();
let mut by_ts = self.by_timestamp.write();
if let Some(ts) = id_to_ts.remove(&id) {
if let Some(ids) = by_ts.get_mut(&ts) {
ids.retain(|&x| x != id);
if ids.is_empty() {
by_ts.remove(&ts);
}
}
}
}
/// Returns IDs of events within a timestamp range (inclusive).
///
/// # Arguments
///
/// * `start` - Start timestamp (inclusive)
/// * `end` - End timestamp (inclusive)
///
/// # Returns
///
/// Vector of (id, timestamp) pairs within the range.
#[must_use]
pub fn range(&self, start: i64, end: i64) -> Vec<TemporalEntry> {
let by_ts = self.by_timestamp.read();
let mut results = Vec::new();
for (&ts, ids) in by_ts.range(start..=end) {
for &id in ids {
results.push(TemporalEntry { id, timestamp: ts });
}
}
drop(by_ts);
results
}
/// Returns the most recent events up to a limit.
///
/// # Arguments
///
/// * `limit` - Maximum number of events to return
/// * `since_timestamp` - Optional filter for events after this time
///
/// # Returns
///
/// Vector of (id, timestamp) pairs, ordered by timestamp descending.
#[must_use]
pub fn recent(&self, limit: usize, since_timestamp: Option<i64>) -> Vec<TemporalEntry> {
// Read `id_to_timestamp` first (insert order) to honour lock ordering and
// to clamp the caller-supplied `limit`: results can never exceed the total
// number of indexed entries, so a huge `limit` must not pre-allocate beyond it.
let total = self.id_to_timestamp.read().len();
let by_ts = self.by_timestamp.read();
let mut results = Vec::with_capacity(limit.min(total));
for (&ts, ids) in by_ts.iter().rev() {
if let Some(since) = since_timestamp {
if ts <= since {
continue;
}
}
for &id in ids.iter().rev() {
results.push(TemporalEntry { id, timestamp: ts });
if results.len() >= limit {
return results;
}
}
}
drop(by_ts);
results
}
/// Returns events older than the specified timestamp.
///
/// # Arguments
///
/// * `before_timestamp` - Cutoff timestamp (exclusive)
/// * `limit` - Maximum number of events to return
///
/// # Returns
///
/// Vector of (id, timestamp) pairs, ordered by timestamp ascending.
#[must_use]
pub fn older_than(&self, before_timestamp: i64, limit: usize) -> Vec<TemporalEntry> {
// Clamp `limit` to the total indexed entries (read first, per lock order)
// so a huge caller-supplied `limit` cannot pre-allocate beyond available data.
let total = self.id_to_timestamp.read().len();
let by_ts = self.by_timestamp.read();
let mut results = Vec::with_capacity(limit.min(total));
for (&ts, ids) in by_ts.iter() {
if ts >= before_timestamp {
break;
}
for &id in ids {
results.push(TemporalEntry { id, timestamp: ts });
if results.len() >= limit {
return results;
}
}
}
drop(by_ts);
results
}
/// Returns the timestamp for an ID if it exists.
#[must_use]
pub fn get_timestamp(&self, id: u64) -> Option<i64> {
self.id_to_timestamp.read().get(&id).copied()
}
/// Returns the total number of indexed entries.
#[must_use]
pub fn len(&self) -> usize {
self.id_to_timestamp.read().len()
}
/// Returns true if the index is empty.
#[must_use]
pub fn is_empty(&self) -> bool {
self.id_to_timestamp.read().is_empty()
}
/// Clears all entries from the index.
// Same pair invariant as `insert`: dropping `by_timestamp` after its clear
// but before `id_to_timestamp`'s would leave a reader seeing a populated
// id map against an empty timestamp map.
#[expect(clippy::significant_drop_tightening)]
pub fn clear(&self) {
let mut id_to_ts = self.id_to_timestamp.write();
let mut by_ts = self.by_timestamp.write();
by_ts.clear();
id_to_ts.clear();
}
/// Returns all IDs in the index.
#[must_use]
pub fn all_ids(&self) -> Vec<u64> {
self.id_to_timestamp.read().keys().copied().collect()
}
/// Serializes the index to bytes for snapshot support.
#[must_use]
pub fn serialize(&self) -> Vec<u8> {
let id_to_ts = self.id_to_timestamp.read();
let count = id_to_ts.len();
let mut buf = Vec::with_capacity(8 + count * 16);
buf.extend_from_slice(&(count as u64).to_le_bytes());
for (&id, &ts) in id_to_ts.iter() {
buf.extend_from_slice(&id.to_le_bytes());
buf.extend_from_slice(&ts.to_le_bytes());
}
drop(id_to_ts);
buf
}
/// Deserializes the index from bytes.
///
/// # Returns
///
/// `None` if the data is malformed.
#[must_use]
pub fn deserialize(data: &[u8]) -> Option<Self> {
let count = super::memory_helpers::validate_binary_header(data, 16)?;
let index = Self::new();
for i in 0..count {
let offset = 8 + i * 16;
let id = u64::from_le_bytes(data[offset..offset + 8].try_into().ok()?);
let ts = i64::from_le_bytes(data[offset + 8..offset + 16].try_into().ok()?);
index.insert(id, ts);
}
Some(index)
}
/// Rebuilds the index from a collection of entries.
///
/// This is useful for initializing the index from existing data.
pub fn rebuild_from_entries(&self, entries: impl IntoIterator<Item = (u64, i64)>) {
self.clear();
for (id, timestamp) in entries {
self.insert(id, timestamp);
}
}
}
/// Statistics about the temporal index.
#[derive(Debug, Clone, Default)]
pub struct TemporalIndexStats {
/// Total number of entries.
pub entry_count: usize,
/// Number of unique timestamps.
pub unique_timestamps: usize,
/// Minimum timestamp in the index.
pub min_timestamp: Option<i64>,
/// Maximum timestamp in the index.
pub max_timestamp: Option<i64>,
}
impl TemporalIndex {
/// Returns statistics about the index.
///
/// Acquires `id_to_timestamp` **before** `by_timestamp` to honour the global
/// lock ordering used by every mutator (`insert`, `remove`, `clear`) and
/// reader (`recent`, `older_than`). Locking them in the opposite order here
/// previously risked a deadlock against a concurrent mutator.
#[must_use]
pub fn stats(&self) -> TemporalIndexStats {
let id_to_ts = self.id_to_timestamp.read();
let by_ts = self.by_timestamp.read();
TemporalIndexStats {
entry_count: id_to_ts.len(),
unique_timestamps: by_ts.len(),
min_timestamp: by_ts.keys().next().copied(),
max_timestamp: by_ts.keys().next_back().copied(),
}
}
}