/home/runner/work/feoxdb/feoxdb/src/core/ttl_sweep.rs

Source
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Weak};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use rand::Rng;

use crate::constants::Operation;
use crate::core::store::FeoxStore;

/// Configuration for TTL cleaner background thread
#[derive(Clone, Debug)]
pub struct TtlConfig {
    /// Number of keys to sample per batch
    pub sample_size: usize,
    /// Continue sampling if expiry rate exceeds this threshold (0.0-1.0)
    pub expiry_threshold: f32,
    /// Maximum iterations per cleaning run
    pub max_iterations: usize,
    /// Maximum time to spend per cleaning run
    pub max_time_per_run: Duration,
    /// Sleep interval between cleaning runs
    pub sleep_interval: Duration,
    /// Whether TTL cleaner is enabled
    pub enabled: bool,
}

impl Default for TtlConfig {
    fn default() -> Self {
        Self {
            sample_size: 100,
            expiry_threshold: 0.25,
            max_iterations: 16,
            max_time_per_run: Duration::from_millis(1),
            sleep_interval: Duration::from_millis(1000),
            enabled: false,
        }
    }
}

impl TtlConfig {
    /// Create a default configuration for persistent stores
    pub fn default_persistent() -> Self {
        Self {
            enabled: true,
            ..Default::default()
        }
    }

    /// Create a default configuration for memory-only stores
    pub fn default_memory() -> Self {
        Self {
            enabled: true,
            ..Default::default()
        }
    }
}

/// Background thread that periodically sweeps expired TTL keys
pub struct TtlSweeper {
    /// Weak reference to the store to avoid circular references
    store: Weak<FeoxStore>,
    /// Configuration
    config: TtlConfig,
    /// Shutdown flag
    shutdown: Arc<AtomicBool>,
    /// Thread handle
    handle: Option<JoinHandle<()>>,
    /// Statistics
    stats: TtlSweeperStats,
}

/// Statistics for TTL sweeper operations
pub struct TtlSweeperStats {
    /// Total keys sampled
    pub total_sampled: Arc<AtomicU64>,
    /// Total keys expired
    pub total_expired: Arc<AtomicU64>,
    /// Total cleaning runs
    pub total_runs: Arc<AtomicU64>,
    /// Last run timestamp (nanoseconds)
    pub last_run: Arc<AtomicU64>,
}

impl TtlSweeperStats {
    fn new() -> Self {
        Self {
            total_sampled: Arc::new(AtomicU64::new(0)),
            total_expired: Arc::new(AtomicU64::new(0)),
            total_runs: Arc::new(AtomicU64::new(0)),
            last_run: Arc::new(AtomicU64::new(0)),
        }
    }
}

impl TtlSweeper {
    /// Create a new TTL sweeper
    pub fn new(store: Weak<FeoxStore>, config: TtlConfig) -> Self {
        Self {
            store,
            config,
            shutdown: Arc::new(AtomicBool::new(false)),
            handle: None,
            stats: TtlSweeperStats::new(),
        }
    }

    /// Start the background sweeper thread
    pub fn start(&mut self) {
        if !self.config.enabled {
            return;
        }

        let store = self.store.clone();
        let config = self.config.clone();
        let shutdown = self.shutdown.clone();
        let stats = TtlSweeperStats {
            total_sampled: self.stats.total_sampled.clone(),
            total_expired: self.stats.total_expired.clone(),
            total_runs: self.stats.total_runs.clone(),
            last_run: self.stats.last_run.clone(),
        };

        let handle = thread::spawn(move || {
            run_sweeper_loop(store, config, shutdown, stats);
        });

        self.handle = Some(handle);
    }

    /// Stop the background sweeper thread
    pub fn stop(&mut self) {
        self.shutdown.store(true, Ordering::Release);

        if let Some(handle1) = self.handle.take() {
            if handle.thread().id() != thread::current().id() {
                let _ = handle.join();
            }
        }
    }

    /// Get sweeper statistics
    pub fn stats(&self) -> SweeperSnapshot {
        SweeperSnapshot {
            total_sampled: self.stats.total_sampled.load(Ordering::Relaxed),
            total_expired: self.stats.total_expired.load(Ordering::Relaxed),
            total_runs: self.stats.total_runs.load(Ordering::Relaxed),
            last_run: self.stats.last_run.load(Ordering::Relaxed),
        }
    }
}

impl Drop for TtlSweeper {
    fn drop(&mut self) {
        self.stop();
    }
}

