harmos_runtime/streams/
definition.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17
18#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
19use std::collections::BTreeMap;
20use std::collections::{HashMap, HashSet};
21use std::sync::atomic::AtomicU64;
22use std::sync::{Arc, Mutex, RwLock, Weak};
23// Third Party Imports
24use serde::Serialize;
25use serde::de::DeserializeOwned;
26use tokio::sync::watch;
27// Local Imports
28use super::demand::Holding;
29use super::window::{Row, Window};
30use crate::streams::{BatchDefinition, Definition, StreamCatalog, StreamRow};
31use crate::{Application, state::Position, storage::Cursor};
32
33/* === Constants === */
34
35/* === Macros === */
36
37/* === Type Aliases === */
38
39pub(super) type Windows = HashMap<Arc<str>, HashMap<u32, Arc<Wired>>>;
40
41/* === Structs === */
42
43pub(crate) struct Delivery {
44    pub(crate) rows: Vec<(Cursor, Definition, Position, Row)>,
45    pub(crate) reaches: Cursor,
46    pub(crate) incomplete: Vec<Definition>,
47}
48
49#[derive(Debug)]
50pub(crate) struct Wired {
51    pub(crate) definition: Definition,
52    pub(crate) window: Arc<Window>,
53    pub(super) validation: Option<Arc<crate::schema::Validator>>,
54    pub(super) publisher: Mutex<Weak<Mutex<bool>>>,
55
56    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
57    pub(super) staged: Option<super::registration::Staged>,
58}
59
60#[derive(Debug)]
61pub(crate) struct Table {
62    pub(super) windows: RwLock<Windows>,
63    pub(super) sequence: Arc<AtomicU64>,
64    pub(super) pump: watch::Sender<()>,
65    pub(super) visibility: Arc<RwLock<()>>,
66    pub(super) publications: Arc<super::publication::Pool>,
67    pub(super) ingest: Arc<super::ingest::Ingest>,
68
69    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
70    pub(super) listeners: Mutex<BTreeMap<String, Arc<super::listeners::Listening>>>,
71}
72
73/* === Enums === */
74
75#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
76pub(crate) enum Refused {
77    #[error("two stream definitions share the id \"{0}\"")]
78    SharedId(Arc<str>),
79
80    #[error("the stream \"{0}\" declares a window of zero rows")]
81    EmptyWindow(Arc<str>),
82
83    #[error("a stream consumer selected the undeclared stream \"{0}\"")]
84    UndeclaredSelection(Arc<str>),
85}
86
87/* === Traits === */
88
89pub trait Stream<A: Application>: Clone + Send + Sync + Serialize + DeserializeOwned + 'static {
90    const ID: &'static str;
91
92    const VERSION: u32;
93
94    const WINDOW: usize;
95
96    fn batch_definition() -> Option<BatchDefinition> {
97        None
98    }
99
100    fn metadata() -> crate::Metadata {
101        crate::Metadata::new(crate::Identifier::declared(Self::ID, "stream"), Self::ID, None)
102    }
103
104    fn row_metadata() -> Vec<crate::StreamField> {
105        Vec::new()
106    }
107
108    fn definition() -> Definition {
109        Definition::declared::<Self>(Self::ID, Self::VERSION, Self::WINDOW)
110            .batching(Self::batch_definition())
111            .presentation(Self::metadata(), Self::row_metadata())
112    }
113}
114
115/* === Implementations === */
116
117impl<A, S> Stream<A> for S
118where
119    A: Application,
120    S: StreamRow,
121    A::Streams: StreamCatalog<S>,
122{
123    const ID: &'static str = S::ID;
124    const VERSION: u32 = S::VERSION;
125    const WINDOW: usize = S::WINDOW;
126    fn metadata() -> crate::Metadata {
127        S::metadata()
128    }
129
130    fn row_metadata() -> Vec<crate::StreamField> {
131        S::row_metadata()
132    }
133
134    fn batch_definition() -> Option<BatchDefinition> {
135        S::batch_definition()
136    }
137}
138
139impl Table {
140    pub(crate) fn resuming(definitions: impl IntoIterator<Item = Definition>, floor: Position, resume: Cursor) -> Result<Self, Refused> {
141        let (pump, _) = watch::channel(());
142        let table = Self {
143            windows: RwLock::new(HashMap::new()),
144            sequence: Arc::new(AtomicU64::new(resume.0)),
145            pump,
146            visibility: Arc::new(RwLock::new(())),
147            publications: Arc::new(super::publication::Pool::new()),
148            ingest: Arc::new(super::ingest::Ingest::new()),
149            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
150            listeners: Mutex::new(BTreeMap::new()),
151        };
152
153        for definition in definitions {
154            table.insert(definition, floor)?;
155        }
156
157        Ok(table)
158    }
159
160    pub(crate) fn insert(&self, definition: Definition, floor: Position) -> Result<Arc<Wired>, Refused> {
161        if definition.window() == 0 {
162            return Err(Refused::EmptyWindow(definition.id().into()));
163        }
164
165        let _visibility = self.visibility.read().expect("stream visibility poisoned");
166        let mut windows = self.windows.write().expect("stream registry poisoned");
167        if windows.contains_key(definition.id()) {
168            return Err(Refused::SharedId(definition.id().into()));
169        }
170
171        let window = Arc::new(
172            Window::wired(definition.window(), floor, Arc::clone(&self.sequence), self.pump.clone())
173                .visible_with(self.visibility.clone())
174                .batching(definition.batch()),
175        );
176        let wired = Arc::new(Wired {
177            validation: None,
178            definition,
179            window,
180            publisher: Mutex::new(Weak::new()),
181            #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
182            staged: None,
183        });
184        windows.insert(Arc::<str>::from(wired.definition.id()), HashMap::from([(wired.definition.version(), Arc::clone(&wired))]));
185
186        Ok(wired)
187    }
188
189    pub(crate) fn wired(&self, id: &str, version: u32) -> Option<Arc<Wired>> {
190        let _visibility = self.visibility.read().expect("stream visibility poisoned");
191        let wired =
192            self.windows.read().expect("stream registry poisoned").get(id).and_then(|revisions| revisions.get(&version)).cloned()?;
193        wired.admitted()
194    }
195
196    pub(crate) fn declared(&self) -> Vec<Arc<Wired>> {
197        let _visibility = self.visibility.read().expect("stream visibility poisoned");
198        visible(
199            self.windows
200                .read()
201                .expect("stream registry poisoned")
202                .values()
203                .flat_map(|revisions| revisions.values())
204                .cloned()
205                .collect::<Vec<_>>(),
206        )
207    }
208
209    pub(crate) fn hold(&self, selections: impl IntoIterator<Item = Vec<Definition>>) -> Result<Vec<Holding>, Refused> {
210        let mut held = Vec::new();
211
212        for selections in selections {
213            let mut distinct = HashSet::new();
214            for definition in selections {
215                if !distinct.insert((Arc::<str>::from(definition.id()), definition.version())) {
216                    continue;
217                }
218                let wired =
219                    self.wired(definition.id(), definition.version()).ok_or(Refused::UndeclaredSelection(definition.id().into()))?;
220                held.push(Holding::engage(Arc::clone(&wired.window)));
221            }
222        }
223
224        Ok(held)
225    }
226
227    pub(crate) fn delivery(&self) -> watch::Receiver<()> {
228        self.pump.subscribe()
229    }
230
231    pub(crate) fn recorded_after(&self, resume: Cursor) -> Delivery {
232        let _visibility = self.visibility.read().expect("stream visibility poisoned");
233        // Hold admission until the global cut is assembled. A new window
234        // publishing below `reaches` must not be absent from this delivery.
235        let windows = self.windows.read().expect("stream registry poisoned");
236        // Include every physical window in the cut. A shared generation may
237        // commit while this read lock is held; its first rows must remain part
238        // of the global cursor span even when catalog discovery began earlier.
239        let reaches = windows
240            .values()
241            .flat_map(|revisions| revisions.values())
242            .map(|wired| wired.window.latest_locked())
243            .max()
244            .unwrap_or(resume.0)
245            .max(resume.0);
246        let mut rows = Vec::new();
247        let mut incomplete = Vec::new();
248        for wired in windows.values().flat_map(|revisions| revisions.values()) {
249            let (retained, lost) = wired.window.recorded_after_locked(resume.0, reaches);
250            let Some(wired) = wired.admitted() else {
251                continue;
252            };
253            rows.extend(retained.into_iter().map(|(cursor, position, row)| (Cursor(cursor), wired.definition.clone(), position, row)));
254            if lost {
255                incomplete.push(wired.definition.clone());
256            }
257        }
258        rows.sort_unstable_by_key(|(cursor, ..)| *cursor);
259
260        Delivery { rows, reaches: Cursor(reaches), incomplete }
261    }
262
263    pub(crate) fn frontier(&self) -> Cursor {
264        let _visibility = self.visibility.read().expect("stream visibility poisoned");
265        let windows = self.windows.read().expect("stream registry poisoned");
266        Cursor(windows.values().flat_map(|revisions| revisions.values()).map(|wired| wired.window.latest_locked()).max().unwrap_or(0))
267    }
268
269    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
270    pub(super) fn can_advance(&self, count: u64) -> bool {
271        self.sequence.load(std::sync::atomic::Ordering::Relaxed).checked_add(count).is_some()
272    }
273
274    pub(crate) fn wake(&self) {
275        self.pump.send_modify(|()| {});
276        for wired in self.declared() {
277            wired.window.knock();
278        }
279    }
280}
281
282impl Wired {
283    fn admitted(self: &Arc<Self>) -> Option<Arc<Self>> {
284        #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
285        if let Some(staged) = &self.staged
286            && !staged.scope.has_committed()
287        {
288            return staged.previous.clone();
289        }
290        Some(self.clone())
291    }
292}
293
294/* === Public Functions === */
295
296/* === Private Functions === */
297
298fn visible(wires: impl IntoIterator<Item = Arc<Wired>>) -> Vec<Arc<Wired>> {
299    let wires = wires.into_iter();
300    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
301    let wires = {
302        let mut commits = std::collections::BTreeMap::new();
303        wires.filter_map(move |wired| {
304            if let Some(staged) = &wired.staged {
305                let committed = *commits.entry(staged.scope.generation()).or_insert_with(|| staged.scope.has_committed());
306                if !committed {
307                    return staged.previous.clone();
308                }
309            }
310            Some(wired)
311        })
312    };
313    wires.collect()
314}
315
316/* === Tests === */
harmos_runtime/streams/definition.rs

harmos_runtime/streams/
definition.rs
