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, Held, Pool, Refusal, Reservation};
28use super::config::SCAN_ROWS;
29use super::manager::{Domain, ENTRY_OVERHEAD, 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    inner: Arc<Inner>,
48    id: u64,
49    held: Held,
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    _charge: Reservation,
60}
61
62pub(super) struct Following {
63    pub(super) after: u64,
64    failure: Option<Failure>,
65}
66
67/* === Enums === */
68
69#[derive(Clone, Copy)]
70enum Failure {
71    Gap { requested: u64, available: u64 },
72    Closed,
73}
74
75/* === Traits === */
76
77/* === Implementations === */
78
79impl<C> StorageManager<C> {
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, Pool::Read)?;
92            let saved = state.status.saved.clone();
93            let accepted = state.status.accepted.clone();
94            if requested > accepted.sequence {
95                return Err(Error::InvalidInput { field: "storage follow", reason: "start exceeds accepted boundary" });
96            }
97            state.followers.insert(id, Following { after: saved.sequence.max(after), failure: None });
98            (saved, accepted)
99        };
100        let mut follower =
101            Follower { inner: Arc::clone(&self.inner), id, held: Held::default(), selection, history: None, after, saved, accepted };
102        if after < follower.saved.sequence {
103            follower.history = Some(self.snapshot(&follower.saved).await?);
104        }
105        Ok(follower)
106    }
107}
108
109impl<C> StorageManager<C> {
110    pub async fn page<T: Entry>(&self, selection: &T::Selection, after: u64, limits: ReadLimits) -> Result<FollowPage<T>> {
111        let held = Held::default();
112        let rows = limits.rows.min(SCAN_ROWS);
113        let bytes = limits.bytes.min(self.inner.settings.scan_bytes());
114        let charge = rows.get().saturating_mul(ENTRY_OVERHEAD).saturating_add(bytes.get());
115        let mut reservation = Reservation::wait(&self.inner, Pool::Read, &held, charge, None).await.map_err(Refusal::into_error)?;
116        let snapshot = self.snapshot(&self.cursor_saved()).await?;
117        let (entries, encoded) = snapshot.scan::<T>(selection, ScanPage { after, rows, bytes }).await?;
118        reservation.shrink_to(encoded.saturating_add(entries.len().saturating_mul(ENTRY_OVERHEAD)));
119        Ok(FollowPage { entries, _charge: reservation })
120    }
121}
122
123impl<T: Entry> Follower<T> {
124    pub fn saved_boundary(&self) -> &StorageCursor {
125        &self.saved
126    }
127
128    pub fn accepted_boundary(&self) -> &StorageCursor {
129        &self.accepted
130    }
131
132    pub async fn next(&mut self) -> Result<FollowPage<T>> {
133        loop {
134            if let Some(page) = self.next_history().await? {
135                return Ok(page);
136            }
137            if let Some(page) = self.next_live().await? {
138                return Ok(page);
139            }
140        }
141    }
142
143    // Saved History Through The Registration Frontier, From The Pinned Image.
144    async fn next_history(&mut self) -> Result<Option<FollowPage<T>>> {
145        let Some(snapshot) = &self.history else {
146            return Ok(None);
147        };
148        let bytes = self.inner.settings.scan_bytes();
149        let mut reservation =
150            Reservation::wait(&self.inner, Pool::Read, &self.held, page_charge(bytes), None).await.map_err(Refusal::into_error)?;
151        let page = ScanPage { after: self.after, rows: SCAN_ROWS, bytes };
152        let (entries, encoded) = snapshot.scan::<T>(&self.selection, page).await?;
153        let Some(last) = entries.last() else {
154            self.history = None;
155            self.after = self.after.max(self.saved.sequence);
156            return Ok(None);
157        };
158        self.after = last.position().sequence;
159        reservation.shrink_to(encoded.saturating_add(entries.len().saturating_mul(ENTRY_OVERHEAD)));
160        Ok(Some(FollowPage { entries, _charge: reservation }))
161    }
162
163    // Accepted Entries From The Queue; Waits Without Allocating While None Match.
164    async fn next_live(&mut self) -> Result<Option<FollowPage<T>>> {
165        let id = self.id;
166        let selection = &self.selection;
167        self.inner
168            .wait(None, |state| match find::<T>(state, id, selection) {
169                Ok(true) => Attempt::Ready(()),
170                Ok(false) => Attempt::Wait,
171                Err(error) => Attempt::Refuse(Refusal::Failed(error)),
172            })
173            .await
174            .map_err(Refusal::into_error)?;
175
176        // Charge The Page Before Copying Entries Out Of The Queue.
177        let bytes = self.inner.settings.scan_bytes();
178        let mut reservation =
179            Reservation::wait(&self.inner, Pool::Read, &self.held, page_charge(bytes), None).await.map_err(Refusal::into_error)?;
180        let (entries, encoded) = {
181            let mut state = self.inner.state.lock().unwrap();
182            copy::<T>(&mut state, self.id, &self.selection, bytes)?
183        };
184        let Some(last) = entries.last() else {
185            return Ok(None);
186        };
187        self.after = last.position().sequence;
188        reservation.shrink_to(encoded.saturating_add(entries.len().saturating_mul(ENTRY_OVERHEAD)));
189        Ok(Some(FollowPage { entries, _charge: reservation }))
190    }
191}
192
193impl<T: Entry> Drop for Follower<T> {
194    fn drop(&mut self) {
195        let mut state = self.inner.state.lock().unwrap();
196        state.followers.remove(&self.id);
197        state.retire_data();
198        state.collect();
199        drop(state);
200        self.inner.notify();
201    }
202}
203
204impl<T> FollowPage<T> {
205    pub fn entries(&self) -> &[T] {
206        &self.entries
207    }
208}
209
210impl Following {
211    fn retains(&self) -> bool {
212        self.failure.is_none()
213    }
214
215    fn check(&self) -> Result<()> {
216        match self.failure {
217            Some(Failure::Gap { requested, available }) => Err(Error::Gap { requested, available }),
218            Some(Failure::Closed) => Err(Error::Closing { domain: "storage" }),
219            None => Ok(()),
220        }
221    }
222}
223
224impl State {
225    pub(super) fn close_followers(&mut self) {
226        for following in self.followers.values_mut() {
227            following.failure.get_or_insert(Failure::Closed);
228        }
229        self.retire_data();
230    }
231
232    pub(super) fn retire_data(&mut self) {
233        let through =
234            self.followers.values().filter(|following| following.retains()).map(|following| following.after).min().unwrap_or(u64::MAX);
235        self.queue.retire_data(self.data_flushed, through);
236    }
237
238    pub(super) fn lag_followers(&mut self) -> bool {
239        let Some(through) = self
240            .queue
241            .iter()
242            .filter(|entry| entry.operation.domain() == Domain::Data && entry.index <= self.data_flushed)
243            .map(|entry| entry.operation.sequence())
244            .max()
245        else {
246            return false;
247        };
248        let mut changed = false;
249        for following in self.followers.values_mut() {
250            if following.retains() && following.after < through {
251                following.failure =
252                    Some(Failure::Gap { requested: following.after.saturating_add(1), available: through.saturating_add(1) });
253                changed = true;
254            }
255        }
256        if changed {
257            self.retire_data();
258            self.collect();
259        }
260        changed
261    }
262}
263
264/* === Public Functions === */
265
266/* === Private Functions === */
267
268fn page_charge(bytes: NonZeroUsize) -> usize {
269    SCAN_ROWS.get().saturating_mul(ENTRY_OVERHEAD).saturating_add(bytes.get())
270}
271
272// Whether A Matching Queued Entry Follows The Subscription Cursor. Queue
273// Entries Without A Match Advance The Cursor So They Can Be Retired.
274fn find<T: Entry>(state: &mut State, id: u64, selection: &T::Selection) -> Result<bool> {
275    super::capacity::admitting(state, Pool::Read)?;
276    let following = state.followers.get(&id).ok_or(Error::Closing { domain: "storage follow" })?;
277    following.check()?;
278    let mut cursor = following.after;
279    let mut found = false;
280    for pending in state.queue.iter().filter(|entry| entry.operation.domain() == Domain::Data) {
281        if pending.operation.sequence() <= cursor {
282            continue;
283        }
284        let entries = T::pending(Queued::new(&pending.operation));
285        let start = entries.partition_point(|entry| entry.position().sequence <= cursor);
286        if entries[start..].iter().any(|entry| entry.matches(selection)) {
287            found = true;
288            break;
289        }
290        cursor = pending.operation.sequence();
291    }
292    let following = state.followers.get_mut(&id).expect("registered follower");
293    if cursor > following.after {
294        following.after = cursor;
295        state.retire_data();
296        state.collect();
297    }
298    Ok(found)
299}
300
301// Copy Matching Queued Entries After The Cursor Within The Page Bounds.
302fn copy<T: Entry>(state: &mut State, id: u64, selection: &T::Selection, bytes: NonZeroUsize) -> Result<(Vec<T>, usize)> {
303    let following = state.followers.get(&id).ok_or(Error::Closing { domain: "storage follow" })?;
304    following.check()?;
305    let mut cursor = following.after;
306    let mut entries = Vec::new();
307    let mut encoded = 0usize;
308    'queue: for pending in state.queue.iter().filter(|entry| entry.operation.domain() == Domain::Data) {
309        if pending.operation.sequence() <= cursor {
310            continue;
311        }
312        let values = T::pending(Queued::new(&pending.operation));
313        let start = values.partition_point(|entry| entry.position().sequence <= cursor);
314        for entry in &values[start..] {
315            if entries.len() >= SCAN_ROWS.get() {
316                break 'queue;
317            }
318            if entry.matches(selection) {
319                let size = encoded_size(entry)?;
320                if !entries.is_empty() && encoded.saturating_add(size) > bytes.get() {
321                    break 'queue;
322                }
323                encoded = encoded.saturating_add(size);
324                entries.push(entry.clone());
325            }
326            cursor = entry.position().sequence;
327        }
328        cursor = cursor.max(pending.operation.sequence());
329    }
330    let following = state.followers.get_mut(&id).expect("registered follower");
331    following.after = following.after.max(cursor);
332    state.retire_data();
333    state.collect();
334    Ok((entries, encoded))
335}
336
337/* === Tests === */
harmos_storage/follow.rs

harmos_storage/
follow.rs
