reifydb_core/state/
timer.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::cmp::Reverse;
5
6use reifydb_codec::{key::encoded::EncodedKey, row::pod::EncodedPodRow};
7use reifydb_value::{
8	Result,
9	byte_size::ByteSize,
10	value::{datetime::DateTime, row_number::RowNumber},
11};
12
13use crate::{
14	key::operator::{
15		keyspace::{RootSibling, root_sibling_of},
16		state::{GroupId, GroupStateKey, GroupStateRange, KeyspaceMask},
17	},
18	state::batch::StateBatch,
19};
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
22#[repr(u8)]
23pub enum TimerKind {
24	Seal = 0,
25	Grace = 1,
26	RowTtl = 2,
27	Maintenance = 3,
28	Reclaim = 4,
29}
30
31impl TimerKind {
32	pub fn is_maintenance(&self) -> bool {
33		matches!(self, Self::Maintenance)
34	}
35
36	pub fn from_u8(value: u8) -> Option<Self> {
37		match value {
38			0 => Some(Self::Seal),
39			1 => Some(Self::Grace),
40			2 => Some(Self::RowTtl),
41			3 => Some(Self::Maintenance),
42			4 => Some(Self::Reclaim),
43			_ => None,
44		}
45	}
46}
47
48pub struct GroupSweep {
49	pub rows: Vec<(GroupStateKey, EncodedPodRow)>,
50	pub complete: bool,
51}
52
53pub fn sweep_order(groups: &[GroupId]) -> Vec<GroupId> {
54	let mut ordered = groups.to_vec();
55	ordered.sort_by_key(|group| Reverse(*group.as_bytes()));
56	ordered.dedup();
57	ordered
58}
59
60pub trait StateStore {
61	fn state_get(&mut self, keys: &[GroupStateKey]) -> Result<Vec<Option<EncodedPodRow>>>;
62
63	fn state_classify(&mut self, _key: &GroupStateKey, _pre: Option<ByteSize>) {}
64
65	fn state_set(&mut self, rows: Vec<(GroupStateKey, EncodedPodRow)>) -> Result<()>;
66
67	fn state_remove(&mut self, keys: &[GroupStateKey]) -> Result<()>;
68
69	fn state_batch(&mut self, keys: Vec<GroupStateKey>) -> Result<StateBatch>;
70
71	fn state_write_batch(&mut self, batch: StateBatch) -> Result<()>;
72
73	fn state_page(
74		&mut self,
75		range: GroupStateRange,
76		limit: Option<usize>,
77	) -> Result<Vec<(GroupStateKey, EncodedPodRow)>> {
78		debug_assert!(
79			matches!(range, GroupStateRange::Keyspace { .. }),
80			"a state page must stay inside one group and one keyspace; {range:?} spans more than one"
81		);
82		self.state_page_inner(range, limit)
83	}
84
85	fn state_page_inner(
86		&mut self,
87		range: GroupStateRange,
88		limit: Option<usize>,
89	) -> Result<Vec<(GroupStateKey, EncodedPodRow)>>;
90
91	fn group_sweep(&mut self, groups: &[GroupId], limit: usize, keyspaces: KeyspaceMask) -> Result<GroupSweep>;
92
93	fn remove_root_siblings(&mut self, swept: &[(GroupStateKey, EncodedPodRow)]) -> Result<()> {
94		let mut siblings = Vec::new();
95		for (key, row) in swept {
96			if let Some(RootSibling::Derived(sibling)) = root_sibling_of(key, row) {
97				siblings.push(sibling);
98			}
99		}
100		self.state_remove(&siblings)
101	}
102
103	fn state_last(&mut self, range: GroupStateRange) -> Result<Option<(GroupStateKey, EncodedPodRow)>> {
104		Ok(self.state_page(range, None)?.pop())
105	}
106
107	fn get_or_create_row_numbers(&mut self, group: GroupId, keys: &[EncodedKey]) -> Result<Vec<(RowNumber, bool)>>;
108
109	fn get_or_create_row_numbers_for_groups(&mut self, groups: &[GroupId]) -> Result<Vec<(RowNumber, bool)>>;
110
111	fn get_row_numbers(&mut self, group: GroupId, keys: &[EncodedKey]) -> Result<Vec<Option<RowNumber>>>;
112
113	fn remove_row_number(&mut self, group: GroupId, keys: &[EncodedKey]) -> Result<()>;
114
115	fn remove_row_number_for_group(&mut self, groups: &[GroupId]) -> Result<()>;
116
117	fn written_at(&self) -> DateTime;
118}
119
120pub trait TimerStore {
121	fn arm_timer(&mut self, due: DateTime, kind: TimerKind, key: &EncodedKey) -> Result<()>;
122
123	fn disarm_timer(&mut self, due: DateTime, kind: TimerKind, key: &EncodedKey) -> Result<()>;
124
125	fn flow_watermark(&mut self) -> Result<Option<DateTime>>;
126}
reifydb_core/state/timer.rs

reifydb_core/state/
timer.rs
