harmos_storage/
follow.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Live Storage Subscriptions
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::num::NonZeroUsize;
20use std::sync::Arc;
21use std::sync::atomic::{AtomicU64, Ordering};
22
23// Third Party Imports
24
25// Local Imports
26use super::backend::{ReadLimits, ScanPage};
27use super::capacity::{Attempt, Refusal};
28use super::config::SCAN_ROWS;
29use super::manager::{Domain, Inner, State, encoded_size};
30pub use super::selection::Entry;
31use super::selection::Queued;
32use super::snapshot::Snapshot;
33use super::{StorageCursor, StorageManager};
34use crate::{Error, Result};
35
36/* === Constants === */
37
38static NEXT_FOLLOWER: AtomicU64 = AtomicU64::new(1);
39
40/* === Macros === */
41
42/* === Type Aliases === */
43
44/* === Structs === */
45
46pub struct Follower<T: Entry> {
47    context: crate::execution::Context,
48    inner: Arc<Inner>,
49    id: u64,
50    selection: T::Selection,
51    history: Option<Snapshot>,
52    after: u64,
53    saved: StorageCursor,
54    accepted: StorageCursor,
55}
56
57pub struct FollowPage<T> {
58    entries: Vec<T>,
59}
60
61pub(super) struct Following {
62    pub(super) after: u64,
63    failure: Option<Failure>,
64}
65
66/* === Enums === */
67
68#[derive(Clone, Copy)]
69enum Failure {
70    Gap { requested: u64, available: u64 },
71    Closed,
72}
73
74/* === Traits === */
75
76/* === Implementations === */
77
78impl<C> StorageManager<C> {
79    #[tracing::instrument(name = "storage.follow.follow", level = "debug", skip_all)]
80    pub async fn follow<T: Entry>(&self, after: Option<&StorageCursor>, selection: T::Selection) -> Result<Follower<T>> {
81        if let Some(after) = after {
82            self.inner.primary.history().check(&after.history)?;
83        }
84        let requested = after.map_or(0, |after| after.sequence);
85        let after = requested.max(T::start(&selection));
86        let id = NEXT_FOLLOWER.fetch_add(1, Ordering::Relaxed);
87
88        // Register Queue Retention From The Saved Frontier Atomically.
89        let (saved, accepted) = {
90            let mut state = self.inner.state.lock().unwrap();
91            super::capacity::admitting(&state, false)?;
92            let saved = state.status.saved.clone();
93            let accepted = state.status.accepted.clone();
94            if requested > accepted.sequence {
95                return Err(Error::InvalidInput {
96                    field: "storage follow",
97                    reason: "start exceeds accepted boundary",
98                });
99            }
100            state.followers.insert(
101                id,
102                Following {
103                    after: saved.sequence.max(after),
104                    failure: None,
105                },
106            );
107            (saved, accepted)
108        };
109        let mut follower = Follower {
110            context: crate::execution::Context::current(),
111            inner: Arc::clone(&self.inner),
112            id,
113            selection,
114            history: None,
115            after,
116            saved,
117            accepted,
118        };
119        if after < follower.saved.sequence {
120            follower.history = Some(self.snapshot(&follower.saved).await?);
121        }
122        Ok(follower)
123    }
124}
125
126impl<C> StorageManager<C> {
127    #[tracing::instrument(name = "storage.follow.page", level = "debug", skip_all)]
128    pub async fn page<T: Entry>(&self, selection: &T::Selection, after: u64, limits: ReadLimits) -> Result<FollowPage<T>> {
129        let rows = limits.rows.min(SCAN_ROWS);
130        let bytes = limits.bytes.min(self.inner.settings.scan_bytes());
131
132        let snapshot = self.snapshot(&self.cursor_saved()).await?;
133        let (entries, _) = snapshot.scan::<T>(selection, ScanPage { after, rows, bytes }).await?;
134        Ok(FollowPage { entries })
135    }
136}
137
138impl<T: Entry> Follower<T> {
139    pub fn saved_boundary(&self) -> &StorageCursor {
140        &self.saved
141    }
142
143    pub fn accepted_boundary(&self) -> &StorageCursor {
144        &self.accepted
145    }
146
147    pub async fn next(&mut self) -> Result<FollowPage<T>> {
148        let context = self.context.clone();
149        let span = context.sync(|| tracing::trace_span!("storage.reader.page"));
150        let future = async {
151            loop {
152                if let Some(page) = self.next_history().await? {
153                    return Ok(page);
154                }
155                if let Some(page) = self.next_live().await? {
156                    return Ok(page);
157                }
158            }
159        };
160        context.scope(tracing::Instrument::instrument(future, span)).await
161    }
162
163    // Saved History Through The Registration Frontier, From The Pinned Image.
164    #[tracing::instrument(name = "storage.follow.next_history", level = "debug", skip_all)]
165    async fn next_history(&mut self) -> Result<Option<FollowPage<T>>> {
166        let Some(snapshot) = &self.history else {
167            return Ok(None);
168        };
169        let bytes = self.inner.settings.scan_bytes();
170
171        let page = ScanPage {
172            after: self.after,
173            rows: SCAN_ROWS,
174            bytes,
175        };
176        let (entries, _) = snapshot.scan::<T>(&self.selection, page).await?;
177        let Some(last) = entries.last() else {
178            self.history = None;
179            self.after = self.after.max(self.saved.sequence);
180            return Ok(None);
181        };
182        self.after = last.position().sequence;
183        Ok(Some(FollowPage { entries }))
184    }
185
186    // Accepted Entries From The Queue; Waits Without Allocating While None Match.
187    #[tracing::instrument(name = "storage.follow.next_live", level = "debug", skip_all)]
188    async fn next_live(&mut self) -> Result<Option<FollowPage<T>>> {
189        let id = self.id;
190        let selection = &self.selection;
191        self.inner
192            .wait(None, |state| match find::<T>(state, id, selection) {
193                Ok(true) => Attempt::Ready(()),
194                Ok(false) => Attempt::Wait,
195                Err(error) => Attempt::Refuse(Refusal::Failed(error)),
196            })
197            .await
198            .map_err(Refusal::into_error)?;
199
200        // Charge The Page Before Copying Entries Out Of The Queue.
201        let bytes = self.inner.settings.scan_bytes();
202
203        let (entries, _) = {
204            let mut state = self.inner.state.lock().unwrap();
205            copy::<T>(&mut state, self.id, &self.selection, bytes)?
206        };
207        let Some(last) = entries.last() else {
208            return Ok(None);
209        };
210        self.after = last.position().sequence;
211        Ok(Some(FollowPage { entries }))
212    }
213}
214
215impl<T: Entry> Drop for Follower<T> {
216    fn drop(&mut self) {
217        let mut state = self.inner.state.lock().unwrap();
218        state.followers.remove(&self.id);
219        state.retire_data();
220        state.collect();
221        drop(state);
222        self.inner.notify();
223    }
224}
225
226impl<T> FollowPage<T> {
227    pub fn entries(&self) -> &[T] {
228        &self.entries
229    }
230}
231
232impl Following {
233    fn retains(&self) -> bool {
234        self.failure.is_none()
235    }
236
237    fn check(&self) -> Result<()> {
238        match self.failure {
239            Some(Failure::Gap { requested, available }) => Err(Error::Gap { requested, available }),
240            Some(Failure::Closed) => Err(Error::Closing { domain: "storage" }),
241            None => Ok(()),
242        }
243    }
244}
245
246impl State {
247    pub(super) fn close_followers(&mut self) {
248        for following in self.followers.values_mut() {
249            following.failure.get_or_insert(Failure::Closed);
250        }
251        self.retire_data();
252    }
253
254    pub(super) fn retire_data(&mut self) {
255        let through =
256            self.followers.values().filter(|following| following.retains()).map(|following| following.after).min().unwrap_or(u64::MAX);
257        self.queue.retire_data(self.data_flushed, through);
258    }
259
260    pub(super) fn lag_followers(&mut self) -> bool {
261        let Some(through) = self
262            .queue
263            .iter()
264            .filter(|entry| entry.operation.domain() == Domain::Data && entry.index <= self.data_flushed)
265            .map(|entry| entry.operation.sequence())
266            .max()
267        else {
268            return false;
269        };
270        let mut changed = false;
271        for following in self.followers.values_mut() {
272            if following.retains() && following.after < through {
273                following.failure = Some(Failure::Gap {
274                    requested: following.after.saturating_add(1),
275                    available: through.saturating_add(1),
276                });
277                changed = true;
278            }
279        }
280        if changed {
281            self.retire_data();
282            self.collect();
283        }
284        changed
285    }
286}
287
288/* === Public Functions === */
289
290/* === Private Functions === */
291
292// Whether A Matching Queued Entry Follows The Subscription Cursor. Queue
293// Entries Without A Match Advance The Cursor So They Can Be Retired.
294fn find<T: Entry>(state: &mut State, id: u64, selection: &T::Selection) -> Result<bool> {
295    super::capacity::admitting(state, false)?;
296    let following = state.followers.get(&id).ok_or(Error::Closing { domain: "storage follow" })?;
297    following.check()?;
298    let mut cursor = following.after;
299    let mut found = false;
300    for pending in state.queue.iter().filter(|entry| entry.operation.domain() == Domain::Data) {
301        if pending.operation.sequence() <= cursor {
302            continue;
303        }
304        let entries = T::pending(Queued::new(&pending.operation));
305        let start = entries.partition_point(|entry| entry.position().sequence <= cursor);
306        if entries[start..].iter().any(|entry| entry.matches(selection)) {
307            found = true;
308            break;
309        }
310        cursor = pending.operation.sequence();
311    }
312    let following = state.followers.get_mut(&id).expect("registered follower");
313    if cursor > following.after {
314        following.after = cursor;
315        state.retire_data();
316        state.collect();
317    }
318    Ok(found)
319}
320
321// Copy Matching Queued Entries After The Cursor Within The Page Bounds.
322fn copy<T: Entry>(state: &mut State, id: u64, selection: &T::Selection, bytes: NonZeroUsize) -> Result<(Vec<T>, usize)> {
323    let following = state.followers.get(&id).ok_or(Error::Closing { domain: "storage follow" })?;
324    following.check()?;
325    let mut cursor = following.after;
326    let mut entries = Vec::new();
327    let mut encoded = 0usize;
328    'queue: for pending in state.queue.iter().filter(|entry| entry.operation.domain() == Domain::Data) {
329        if pending.operation.sequence() <= cursor {
330            continue;
331        }
332        let values = T::pending(Queued::new(&pending.operation));
333        let start = values.partition_point(|entry| entry.position().sequence <= cursor);
334        for entry in &values[start..] {
335            if entries.len() >= SCAN_ROWS.get() {
336                break 'queue;
337            }
338            if entry.matches(selection) {
339                let size = encoded_size(entry)?;
340                if !entries.is_empty() && encoded.saturating_add(size) > bytes.get() {
341                    break 'queue;
342                }
343                encoded = encoded.saturating_add(size);
344                entries.push(entry.clone());
345            }
346            cursor = entry.position().sequence;
347        }
348        cursor = cursor.max(pending.operation.sequence());
349    }
350    let following = state.followers.get_mut(&id).expect("registered follower");
351    following.after = following.after.max(cursor);
352    state.retire_data();
353    state.collect();
354    Ok((entries, encoded))
355}
356
357/* === Tests === */
harmos_storage/follow.rs

harmos_storage/
follow.rs