/// Snapshot of sweeper statistics
#[derive(Debug, Clone)]
pub struct SweeperSnapshot {
    pub total_sampled: u64,
    pub total_expired: u64,
    pub total_runs: u64,
    pub last_run: u64,
}

/// Main sweeper loop that runs in the background thread
fn run_sweeper_loop(
    store: Weak<FeoxStore>,
    config: TtlConfig,
    shutdown: Arc<AtomicBool>,
    stats: TtlSweeperStats,
) {
    while !shutdown.load(Ordering::Acquire) {
        // Sleep between runs
        thread::sleep(config.sleep_interval);

        // Try to get strong reference to store
        let Some(store) = store.upgrade() else {
            // Store has been dropped, exit
            break;
        };

        // Perform sweeping run
        let start = Instant::now();
        let mut iterations = 0;
        let mut total_sampled = 0;
        let mut total_expired = 0;

        loop {
            // Sample and expire a batch
            let (sampled, expired) = sample_and_expire_batch(&store, &config);
            total_sampled += sampled;
            total_expired += expired;
            iterations += 1;

            // Calculate expiry rate
            let expiry_rate = if sampled > 0 {
                expired as f32 / sampled as f32
            } else {
                0.0
            };

            // Check stop conditions
            if expiry_rate < config.expiry_threshold {
                break; // Few expired keys, we're done
            }
            if iterations >= config.max_iterations {
                break; // Bounded iterations
            }
            if start.elapsed() > config.max_time_per_run {
                break; // Bounded time
            }
        }

        // Update statistics
        if total_sampled > 0 {
            stats
                .total_sampled
                .fetch_add(total_sampled, Ordering::Relaxed);
            stats
                .total_expired
                .fetch_add(total_expired, Ordering::Relaxed);
            stats.total_runs.fetch_add(1, Ordering::Relaxed);
            stats.last_run.store(
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_nanos() as u64,
                Ordering::Relaxed,
            );
        }

        // Check shutdown flag again
        if shutdown.load(Ordering::Acquire) {
            break;
        }
    }
}

/// Sample keys and expire those that have exceeded their TTL
fn sample_and_expire_batch(store: &Arc<FeoxStore>, config: &TtlConfig) -> (u64, u64) {
    let now = store.get_timestamp_pub();
    let mut expired = 0;
    let mut rng = rand::rng();

    // Get access to the hash table
    let hash_table = store.get_hash_table();

    let candidates = sample_ttl_entries(hash_table, config.sample_size, &mut rng);
    let sampled = candidates.len() as u64;

    for (key, record) in candidates {
        let ttl_expiry = record.ttl_expiry.load(Ordering::Relaxed);

        if ttl_expiry > 0 && ttl_expiry < now {
            #[cfg(test)]
            crate::test_hooks::pause_at(crate::test_hooks::TTL_AFTER_EXPIRED_SAMPLE);

            let old_value_len = record.value_len;
            let record_size = record.calculate_size();
            let retired = match hash_table.entry(key.clone()) {
                scc::hash_map::Entry::Occupied(entry) => {
                    let current_expiry = record.ttl_expiry.load(Ordering::Acquire);
                    if Arc::ptr_eq(entry.get(), &record)
                        && current_expiry > 0
                        && current_expiry < now
                    {
                        record.retired_at.store(now, Ordering::Release);
                        record.refcount.store(0, Ordering::Release);
                        store.remove_from_tree(&key);
                        let _ = entry.remove();
                        true
                    } else {
                        false
                    }
                }
                scc::hash_map::Entry::Vacant(_) => false,
            };

            if retired {
                store.remove_cached(&key, &record);
                store.note_expired_record(record_size);
                expired += 1;

                if let Some(wb0) = store.get_write_buffer() {
                    let _ = wb.add_write(Operation::Delete, record, old_value_len);
                }
            }
        }
    }

    (sampled, expired)
}

#[cfg(test)]
pub(crate) fn sample_and_expire_for_test(store: &Arc<FeoxStore>) -> (u64, u64) {
    sample_and_expire_batch(
        store,
        &TtlConfig {
            sample_size: 1,
            ..TtlConfig::default()
        },
    )
}

