reifydb_sub_flow/execution/
tick.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::collections::HashMap;
5
6use reifydb_core::{
7	actors::pending::PendingWrite,
8	event::row::OperatorRowsExpiredEvent,
9	interface::{
10		catalog::flow::{FlowId, FlowNodeId},
11		change::Change,
12	},
13	key::{EncodableKey, flow_node_internal_state::FlowNodeInternalStateKey, flow_node_state::FlowNodeStateKey},
14};
15use reifydb_rql::flow::node::FlowNode;
16use reifydb_sdk::operator::Tick;
17use reifydb_value::{Result, value::datetime::DateTime};
18use tracing::instrument;
19
20use crate::{engine::FlowEngineInner, operator::Operators, transaction::FlowTransaction};
21
22impl FlowEngineInner {
23	#[instrument(name = "flow::engine::process_tick", level = "debug", skip(self, txn), fields(
24		flow_id = ?flow_id,
25		timestamp = %timestamp
26	))]
27	pub fn process_tick(&self, txn: &mut FlowTransaction, flow_id: FlowId, timestamp: DateTime) -> Result<()> {
28		let flow = match self.flows.get(&flow_id) {
29			Some(f) => f.clone(),
30			None => return Ok(()),
31		};
32
33		let mut pending: HashMap<FlowNodeId, Vec<Change>> = HashMap::new();
34		for node_id in flow.topological_order()? {
35			let node = match flow.get_node(&node_id) {
36				Some(n) => n.clone(),
37				None => continue,
38			};
39
40			self.dispatch_inbox(txn, &node, node_id, &mut pending)?;
41			self.fire_operator_tick(txn, &node, node_id, timestamp, &mut pending)?;
42		}
43
44		self.emit_operator_drop_metrics(txn);
45		Ok(())
46	}
47
48	#[inline]
49	fn dispatch_inbox(
50		&self,
51		txn: &mut FlowTransaction,
52		node: &FlowNode,
53		node_id: FlowNodeId,
54		pending: &mut HashMap<FlowNodeId, Vec<Change>>,
55	) -> Result<()> {
56		let Some(inbox) = pending.remove(&node_id).filter(|v| !v.is_empty()) else {
57			return Ok(());
58		};
59		let combined_output = self.dispatch_node(txn, node, inbox)?;
60		if !combined_output.diffs.is_empty() {
61			for child_id in &node.outputs {
62				pending.entry(*child_id).or_default().push(combined_output.clone());
63			}
64		}
65		Ok(())
66	}
67
68	#[inline]
69	fn fire_operator_tick(
70		&self,
71		txn: &mut FlowTransaction,
72		node: &FlowNode,
73		node_id: FlowNodeId,
74		timestamp: DateTime,
75		pending: &mut HashMap<FlowNodeId, Vec<Change>>,
76	) -> Result<()> {
77		let operator = match self.operators.get(&node_id) {
78			Some(op) => op.clone(),
79			None => return Ok(()),
80		};
81		let interval = match operator.ticks() {
82			Some(interval) => interval,
83			None => return Ok(()),
84		};
85		if matches!(&*operator, Operators::Custom(_) | Operators::Apply(_))
86			&& !self.operator_due(node_id, timestamp.to_nanos(), interval)
87		{
88			return Ok(());
89		}
90		if let Some(tick_emission) = operator.tick(
91			txn,
92			Tick {
93				now: timestamp,
94			},
95		)? && !tick_emission.diffs.is_empty()
96		{
97			for child_id in &node.outputs {
98				pending.entry(*child_id).or_default().push(tick_emission.clone());
99			}
100		}
101		Ok(())
102	}
103
104	fn emit_operator_drop_metrics(&self, txn: &FlowTransaction) {
105		let mut per_node: HashMap<FlowNodeId, u64> = HashMap::new();
106		for (key, write) in txn.pending().iter_sorted() {
107			if !matches!(write, PendingWrite::Drop) {
108				continue;
109			}
110			let node = FlowNodeStateKey::decode(key)
111				.map(|k| k.node)
112				.or_else(|| FlowNodeInternalStateKey::decode(key).map(|k| k.node));
113			if let Some(node) = node {
114				*per_node.entry(node).or_default() += 1;
115			}
116		}
117
118		if per_node.is_empty() {
119			return;
120		}
121
122		let rows: u64 = per_node.values().copied().sum();
123		self.event_bus.emit(OperatorRowsExpiredEvent::new(
124			per_node.len() as u64,
125			0,
126			rows,
127			rows,
128			per_node.clone(),
129			per_node,
130		));
131	}
132}
reifydb_sub_flow/execution/tick.rs

reifydb_sub_flow/execution/
tick.rs
