reifydb_core/interface/
flow.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::sync::Arc;
5
6use crate::interface::catalog::{flow::FlowId, object::ObjectId};
7
8#[repr(u32)]
9#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
10pub enum OperatorCapability {
11	Insert = 1 << 0,
12	Update = 1 << 1,
13	Delete = 1 << 2,
14	Expire = 1 << 5,
15}
16
17impl OperatorCapability {
18	pub const STANDARD: &'static [OperatorCapability] =
19		&[OperatorCapability::Insert, OperatorCapability::Update, OperatorCapability::Delete];
20
21	pub const ALL: &'static [OperatorCapability] = &[
22		OperatorCapability::Insert,
23		OperatorCapability::Update,
24		OperatorCapability::Delete,
25		OperatorCapability::Expire,
26	];
27
28	pub const fn bit(self) -> u32 {
29		self as u32
30	}
31}
32
33pub fn to_bitmask(caps: &[OperatorCapability]) -> u32 {
34	let mut mask = 0;
35	for cap in caps {
36		mask |= cap.bit();
37	}
38	mask
39}
40
41pub fn from_bitmask(mask: u32) -> Vec<OperatorCapability> {
42	OperatorCapability::ALL.iter().copied().filter(|cap| mask & cap.bit() != 0).collect()
43}
44
45#[derive(Debug, Clone)]
46pub struct FlowWatermarkRow {
47	pub flow_id: FlowId,
48
49	pub object_id: ObjectId,
50
51	pub lag: u64,
52
53	pub outstanding: u64,
54}
55
56#[derive(Clone)]
57pub struct FlowWatermarkSampler {
58	fetch: Arc<dyn Fn() -> Vec<FlowWatermarkRow> + Send + Sync>,
59}
60
61impl FlowWatermarkSampler {
62	pub fn new<F>(fetch: F) -> Self
63	where
64		F: Fn() -> Vec<FlowWatermarkRow> + Send + Sync + 'static,
65	{
66		Self {
67			fetch: Arc::new(fetch),
68		}
69	}
70
71	pub fn all(&self) -> Vec<FlowWatermarkRow> {
72		(self.fetch)()
73	}
74}
75
76#[derive(Debug, Clone)]
77pub struct FlowUnitRow {
78	pub flow_id: FlowId,
79
80	pub unit: FlowId,
81}
82
83#[derive(Clone)]
84pub struct FlowUnitSampler {
85	fetch: Arc<dyn Fn() -> Vec<FlowUnitRow> + Send + Sync>,
86}
87
88impl FlowUnitSampler {
89	pub fn new<F>(fetch: F) -> Self
90	where
91		F: Fn() -> Vec<FlowUnitRow> + Send + Sync + 'static,
92	{
93		Self {
94			fetch: Arc::new(fetch),
95		}
96	}
97
98	pub fn all(&self) -> Vec<FlowUnitRow> {
99		(self.fetch)()
100	}
101}
102
103#[cfg(test)]
104mod tests {
105	use super::{OperatorCapability, from_bitmask, to_bitmask};
106
107	#[test]
108	fn every_capability_bit_is_distinct() {
109		// A shared bit would make two capabilities indistinguishable in the descriptor
110		// bitmask, silently gating the wrong method on the plugin side.
111		for (i, a) in OperatorCapability::ALL.iter().enumerate() {
112			for b in &OperatorCapability::ALL[i + 1..] {
113				assert_ne!(a.bit(), b.bit(), "{a:?} collides with {b:?}");
114			}
115		}
116	}
117
118	#[test]
119	fn every_declared_capability_is_reachable_through_all() {
120		// from_bitmask filters over ALL, so a variant missing from ALL is dropped on every
121		// descriptor round trip and the operator loses that capability with no error anywhere. The
122		// match is exhaustive so a new variant fails to compile here rather than vanishing at runtime.
123		for capability in [
124			OperatorCapability::Insert,
125			OperatorCapability::Update,
126			OperatorCapability::Delete,
127			OperatorCapability::Expire,
128		] {
129			match capability {
130				OperatorCapability::Insert
131				| OperatorCapability::Update
132				| OperatorCapability::Delete
133				| OperatorCapability::Expire => {}
134			}
135			assert!(
136				OperatorCapability::ALL.contains(&capability),
137				"{capability:?} is missing from ALL, so from_bitmask silently drops it"
138			);
139			assert!(
140				from_bitmask(to_bitmask(&[capability])).contains(&capability),
141				"{capability:?} does not survive a bitmask round trip"
142			);
143		}
144	}
145
146	#[test]
147	fn presets_survive_a_bitmask_round_trip() {
148		let restored = from_bitmask(to_bitmask(OperatorCapability::STANDARD));
149		assert!(restored.contains(&OperatorCapability::Insert));
150		assert!(restored.contains(&OperatorCapability::Update));
151		assert!(restored.contains(&OperatorCapability::Delete));
152	}
153
154	#[test]
155	fn expire_is_reachable_through_all_but_never_through_standard() {
156		// STANDARD must never carry Expire, or every in-tree operator silently opts into rows it has no arm
157		// for.
158		assert!(!OperatorCapability::STANDARD.contains(&OperatorCapability::Expire));
159		assert!(OperatorCapability::ALL.contains(&OperatorCapability::Expire));
160		assert!(from_bitmask(to_bitmask(&[OperatorCapability::Expire])).contains(&OperatorCapability::Expire));
161	}
162
163	#[test]
164	fn expire_does_not_claim_the_retired_reclaim_bit() {
165		// Expire must never take the retired Reclaim bit, or a stale guest still setting it reads as opted in.
166		assert_ne!(OperatorCapability::Expire.bit(), 1 << 4);
167		assert!(!from_bitmask(1 << 4).contains(&OperatorCapability::Expire));
168	}
169
170	#[test]
171	fn an_unknown_descriptor_bit_is_dropped_rather_than_misread() {
172		// Guests built against an older ABI may still set retired bits (the removed Reclaim
173		// capability was 1 << 4); from_bitmask must ignore them instead of aliasing them onto a
174		// live capability.
175		let stale = to_bitmask(OperatorCapability::STANDARD) | (1 << 4);
176		let restored = from_bitmask(stale);
177		assert_eq!(restored, OperatorCapability::STANDARD.to_vec());
178	}
179}
reifydb_core/interface/flow.rs

reifydb_core/interface/
flow.rs