fn sample_ttl_entries<R: Rng + ?Sized>(
    hash_table: &scc::HashMap<Vec<u8>, Arc<crate::core::record::Record>, ahash::RandomState>,
    sample_size: usize,
    rng: &mut R,
) -> Vec<(Vec<u8>, Arc<crate::core::record::Record>)> {
    if sample_size == 0 {
        return Vec::new();
    }

    let mut candidates = Vec::with_capacity(sample_size.min(hash_table.len()));
    let mut seen = 0usize;
    hash_table5.scan5(|key: &Vec<u8>, value: &Arc<crate::core::record::Record>| {
        if value.ttl_expiry.load(Ordering::Relaxed) > 0 {
            seen += 1;
            if candidates.len() < sample_size {
                candidates.push((key.clone(), Arc::clone(value)));
            } else {
                let index = rng.random_range(0..seen);
                if index < sample_size {
                    candidates[index] = (key.clone(), Arc::clone(value));
                }
            }
        }
    });

    candidates
}

#[cfg(test)]
pub(crate) fn sample_ttl_keys_for_test(store: &FeoxStore, sample_size: usize) -> Vec<Vec<u8>> {
    let mut rng = rand::rng();
    sample_ttl_entries(store.get_hash_table(), sample_size, &mut rng)
        .into_iter()
        .map(|(key, _)| key)
        .collect()
}

#[cfg(test)]
#[path = "../tests/ttl_sweep_safety_tests.rs"]
mod tests;

Coverage Report

Created: 2026-07-31 00:13

