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 listener_summary;
11mod publication;
12mod registry;
13pub use listener_summary::{NotificationListenerSummary, NotificationWakePorts};
14pub use publication::{
15    NotificationMessageRef, NotificationPublication, NotificationPublicationHeader,
16    NotificationPublicationStart, NotificationPublicationView, NotificationSubscriptionView,
17};
18pub use registry::{
19    NotificationListenerKey, NotificationListenerMetadata, NotificationListenerRow,
20    NotificationQueueEntry, NotificationQueueEntryRef, NotificationQueueScan,
21    NotificationQueueState,
22};
23
24/// 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.
25pub trait NotificationPublicationStore: Send + Sync {
26    fn stage_notification_publication(
27        &self,
28        publication: &NotificationPublication,
29    ) -> crate::StorageBackendResult<()>;
30
31    /// 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.
32    fn stage_notification_publication_after(
33        &self,
34        publication: &NotificationPublication,
35        acknowledged: Option<[u8; 32]>,
36    ) -> crate::StorageBackendResult<()> {
37        if acknowledged.is_some() {
38            return Err(crate::StorageBackendError::Other(
39                "notification publication requires autonomous acknowledgement cleanup".into(),
40            ));
41        }
42        self.stage_notification_publication(publication)
43    }
44
45    fn visit_notification_publication(
46        &self,
47        control: &crate::read_control::StorageReadControl,
48        visit: &mut dyn FnMut(
49            Option<NotificationPublicationView<'_>>,
50        ) -> crate::StorageBackendResult<()>,
51    ) -> crate::StorageBackendResult<()>;
52
53    /// Clear only the acknowledged fingerprint through autonomous committed state. The caller must serialize queue acknowledgement and new publication and supply uncancelled completion control.
54    fn acknowledge_notification_publication(
55        &self,
56        fingerprint: [u8; 32],
57        control: &crate::read_control::StorageReadControl,
58    ) -> crate::StorageBackendResult<()>;
59
60    /// 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.
61    fn try_acknowledge_notification_publication(
62        &self,
63        fingerprint: [u8; 32],
64        control: &crate::read_control::StorageReadControl,
65    ) -> crate::StorageBackendResult<bool> {
66        self.acknowledge_notification_publication(fingerprint, control)?;
67        Ok(true)
68    }
69}
70
71pub const MAX_NOTIFICATION_CHANNEL_BYTES: usize = 64;
72pub const MAX_NOTIFICATION_PAYLOAD_BYTES: usize = 8_000;
73pub const NOTIFICATION_QUEUE_PAGE_BYTES: u64 = 8_192;
74pub const MAX_NOTIFICATION_QUEUE_PAGES: u64 = 1_048_576;
75const NOTIFICATION_ENTRY_HEADER_BYTES: u64 = 16;
76const MIN_NOTIFICATION_ENTRY_BYTES: u64 = 20;
77
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct PendingNotification {
80    pub channel: String,
81    pub payload: String,
82}
83
84pub fn notification_end_position(position: u64, notification: &PendingNotification) -> u64 {
85    notification_parts_end_position(position, &notification.channel, &notification.payload)
86}
87
88/// Advance physical queue accounting from borrowed content without copying a pending notification.
89pub fn notification_parts_end_position(position: u64, channel: &str, payload: &str) -> u64 {
90    end_position(position, channel.len(), payload.len())
91}
92
93fn end_position(position: u64, channel_bytes: usize, payload_bytes: usize) -> u64 {
94    let content = NOTIFICATION_ENTRY_HEADER_BYTES
95        .saturating_add(channel_bytes as u64)
96        .saturating_add(1)
97        .saturating_add(payload_bytes as u64)
98        .saturating_add(1);
99    let length = content.saturating_add(3) & !3;
100    let offset = position % NOTIFICATION_QUEUE_PAGE_BYTES;
101    let aligned_position = if offset.saturating_add(length) > NOTIFICATION_QUEUE_PAGE_BYTES {
102        position.saturating_add(NOTIFICATION_QUEUE_PAGE_BYTES - offset)
103    } else {
104        position
105    };
106    let end = aligned_position.saturating_add(length);
107    let end_offset = end % NOTIFICATION_QUEUE_PAGE_BYTES;
108    if end_offset.saturating_add(MIN_NOTIFICATION_ENTRY_BYTES) > NOTIFICATION_QUEUE_PAGE_BYTES {
109        end.saturating_add(NOTIFICATION_QUEUE_PAGE_BYTES - end_offset)
110    } else {
111        end
112    }
113}
114
115pub fn notifications_fit_queue(
116    mut head: u64,
117    tail: u64,
118    max_queue_pages: u64,
119    pending: &[PendingNotification],
120) -> bool {
121    let tail_page = queue_page(tail);
122    for notification in pending {
123        if queue_page(head).saturating_sub(tail_page) >= max_queue_pages {
124            return false;
125        }
126        let content = NOTIFICATION_ENTRY_HEADER_BYTES
127            .saturating_add(notification.channel.len() as u64)
128            .saturating_add(1)
129            .saturating_add(notification.payload.len() as u64)
130            .saturating_add(1);
131        let length = content.saturating_add(3) & !3;
132        let offset = head % NOTIFICATION_QUEUE_PAGE_BYTES;
133        if offset.saturating_add(length) > NOTIFICATION_QUEUE_PAGE_BYTES {
134            head = head.saturating_add(NOTIFICATION_QUEUE_PAGE_BYTES - offset);
135            if queue_page(head).saturating_sub(tail_page) >= max_queue_pages {
136                return false;
137            }
138        }
139        head = notification_end_position(head, notification);
140    }
141    true
142}
143
144pub fn queue_page(position: u64) -> u64 {
145    position / NOTIFICATION_QUEUE_PAGE_BYTES
146}
147
148#[cfg(test)]
149mod tests {
150    use super::*;
151
152    fn pending(channel_bytes: usize, payload_bytes: usize) -> PendingNotification {
153        PendingNotification {
154            channel: "c".repeat(channel_bytes),
155            payload: "p".repeat(payload_bytes),
156        }
157    }
158
159    #[test]
160    fn queue_layout_accounts_for_alignment_page_padding_and_capacity() {
161        let largest = pending(63, 7_999);
162        let smallest = pending(1, 0);
163        assert_eq!(notification_end_position(0, &largest), 8_080);
164        assert_eq!(notification_end_position(8_080, &smallest), 8_100);
165
166        let mut one_page = vec![largest];
167        one_page.extend(std::iter::repeat_n(smallest.clone(), 5));
168        assert!(notifications_fit_queue(0, 0, 1, &one_page));
169        assert!(!notifications_fit_queue(
170            0,
171            0,
172            1,
173            &[one_page, vec![smallest]].concat()
174        ));
175
176        let entry_that_requires_the_next_page = pending(1, 30);
177        assert!(!notifications_fit_queue(
178            8_160,
179            0,
180            1,
181            &[entry_that_requires_the_next_page]
182        ));
183    }
184}
uqa_storage/notifications.rs

uqa_storage/
notifications.rs
