reifydb_core/actors/
flow.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	collections::{BTreeMap, HashMap},
6	sync::Arc,
7};
8
9use reifydb_codec::encoded::shape::RowShape;
10use reifydb_runtime::actor::system::ActorHandle;
11use reifydb_value::{Result, error::Error, value::datetime::DateTime};
12
13use super::pending::Pending;
14use crate::{
15	common::CommitVersion,
16	interface::{
17		catalog::{flow::FlowId, shape::ShapeId},
18		cdc::Cdc,
19		change::Change,
20	},
21};
22
23#[derive(Clone, Debug)]
24pub struct FlowInstruction {
25	pub flow_id: FlowId,
26
27	pub to_version: CommitVersion,
28
29	pub changes: Vec<Change>,
30}
31
32impl FlowInstruction {
33	pub fn new(flow_id: FlowId, to_version: CommitVersion, changes: Vec<Change>) -> Self {
34		Self {
35			flow_id,
36			to_version,
37			changes,
38		}
39	}
40}
41
42#[derive(Clone, Debug)]
43pub struct WorkerBatch {
44	pub state_version: CommitVersion,
45
46	pub instructions: Vec<FlowInstruction>,
47}
48
49impl WorkerBatch {
50	pub fn new(state_version: CommitVersion) -> Self {
51		Self {
52			state_version,
53			instructions: Vec::new(),
54		}
55	}
56
57	pub fn add_instruction(&mut self, instruction: FlowInstruction) {
58		self.instructions.push(instruction);
59	}
60}
61
62pub enum FlowResponse {
63	Success {
64		pending: Pending,
65		pending_shapes: Vec<RowShape>,
66		view_changes: Vec<Change>,
67	},
68
69	Error(Error),
70}
71
72pub type FlowHandle = ActorHandle<FlowMessage>;
73
74pub enum FlowMessage {
75	Process {
76		batch: WorkerBatch,
77		reply: Box<dyn FnOnce(FlowResponse) + Send>,
78	},
79
80	Dispatch {
81		state_version: CommitVersion,
82		to_version: CommitVersion,
83		changes: Arc<Vec<Change>>,
84		index: Arc<HashMap<ShapeId, Vec<FlowId>>>,
85		active: Arc<Vec<FlowId>>,
86		reply: Box<dyn FnOnce(FlowResponse) + Send>,
87	},
88
89	Register {
90		flow_id: FlowId,
91		reply: Box<dyn FnOnce(FlowResponse) + Send>,
92	},
93
94	Tick {
95		flow_ids: Vec<FlowId>,
96		timestamp: DateTime,
97		state_version: CommitVersion,
98		reply: Box<dyn FnOnce(FlowResponse) + Send>,
99	},
100	Rebalance {
101		flow_ids: Vec<FlowId>,
102		reply: Box<dyn FnOnce(FlowResponse) + Send>,
103	},
104}
105
106pub enum PoolResponse {
107	Success {
108		pending: Pending,
109		pending_shapes: Vec<RowShape>,
110		view_changes: Vec<Change>,
111	},
112
113	RegisterSuccess,
114
115	Error(Error),
116}
117
118pub type FlowPoolHandle = ActorHandle<FlowPoolMessage>;
119
120pub enum FlowPoolMessage {
121	RegisterFlow {
122		flow_id: FlowId,
123		reply: Box<dyn FnOnce(PoolResponse) + Send>,
124	},
125
126	Submit {
127		batches: BTreeMap<usize, WorkerBatch>,
128		reply: Box<dyn FnOnce(PoolResponse) + Send>,
129	},
130
131	Broadcast {
132		state_version: CommitVersion,
133		to_version: CommitVersion,
134		changes: Arc<Vec<Change>>,
135		index: Arc<HashMap<ShapeId, Vec<FlowId>>>,
136		active: Arc<Vec<FlowId>>,
137		reply: Box<dyn FnOnce(PoolResponse) + Send>,
138	},
139
140	SubmitToWorker {
141		worker_id: usize,
142		batch: WorkerBatch,
143		reply: Box<dyn FnOnce(PoolResponse) + Send>,
144	},
145
146	Tick {
147		ticks: BTreeMap<usize, Vec<FlowId>>,
148		timestamp: DateTime,
149		state_version: CommitVersion,
150		reply: Box<dyn FnOnce(PoolResponse) + Send>,
151	},
152	Rebalance {
153		assignments: BTreeMap<usize, Vec<FlowId>>,
154		reply: Box<dyn FnOnce(PoolResponse) + Send>,
155	},
156
157	WorkerReply {
158		worker_id: usize,
159		response: FlowResponse,
160	},
161}
162
163pub type FlowCoordinatorHandle = ActorHandle<FlowCoordinatorMessage>;
164
165pub enum FlowCoordinatorMessage {
166	Consume {
167		cdcs: Vec<Cdc>,
168		current_version: CommitVersion,
169		reply: Box<dyn FnOnce(Result<()>) + Send>,
170	},
171
172	PoolReply(PoolResponse),
173
174	Tick,
175
176	Bootstrap {
177		flows: Vec<(FlowId, bool)>,
178	},
179}
reifydb_core/actors/flow.rs

reifydb_core/actors/
flow.rs
