acyclic_fs/
notification.rs

1//! Disposable authority-head notification acceleration.
2//!
3//! Notifications are hints only. They can suppress unnecessary authority
4//! polls or wake a tracking checkout, but cannot establish authority, fill a
5//! replay gap, or permit publication. Every consumer must authenticate the
6//! indicated head through [`crate::AsyncAuthorityStore`].
7
8use crate::cancellation::CancellationToken;
9use crate::foundation::{AuthorityId, Head};
10use crate::performance::{OperationFailure, OperationReceipt, WorkBudget, WorkCounters, WorkError};
11use std::collections::HashMap;
12use std::future::Future;
13use std::sync::Mutex;
14use thiserror::Error;
15
16/// Result of polling a disposable notification cursor.
17#[derive(Clone, Copy, Debug, Eq, PartialEq)]
18pub enum NotificationPoll {
19    /// No newer hint is currently visible; callers may poll authority directly.
20    Unchanged,
21    /// A possibly newer authority head should be authenticated and replayed.
22    Advanced(Head),
23}
24
25/// Notification adapter failures never alter filesystem truth.
26#[derive(Debug, Error)]
27pub enum NotificationError {
28    /// Cooperative cancellation occurred before adapter access.
29    #[error("notification operation was cancelled")]
30    Cancelled,
31    /// Adapter synchronization state is unavailable.
32    #[error("notification adapter state is unavailable")]
33    Unavailable,
34    /// Exact work overflowed or exceeded the admitted budget.
35    #[error(transparent)]
36    Work(#[from] WorkError),
37}
38
39/// Receipt-bearing notification result.
40pub type NotificationResult<T> = Result<OperationReceipt<T>, OperationFailure<NotificationError>>;
41
42/// Optional synchronous authority-head hint adapter.
43///
44/// Implementations may coalesce, duplicate, delay, or lose hints. They must
45/// never invent a head not supplied to [`Self::publish`], and returning
46/// [`NotificationPoll::Unchanged`] must never be interpreted as proof that the
47/// authority head is unchanged.
48pub trait NotificationStore {
49    /// Publishes a disposable monotonic hint.
50    ///
51    /// # Errors
52    ///
53    /// Returns typed adapter or work-budget failure without changing truth.
54    fn publish(
55        &self,
56        authority_id: AuthorityId,
57        head: Head,
58        budget: WorkBudget,
59    ) -> NotificationResult<()>;
60
61    /// Polls for a hint strictly newer than `after` in epoch/sequence order.
62    ///
63    /// # Errors
64    ///
65    /// Returns typed adapter or work-budget failure without changing truth.
66    fn poll_after(
67        &self,
68        authority_id: AuthorityId,
69        after: Head,
70        budget: WorkBudget,
71    ) -> NotificationResult<NotificationPoll>;
72}
73
74/// Runtime-neutral notification adapter suitable for browser and remote APIs.
75pub trait AsyncNotificationStore {
76    /// Asynchronously publishes one disposable head hint.
77    ///
78    /// # Errors
79    ///
80    /// Returns cancellation, adapter, or exact work-budget failure.
81    fn publish_hint(
82        &self,
83        authority_id: AuthorityId,
84        head: Head,
85        budget: WorkBudget,
86        cancellation: &CancellationToken,
87    ) -> impl Future<Output = NotificationResult<()>>;
88
89    /// Asynchronously polls for a possibly newer hint.
90    ///
91    /// # Errors
92    ///
93    /// Returns cancellation, adapter, or exact work-budget failure.
94    fn poll_hint_after(
95        &self,
96        authority_id: AuthorityId,
97        after: Head,
98        budget: WorkBudget,
99        cancellation: &CancellationToken,
100    ) -> impl Future<Output = NotificationResult<NotificationPoll>>;
101}
102
103/// Explicit opt-in for synchronous notification stores that complete inline.
104pub trait ImmediateNotificationStore: NotificationStore {}
105
106impl<T: ImmediateNotificationStore + ?Sized> AsyncNotificationStore for T {
107    async fn publish_hint(
108        &self,
109        authority_id: AuthorityId,
110        head: Head,
111        budget: WorkBudget,
112        cancellation: &CancellationToken,
113    ) -> NotificationResult<()> {
114        cancellation
115            .check()
116            .map_err(|_| OperationFailure::before_work(NotificationError::Cancelled))?;
117        NotificationStore::publish(self, authority_id, head, budget)
118    }
119
120    async fn poll_hint_after(
121        &self,
122        authority_id: AuthorityId,
123        after: Head,
124        budget: WorkBudget,
125        cancellation: &CancellationToken,
126    ) -> NotificationResult<NotificationPoll> {
127        cancellation
128            .check()
129            .map_err(|_| OperationFailure::before_work(NotificationError::Cancelled))?;
130        NotificationStore::poll_after(self, authority_id, after, budget)
131    }
132}
133
134/// Deterministic process-local coalescing notification adapter.
135#[derive(Default)]
136pub struct MemoryNotificationStore {
137    heads: Mutex<HashMap<AuthorityId, Head>>,
138}
139
140impl MemoryNotificationStore {
141    /// Creates an empty adapter.
142    #[must_use]
143    pub fn new() -> Self {
144        Self::default()
145    }
146}
147
148impl NotificationStore for MemoryNotificationStore {
149    fn publish(
150        &self,
151        authority_id: AuthorityId,
152        head: Head,
153        budget: WorkBudget,
154    ) -> NotificationResult<()> {
155        let work = WorkCounters {
156            backend_write_operations: 1,
157            items_examined: 1,
158            ..WorkCounters::default()
159        };
160        work.verify(budget)
161            .map_err(|error| OperationFailure::before_work(error.into()))?;
162        let mut heads = self
163            .heads
164            .lock()
165            .map_err(|_| OperationFailure::new(NotificationError::Unavailable, work))?;
166        match heads.get(&authority_id) {
167            Some(current) if !head_is_newer(head, *current) => {}
168            _ => {
169                heads.insert(authority_id, head);
170            }
171        }
172        Ok(OperationReceipt { value: (), work })
173    }
174
175    fn poll_after(
176        &self,
177        authority_id: AuthorityId,
178        after: Head,
179        budget: WorkBudget,
180    ) -> NotificationResult<NotificationPoll> {
181        let work = WorkCounters {
182            backend_read_operations: 1,
183            items_examined: 1,
184            ..WorkCounters::default()
185        };
186        work.verify(budget)
187            .map_err(|error| OperationFailure::before_work(error.into()))?;
188        let heads = self
189            .heads
190            .lock()
191            .map_err(|_| OperationFailure::new(NotificationError::Unavailable, work))?;
192        let value = heads
193            .get(&authority_id)
194            .copied()
195            .filter(|head| head_is_newer(*head, after))
196            .map_or(NotificationPoll::Unchanged, NotificationPoll::Advanced);
197        Ok(OperationReceipt { value, work })
198    }
199}
200
201impl ImmediateNotificationStore for MemoryNotificationStore {}
202
203fn head_is_newer(candidate: Head, current: Head) -> bool {
204    candidate.epoch > current.epoch
205        || (candidate.epoch == current.epoch && candidate.sequence > current.sequence)
206}
207
208#[cfg(test)]
209#[path = "tests/notification.rs"]
210mod tests;
acyclic_fs/notification.rs

acyclic_fs/
notification.rs
