uqa_storage/mvcc/
notifications.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Materialize one auxiliary publication slot against fresh committed state without replaying evaluated user writes.
8
9use sha2::{Digest, Sha256};
10use std::sync::Arc;
11use uqa_core::memory::BudgetedVec;
12
13use super::{
14    key::RecordKey, CommittedRecordSnapshot, PreparedRecordCommit, PreparedRecordWrite,
15    SharedRecordValue, VersionError, VersionResult,
16};
17use crate::{
18    notifications::{NotificationPublication, NotificationPublicationView},
19    read_control::StorageReadControl,
20};
21
22pub const NOTIFICATION_PUBLICATION_KEY: &str = "_uqa_notification_publication";
23
24/// Provider-owned metadata addressing and lossless row encoding. Native adapters must use their stable data namespace, independent of restored transaction history.
25pub trait NotificationRecordLayout: Send + Sync {
26    fn key(&self, control: &StorageReadControl) -> VersionResult<BudgetedVec<u8>>;
27    fn encode(
28        &self,
29        publication: &NotificationPublication,
30        control: &StorageReadControl,
31    ) -> VersionResult<SharedRecordValue>;
32    fn decode<'a>(
33        &self,
34        key: &[u8],
35        record: &'a [u8],
36        control: &StorageReadControl,
37    ) -> VersionResult<NotificationPublicationView<'a>>;
38}
39
40pub(super) struct NotificationEffect {
41    key: RecordKey,
42    record: Option<SharedRecordValue>,
43    publication: [u8; 32],
44}
45
46impl NotificationEffect {
47    pub(super) fn publish(
48        publication: &NotificationPublication,
49        layout: &dyn NotificationRecordLayout,
50        control: &StorageReadControl,
51    ) -> VersionResult<Arc<Self>> {
52        control.check()?;
53        Ok(Arc::new(Self {
54            key: RecordKey::new(&layout.key(control)?, control.memory())?,
55            record: Some(layout.encode(publication, control)?),
56            publication: publication.fingerprint(),
57        }))
58    }
59
60    pub(super) fn acknowledge(
61        publication: [u8; 32],
62        layout: &dyn NotificationRecordLayout,
63        control: &StorageReadControl,
64    ) -> VersionResult<Arc<Self>> {
65        control.check()?;
66        Ok(Arc::new(Self {
67            key: RecordKey::new(&layout.key(control)?, control.memory())?,
68            record: None,
69            publication,
70        }))
71    }
72
73    pub(super) fn same_publication(&self, other: &Self) -> bool {
74        self.key.bytes() == other.key.bytes()
75            && self.record.is_some() == other.record.is_some()
76            && self.publication == other.publication
77    }
78
79    pub(super) fn seal(&self, original: super::CommitFingerprint) -> super::CommitFingerprint {
80        let mut digest = Sha256::new();
81        digest.update(b"UQA prepared notification effect 1");
82        digest.update(original);
83        digest.update((self.key.bytes().len() as u64).to_be_bytes());
84        digest.update(self.key.bytes());
85        digest.update([u8::from(self.record.is_some())]);
86        digest.update(self.publication);
87        digest.finalize().into()
88    }
89}
90
91pub(super) fn resolve(
92    original: &PreparedRecordCommit,
93    effect: &NotificationEffect,
94    current: &dyn CommittedRecordSnapshot,
95    layout: &dyn NotificationRecordLayout,
96    control: &StorageReadControl,
97) -> VersionResult<PreparedRecordCommit> {
98    control.check()?;
99    let key = effect.key.bytes();
100    if original.contains_key(key, control)? {
101        return Err(VersionError::InvalidEncoding(
102            "notification publication conflicts with an ordinary record write",
103        ));
104    }
105    let latest = current.get(key, control)?;
106    let pending = latest
107        .as_ref()
108        .and_then(super::RecordVersion::value)
109        .map(|value| layout.decode(key, value, control))
110        .transpose()?;
111    let replace = if effect.record.is_some() {
112        if pending.is_some() {
113            return Err(VersionError::InvalidEncoding(
114                "committed notification publication must be acknowledged before replacement",
115            ));
116        }
117        true
118    } else {
119        pending.is_some_and(|publication| publication.fingerprint() == effect.publication)
120    };
121    // The publication takes its place in key order, which a spilled batch keeps.
122    let mut publication = replace.then(|| {
123        PreparedRecordWrite::from_shared(
124            effect.key.clone(),
125            latest.as_ref().map(super::RecordVersion::sequence),
126            effect.record.clone(),
127        )
128    });
129    let mut writes = super::commit::PreparedWritesBuilder::like(original, control)?;
130    let mut originals = original.writes();
131    while let Some(write) = originals.next(control)? {
132        control.check()?;
133        if write.key() > key {
134            if let Some(publication) = publication.take() {
135                writes.push(publication, control)?;
136            }
137        }
138        writes.push(write, control)?;
139    }
140    if let Some(publication) = publication {
141        writes.push(publication, control)?;
142    }
143    Ok(writes
144        .finish(control)?
145        .resolved(original, current.sequence()))
146}
uqa_storage/mvcc/notifications.rs

uqa_storage/mvcc/
notifications.rs