Line	Count	Source
1		use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
2		use std::sync::{Arc, Weak};
3		use std::thread::{self, JoinHandle};
4		use std::time::{Duration, Instant};
5
6		use rand::Rng;
7
8		use crate::constants::Operation;
9		use crate::core::store::FeoxStore;
10
11		/// Configuration for TTL cleaner background thread
12		#[derive(Clone, Debug)]
13		pub struct TtlConfig {
14		/// Number of keys to sample per batch
15		pub sample_size: usize,
16		/// Continue sampling if expiry rate exceeds this threshold (0.0-1.0)
17		pub expiry_threshold: f32,
18		/// Maximum iterations per cleaning run
19		pub max_iterations: usize,
20		/// Maximum time to spend per cleaning run
21		pub max_time_per_run: Duration,
22		/// Sleep interval between cleaning runs
23		pub sleep_interval: Duration,
24		/// Whether TTL cleaner is enabled
25		pub enabled: bool,
26		}
27
28		impl Default for TtlConfig {
29	39	fn default() -> Self {
30	39	Self {
31	39	sample_size: 100,
32	39	expiry_threshold: 0.25,
33	39	max_iterations: 16,
34	39	max_time_per_run: Duration::from_millis(1),
35	39	sleep_interval: Duration::from_millis(1000),
36	39	enabled: false,
37	39	}
38	39	}
39		}
40
41		impl TtlConfig {
42		/// Create a default configuration for persistent stores
43	0	pub fn default_persistent() -> Self {
44	0	Self {
45	0	enabled: true,
46	0	..Default::default()
47	0	}
48	0	}
49
50		/// Create a default configuration for memory-only stores
51	0	pub fn default_memory() -> Self {
52	0	Self {
53	0	enabled: true,
54	0	..Default::default()
55	0	}
56	0	}
57		}
58
59		/// Background thread that periodically sweeps expired TTL keys
60		pub struct TtlSweeper {
61		/// Weak reference to the store to avoid circular references
62		store: Weak<FeoxStore>,
63		/// Configuration
64		config: TtlConfig,
65		/// Shutdown flag
66		shutdown: Arc<AtomicBool>,
67		/// Thread handle
68		handle: Option<JoinHandle<()>>,
69		/// Statistics
70		stats: TtlSweeperStats,
71		}
72
73		/// Statistics for TTL sweeper operations
74		pub struct TtlSweeperStats {
75		/// Total keys sampled
76		pub total_sampled: Arc<AtomicU64>,
77		/// Total keys expired
78		pub total_expired: Arc<AtomicU64>,
79		/// Total cleaning runs
80		pub total_runs: Arc<AtomicU64>,
81		/// Last run timestamp (nanoseconds)
82		pub last_run: Arc<AtomicU64>,
83		}
84
85		impl TtlSweeperStats {
86	1	fn new() -> Self {
87	1	Self {
88	1	total_sampled: Arc::new(AtomicU64::new(0)),
89	1	total_expired: Arc::new(AtomicU64::new(0)),
90	1	total_runs: Arc::new(AtomicU64::new(0)),
91	1	last_run: Arc::new(AtomicU64::new(0)),
92	1	}
93	1	}
94		}
95
96		impl TtlSweeper {
97		/// Create a new TTL sweeper
98	0	pub fn new(store: Weak<FeoxStore>, config: TtlConfig) -> Self {
99	0	Self {
100	0	store,
101	0	config,
102	0	shutdown: Arc::new(AtomicBool::new(false)),
103	0	handle: None,
104	0	stats: TtlSweeperStats::new(),
105	0	}
106	0	}
107
108		/// Start the background sweeper thread
109	0	pub fn start(&mut self) {
110	0	if !self.config.enabled {
111	0	return;
112	0	}
113
114	0	let store = self.store.clone();
115	0	let config = self.config.clone();
116	0	let shutdown = self.shutdown.clone();
117	0	let stats = TtlSweeperStats {
118	0	total_sampled: self.stats.total_sampled.clone(),
119	0	total_expired: self.stats.total_expired.clone(),
120	0	total_runs: self.stats.total_runs.clone(),
121	0	last_run: self.stats.last_run.clone(),
122	0	};
123
124	0	let handle = thread::spawn(move \|\| {
125	0	run_sweeper_loop(store, config, shutdown, stats);
126	0	});
127
128	0	self.handle = Some(handle);
129	0	}
130
131		/// Stop the background sweeper thread
132	2	pub fn stop(&mut self) {
133	2	self.shutdown.store(true, Ordering::Release);
134
135	2	if let Some( handle1 ) = self.handle.take() {
136	1	if handle.thread().id() != thread::current().id() {
137	0	let _ = handle.join();
138	1	}
139	1	}
140	2	}
141
142		/// Get sweeper statistics
143	0	pub fn stats(&self) -> SweeperSnapshot {
144	0	SweeperSnapshot {
145	0	total_sampled: self.stats.total_sampled.load(Ordering::Relaxed),
146	0	total_expired: self.stats.total_expired.load(Ordering::Relaxed),
147	0	total_runs: self.stats.total_runs.load(Ordering::Relaxed),
148	0	last_run: self.stats.last_run.load(Ordering::Relaxed),
149	0	}
150	0	}
151		}
152
153		impl Drop for TtlSweeper {
154	1	fn drop(&mut self) {
155	1	self.stop();
156	1	}
157		}
158
159		/// Snapshot of sweeper statistics
160		#[derive(Debug, Clone)]
161		pub struct SweeperSnapshot {
162		pub total_sampled: u64,
163		pub total_expired: u64,
164		pub total_runs: u64,
165		pub last_run: u64,
166		}
167
168		/// Main sweeper loop that runs in the background thread
169	0	fn run_sweeper_loop(
170	0	store: Weak<FeoxStore>,
171	0	config: TtlConfig,
172	0	shutdown: Arc<AtomicBool>,
173	0	stats: TtlSweeperStats,
174	0	) {
175	0	while !shutdown.load(Ordering::Acquire) {
176		// Sleep between runs
177	0	thread::sleep(config.sleep_interval);
178
179		// Try to get strong reference to store
180	0	let Some(store) = store.upgrade() else {
181		// Store has been dropped, exit
182	0	break;
183		};
184
185		// Perform sweeping run
186	0	let start = Instant::now();
187	0	let mut iterations = 0;
188	0	let mut total_sampled = 0;
189	0	let mut total_expired = 0;
190
191		loop {
192		// Sample and expire a batch
193	0	let (sampled, expired) = sample_and_expire_batch(&store, &config);
194	0	total_sampled += sampled;
195	0	total_expired += expired;
196	0	iterations += 1;
197
198		// Calculate expiry rate
199	0	let expiry_rate = if sampled > 0 {
200	0	expired as f32 / sampled as f32
201		} else {
202	0	0.0
203		};
204
205		// Check stop conditions
206	0	if expiry_rate < config.expiry_threshold {
207	0	break; // Few expired keys, we're done
208	0	}
209	0	if iterations >= config.max_iterations {
210	0	break; // Bounded iterations
211	0	}
212	0	if start.elapsed() > config.max_time_per_run {
213	0	break; // Bounded time
214	0	}
215		}
216
217		// Update statistics
218	0	if total_sampled > 0 {
219	0	stats
220	0	.total_sampled
221	0	.fetch_add(total_sampled, Ordering::Relaxed);
222	0	stats
223	0	.total_expired
224	0	.fetch_add(total_expired, Ordering::Relaxed);
225	0	stats.total_runs.fetch_add(1, Ordering::Relaxed);
226	0	stats.last_run.store(
227	0	std::time::SystemTime::now()
228	0	.duration_since(std::time::UNIX_EPOCH)
229	0	.unwrap()
230	0	.as_nanos() as u64,
231	0	Ordering::Relaxed,
232	0	);
233	0	}
234
235		// Check shutdown flag again
236	0	if shutdown.load(Ordering::Acquire) {
237	0	break;
238	0	}
239		}
240	0	}
241
242		/// Sample keys and expire those that have exceeded their TTL
243	3	fn sample_and_expire_batch(store: &Arc<FeoxStore>, config: &TtlConfig) -> (u64, u64) {
244	3	let now = store.get_timestamp_pub();
245	3	let mut expired = 0;
246	3	let mut rng = rand::rng();
247
248		// Get access to the hash table
249	3	let hash_table = store.get_hash_table();
250
251	3	let candidates = sample_ttl_entries(hash_table, config.sample_size, &mut rng);
252	3	let sampled = candidates.len() as u64;
253
254	3	for (key, record) in candidates {
255	3	let ttl_expiry = record.ttl_expiry.load(Ordering::Relaxed);
256
257	3	if ttl_expiry > 0 && ttl_expiry < now {
258		#[cfg(test)]
259	3	crate::test_hooks::pause_at(crate::test_hooks::TTL_AFTER_EXPIRED_SAMPLE);
260
261	3	let old_value_len = record.value_len;
262	3	let record_size = record.calculate_size();
263	3	let retired = match hash_table.entry(key.clone()) {
264	3	scc::hash_map::Entry::Occupied(entry) => {
265	3	let current_expiry = record.ttl_expiry.load(Ordering::Acquire);
266	3	if Arc::ptr_eq(entry.get(), &record)
267	2	&& current_expiry > 0
268	2	&& current_expiry < now
269		{
270	2	record.retired_at.store(now, Ordering::Release);
271	2	record.refcount.store(0, Ordering::Release);
272	2	store.remove_from_tree(&key);
273	2	let _ = entry.remove();
274	2	true
275		} else {
276	1	false
277		}
278		}
279	0	scc::hash_map::Entry::Vacant(_) => false,
280		};
281
282	3	if retired {
283	2	store.remove_cached(&key, &record);
284	2	store.note_expired_record(record_size);
285	2	expired += 1;
286
287	2	if let Some( wb0 ) = store.get_write_buffer() {
288	0	let _ = wb.add_write(Operation::Delete, record, old_value_len);
289	2	}
290	1	}
291	0	}
292		}
293
294	3	(sampled, expired)
295	3	}
296
297		#[cfg(test)]
298	3	pub(crate) fn sample_and_expire_for_test(store: &Arc<FeoxStore>) -> (u64, u64) {
299	3	sample_and_expire_batch(
300	3	store,
301	3	&TtlConfig {
302	3	sample_size: 1,
303	3	..TtlConfig::default()
304	3	},
305		)
306	3	}
307
308	6	fn sample_ttl_entries<R: Rng + ?Sized>(
309	6	hash_table: &scc::HashMap<Vec<u8>, Arc<crate::core::record::Record>, ahash::RandomState>,
310	6	sample_size: usize,
311	6	rng: &mut R,
312	6	) -> Vec<(Vec<u8>, Arc<crate::core::record::Record>)> {
313	6	if sample_size == 0 {
314	1	return Vec::new();
315	5	}
316
317	5	let mut candidates = Vec::with_capacity(sample_size.min(hash_table.len()));
318	5	let mut seen = 0usize;
319	131	hash_table5 . scan5 (\|key: &Vec<u8>, value: &Arc<crate::core::record::Record>\| {
320	131	if value.ttl_expiry.load(Ordering::Relaxed) > 0 {
321	67	seen += 1;
322	67	if candidates.len() < sample_size {
323	42	candidates.push((key.clone(), Arc::clone(value)));
324	42	} else {
325	25	let index = rng.random_range(0..seen);
326	25	if index < sample_size {
327	8	candidates[index] = (key.clone(), Arc::clone(value));
328	17	}
329		}
330	64	}
331	131	});
332
333	5	candidates
334	6	}
335
336		#[cfg(test)]
337	3	pub(crate) fn sample_ttl_keys_for_test(store: &FeoxStore, sample_size: usize) -> Vec<Vec<u8>> {
338	3	let mut rng = rand::rng();
339	3	sample_ttl_entries(store.get_hash_table(), sample_size, &mut rng)
340	3	.into_iter()
341	3	.map(\|(key, _)\| key)
342	3	.collect()
343	3	}
344
345		#[cfg(test)]
346		#[path = "../tests/ttl_sweep_safety_tests.rs"]
347		mod tests;