uqa_storage/notifications/
publication.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Admitted immutable publication records with borrowed, bounded recovery decoding.
8
9use std::sync::Arc;
10
11use sha2::{Digest, Sha256};
12use uqa_core::memory::BudgetedVec;
13
14use super::{
15    end_position, NotificationListenerRow, PendingNotification, MAX_NOTIFICATION_CHANNEL_BYTES,
16    MAX_NOTIFICATION_PAYLOAD_BYTES,
17};
18use crate::{
19    mvcc::{VersionError, VersionResult},
20    read_control::StorageReadControl,
21};
22
23mod subscription;
24pub use subscription::NotificationSubscriptionView;
25
26const MAGIC: &[u8] = b"UQA notification publication 2\n";
27
28#[derive(Clone, Copy, Debug, PartialEq, Eq)]
29pub struct NotificationPublicationStart {
30    pub registry_id: [u8; 16],
31    pub publication_sequence: u64,
32    pub first_sequence: u64,
33    pub first_position: u64,
34    pub process_id: i32,
35}
36
37#[derive(Clone, Copy, Debug, PartialEq, Eq)]
38pub struct NotificationPublicationHeader {
39    pub registry_id: [u8; 16],
40    pub publication_sequence: u64,
41    pub first_sequence: u64,
42    pub next_sequence: u64,
43    pub first_position: u64,
44    pub end_position: u64,
45    pub process_id: i32,
46}
47
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
49pub struct NotificationMessageRef<'a> {
50    pub channel: &'a str,
51    pub payload: &'a str,
52}
53
54/// One immutable record shares its original admission across clones. Length-framed UTF-8 fields can also occupy native metadata TEXT without escaping or allocating decoded message strings.
55#[derive(Clone)]
56pub struct NotificationPublication {
57    bytes: Arc<BudgetedVec<u8>>,
58    header: NotificationPublicationHeader,
59    messages_offset: usize,
60    subscription: Option<subscription::Subscription>,
61    fingerprint: [u8; 32],
62}
63
64impl NotificationPublication {
65    pub fn encode(
66        start: NotificationPublicationStart,
67        messages: &[PendingNotification],
68        listener: Option<&NotificationListenerRow>,
69        control: &StorageReadControl,
70    ) -> VersionResult<Self> {
71        control.cancellation().check()?;
72        let NotificationPublicationStart {
73            registry_id,
74            publication_sequence,
75            first_sequence,
76            first_position,
77            process_id,
78        } = start;
79        let count = u64::try_from(messages.len()).map_err(|_| invalid())?;
80        validate_header(
81            registry_id,
82            publication_sequence,
83            first_sequence,
84            first_position,
85            process_id,
86            count,
87        )?;
88        let mut bytes = BudgetedVec::new(control.memory());
89        bytes.extend_from_slice(MAGIC)?;
90        for byte in registry_id {
91            bytes.push(b"0123456789abcdef"[usize::from(byte >> 4)])?;
92            bytes.push(b"0123456789abcdef"[usize::from(byte & 15)])?;
93        }
94        bytes.push(b'\n')?;
95        for value in [
96            publication_sequence,
97            first_sequence,
98            first_position,
99            u64::try_from(process_id).map_err(|_| invalid())?,
100            count,
101        ] {
102            write_number(&mut bytes, value, b'\n')?;
103        }
104        for message in messages {
105            control.cancellation().check()?;
106            if message.channel.is_empty()
107                || message.channel.len() >= MAX_NOTIFICATION_CHANNEL_BYTES
108                || message.payload.len() >= MAX_NOTIFICATION_PAYLOAD_BYTES
109            {
110                return Err(invalid());
111            }
112            for field in [&message.channel, &message.payload] {
113                write_number(
114                    &mut bytes,
115                    u64::try_from(field.len()).map_err(|_| invalid())?,
116                    b':',
117                )?;
118                bytes.extend_from_slice(field.as_bytes())?;
119            }
120        }
121        subscription::encode(&mut bytes, listener, process_id, control)?;
122        let view = NotificationPublicationView::decode(&bytes, control)?;
123        let header = view.header;
124        let messages_offset = view.messages_offset;
125        let fingerprint = view.fingerprint();
126        let subscription = view.subscription;
127        Ok(Self {
128            bytes: Arc::new(bytes),
129            header,
130            messages_offset,
131            subscription,
132            fingerprint,
133        })
134    }
135
136    pub fn bytes(&self) -> &[u8] {
137        &self.bytes
138    }
139
140    /// Retain the immutable encoded record and its original memory admission.
141    pub fn shared_bytes(&self) -> Arc<BudgetedVec<u8>> {
142        Arc::clone(&self.bytes)
143    }
144
145    pub const fn header(&self) -> NotificationPublicationHeader {
146        self.header
147    }
148
149    pub const fn fingerprint(&self) -> [u8; 32] {
150        self.fingerprint
151    }
152
153    pub fn view(&self) -> NotificationPublicationView<'_> {
154        NotificationPublicationView {
155            bytes: self.bytes(),
156            header: self.header,
157            messages_offset: self.messages_offset,
158            subscription: self.subscription,
159        }
160    }
161}
162
163/// A validated view borrows its entire record. Recovery can visit one message at a time without materializing a second payload list.
164#[derive(Clone, Copy)]
165pub struct NotificationPublicationView<'a> {
166    bytes: &'a [u8],
167    header: NotificationPublicationHeader,
168    messages_offset: usize,
169    subscription: Option<subscription::Subscription>,
170}
171
172impl<'a> NotificationPublicationView<'a> {
173    pub fn decode(bytes: &'a [u8], control: &StorageReadControl) -> VersionResult<Self> {
174        control.cancellation().check()?;
175        let mut cursor = Cursor {
176            rest: bytes.strip_prefix(MAGIC).ok_or_else(invalid)?,
177        };
178        let mut registry_id = [0; 16];
179        for byte in &mut registry_id {
180            let hex = cursor.take(2)?;
181            *byte = (hex_digit(hex[0])? << 4) | hex_digit(hex[1])?;
182        }
183        if cursor.take(1)? != b"\n" {
184            return Err(invalid());
185        }
186        let publication_sequence = cursor.number(b'\n')?;
187        let first_sequence = cursor.number(b'\n')?;
188        let first_position = cursor.number(b'\n')?;
189        let process_id = i32::try_from(cursor.number(b'\n')?).map_err(|_| invalid())?;
190        let count = cursor.number(b'\n')?;
191        let next_sequence = validate_header(
192            registry_id,
193            publication_sequence,
194            first_sequence,
195            first_position,
196            process_id,
197            count,
198        )?;
199        if count > u64::try_from(cursor.rest.len() / 5).map_err(|_| invalid())? {
200            return Err(invalid());
201        }
202        let messages_offset = bytes.len() - cursor.rest.len();
203        let mut position = first_position;
204        for _ in 0..count {
205            control.cancellation().check()?;
206            let message = cursor.message()?;
207            position = end_position(position, message.channel.len(), message.payload.len());
208            if i64::try_from(position).is_err() {
209                return Err(invalid());
210            }
211        }
212        let subscription = subscription::decode(&mut cursor, bytes.len(), control)?;
213        if !cursor.rest.is_empty() {
214            return Err(invalid());
215        }
216        if subscription.is_some_and(|subscription| {
217            let subscription = subscription.view(bytes);
218            subscription.next_sequence > first_sequence || subscription.position > first_position
219        }) {
220            return Err(invalid());
221        }
222        Ok(Self {
223            bytes,
224            header: NotificationPublicationHeader {
225                registry_id,
226                publication_sequence,
227                first_sequence,
228                next_sequence,
229                first_position,
230                end_position: position,
231                process_id,
232            },
233            messages_offset,
234            subscription,
235        })
236    }
237
238    pub const fn header(&self) -> NotificationPublicationHeader {
239        self.header
240    }
241
242    pub fn fingerprint(&self) -> [u8; 32] {
243        Sha256::digest(self.bytes).into()
244    }
245
246    pub fn subscription(&self) -> Option<NotificationSubscriptionView<'a>> {
247        self.subscription
248            .map(|subscription| subscription.view(self.bytes))
249    }
250
251    pub fn messages(&self) -> impl Iterator<Item = VersionResult<NotificationMessageRef<'a>>> + 'a {
252        let mut cursor = Cursor {
253            rest: &self.bytes[self.messages_offset..],
254        };
255        let mut count = self.header.next_sequence - self.header.first_sequence;
256        std::iter::from_fn(move || {
257            if count == 0 {
258                return None;
259            }
260            count -= 1;
261            Some(cursor.message())
262        })
263    }
264}
265
266fn validate_header(
267    registry: [u8; 16],
268    publication_sequence: u64,
269    sequence: u64,
270    position: u64,
271    process_id: i32,
272    count: u64,
273) -> VersionResult<u64> {
274    let next_sequence = sequence.checked_add(count).ok_or_else(invalid)?;
275    if registry == [0; 16]
276        || process_id <= 0
277        || publication_sequence >= i64::MAX as u64
278        || i64::try_from(next_sequence).is_err()
279        || i64::try_from(position).is_err()
280    {
281        return Err(invalid());
282    }
283    Ok(next_sequence)
284}
285
286fn write_number(output: &mut BudgetedVec<u8>, mut number: u64, separator: u8) -> VersionResult<()> {
287    let mut digits = [0; 20];
288    let mut start = digits.len();
289    loop {
290        start -= 1;
291        digits[start] = b'0' + u8::try_from(number % 10).map_err(|_| invalid())?;
292        number /= 10;
293        if number == 0 {
294            break;
295        }
296    }
297    output.extend_from_slice(&digits[start..])?;
298    output.push(separator)?;
299    Ok(())
300}
301
302struct Cursor<'a> {
303    rest: &'a [u8],
304}
305
306impl<'a> Cursor<'a> {
307    fn take(&mut self, length: usize) -> VersionResult<&'a [u8]> {
308        let (value, rest) = self.rest.split_at_checked(length).ok_or_else(invalid)?;
309        self.rest = rest;
310        Ok(value)
311    }
312
313    fn number(&mut self, separator: u8) -> VersionResult<u64> {
314        let mut number = 0u64;
315        for length in 0..=20 {
316            let byte = *self.rest.get(length).ok_or_else(invalid)?;
317            if byte == separator && length != 0 {
318                self.take(length + 1)?;
319                return Ok(number);
320            }
321            if !byte.is_ascii_digit() || (length > 0 && self.rest[0] == b'0') {
322                return Err(invalid());
323            }
324            number = number
325                .checked_mul(10)
326                .and_then(|value| value.checked_add(u64::from(byte - b'0')))
327                .ok_or_else(invalid)?;
328        }
329        Err(invalid())
330    }
331
332    fn text(&mut self, maximum: usize) -> VersionResult<&'a str> {
333        let length = usize::try_from(self.number(b':')?).map_err(|_| invalid())?;
334        if length >= maximum {
335            return Err(invalid());
336        }
337        std::str::from_utf8(self.take(length)?).map_err(|_| invalid())
338    }
339
340    fn message(&mut self) -> VersionResult<NotificationMessageRef<'a>> {
341        let channel = self.text(MAX_NOTIFICATION_CHANNEL_BYTES)?;
342        if channel.is_empty() {
343            return Err(invalid());
344        }
345        Ok(NotificationMessageRef {
346            channel,
347            payload: self.text(MAX_NOTIFICATION_PAYLOAD_BYTES)?,
348        })
349    }
350}
351
352fn hex_digit(byte: u8) -> VersionResult<u8> {
353    match byte {
354        b'0'..=b'9' => Ok(byte - b'0'),
355        b'a'..=b'f' => Ok(byte - b'a' + 10),
356        _ => Err(invalid()),
357    }
358}
359
360fn invalid() -> VersionError {
361    VersionError::InvalidEncoding("invalid notification publication record")
362}
363
364#[cfg(test)]
365mod tests;
uqa_storage/notifications/publication.rs

uqa_storage/notifications/
publication.rs
