feoxdb/core/store/
operations.rs

1use bytes::Bytes;
2use std::sync::atomic::Ordering;
3use std::sync::Arc;
4use std::time::{SystemTime, UNIX_EPOCH};
5
6use crate::constants::*;
7use crate::core::record::Record;
8use crate::error::{FeoxError, Result};
9
10use super::{FeoxStore, MemoryReservation};
11
12impl FeoxStore {
13    /// Insert or update a key-value pair.
14    ///
15    /// If the key already exists with a TTL, the TTL is removed (key becomes permanent).
16    /// To preserve or set TTL, use `insert_with_ttl()` instead.
17    ///
18    /// # Arguments
19    ///
20    /// * `key` - The key to insert
21    /// * `value` - The value to store
22    /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
23    ///
24    /// # Returns
25    ///
26    /// Returns `Ok(true)` if a new key was inserted, `Ok(false)` if an existing key was updated.
27    ///
28    /// # Errors
29    ///
30    /// * `InvalidKey` - Key is empty or too large
31    /// * `InvalidValue` - Value is too large
32    /// * `OlderTimestamp` - Timestamp is not newer than existing record
33    /// * `OutOfMemory` - Memory limit exceeded
34    ///
35    /// # Example
36    ///
37    /// ```rust
38    /// # use feoxdb::FeoxStore;
39    /// # fn main() -> feoxdb::Result<()> {
40    /// # let store = FeoxStore::new(None)?;
41    /// store.insert(b"user:123", b"{\"name\":\"Mehran\"}")?;
42    /// # Ok(())
43    /// # }
44    /// ```
45    ///
46    /// # Performance
47    ///
48    /// * Memory mode: ~600ns
49    /// * Persistent mode: ~800ns (buffered write)
50    pub fn insert(&self, key: &[u8], value: &[u8]) -> Result<bool> {
51        self.insert_with_timestamp(key, value, None)
52    }
53
54    /// Insert or update a key-value pair with explicit timestamp.
55    ///
56    /// This is the advanced version that allows manual timestamp control for
57    /// conflict resolution. Most users should use `insert()` instead.
58    ///
59    /// # Arguments
60    ///
61    /// * `key` - The key to insert
62    /// * `value` - The value to store
63    /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
64    ///
65    /// # Errors
66    ///
67    /// * `OlderTimestamp` - Timestamp is not newer than existing record
68    pub fn insert_with_timestamp(
69        &self,
70        key: &[u8],
71        value: &[u8],
72        timestamp: Option<u64>,
73    ) -> Result<bool> {
74        self.insert_with_timestamp_and_ttl_internal(key, value, timestamp, 0)
75    }
76
77    /// Insert or update a key-value pair using zero-copy Bytes.
78    ///
79    /// This method avoids copying the value data by directly using the Bytes type,
80    /// which provides reference-counted zero-copy semantics. Useful when inserting
81    /// data that was already read from network or disk as Bytes.
82    ///
83    /// If the key already exists with a TTL, the TTL is removed (key becomes permanent).
84    /// To preserve or set TTL, use `insert_bytes_with_ttl()` instead.
85    ///
86    /// # Arguments
87    ///
88    /// * `key` - The key to insert
89    /// * `value` - The value to store as Bytes
90    ///
91    /// # Returns
92    ///
93    /// Returns `Ok(true)` if a new key was inserted, `Ok(false)` if an existing key was updated.
94    ///
95    /// # Errors
96    ///
97    /// * `InvalidKey` - Key is empty or too large
98    /// * `InvalidValue` - Value is too large
99    /// * `OlderTimestamp` - Timestamp is not newer than existing record
100    /// * `OutOfMemory` - Memory limit exceeded
101    ///
102    /// # Example
103    ///
104    /// ```rust
105    /// # use feoxdb::FeoxStore;
106    /// # use bytes::Bytes;
107    /// # fn main() -> feoxdb::Result<()> {
108    /// # let store = FeoxStore::new(None)?;
109    /// let data = Bytes::from_static(b"{\"name\":\"Mehran\"}");
110    /// store.insert_bytes(b"user:123", data)?;
111    /// # Ok(())
112    /// # }
113    /// ```
114    ///
115    /// # Performance
116    ///
117    /// * Memory mode: ~600ns (avoids value copy)
118    /// * Persistent mode: ~800ns (buffered write, avoids value copy)
119    pub fn insert_bytes(&self, key: &[u8], value: Bytes) -> Result<bool> {
120        self.insert_bytes_with_timestamp(key, value, None)
121    }
122
123    /// Insert or update a key-value pair using zero-copy Bytes with explicit timestamp.
124    ///
125    /// This is the advanced version that allows manual timestamp control for
126    /// conflict resolution. Most users should use `insert_bytes()` instead.
127    ///
128    /// # Arguments
129    ///
130    /// * `key` - The key to insert
131    /// * `value` - The value to store as Bytes
132    /// * `timestamp` - Optional timestamp for conflict resolution. If `None`, uses current time.
133    ///
134    /// # Errors
135    ///
136    /// * `OlderTimestamp` - Timestamp is not newer than existing record
137    pub fn insert_bytes_with_timestamp(
138        &self,
139        key: &[u8],
140        value: Bytes,
141        timestamp: Option<u64>,
142    ) -> Result<bool> {
143        self.insert_bytes_with_timestamp_and_ttl_internal(key, value, timestamp, 0)
144    }
145
146    pub(super) fn insert_with_timestamp_and_ttl_internal(
147        &self,
148        key: &[u8],
149        value: &[u8],
150        timestamp: Option<u64>,
151        ttl_seconds: u64,
152    ) -> Result<bool> {
153        let start = std::time::Instant::now();
154        self.validate_key_value(key, value)?;
155        let (timestamp, explicit_timestamp) = self.resolve_timestamp(key, timestamp);
156        let ttl_expiry = if ttl_seconds > 0 && self.enable_ttl {
157            timestamp.saturating_add(ttl_seconds.saturating_mul(1_000_000_000))
158        } else {
159            0
160        };
161
162        let record_size = self.calculate_record_size(key.len(), value.len());
163
164        loop {
165            let existing_record = self.hash_table.read(key, |_, v| v.clone());
166            if let Some(existing_record) = existing_record {
167                if timestamp <= existing_record.timestamp {
168                    return Err(FeoxError::OlderTimestamp);
169                }
170                crate::test_hooks::pause_at(crate::test_hooks::AFTER_UPSERT_READ);
171
172                match self.update_record_with_ttl(
173                    &existing_record,
174                    value,
175                    timestamp,
176                    explicit_timestamp,
177                    ttl_expiry,
178                ) {
179                    Err(FeoxError::KeyNotFound) => continue,
180                    result => return result,
181                }
182            }
183
184            let reservation = self.reserve_memory(record_size)?;
185
186            let record = if ttl_expiry > 0 && self.enable_ttl {
187                Arc::new(Record::new_with_timestamp_ttl(
188                    key.to_vec(),
189                    value.to_vec(),
190                    timestamp,
191                    ttl_expiry,
192                ))
193            } else {
194                Arc::new(Record::new(key.to_vec(), value.to_vec(), timestamp))
195            };
196
197            let key_vec = record.key.clone();
198
199            let buffered_record = match self.hash_table.entry(key_vec.clone()) {
200                scc::hash_map::Entry::Vacant(entry) => {
201                    let buffered_record = self
202                        .write_buffer
203                        .as_ref()
204                        .filter(|_| !self.memory_only)
205                        .map(|_| Arc::clone(&record));
206                    let _entry = entry.insert_entry(Arc::clone(&record));
207                    self.insert_into_tree(key_vec, record);
208                    self.observe_published_timestamp(key, timestamp, explicit_timestamp);
209                    reservation.commit();
210                    if ttl_expiry > 0 && self.enable_ttl {
211                        self.stats.keys_with_ttl.fetch_add(1, Ordering::Relaxed);
212                    }
213                    self.stats.record_count.fetch_add(1, Ordering::Relaxed);
214                    buffered_record
215                }
216                scc::hash_map::Entry::Occupied(_) => continue,
217            };
218
219            self.stats
220                .record_insert(start.elapsed().as_nanos() as u64, false);
221
222            if let (Some(wb), Some(record)) = (&self.write_buffer, buffered_record) {
223                wb.add_write(Operation::Insert, record, 0)?;
224            }
225
226            return Ok(true);
227        }
228    }
229
230    /// Internal method to insert a Bytes value with timestamp and TTL (zero-copy)
231    pub(super) fn insert_bytes_with_timestamp_and_ttl_internal(
232        &self,
233        key: &[u8],
234        value: Bytes,
235        timestamp: Option<u64>,
236        ttl_seconds: u64,
237    ) -> Result<bool> {
238        let start = std::time::Instant::now();
239        self.validate_new_key(key)?;
240        let value_len = value.len();
241        if value_len == 0 || value_len > MAX_VALUE_SIZE {
242            return Err(FeoxError::InvalidValueSize);
243        }
244        let (timestamp, explicit_timestamp) = self.resolve_timestamp(key, timestamp);
245
246        let ttl_expiry = if ttl_seconds > 0 && self.enable_ttl {
247            timestamp.saturating_add(ttl_seconds.saturating_mul(1_000_000_000))
248        } else {
249            0
250        };
251        self.insert_bytes_with_expiry(key, value, timestamp, explicit_timestamp, ttl_expiry, start)
252    }
253
254    pub(super) fn insert_migrated_bytes(
255        &self,
256        key: &[u8],
257        value: Bytes,
258        timestamp: u64,
259        ttl_expiry: u64,
260    ) -> Result<bool> {
261        let start = std::time::Instant::now();
262        self.validate_new_key(key)?;
263        if value.is_empty() || value.len() > MAX_VALUE_SIZE {
264            return Err(FeoxError::InvalidValueSize);
265        }
266        self.insert_bytes_with_expiry(key, value, timestamp, true, ttl_expiry, start)
267    }
268
269    #[inline]
270    fn insert_bytes_with_expiry(
271        &self,
272        key: &[u8],
273        value: Bytes,
274        timestamp: u64,
275        explicit_timestamp: bool,
276        ttl_expiry: u64,
277        start: std::time::Instant,
278    ) -> Result<bool> {
279        let new_size = self.calculate_record_size(key.len(), value.len());
280        loop {
281            let existing_record = self.hash_table.read(key, |_, v| v.clone());
282            if let Some(existing_record) = existing_record {
283                if timestamp <= existing_record.timestamp {
284                    return Err(FeoxError::OlderTimestamp);
285                }
286
287                match self.update_record_with_ttl_bytes(
288                    &existing_record,
289                    value.clone(),
290                    timestamp,
291                    explicit_timestamp,
292                    ttl_expiry,
293                ) {
294                    Err(FeoxError::KeyNotFound) => continue,
295                    result => return result,
296                }
297            }
298
299            let reservation = self.reserve_memory(new_size)?;
300
301            let record = if ttl_expiry > 0 {
302                Arc::new(Record::new_from_bytes_with_ttl(
303                    key.to_vec(),
304                    value.clone(),
305                    timestamp,
306                    ttl_expiry,
307                ))
308            } else {
309                Arc::new(Record::new_from_bytes(
310                    key.to_vec(),
311                    value.clone(),
312                    timestamp,
313                ))
314            };
315
316            let key_vec = record.key.clone();
317
318            let buffered_record = match self.hash_table.entry(key_vec.clone()) {
319                scc::hash_map::Entry::Vacant(entry) => {
320                    let buffered_record = self
321                        .write_buffer
322                        .as_ref()
323                        .filter(|_| !self.memory_only)
324                        .map(|_| Arc::clone(&record));
325                    let _entry = entry.insert_entry(Arc::clone(&record));
326                    self.insert_into_tree(key_vec, record);
327                    self.observe_published_timestamp(key, timestamp, explicit_timestamp);
328                    reservation.commit();
329                    if ttl_expiry > 0 {
330                        self.stats.keys_with_ttl.fetch_add(1, Ordering::Relaxed);
331                    }
332                    self.stats.record_count.fetch_add(1, Ordering::Relaxed);
333                    buffered_record
334                }
335                scc::hash_map::Entry::Occupied(_) => continue,
336            };
337
338            self.stats
339                .record_insert(start.elapsed().as_nanos() as u64, false);
340
341            if let (Some(wb), Some(record)) = (&self.write_buffer, buffered_record) {
342                wb.add_write(Operation::Insert, record, 0)?;
343            }
344
345            return Ok(true);
346        }
347    }
348
349    /// Retrieve a value by key.
350    ///
351    /// # Arguments
352    ///
353    /// * `key` - The key to look up
354    /// * `expected_size` - Optional expected value size for validation
355    ///
356    /// # Returns
357    ///
358    /// Returns the value as a `Vec<u8>` if found.
359    ///
360    /// # Errors
361    ///
362    /// * `KeyNotFound` - Key does not exist
363    /// * `InvalidKey` - Key is invalid
364    /// * `SizeMismatch` - Value size doesn't match expected size
365    /// * `IoError` - Failed to read from disk (persistent mode)
366    ///
367    /// # Example
368    ///
369    /// ```rust
370    /// # use feoxdb::FeoxStore;
371    /// # fn main() -> feoxdb::Result<()> {
372    /// # let store = FeoxStore::new(None)?;
373    /// # store.insert(b"key", b"value")?;
374    /// let value = store.get(b"key")?;
375    /// assert_eq!(value, b"value");
376    /// # Ok(())
377    /// # }
378    /// ```
379    ///
380    /// # Performance
381    ///
382    /// * Memory mode: ~100ns
383    /// * Persistent mode (cached): ~150ns
384    /// * Persistent mode (disk read): ~500ns
385    pub fn get(&self, key: &[u8]) -> Result<Vec<u8>> {
386        let start = std::time::Instant::now();
387        self.validate_key(key)?;
388
389        let record = self
390            .hash_table
391            .read(key, |_, v| v.clone())
392            .ok_or(FeoxError::KeyNotFound)?;
393
394        let (value, cache_hit, source) = self.resolve_value(key, record)?;
395
396        if !cache_hit {
397            if let Some(ref cache) = self.cache {
398                cache.insert_for_record(key.to_vec(), value.clone(), &source);
399            }
400        }
401
402        self.stats
403            .record_get(start.elapsed().as_nanos() as u64, cache_hit);
404        Ok(value.to_vec())
405    }
406
407    /// Get a value by key without copying (zero-copy).
408    ///
409    /// Returns `Bytes` which avoids the memory copy that `get()` performs
410    /// when converting to `Vec<u8>`.
411    ///
412    /// # Arguments
413    ///
414    /// * `key` - The key to look up
415    ///
416    /// # Returns
417    ///
418    /// Returns the value as `Bytes` if found.
419    ///
420    /// # Example
421    ///
422    /// ```rust
423    /// # use feoxdb::FeoxStore;
424    /// # fn main() -> feoxdb::Result<()> {
425    /// # let store = FeoxStore::new(None)?;
426    /// # store.insert(b"key", b"value")?;
427    /// let bytes = store.get_bytes(b"key")?;
428    /// // Use bytes directly without copying
429    /// assert_eq!(&bytes[..], b"value");
430    /// # Ok(())
431    /// # }
432    /// ```
433    ///
434    /// # Performance
435    ///
436    /// Significantly faster than `get()` for large values:
437    /// * 100 bytes: ~15% faster
438    /// * 1KB: ~50% faster  
439    /// * 10KB: ~90% faster
440    /// * 100KB: ~95% faster
441    pub fn get_bytes(&self, key: &[u8]) -> Result<Bytes> {
442        let start = std::time::Instant::now();
443        self.validate_key(key)?;
444
445        let record = self
446            .hash_table
447            .read(key, |_, v| v.clone())
448            .ok_or(FeoxError::KeyNotFound)?;
449
450        let (value, cache_hit, source) = self.resolve_value(key, record)?;
451
452        if !cache_hit {
453            if let Some(ref cache) = self.cache {
454                cache.insert_for_record(key.to_vec(), value.clone(), &source);
455            }
456        }
457
458        self.stats
459            .record_get(start.elapsed().as_nanos() as u64, cache_hit);
460        Ok(value)
461    }
462
463    /// Delete a key-value pair.
464    ///
465    /// # Arguments
466    ///
467    /// * `key` - The key to delete
468    /// * `timestamp` - Optional timestamp for conflict resolution
469    ///
470    /// # Returns
471    ///
472    /// Returns `Ok(())` if the key was deleted.
473    ///
474    /// # Errors
475    ///
476    /// * `KeyNotFound` - Key does not exist
477    /// * `OlderTimestamp` - Timestamp is not newer than existing record
478    ///
479    /// # Example
480    ///
481    /// ```rust
482    /// # use feoxdb::FeoxStore;
483    /// # fn main() -> feoxdb::Result<()> {
484    /// # let store = FeoxStore::new(None)?;
485    /// # store.insert(b"temp", b"data")?;
486    /// store.delete(b"temp")?;
487    /// # Ok(())
488    /// # }
489    /// ```
490    ///
491    /// # Performance
492    ///
493    /// * Memory mode: ~300ns
494    /// * Persistent mode: ~400ns
495    pub fn delete(&self, key: &[u8]) -> Result<()> {
496        self.delete_with_timestamp(key, None)
497    }
498
499    /// Delete a key-value pair with explicit timestamp.
500    ///
501    /// This is the advanced version that allows manual timestamp control.
502    /// Most users should use `delete()` instead.
503    ///
504    /// # Arguments
505    ///
506    /// * `key` - The key to delete
507    /// * `timestamp` - Optional timestamp. If `None`, uses current time.
508    ///
509    /// # Errors
510    ///
511    /// * `OlderTimestamp` - Timestamp is not newer than existing record
512    pub fn delete_with_timestamp(&self, key: &[u8], timestamp: Option<u64>) -> Result<()> {
513        let start = std::time::Instant::now();
514        self.validate_key(key)?;
515        let (timestamp, explicit_timestamp) = self.resolve_timestamp(key, timestamp);
516
517        let (record, old_value_len) = match self.hash_table.entry(key.to_vec()) {
518            scc::hash_map::Entry::Occupied(entry) => {
519                let record = Arc::clone(entry.get());
520                if timestamp <= record.timestamp {
521                    return Err(FeoxError::OlderTimestamp);
522                }
523                let record_size = record.calculate_size();
524                let old_value_len = record.value_len;
525                record.retired_at.store(timestamp, Ordering::Release);
526                record.refcount.store(0, Ordering::Release);
527                // Ordered index first: a key vanishing early from a range scan is
528                // benign, whereas a deleted key lingering there is a phantom.
529                self.tree.remove(key);
530                self.stats.record_count.fetch_sub(1, Ordering::Relaxed);
531                self.stats
532                    .memory_usage
533                    .fetch_sub(record_size, Ordering::Relaxed);
534                self.note_ttl_transition(record.ttl_expiry.load(Ordering::Acquire), 0);
535                self.observe_published_timestamp(key, timestamp, explicit_timestamp);
536                let _ = entry.remove();
537                (record, old_value_len)
538            }
539            scc::hash_map::Entry::Vacant(_) => return Err(FeoxError::KeyNotFound),
540        };
541
542        self.remove_cached(key, &record);
543
544        // Queue deletion for persistence if write buffer exists and not memory-only
545        if !self.memory_only {
546            if let Some(ref wb) = self.write_buffer {
547                wb.add_write(Operation::Delete, record, old_value_len)?;
548            }
549        }
550
551        self.stats.record_delete(start.elapsed().as_nanos() as u64);
552        Ok(())
553    }
554
555    /// Get the size of a value without loading it.
556    ///
557    /// Useful for checking value size before loading large values from disk.
558    ///
559    /// # Arguments
560    ///
561    /// * `key` - The key to check
562    ///
563    /// # Returns
564    ///
565    /// Returns the size in bytes of the value.
566    ///
567    /// # Errors
568    ///
569    /// * `KeyNotFound` - Key does not exist
570    ///
571    /// # Example
572    ///
573    /// ```rust
574    /// # use feoxdb::FeoxStore;
575    /// # fn main() -> feoxdb::Result<()> {
576    /// # let store = FeoxStore::new(None)?;
577    /// store.insert(b"large_file", &vec![0u8; 1_000_000])?;
578    ///
579    /// // Check size before loading
580    /// let size = store.get_size(b"large_file")?;
581    /// assert_eq!(size, 1_000_000);
582    /// # Ok(())
583    /// # }
584    /// ```
585    pub fn get_size(&self, key: &[u8]) -> Result<usize> {
586        self.validate_key(key)?;
587
588        let record = self
589            .hash_table
590            .read(key, |_, v| v.clone())
591            .ok_or(FeoxError::KeyNotFound)?;
592
593        Ok(record.value_len)
594    }
595
596    // Internal helper methods
597
598    pub(super) fn validate_key_value(&self, key: &[u8], value: &[u8]) -> Result<()> {
599        self.validate_new_key(key)?;
600
601        if value.is_empty() || value.len() > MAX_VALUE_SIZE {
602            return Err(FeoxError::InvalidValueSize);
603        }
604
605        Ok(())
606    }
607
608    /// Bound for a key that is about to create a record. A persistent store must
609    /// refuse keys it could never rebuild an index from: the record would be written,
610    /// silently dropped on restart, and its extent handed out again while the old
611    /// bytes are still on disk to be misparsed as headers.
612    pub(super) fn validate_new_key(&self, key: &[u8]) -> Result<()> {
613        if key.is_empty() || key.len() > MAX_KEY_SIZE {
614            return Err(FeoxError::InvalidKeySize);
615        }
616        if self.memory_only || key.len() <= MAX_RECOVERABLE_KEY_SIZE {
617            return Ok(());
618        }
619        if self.format_version == 1 && key.len() <= MAX_RECOVERABLE_KEY_SIZE_V1 {
620            return Ok(());
621        }
622
623        Err(FeoxError::InvalidKeySize)
624    }
625
626    pub(super) fn validate_key(&self, key: &[u8]) -> Result<()> {
627        if key.is_empty() || key.len() > MAX_KEY_SIZE {
628            return Err(FeoxError::InvalidKeySize);
629        }
630
631        Ok(())
632    }
633
634    #[inline]
635    pub(super) fn reserve_memory(&self, amount: usize) -> Result<MemoryReservation<'_>> {
636        let usage = &self.stats.memory_usage;
637        if amount == 0 {
638            return Ok(MemoryReservation { usage, amount });
639        }
640        let Some(limit) = self.max_memory else {
641            usage.fetch_add(amount, Ordering::Relaxed);
642            return Ok(MemoryReservation { usage, amount });
643        };
644        let mut current = usage.load(Ordering::Relaxed);
645        loop {
646            let next = current.checked_add(amount).ok_or(FeoxError::OutOfMemory)?;
647            if next > limit {
648                return Err(FeoxError::OutOfMemory);
649            }
650            match usage.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) {
651                Ok(_) => return Ok(MemoryReservation { usage, amount }),
652                Err(observed) => current = observed,
653            }
654        }
655    }
656
657    #[inline]
658    pub(super) fn release_memory(&self, amount: usize) {
659        self.stats.memory_usage.fetch_sub(amount, Ordering::Relaxed);
660    }
661
662    pub(super) fn calculate_record_size(&self, key_len: usize, value_len: usize) -> usize {
663        std::mem::size_of::<Record>() + key_len + value_len
664    }
665
666    /// Resolve a key's value from memory, cache, or disk.
667    ///
668    /// A disk read can come back rejected when the record was retired and its
669    /// extent reused while this reader held it. That is a stale handle rather than
670    /// a failure, so the current generation is fetched from the hash table and the
671    /// read retried. Returns the record the value actually came from, so callers
672    /// populate the cache against the right generation.
673    pub(super) fn resolve_value(
674        &self,
675        key: &[u8],
676        record: Arc<Record>,
677    ) -> Result<(Bytes, bool, Arc<Record>)> {
678        let mut record = record;
679        for _ in 0..STALE_READ_RETRY_LIMIT {
680            match self.resolve_record_value(key, &record)? {
681                Some((value, cache_hit)) => return Ok((value, cache_hit, record)),
682                None => {
683                    record = self
684                        .hash_table
685                        .read(key, |_, v| v.clone())
686                        .ok_or(FeoxError::KeyNotFound)?;
687                }
688            }
689        }
690        Err(FeoxError::StaleExtent)
691    }
692
693    pub(super) fn resolve_value_ref(&self, key: &[u8], record: &Arc<Record>) -> Result<Bytes> {
694        if let Some((value, _)) = self.resolve_record_value(key, record)? {
695            return Ok(value);
696        }
697
698        let record = self
699            .hash_table
700            .read(key, |_, record| Arc::clone(record))
701            .ok_or(FeoxError::KeyNotFound)?;
702        self.resolve_value(key, record).map(|(value, _, _)| value)
703    }
704
705    fn resolve_record_value(
706        &self,
707        key: &[u8],
708        record: &Arc<Record>,
709    ) -> Result<Option<(Bytes, bool)>> {
710        if self.enable_ttl {
711            let ttl_expiry = record.ttl_expiry.load(Ordering::Acquire);
712            if ttl_expiry > 0 {
713                let now = SystemTime::now()
714                    .duration_since(UNIX_EPOCH)
715                    .unwrap_or_default()
716                    .as_nanos() as u64;
717                if now > ttl_expiry {
718                    self.stats.ttl_expired_lazy.fetch_add(1, Ordering::Relaxed);
719                    return Err(FeoxError::KeyNotFound);
720                }
721            }
722        }
723        if let Some(value) = record.get_value() {
724            return Ok(Some((value, true)));
725        }
726        if let Some(value) = self
727            .cache
728            .as_ref()
729            .and_then(|cache| cache.get_for_record(key, record))
730        {
731            return Ok(Some((value, true)));
732        }
733        match self.load_value_from_disk(record) {
734            Ok(value) => Ok(Some((value, false))),
735            Err(FeoxError::StaleExtent) => Ok(None),
736            Err(error) => Err(error),
737        }
738    }
739
740    /// Timestamps double as record version numbers and must increase for a key.
741    #[inline]
742    pub(super) fn get_timestamp(&self, key: &[u8]) -> u64 {
743        self.version_clock.next(key, self.get_timestamp_pub())
744    }
745
746    #[inline]
747    pub(super) fn resolve_timestamp(&self, key: &[u8], timestamp: Option<u64>) -> (u64, bool) {
748        match timestamp {
749            Some(timestamp) if timestamp != 0 => (timestamp, true),
750            _ => (self.get_timestamp(key), false),
751        }
752    }
753
754    #[inline]
755    pub(super) fn observe_published_timestamp(&self, key: &[u8], timestamp: u64, explicit: bool) {
756        if explicit {
757            self.version_clock.observe(key, timestamp);
758        }
759    }
760
761    #[cfg(test)]
762    pub(crate) fn timestamp_shard_for_test(&self, key: &[u8]) -> usize {
763        self.version_clock.shard_index(key)
764    }
765}
feoxdb/core/store/operations.rs

feoxdb/core/store/
operations.rs
