feoxdb/core/store/ttl.rs
1use bytes::Bytes;
2use std::sync::atomic::Ordering;
3use std::sync::Arc;
4use std::time::{SystemTime, UNIX_EPOCH};
5
6use crate::core::record::Record;
7use crate::core::ttl_sweep::TtlConfig;
8use crate::error::{FeoxError, Result};
9
10use super::FeoxStore;
11
12enum TtlReplacementValue {
13 Resident(Bytes),
14 Deferred(Arc<Record>),
15}
16
17impl FeoxStore {
18 /// Insert or update a key-value pair with TTL (Time-To-Live).
19 ///
20 /// # Arguments
21 ///
22 /// * `key` - The key to insert
23 /// * `value` - The value to store
24 /// * `ttl_seconds` - Time-to-live in seconds
25 ///
26 /// # Returns
27 ///
28 /// Returns `Ok(())` if successful.
29 ///
30 /// # Example
31 ///
32 /// ```rust
33 /// # use feoxdb::FeoxStore;
34 /// # fn main() -> feoxdb::Result<()> {
35 /// # let store = FeoxStore::builder().enable_ttl(true).build()?;
36 /// // Key expires after 60 seconds
37 /// store.insert_with_ttl(b"session:123", b"data", 60)?;
38 /// # Ok(())
39 /// # }
40 /// ```
41 ///
42 /// # Performance
43 ///
44 /// * Memory mode: ~800ns
45 /// * Persistent mode: ~1µs (buffered write)
46 pub fn insert_with_ttl(&self, key: &[u8], value: &[u8], ttl_seconds: u64) -> Result<bool> {
47 if !self.enable_ttl {
48 return Err(FeoxError::TtlNotEnabled);
49 }
50 self.insert_with_ttl_and_timestamp(key, value, ttl_seconds, None)
51 }
52
53 /// Insert or update a key-value pair with TTL and explicit timestamp.
54 ///
55 /// # Arguments
56 ///
57 /// * `key` - The key to insert
58 /// * `value` - The value to store
59 /// * `ttl_seconds` - Time-to-live in seconds
60 /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
61 ///
62 /// # Returns
63 ///
64 /// Returns `Ok(())` if successful.
65 pub fn insert_with_ttl_and_timestamp(
66 &self,
67 key: &[u8],
68 value: &[u8],
69 ttl_seconds: u64,
70 timestamp: Option<u64>,
71 ) -> Result<bool> {
72 if !self.enable_ttl {
73 return Err(FeoxError::TtlNotEnabled);
74 }
75 self.ensure_ttl_write_supported()?;
76 self.insert_with_timestamp_and_ttl_internal(key, value, timestamp, ttl_seconds)
77 }
78
79 /// Insert or update a key-value pair with TTL using zero-copy Bytes.
80 ///
81 /// This method avoids copying the value data by directly using the Bytes type,
82 /// which provides reference-counted zero-copy semantics.
83 ///
84 /// # Arguments
85 ///
86 /// * `key` - The key to insert
87 /// * `value` - The value to store as Bytes
88 /// * `ttl_seconds` - Time-to-live in seconds
89 ///
90 /// # Returns
91 ///
92 /// Returns `Ok(())` if successful.
93 ///
94 /// # Example
95 ///
96 /// ```rust
97 /// # use feoxdb::FeoxStore;
98 /// # use bytes::Bytes;
99 /// # fn main() -> feoxdb::Result<()> {
100 /// # let store = FeoxStore::builder().enable_ttl(true).build()?;
101 /// let data = Bytes::from_static(b"session_data");
102 /// // Key expires after 60 seconds
103 /// store.insert_bytes_with_ttl(b"session:123", data, 60)?;
104 /// # Ok(())
105 /// # }
106 /// ```
107 ///
108 /// # Performance
109 ///
110 /// * Memory mode: ~800ns (avoids value copy)
111 /// * Persistent mode: ~1µs (buffered write, avoids value copy)
112 pub fn insert_bytes_with_ttl(
113 &self,
114 key: &[u8],
115 value: Bytes,
116 ttl_seconds: u64,
117 ) -> Result<bool> {
118 if !self.enable_ttl {
119 return Err(FeoxError::TtlNotEnabled);
120 }
121 self.insert_bytes_with_ttl_and_timestamp(key, value, ttl_seconds, None)
122 }
123
124 /// Insert or update a key-value pair with TTL and explicit timestamp using zero-copy Bytes.
125 ///
126 /// # Arguments
127 ///
128 /// * `key` - The key to insert
129 /// * `value` - The value to store as Bytes
130 /// * `ttl_seconds` - Time-to-live in seconds
131 /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
132 ///
133 /// # Returns
134 ///
135 /// Returns `Ok(())` if successful.
136 pub fn insert_bytes_with_ttl_and_timestamp(
137 &self,
138 key: &[u8],
139 value: Bytes,
140 ttl_seconds: u64,
141 timestamp: Option<u64>,
142 ) -> Result<bool> {
143 if !self.enable_ttl {
144 return Err(FeoxError::TtlNotEnabled);
145 }
146 self.ensure_ttl_write_supported()?;
147 self.insert_bytes_with_timestamp_and_ttl_internal(key, value, timestamp, ttl_seconds)
148 }
149
150 /// Get the remaining TTL (Time-To-Live) for a key in seconds.
151 ///
152 /// # Arguments
153 ///
154 /// * `key` - The key to check
155 ///
156 /// # Returns
157 ///
158 /// Returns `Some(seconds)` if the key has TTL set, `None` if no TTL or key not found.
159 ///
160 /// # Example
161 ///
162 /// ```rust
163 /// # use feoxdb::FeoxStore;
164 /// # fn main() -> feoxdb::Result<()> {
165 /// # let store = FeoxStore::builder().enable_ttl(true).build()?;
166 /// store.insert_with_ttl(b"session", b"data", 3600)?;
167 ///
168 /// // Check remaining TTL
169 /// if let Ok(Some(ttl)) = store.get_ttl(b"session") {
170 /// println!("Session expires in {} seconds", ttl);
171 /// }
172 /// # Ok(())
173 /// # }
174 /// ```
175 pub fn get_ttl(&self, key: &[u8]) -> Result<Option<u64>> {
176 if !self.enable_ttl {
177 return Err(FeoxError::TtlNotEnabled);
178 }
179 self.validate_key(key)?;
180
181 let record = self
182 .hash_table
183 .read(key, |_, v| v.clone())
184 .ok_or(FeoxError::KeyNotFound)?;
185 let ttl_expiry = record.ttl_expiry.load(Ordering::Acquire);
186
187 if ttl_expiry == 0 {
188 return Ok(None); // No TTL set
189 }
190
191 let now = self.get_timestamp_pub();
192 if now >= ttl_expiry {
193 return Ok(Some(0)); // Already expired
194 }
195
196 // Return remaining seconds
197 Ok(Some((ttl_expiry - now) / 1_000_000_000))
198 }
199
200 /// Update the TTL for an existing key.
201 ///
202 /// # Arguments
203 ///
204 /// * `key` - The key to update
205 /// * `ttl_seconds` - New TTL in seconds (0 to remove TTL)
206 ///
207 /// # Returns
208 ///
209 /// Returns `Ok(())` if successful.
210 ///
211 /// # Errors
212 ///
213 /// * `KeyNotFound` - Key does not exist
214 ///
215 /// # Example
216 ///
217 /// ```rust
218 /// # use feoxdb::FeoxStore;
219 /// # fn main() -> feoxdb::Result<()> {
220 /// # let store = FeoxStore::builder().enable_ttl(true).build()?;
221 /// # store.insert(b"key", b"value")?;
222 /// // Extend TTL to 1 hour
223 /// store.update_ttl(b"key", 3600)?;
224 /// # Ok(())
225 /// # }
226 /// ```
227 pub fn update_ttl(&self, key: &[u8], ttl_seconds: u64) -> Result<()> {
228 if !self.enable_ttl {
229 return Err(FeoxError::TtlNotEnabled);
230 }
231 self.ensure_ttl_write_supported()?;
232 self.validate_key(key)?;
233
234 let (new_record, old_record, cache_guarded) = self
235 .hash_table
236 .update(key, |stored_key, current| {
237 let old_record = Arc::clone(current);
238 let old_expiry = old_record.ttl_expiry.load(Ordering::Acquire);
239 let now = self.get_timestamp_pub();
240 if old_expiry > 0 && now > old_expiry {
241 return Err(FeoxError::KeyNotFound);
242 }
243 let timestamp = self.version_clock.next(stored_key, now).max(
244 old_record
245 .timestamp
246 .checked_add(1)
247 .ok_or(FeoxError::OlderTimestamp)?,
248 );
249 let expiry = ttl_expiry(now, ttl_seconds);
250 let resident = old_record.get_value();
251 let cache_entry = resident.is_none().then(|| {
252 self.cache
253 .as_ref()
254 .map(|cache| cache.record_entry(stored_key, &old_record))
255 });
256 let cache_entry = cache_entry.flatten();
257 let cache_guarded = cache_entry.is_some();
258 let value = Self::ttl_replacement_value(
259 &old_record,
260 resident.or_else(|| cache_entry.as_ref().and_then(|entry| entry.value())),
261 );
262 let new_record = match value {
263 TtlReplacementValue::Resident(value) if expiry == 0 => {
264 Arc::new(Record::new_from_bytes(stored_key.clone(), value, timestamp))
265 }
266 TtlReplacementValue::Resident(value) => {
267 Arc::new(Record::new_from_bytes_with_ttl(
268 stored_key.clone(),
269 value,
270 timestamp,
271 expiry,
272 ))
273 }
274 TtlReplacementValue::Deferred(predecessor) => Arc::new(
275 Record::new_deferred_with_ttl(&predecessor, timestamp, expiry),
276 ),
277 };
278
279 old_record.link_successor(&new_record);
280 old_record.refcount.store(0, Ordering::Release);
281 *current = Arc::clone(&new_record);
282 if let Some(entry) = cache_entry {
283 entry.remove();
284 }
285 self.publish_to_tree(stored_key, Arc::clone(&new_record));
286 self.note_ttl_transition(old_expiry, expiry);
287
288 Ok((new_record, old_record, cache_guarded))
289 })
290 .ok_or(FeoxError::KeyNotFound)??;
291
292 if !cache_guarded {
293 self.remove_cached(key, &old_record);
294 }
295
296 if let Some(write_buffer) = self.write_buffer.as_ref() {
297 write_buffer.add_replacement(new_record, old_record)?;
298 }
299
300 Ok(())
301 }
302
303 fn ttl_replacement_value(record: &Arc<Record>, value: Option<Bytes>) -> TtlReplacementValue {
304 if let Some(value) = value {
305 return TtlReplacementValue::Resident(value);
306 }
307
308 let predecessor = Arc::clone(record);
309 #[cfg(test)]
310 crate::test_hooks::pause_at(crate::test_hooks::AFTER_TTL_DEFERRED_SOURCE);
311 TtlReplacementValue::Deferred(predecessor)
312 }
313
314 pub(super) fn note_ttl_transition(&self, previous: u64, current: u64) {
315 match (previous > 0, current > 0) {
316 (false, true) => {
317 self.stats.keys_with_ttl.fetch_add(1, Ordering::Relaxed);
318 }
319 (true, false) => {
320 let _ = self.stats.keys_with_ttl.fetch_update(
321 Ordering::Relaxed,
322 Ordering::Relaxed,
323 |count| Some(count.saturating_sub(1)),
324 );
325 }
326 _ => {}
327 }
328 }
329
330 pub(super) fn ensure_ttl_write_supported(&self) -> Result<()> {
331 if !self.memory_only && self.format_version == 1 {
332 return Err(FeoxError::Unsupported);
333 }
334 Ok(())
335 }
336
337 /// Remove TTL from a key, making it persistent.
338 ///
339 /// # Arguments
340 ///
341 /// * `key` - The key to persist
342 ///
343 /// # Returns
344 ///
345 /// Returns `Ok(())` if successful.
346 ///
347 /// # Errors
348 ///
349 /// * `KeyNotFound` - Key does not exist
350 ///
351 /// # Example
352 ///
353 /// ```rust
354 /// # use feoxdb::FeoxStore;
355 /// # fn main() -> feoxdb::Result<()> {
356 /// # let store = FeoxStore::builder().enable_ttl(true).build()?;
357 /// # store.insert_with_ttl(b"temp", b"data", 60)?;
358 /// // Remove TTL, make permanent
359 /// store.persist(b"temp")?;
360 /// # Ok(())
361 /// # }
362 /// ```
363 pub fn persist(&self, key: &[u8]) -> Result<()> {
364 if !self.enable_ttl {
365 return Err(FeoxError::TtlNotEnabled);
366 }
367 self.update_ttl(key, 0)
368 }
369
370 /// Start the TTL sweeper if configured
371 /// This must be called with an `Arc<Self>` after construction
372 pub fn start_ttl_sweeper(self: &Arc<Self>, config: Option<TtlConfig>) {
373 // Only start TTL sweeper if TTL is enabled
374 if !self.enable_ttl {
375 return;
376 }
377
378 let ttl_config = config.unwrap_or_else(|| {
379 if self.memory_only {
380 TtlConfig::default_memory()
381 } else {
382 TtlConfig::default_persistent()
383 }
384 });
385
386 if ttl_config.enabled {
387 let weak_store = Arc::downgrade(self);
388 let mut sweeper = crate::core::ttl_sweep::TtlSweeper::new(weak_store, ttl_config);
389 sweeper.start();
390
391 // Store the sweeper
392 *self.ttl_sweeper.write() = Some(sweeper);
393 }
394 }
395
396 /// Get current timestamp (public for TTL cleaner)
397 pub fn get_timestamp_pub(&self) -> u64 {
398 SystemTime::now()
399 .duration_since(UNIX_EPOCH)
400 .unwrap()
401 .as_nanos() as u64
402 }
403}
404
405#[inline]
406fn ttl_expiry(timestamp: u64, ttl_seconds: u64) -> u64 {
407 if ttl_seconds == 0 {
408 0
409 } else {
410 timestamp.saturating_add(ttl_seconds.saturating_mul(1_000_000_000))
411 }
412}