reifydb-sub-flow 0.4.5

Flow subsystem for stream processing and data flows
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2025 ReifyDB

//! Flow state tracking for the coordinator.
//!
//! Tracks the status and checkpoint of each flow, enabling the coordinator
//! to route changes appropriately and manage backfilling.

use std::collections::BTreeMap;

use reifydb_core::{common::CommitVersion, interface::catalog::flow::FlowId};

/// Status of a flow in the coordinator.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FlowStatus {
	/// Flow is active and receiving live CDC events.
	Active,
	/// Flow is catching up from historical CDC data.
	/// It will not receive live CDC events until it catches up to the current version.
	Backfilling,
}

/// State of a single flow tracked by the coordinator.
#[derive(Clone, Debug)]
pub struct FlowState {
	/// Current checkpoint version - the last version this flow has processed.
	pub checkpoint: CommitVersion,
	/// Current status of the flow.
	pub status: FlowStatus,
}

impl FlowState {
	/// Create a new flow state for a newly registered flow.
	///
	/// New flows start in Backfilling status with checkpoint 0.
	pub fn new_backfilling() -> Self {
		Self {
			checkpoint: CommitVersion(0),
			status: FlowStatus::Backfilling,
		}
	}

	/// Create a new flow state that starts immediately active at the given checkpoint.
	pub fn new_active(checkpoint: CommitVersion) -> Self {
		Self {
			checkpoint,
			status: FlowStatus::Active,
		}
	}

	/// Check if the flow is currently backfilling.
	pub fn is_backfilling(&self) -> bool {
		self.status == FlowStatus::Backfilling
	}

	/// Check if the flow is active (receiving live CDC).
	pub fn is_active(&self) -> bool {
		self.status == FlowStatus::Active
	}

	/// Transition the flow to active status.
	pub fn activate(&mut self) {
		self.status = FlowStatus::Active;
	}

	/// Transition the flow back to backfilling status.
	pub fn deactivate(&mut self) {
		self.status = FlowStatus::Backfilling;
	}

	/// Update the checkpoint to a new version.
	pub fn update_checkpoint(&mut self, version: CommitVersion) {
		self.checkpoint = version;
	}
}

/// Collection of flow states managed by the coordinator.
#[derive(Debug, Default)]
pub struct FlowStates {
	states: BTreeMap<FlowId, FlowState>,
}

impl FlowStates {
	/// Create a new empty flow states collection.
	pub fn new() -> Self {
		Self {
			states: BTreeMap::new(),
		}
	}

	/// Get mutable access to the state of a flow.
	pub fn get_mut(&mut self, flow_id: &FlowId) -> Option<&mut FlowState> {
		self.states.get_mut(flow_id)
	}

	/// Returns true if this collection tracks the given flow.
	pub fn contains(&self, flow_id: &FlowId) -> bool {
		self.states.contains_key(flow_id)
	}

	/// Register a new flow in backfilling status.
	pub fn register_backfilling(&mut self, flow_id: FlowId) {
		self.states.insert(flow_id, FlowState::new_backfilling());
	}

	/// Register a new flow as immediately active at the given checkpoint.
	pub fn register_active(&mut self, flow_id: FlowId, checkpoint: CommitVersion) {
		self.states.insert(flow_id, FlowState::new_active(checkpoint));
	}

	/// Get all active flow IDs.
	pub fn active_flow_ids(&self) -> Vec<FlowId> {
		self.states.iter().filter(|(_, state)| state.is_active()).map(|(id, _)| *id).collect()
	}

	/// Get all backfilling flow IDs.
	pub fn backfilling_flow_ids(&self) -> Vec<FlowId> {
		self.states.iter().filter(|(_, state)| state.is_backfilling()).map(|(id, _)| *id).collect()
	}
}