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 {
397            cut: super::ReadCut::Window {
398                delivered: crate::storage::Cursor(ring.last),
399            },
400            rows: rows.into_iter(),
401        }))
402    }
403
404    pub(crate) fn since(&self, frontier: u64) -> Advanced {
405        let _visibility = self.visibility.read().expect("stream visibility poisoned");
406        let ring = self.ring.read().expect("the ring is never poisoned");
407        let rows: Vec<_> =
408            ring.rows.iter().filter(|(_, sequence, ..)| *sequence > frontier).map(|(_, _, stamp, row)| (*stamp, Arc::clone(row))).collect();
409        let missed = ring.next.saturating_sub(frontier).saturating_sub(rows.len() as u64);
410        Advanced {
411            missed,
412            rows,
413            frontier: ring.next.max(frontier),
414        }
415    }
416
417    pub(super) fn latest_locked(&self) -> u64 {
418        self.ring.read().expect("the ring is never poisoned").last
419    }
420
421    pub(super) fn recorded_after_locked(&self, resume: u64, reaches: u64) -> (Vec<(u64, Position, Row)>, bool) {
422        let ring = self.ring.read().expect("the ring is never poisoned");
423        let rows = ring
424            .rows
425            .iter()
426            .filter(|(cursor, ..)| *cursor > resume && *cursor <= reaches)
427            .map(|(cursor, _, stamp, row)| (*cursor, *stamp, Arc::clone(row)))
428            .collect();
429        // Under this same lock: evicting a row after cloning it cannot later
430        // turn this delivery into a gap. A cut overtaken before this lock is
431        // conservatively incomplete rather than falsely certified.
432        (rows, ring.evicted > resume || ring.missed > resume)
433    }
434
435    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
436    pub(super) fn listener_candidates(&self, after: u64, through: u64, source: usize, limit: usize, selected: &mut Vec<ListenerCandidate>) {
437        let ring = self.ring.read().expect("the ring is never poisoned");
438        let mut lower = 0;
439        let mut upper = ring.rows.len();
440        while lower < upper {
441            let mid = lower + (upper - lower) / 2;
442            if ring.rows[mid].0 <= after {
443                lower = mid + 1;
444            } else {
445                upper = mid;
446            }
447        }
448        for (cursor, sequence, ..) in ring.rows.iter().skip(lower).take(limit) {
449            if *cursor > through || selected.len() == limit && *cursor >= selected.last().unwrap().cursor {
450                break;
451            }
452            let index = selected.partition_point(|candidate| candidate.cursor < *cursor);
453            if selected.len() == limit {
454                selected.pop();
455            }
456            selected.insert(
457                index,
458                ListenerCandidate {
459                    cursor: *cursor,
460                    sequence: *sequence,
461                    source,
462                },
463            );
464        }
465    }
466
467    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
468    pub(super) fn listener_row(&self, cursor: u64) -> Option<Row> {
469        let ring = self.ring.read().expect("the ring is never poisoned");
470        let mut lower = 0;
471        let mut upper = ring.rows.len();
472        while lower < upper {
473            let mid = lower + (upper - lower) / 2;
474            if ring.rows[mid].0 < cursor {
475                lower = mid + 1;
476            } else {
477                upper = mid;
478            }
479        }
480        ring.rows.get(lower).filter(|row| row.0 == cursor).map(|row| Arc::clone(&row.3))
481    }
482
483    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
484    pub(super) fn logical_frontier(&self) -> u64 {
485        self.ring.read().expect("the ring is never poisoned").next
486    }
487
488    pub(super) fn wake_delivery(&self) {
489        self.pump.send_modify(|()| {});
490    }
491
492    pub(crate) fn knock(&self) {
493        self.doorbell.send_modify(|_| {});
494    }
495
496    pub(crate) fn interest(&self) -> watch::Receiver<usize> {
497        self.demand.subscribe()
498    }
499
500    pub(crate) fn engaged(&self) {
501        self.demand.send_modify(|count| *count += 1);
502    }
503
504    pub(crate) fn released(&self) {
505        self.demand.send_modify(|count| *count -= 1);
506    }
507}
508
509impl std::fmt::Debug for Window {
510    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
511        formatter.debug_struct("Window").finish_non_exhaustive()
512    }
513}
514
515impl<S> Unpin for Following<S> {}
516
517impl<S: Clone + 'static> futures_core::Stream for Following<S> {
518    type Item = Item<S>;
519
520    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
521        let following = self.get_mut();
522
523        loop {
524            if following.gap > 0 {
525                let missed = std::mem::take(&mut following.gap);
526                return Poll::Ready(Some(Item::Gap { missed }));
527            }
528            if let Some((position, row)) = following.pending.pop_front() {
529                let value = super::row_value::<S>(row.as_ref()).expect("a window holds only its own stream's rows").clone();
530                return Poll::Ready(Some(Item::Row(At { value, position })));
531            }
532            if following.done {
533                return Poll::Ready(None);
534            }
535
536            let mut step = match following.step.take() {
537                Some(step) => step,
538                None => match following.parts.take() {
539                    Some(parts) => Box::pin(turn(parts)),
540                    None => {
541                        following.done = true;
542                        return Poll::Ready(None);
543                    }
544                },
545            };
546
547            match step.as_mut().poll(context) {
548                Poll::Pending => {
549                    following.step = Some(step);
550                    return Poll::Pending;
551                }
552                Poll::Ready((parts, Turn::Advanced(advanced))) => {
553                    following.parts = Some(parts);
554                    following.gap = advanced.missed;
555                    following.pending = advanced.rows.into();
556                }
557                Poll::Ready((parts, Turn::Ended)) => {
558                    following.parts = Some(parts);
559                    following.done = true;
560                }
561            }
562        }
563    }
564}
565
566/* === Public Functions === */
567
568pub(crate) fn following<S: Clone + 'static>(
569    window: Arc<Window>, closed: Arc<AtomicBool>, held: Holding,
570) -> impl futures_core::Stream<Item = Item<S>> + Send + Unpin {
571    let doorbell = window.doorbell.subscribe();
572    let frontier = {
573        let _visibility = window.visibility.read().expect("stream visibility poisoned");
574        window.ring.read().expect("the ring is never poisoned").next
575    };
576
577    Following {
578        parts: Some(Parts { window, doorbell, frontier, closed, held }),
579        step: None,
580        gap: 0,
581        pending: VecDeque::new(),
582        done: false,
583        marker: PhantomData,
584    }
585}
586
587/* === Private Functions === */
588
589fn reaches_beneath(start: Bound<Position>, floor: Position) -> bool {
590    match start {
591        Bound::Included(at) => at <= floor,
592        Bound::Excluded(at) => at < floor,
593        Bound::Unbounded => true,
594    }
595}
596
597fn admitted(start: Bound<Position>, end: Bound<Position>, at: Position) -> bool {
598    let above = match start {
599        Bound::Included(from) => at >= from,
600        Bound::Excluded(from) => at > from,
601        Bound::Unbounded => true,
602    };
603    let below = match end {
604        Bound::Included(to) => at <= to,
605        Bound::Excluded(to) => at < to,
606        Bound::Unbounded => true,
607    };
608
609    above && below
610}
611
612async fn turn(mut parts: Parts) -> (Parts, Turn) {
613    loop {
614        let was_closed = parts.closed.load(Ordering::Acquire);
615        let advanced = parts.window.since(parts.frontier);
616        parts.frontier = advanced.frontier;
617
618        if advanced.missed > 0 || !advanced.rows.is_empty() {
619            return (parts, Turn::Advanced(advanced));
620        }
621        if was_closed || parts.doorbell.changed().await.is_err() {
622            return (parts, Turn::Ended);
623        }
624    }
625}
626
627/* === Tests === */
harmos_runtime/streams/window.rs

harmos_runtime/streams/
window.rs
