harmos_runtime/streams/
window.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
17use std::any::Any;
18use std::collections::VecDeque;
19use std::future::Future;
20use std::marker::PhantomData;
21use std::ops::Bound;
22use std::pin::Pin;
23use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
24use std::sync::{Arc, RwLock};
25use std::task::{Context, Poll};
26// Third Party Imports
27use tokio::sync::watch;
28// Local Imports
29use super::demand::Holding;
30use crate::{state::At, state::Position};
31
32/* === Constants === */
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38pub(crate) type Row = Arc<dyn Any + Send + Sync>;
39
40type Step = Pin<Box<dyn Future<Output = (Parts, Turn)> + Send>>;
41
42/* === Structs === */
43
44pub(crate) struct Window {
45    ring: RwLock<Ring>,
46    pub(super) visibility: Arc<RwLock<()>>,
47    sequence: Arc<AtomicU64>,
48    doorbell: watch::Sender<u64>,
49    pump: watch::Sender<()>,
50    demand: watch::Sender<usize>,
51}
52
53struct Ring {
54    batches: Option<super::batches::State>,
55    rows: VecDeque<(u64, u64, Position, Row)>,
56    capacity: usize,
57    next: u64,
58    last: u64,
59    // Latest overwritten delivery cursor; independent of the durable floor.
60    evicted: u64,
61    // Sidecar loss has no recoverable Position or row identity.
62    missed: u64,
63    floor: Position,
64    standing: bool,
65}
66
67#[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
68pub(super) struct ListenerCandidate {
69    pub(super) cursor: u64,
70    pub(super) sequence: u64,
71    pub(super) source: usize,
72}
73
74pub(crate) struct Retained {
75    pub(crate) rows: Vec<(Position, Row)>,
76    pub(crate) floor: Position,
77    pub(crate) beneath: bool,
78}
79
80pub(crate) struct Advanced {
81    pub(crate) missed: u64,
82    pub(crate) rows: Vec<(Position, Row)>,
83    pub(crate) frontier: u64,
84}
85
86struct Parts {
87    window: Arc<Window>,
88    doorbell: watch::Receiver<u64>,
89    frontier: u64,
90    closed: Arc<AtomicBool>,
91
92    #[allow(dead_code, reason = "held for its drop, which releases the demand count")]
93    held: Holding,
94}
95
96struct Following<S> {
97    parts: Option<Parts>,
98    step: Option<Step>,
99    gap: u64,
100    pending: VecDeque<(Position, Row)>,
101    done: bool,
102    marker: PhantomData<fn() -> S>,
103}
104
105/* === Enums === */
106
107#[derive(Clone, Copy, Debug, PartialEq, Eq)]
108pub enum Item<S> {
109    Row(At<S>),
110    Gap { missed: u64 },
111}
112
113#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
114pub enum BeneathWindow {
115    #[error("no series adapter serves beneath the retained window")]
116    Unserved,
117
118    #[error("the served history begins after position {}", floor.0)]
119    Floor { floor: Position },
120
121    #[error("the series adapter could not read its own history: {0}")]
122    Unreadable(String),
123}
124
125enum Turn {
126    Advanced(Advanced),
127    Ended,
128}
129
130/* === Traits === */
131
132/* === Implementations === */
133
134impl Window {
135    pub(crate) fn wired(capacity: usize, floor: Position, sequence: Arc<AtomicU64>, pump: watch::Sender<()>) -> Self {
136        let (doorbell, _) = watch::channel(0);
137        let (demand, _) = watch::channel(0);
138
139        Self {
140            visibility: Arc::new(RwLock::new(())),
141            ring: RwLock::new(Ring {
142                batches: None,
143                rows: VecDeque::with_capacity(capacity),
144                capacity,
145                next: 0,
146                last: 0,
147                evicted: 0,
148                missed: 0,
149                floor,
150                standing: false,
151            }),
152            sequence,
153            doorbell,
154            pump,
155            demand,
156        }
157    }
158
159    pub(super) fn visible_with(mut self, visibility: Arc<RwLock<()>>) -> Self {
160        self.visibility = visibility;
161        self
162    }
163
164    pub(crate) fn batching(self, definition: Option<super::BatchDefinition>) -> Self {
165        self.ring.write().expect("ring is healthy").batches = definition.map(super::batches::State::new);
166        self
167    }
168
169    pub(crate) fn batches<S>(&self, held: Holding) -> super::Batches<S> {
170        let ring = self.ring.read().expect("ring is healthy");
171        ring.batches.as_ref().expect("stream declared batching").feed.following(held)
172    }
173
174    pub(crate) fn close_batches(&self) {
175        if let Some(batches) = &mut self.ring.write().expect("ring is healthy").batches {
176            batches.close();
177        }
178    }
179
180    pub(crate) fn publish(&self, at: Position, row: Row) {
181        self.append(at, row, 0).expect("native stream cursor exhausted");
182    }
183
184    pub(super) fn append(&self, at: Position, row: Row, missed: u64) -> Result<(), super::StreamPublicationError> {
185        let _visibility = self.visibility.read().expect("stream visibility poisoned");
186        let evicted = self.append_locked(at, row, missed)?;
187        drop(_visibility);
188        drop(evicted);
189        self.knock();
190        self.pump.send_modify(|()| {});
191        Ok(())
192    }
193
194    // Caller holds the facility visibility fence. Notifications follow scope commit.
195    pub(super) fn append_locked(&self, at: Position, row: Row, missed: u64) -> Result<Option<Row>, super::StreamPublicationError> {
196        let evicted;
197        {
198            let mut ring = self.ring.write().expect("the ring is never poisoned");
199            let advance = missed.checked_add(1).ok_or(super::StreamPublicationError::Sequence)?;
200            let next = ring.next.checked_add(advance).ok_or(super::StreamPublicationError::Sequence)?;
201            let previous = self
202                .sequence
203                .try_update(Ordering::Relaxed, Ordering::Relaxed, |cursor| cursor.checked_add(advance))
204                .map_err(|_| super::StreamPublicationError::Sequence)?;
205            let cursor = previous + advance;
206            if missed > 0 {
207                ring.missed = cursor - 1;
208                if let Some(batches) = &mut ring.batches {
209                    batches.miss(missed);
210                }
211            }
212            evicted = if ring.rows.len() == ring.capacity {
213                let (lost, _, stamp, row) = ring.rows.pop_front().expect("full ring");
214                ring.evicted = lost;
215                ring.floor = stamp;
216                ring.standing = true;
217                Some(row)
218            } else {
219                None
220            };
221            if let Some(batches) = &mut ring.batches {
222                batches.publish(row.as_ref());
223            }
224            if let Some(publication) = row.downcast_ref::<super::ManagedPublication>() {
225                publication.assign_sequence(next);
226            }
227            ring.rows.push_back((cursor, next, at, row));
228            ring.next = next;
229            ring.last = cursor;
230        }
231        Ok(evicted)
232    }
233
234    // Pool admission is serialized, and the caller owns facility visibility.
235    // Preflight the whole prefix so a refused row changes no retention or gaps.
236    pub(super) fn can_advance(&self, count: u64) -> bool {
237        self.sequence.load(Ordering::Relaxed).checked_add(count).is_some()
238            && self.ring.read().expect("the ring is never poisoned").next.checked_add(count).is_some()
239    }
240
241    // Rows past `taken` are not in every feed's hands yet and stay retained.
242    pub(super) fn reclaim(&self, needed_bytes: usize, needed_rows: usize, taken: u64) -> Option<()> {
243        let mut ring = self.ring.write().expect("the ring is never poisoned");
244        let mut bytes = 0usize;
245        let mut rows = 0usize;
246        let mut prefix = 0usize;
247        for (cursor, _, _, row) in &ring.rows {
248            if *cursor > taken {
249                break;
250            }
251            let publication = row.downcast_ref::<super::ManagedPublication>()?;
252            prefix += 1;
253            if Arc::strong_count(row) == 1
254                && let Some(charge) = publication.reclaimable_bytes()
255            {
256                bytes = bytes.saturating_add(charge);
257                rows += 1;
258            }
259            if bytes >= needed_bytes && rows >= needed_rows {
260                break;
261            }
262        }
263        if bytes < needed_bytes || rows < needed_rows {
264            return None;
265        }
266        for _ in 0..prefix {
267            let (cursor, _, position, row) = ring.rows.pop_front().expect("validated retention prefix");
268            ring.evicted = cursor;
269            ring.floor = position;
270            ring.standing = true;
271            // Only unshared ManagedPublication owners release credit here;
272            // no application destructor or pending receipt sender is invoked.
273            drop(row);
274        }
275        Some(())
276    }
277
278    // Appending one row would evict a row some live feed has not taken in.
279    pub(super) fn evicts_unread(&self, taken: u64) -> bool {
280        let ring = self.ring.read().expect("the ring is never poisoned");
281        ring.rows.len() == ring.capacity && ring.rows.front().is_some_and(|(cursor, ..)| *cursor > taken)
282    }
283
284    // Rows the window still takes without evicting one past `taken`.
285    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
286    pub(super) fn room(&self, taken: u64) -> usize {
287        let ring = self.ring.read().expect("the ring is never poisoned");
288        let read = ring.rows.partition_point(|(cursor, ..)| *cursor <= taken);
289        ring.capacity - (ring.rows.len() - read)
290    }
291
292    #[cfg(feature = "plugin-sidecar")]
293    pub(crate) fn miss(&self, count: u64) {
294        let _visibility = self.visibility.read().expect("stream visibility poisoned");
295        if count == 0 {
296            return;
297        }
298
299        let published;
300        {
301            let mut ring = self.ring.write().expect("the ring is never poisoned");
302            let next = ring.next.checked_add(count).expect("native stream cursor exhausted");
303            let previous = self
304                .sequence
305                .try_update(Ordering::Relaxed, Ordering::Relaxed, |cursor| cursor.checked_add(count))
306                .expect("native stream cursor exhausted");
307            let cursor = previous + count;
308            if let Some(batches) = &mut ring.batches {
309                batches.miss(count);
310            }
311            ring.missed = cursor;
312            ring.next = next;
313            ring.last = cursor;
314            published = ring.next;
315        }
316        drop(_visibility);
317        self.doorbell.send_replace(published);
318        self.pump.send_modify(|()| {});
319    }
320
321    pub(crate) fn stands_on(&self, floor: Position) {
322        let _visibility = self.visibility.read().expect("stream visibility poisoned");
323        let mut ring = self.ring.write().expect("the ring is never poisoned");
324
325        ring.floor = floor;
326        ring.standing = true;
327    }
328
329    pub(crate) fn snapshot(&self, start: Bound<Position>, end: Bound<Position>, as_of: Option<Position>) -> Retained {
330        self.snapshot_scope(start, end, as_of, true)
331    }
332
333    pub(crate) fn snapshot_scope(&self, start: Bound<Position>, end: Bound<Position>, as_of: Option<Position>, history: bool) -> Retained {
334        let _visibility = self.visibility.read().expect("stream visibility poisoned");
335        let ring = self.ring.read().expect("the ring is never poisoned");
336        let beneath = history && ring.standing && reaches_beneath(start, ring.floor);
337
338        Retained {
339            rows: ring
340                .rows
341                .iter()
342                .filter(|(_, _, stamp, _)| !beneath || *stamp > ring.floor)
343                .filter(|(_, _, stamp, _)| admitted(start, end, *stamp))
344                .filter(|(_, _, stamp, _)| as_of.is_none_or(|cutoff| *stamp <= cutoff))
345                .map(|(_, _, stamp, row)| (*stamp, Arc::clone(row)))
346                .collect(),
347            floor: ring.floor,
348            beneath,
349        }
350    }
351
352    pub(crate) fn reader<S: serde::Serialize + 'static>(
353        &self, start: Bound<Position>, end: Bound<Position>, limits: super::ReadLimits,
354    ) -> Result<Box<dyn super::RowSource>, super::ReadError> {
355        struct Frozen {
356            cut: super::ReadCut,
357            rows: std::collections::vec_deque::IntoIter<(Position, Row)>,
358        }
359        impl super::RowSource for Frozen {
360            fn cut(&self) -> super::ReadCut {
361                self.cut
362            }
363            fn next(&mut self, cancelled: &AtomicBool) -> Result<Option<crate::state::At<Row>>, super::ReadError> {
364                if cancelled.load(Ordering::Acquire) {
365                    return Err(super::ReadError::Closed);
366                }
367                Ok(self.rows.next().map(|(position, value)| crate::state::At { position, value }))
368            }
369        }
370        let _visibility = self.visibility.read().expect("stream visibility poisoned");
371        let ring = self.ring.read().expect("the ring is never poisoned");
372        if ring.missed > 0 {
373            return Err(super::ReadError::IncompleteCapture);
374        }
375        if ring.standing && reaches_beneath(start, ring.floor) {
376            return Err(super::ReadError::Retained { floor: ring.floor });
377        }
378        let mut bytes = 0usize;
379        let mut rows = VecDeque::new();
380        for (scanned, (_, _, position, row)) in ring.rows.iter().enumerate() {
381            if scanned == limits.rows {
382                return Err(super::ReadError::Limit("window row scan"));
383            }
384            if !admitted(start, end, *position) {
385                continue;
386            }
387            let typed =
388                super::row_value::<S>(row.as_ref()).ok_or_else(|| super::ReadError::Unreadable("window row type mismatch".into()))?;
389            bytes = bytes
390                .checked_add(super::reader::encoded_size(typed, limits.row_bytes)?)
391                .and_then(|bytes| bytes.checked_add(std::mem::size_of::<(Position, Row)>()))
392                .filter(|bytes| *bytes <= limits.pinned_bytes)
393                .ok_or(super::ReadError::Limit("pinned bytes"))?;
394            rows.push_back((*position, Arc::clone(row)));
395        }
396        Ok(Box::new(Frozen { cut: super::ReadCut::Window { delivered: crate::storage::Cursor(ring.last) }, rows: rows.into_iter() }))
397    }
398
399    pub(crate) fn since(&self, frontier: u64) -> Advanced {
400        let _visibility = self.visibility.read().expect("stream visibility poisoned");
401        let ring = self.ring.read().expect("the ring is never poisoned");
402        let rows: Vec<_> =
403            ring.rows.iter().filter(|(_, sequence, ..)| *sequence > frontier).map(|(_, _, stamp, row)| (*stamp, Arc::clone(row))).collect();
404        let missed = ring.next.saturating_sub(frontier).saturating_sub(rows.len() as u64);
405        Advanced { missed, rows, frontier: ring.next.max(frontier) }
406    }
407
408    pub(super) fn latest_locked(&self) -> u64 {
409        self.ring.read().expect("the ring is never poisoned").last
410    }
411
412    pub(super) fn recorded_after_locked(&self, resume: u64, reaches: u64) -> (Vec<(u64, Position, Row)>, bool) {
413        let ring = self.ring.read().expect("the ring is never poisoned");
414        let rows = ring
415            .rows
416            .iter()
417            .filter(|(cursor, ..)| *cursor > resume && *cursor <= reaches)
418            .map(|(cursor, _, stamp, row)| (*cursor, *stamp, Arc::clone(row)))
419            .collect();
420        // Under this same lock: evicting a row after cloning it cannot later
421        // turn this delivery into a gap. A cut overtaken before this lock is
422        // conservatively incomplete rather than falsely certified.
423        (rows, ring.evicted > resume || ring.missed > resume)
424    }
425
426    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
427    pub(super) fn listener_candidates(&self, after: u64, through: u64, source: usize, limit: usize, selected: &mut Vec<ListenerCandidate>) {
428        let ring = self.ring.read().expect("the ring is never poisoned");
429        let mut lower = 0;
430        let mut upper = ring.rows.len();
431        while lower < upper {
432            let mid = lower + (upper - lower) / 2;
433            if ring.rows[mid].0 <= after {
434                lower = mid + 1;
435            } else {
436                upper = mid;
437            }
438        }
439        for (cursor, sequence, ..) in ring.rows.iter().skip(lower).take(limit) {
440            if *cursor > through || selected.len() == limit && *cursor >= selected.last().unwrap().cursor {
441                break;
442            }
443            let index = selected.partition_point(|candidate| candidate.cursor < *cursor);
444            if selected.len() == limit {
445                selected.pop();
446            }
447            selected.insert(index, ListenerCandidate { cursor: *cursor, sequence: *sequence, source });
448        }
449    }
450
451    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
452    pub(super) fn listener_row(&self, cursor: u64) -> Option<Row> {
453        let ring = self.ring.read().expect("the ring is never poisoned");
454        let mut lower = 0;
455        let mut upper = ring.rows.len();
456        while lower < upper {
457            let mid = lower + (upper - lower) / 2;
458            if ring.rows[mid].0 < cursor {
459                lower = mid + 1;
460            } else {
461                upper = mid;
462            }
463        }
464        ring.rows.get(lower).filter(|row| row.0 == cursor).map(|row| Arc::clone(&row.3))
465    }
466
467    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
468    pub(super) fn logical_frontier(&self) -> u64 {
469        self.ring.read().expect("the ring is never poisoned").next
470    }
471
472    pub(super) fn wake_delivery(&self) {
473        self.pump.send_modify(|()| {});
474    }
475
476    pub(crate) fn knock(&self) {
477        self.doorbell.send_modify(|_| {});
478    }
479
480    pub(crate) fn interest(&self) -> watch::Receiver<usize> {
481        self.demand.subscribe()
482    }
483
484    pub(crate) fn engaged(&self) {
485        self.demand.send_modify(|count| *count += 1);
486    }
487
488    pub(crate) fn released(&self) {
489        self.demand.send_modify(|count| *count -= 1);
490    }
491}
492
493impl std::fmt::Debug for Window {
494    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
495        formatter.debug_struct("Window").finish_non_exhaustive()
496    }
497}
498
499impl<S> Unpin for Following<S> {}
500
501impl<S: Clone + 'static> futures_core::Stream for Following<S> {
502    type Item = Item<S>;
503
504    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
505        let following = self.get_mut();
506
507        loop {
508            if following.gap > 0 {
509                let missed = std::mem::take(&mut following.gap);
510                return Poll::Ready(Some(Item::Gap { missed }));
511            }
512            if let Some((position, row)) = following.pending.pop_front() {
513                let value = super::row_value::<S>(row.as_ref()).expect("a window holds only its own stream's rows").clone();
514                return Poll::Ready(Some(Item::Row(At { value, position })));
515            }
516            if following.done {
517                return Poll::Ready(None);
518            }
519
520            let mut step = match following.step.take() {
521                Some(step) => step,
522                None => match following.parts.take() {
523                    Some(parts) => Box::pin(turn(parts)),
524                    None => {
525                        following.done = true;
526                        return Poll::Ready(None);
527                    }
528                },
529            };
530
531            match step.as_mut().poll(context) {
532                Poll::Pending => {
533                    following.step = Some(step);
534                    return Poll::Pending;
535                }
536                Poll::Ready((parts, Turn::Advanced(advanced))) => {
537                    following.parts = Some(parts);
538                    following.gap = advanced.missed;
539                    following.pending = advanced.rows.into();
540                }
541                Poll::Ready((parts, Turn::Ended)) => {
542                    following.parts = Some(parts);
543                    following.done = true;
544                }
545            }
546        }
547    }
548}
549
550/* === Public Functions === */
551
552pub(crate) fn following<S: Clone + 'static>(
553    window: Arc<Window>, closed: Arc<AtomicBool>, held: Holding,
554) -> impl futures_core::Stream<Item = Item<S>> + Send + Unpin {
555    let doorbell = window.doorbell.subscribe();
556    let frontier = {
557        let _visibility = window.visibility.read().expect("stream visibility poisoned");
558        window.ring.read().expect("the ring is never poisoned").next
559    };
560
561    Following {
562        parts: Some(Parts { window, doorbell, frontier, closed, held }),
563        step: None,
564        gap: 0,
565        pending: VecDeque::new(),
566        done: false,
567        marker: PhantomData,
568    }
569}
570
571/* === Private Functions === */
572
573fn reaches_beneath(start: Bound<Position>, floor: Position) -> bool {
574    match start {
575        Bound::Included(at) => at <= floor,
576        Bound::Excluded(at) => at < floor,
577        Bound::Unbounded => true,
578    }
579}
580
581fn admitted(start: Bound<Position>, end: Bound<Position>, at: Position) -> bool {
582    let above = match start {
583        Bound::Included(from) => at >= from,
584        Bound::Excluded(from) => at > from,
585        Bound::Unbounded => true,
586    };
587    let below = match end {
588        Bound::Included(to) => at <= to,
589        Bound::Excluded(to) => at < to,
590        Bound::Unbounded => true,
591    };
592
593    above && below
594}
595
596async fn turn(mut parts: Parts) -> (Parts, Turn) {
597    loop {
598        let was_closed = parts.closed.load(Ordering::Acquire);
599        let advanced = parts.window.since(parts.frontier);
600        parts.frontier = advanced.frontier;
601
602        if advanced.missed > 0 || !advanced.rows.is_empty() {
603            return (parts, Turn::Advanced(advanced));
604        }
605        if was_closed || parts.doorbell.changed().await.is_err() {
606            return (parts, Turn::Ended);
607        }
608    }
609}
610
611/* === Tests === */
harmos_runtime/streams/window.rs

harmos_runtime/streams/
window.rs
