uqa_storage/
notifications.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Notification payloads and bounded queue page accounting.
8
9pub mod conformance;
10mod publication;
11mod registry;
12pub use publication::{
13    NotificationMessageRef, NotificationPublication, NotificationPublicationHeader,
14    NotificationPublicationStart, NotificationPublicationView, NotificationSubscriptionView,
15};
16pub use registry::{NotificationListenerRow, NotificationQueueEntry, NotificationQueueState};
17
18/// Fresh committed publication access is independent of the caller's SQL snapshot. Publication is an auxiliary effect of its current transaction, not a user-data write.
19pub trait NotificationPublicationStore: Send + Sync {
20    fn stage_notification_publication(
21        &self,
22        publication: &NotificationPublication,
23    ) -> crate::StorageBackendResult<()>;
24
25    /// Replace a previous slot only when the registry has already committed that exact acknowledgement. Serialized physical writers may need to carry this cleanup in their own atomic data transaction; the default requires autonomous cleanup first.
26    fn stage_notification_publication_after(
27        &self,
28        publication: &NotificationPublication,
29        acknowledged: Option<[u8; 32]>,
30    ) -> crate::StorageBackendResult<()> {
31        if acknowledged.is_some() {
32            return Err(crate::StorageBackendError::Other(
33                "notification publication requires autonomous acknowledgement cleanup".into(),
34            ));
35        }
36        self.stage_notification_publication(publication)
37    }
38
39    fn visit_notification_publication(
40        &self,
41        control: &crate::read_control::StorageReadControl,
42        visit: &mut dyn FnMut(
43            Option<NotificationPublicationView<'_>>,
44        ) -> crate::StorageBackendResult<()>,
45    ) -> crate::StorageBackendResult<()>;
46
47    /// Clear only the acknowledged fingerprint through autonomous committed state. The caller must serialize queue acknowledgement and new publication and supply uncancelled completion control.
48    fn acknowledge_notification_publication(
49        &self,
50        fingerprint: [u8; 32],
51        control: &crate::read_control::StorageReadControl,
52    ) -> crate::StorageBackendResult<()>;
53
54    /// Return false when an existing physical writer prevents autonomous cleanup. The registry acknowledgement remains authoritative, and a later publisher may replace only that acknowledged fingerprint.
55    fn try_acknowledge_notification_publication(
56        &self,
57        fingerprint: [u8; 32],
58        control: &crate::read_control::StorageReadControl,
59    ) -> crate::StorageBackendResult<bool> {
60        self.acknowledge_notification_publication(fingerprint, control)?;
61        Ok(true)
62    }
63}
64
65pub const MAX_NOTIFICATION_CHANNEL_BYTES: usize = 64;
66pub const MAX_NOTIFICATION_PAYLOAD_BYTES: usize = 8_000;
67pub const NOTIFICATION_QUEUE_PAGE_BYTES: u64 = 8_192;
68pub const MAX_NOTIFICATION_QUEUE_PAGES: u64 = 1_048_576;
69const NOTIFICATION_ENTRY_HEADER_BYTES: u64 = 16;
70const MIN_NOTIFICATION_ENTRY_BYTES: u64 = 20;
71
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub struct PendingNotification {
74    pub channel: String,
75    pub payload: String,
76}
77
78pub fn notification_end_position(position: u64, notification: &PendingNotification) -> u64 {
79    end_position(
80        position,
81        notification.channel.len(),
82        notification.payload.len(),
83    )
84}
85
86fn end_position(position: u64, channel_bytes: usize, payload_bytes: usize) -> u64 {
87    let content = NOTIFICATION_ENTRY_HEADER_BYTES
88        .saturating_add(channel_bytes as u64)
89        .saturating_add(1)
90        .saturating_add(payload_bytes as u64)
91        .saturating_add(1);
92    let length = content.saturating_add(3) & !3;
93    let offset = position % NOTIFICATION_QUEUE_PAGE_BYTES;
94    let aligned_position = if offset.saturating_add(length) > NOTIFICATION_QUEUE_PAGE_BYTES {
95        position.saturating_add(NOTIFICATION_QUEUE_PAGE_BYTES - offset)
96    } else {
97        position
98    };
99    let end = aligned_position.saturating_add(length);
100    let end_offset = end % NOTIFICATION_QUEUE_PAGE_BYTES;
101    if end_offset.saturating_add(MIN_NOTIFICATION_ENTRY_BYTES) > NOTIFICATION_QUEUE_PAGE_BYTES {
102        end.saturating_add(NOTIFICATION_QUEUE_PAGE_BYTES - end_offset)
103    } else {
104        end
105    }
106}
107
108pub fn notifications_fit_queue(
109    mut head: u64,
110    tail: u64,
111    max_queue_pages: u64,
112    pending: &[PendingNotification],
113) -> bool {
114    let tail_page = queue_page(tail);
115    for notification in pending {
116        if queue_page(head).saturating_sub(tail_page) >= max_queue_pages {
117            return false;
118        }
119        let content = NOTIFICATION_ENTRY_HEADER_BYTES
120            .saturating_add(notification.channel.len() as u64)
121            .saturating_add(1)
122            .saturating_add(notification.payload.len() as u64)
123            .saturating_add(1);
124        let length = content.saturating_add(3) & !3;
125        let offset = head % NOTIFICATION_QUEUE_PAGE_BYTES;
126        if offset.saturating_add(length) > NOTIFICATION_QUEUE_PAGE_BYTES {
127            head = head.saturating_add(NOTIFICATION_QUEUE_PAGE_BYTES - offset);
128            if queue_page(head).saturating_sub(tail_page) >= max_queue_pages {
129                return false;
130            }
131        }
132        head = notification_end_position(head, notification);
133    }
134    true
135}
136
137pub fn queue_page(position: u64) -> u64 {
138    position / NOTIFICATION_QUEUE_PAGE_BYTES
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144
145    fn pending(channel_bytes: usize, payload_bytes: usize) -> PendingNotification {
146        PendingNotification {
147            channel: "c".repeat(channel_bytes),
148            payload: "p".repeat(payload_bytes),
149        }
150    }
151
152    #[test]
153    fn queue_layout_accounts_for_alignment_page_padding_and_capacity() {
154        let largest = pending(63, 7_999);
155        let smallest = pending(1, 0);
156        assert_eq!(notification_end_position(0, &largest), 8_080);
157        assert_eq!(notification_end_position(8_080, &smallest), 8_100);
158
159        let mut one_page = vec![largest];
160        one_page.extend(std::iter::repeat_n(smallest.clone(), 5));
161        assert!(notifications_fit_queue(0, 0, 1, &one_page));
162        assert!(!notifications_fit_queue(
163            0,
164            0,
165            1,
166            &[one_page, vec![smallest]].concat()
167        ));
168
169        let entry_that_requires_the_next_page = pending(1, 30);
170        assert!(!notifications_fit_queue(
171            8_160,
172            0,
173            1,
174            &[entry_that_requires_the_next_page]
175        ));
176    }
177}
uqa_storage/notifications.rs

uqa_storage/
notifications.rs
