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#[cfg(test)]
77mod tests {
78	use super::{OperatorCapability, from_bitmask, to_bitmask};
79
80	#[test]
81	fn every_capability_bit_is_distinct() {
82		// A shared bit would make two capabilities indistinguishable in the descriptor
83		// bitmask, silently gating the wrong method on the plugin side.
84		for (i, a) in OperatorCapability::ALL.iter().enumerate() {
85			for b in &OperatorCapability::ALL[i + 1..] {
86				assert_ne!(a.bit(), b.bit(), "{a:?} collides with {b:?}");
87			}
88		}
89	}
90
91	#[test]
92	fn every_declared_capability_is_reachable_through_all() {
93		// from_bitmask filters over ALL, so a variant missing from ALL is dropped on every
94		// descriptor round trip and the operator loses that capability with no error anywhere. The
95		// match is exhaustive so a new variant fails to compile here rather than vanishing at runtime.
96		for capability in [
97			OperatorCapability::Insert,
98			OperatorCapability::Update,
99			OperatorCapability::Delete,
100			OperatorCapability::Expire,
101		] {
102			match capability {
103				OperatorCapability::Insert
104				| OperatorCapability::Update
105				| OperatorCapability::Delete
106				| OperatorCapability::Expire => {}
107			}
108			assert!(
109				OperatorCapability::ALL.contains(&capability),
110				"{capability:?} is missing from ALL, so from_bitmask silently drops it"
111			);
112			assert!(
113				from_bitmask(to_bitmask(&[capability])).contains(&capability),
114				"{capability:?} does not survive a bitmask round trip"
115			);
116		}
117	}
118
119	#[test]
120	fn presets_survive_a_bitmask_round_trip() {
121		let restored = from_bitmask(to_bitmask(OperatorCapability::STANDARD));
122		assert!(restored.contains(&OperatorCapability::Insert));
123		assert!(restored.contains(&OperatorCapability::Update));
124		assert!(restored.contains(&OperatorCapability::Delete));
125	}
126
127	#[test]
128	fn expire_is_reachable_through_all_but_never_through_standard() {
129		// STANDARD must never carry Expire, or every in-tree operator silently opts into rows it has no arm
130		// for.
131		assert!(!OperatorCapability::STANDARD.contains(&OperatorCapability::Expire));
132		assert!(OperatorCapability::ALL.contains(&OperatorCapability::Expire));
133		assert!(from_bitmask(to_bitmask(&[OperatorCapability::Expire])).contains(&OperatorCapability::Expire));
134	}
135
136	#[test]
137	fn expire_does_not_claim_the_retired_reclaim_bit() {
138		// Expire must never take the retired Reclaim bit, or a stale guest still setting it reads as opted in.
139		assert_ne!(OperatorCapability::Expire.bit(), 1 << 4);
140		assert!(!from_bitmask(1 << 4).contains(&OperatorCapability::Expire));
141	}
142
143	#[test]
144	fn an_unknown_descriptor_bit_is_dropped_rather_than_misread() {
145		// Guests built against an older ABI may still set retired bits (the removed Reclaim
146		// capability was 1 << 4); from_bitmask must ignore them instead of aliasing them onto a
147		// live capability.
148		let stale = to_bitmask(OperatorCapability::STANDARD) | (1 << 4);
149		let restored = from_bitmask(stale);
150		assert_eq!(restored, OperatorCapability::STANDARD.to_vec());
151	}
152}
reifydb_core/interface/flow.rs

reifydb_core/interface/
flow.rs
