uqa_storage/mvcc/session/
notifications.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Notification staging shares the data outcome; recovery and acknowledgement use fresh autonomous state.
8
9use super::{no_transaction, Transaction, VersionedKeyValueStore};
10use crate::{
11    mvcc::{notifications::NotificationEffect, VersionError},
12    notifications::{
13        NotificationPublication, NotificationPublicationStore, NotificationPublicationView,
14    },
15    read_control::StorageReadControl,
16    StorageBackendResult,
17};
18
19impl NotificationPublicationStore for VersionedKeyValueStore {
20    fn stage_notification_publication(
21        &self,
22        publication: &NotificationPublication,
23    ) -> StorageBackendResult<()> {
24        self.require_mutable_session()?;
25        let control = self.write_control();
26        let effect = NotificationEffect::publish(
27            publication,
28            self.persistence.notification_record_layout(),
29            &control,
30        )
31        .map_err(VersionError::into_storage_error)?;
32        self.active
33            .lock()
34            .as_mut()
35            .ok_or_else(no_transaction)?
36            .stage_notification(effect)
37            .map_err(VersionError::into_storage_error)
38    }
39
40    fn visit_notification_publication(
41        &self,
42        control: &StorageReadControl,
43        visit: &mut dyn FnMut(Option<NotificationPublicationView<'_>>) -> StorageBackendResult<()>,
44    ) -> StorageBackendResult<()> {
45        control.check()?;
46        let layout = self.persistence.notification_record_layout();
47        let key = layout
48            .key(control)
49            .map_err(VersionError::into_storage_error)?;
50        let snapshot = self
51            .persistence
52            .snapshot(control)
53            .map_err(VersionError::into_storage_error)?;
54        snapshot
55            .visit_value(&key, control, &mut |record| {
56                let publication = record
57                    .and_then(|record| record.value)
58                    .map(|value| layout.decode(&key, value, control))
59                    .transpose()?;
60                visit(publication).map_err(Into::into)
61            })
62            .map_err(VersionError::into_storage_error)?;
63        control.check()
64    }
65
66    fn acknowledge_notification_publication(
67        &self,
68        fingerprint: [u8; 32],
69        control: &StorageReadControl,
70    ) -> StorageBackendResult<()> {
71        self.require_mutable_session()?;
72        control.check()?;
73        let mut pending = false;
74        self.visit_notification_publication(control, &mut |publication| {
75            pending =
76                publication.is_some_and(|publication| publication.fingerprint() == fingerprint);
77            Ok(())
78        })?;
79        if !pending {
80            return Ok(());
81        }
82        let effect = NotificationEffect::acknowledge(
83            fingerprint,
84            self.persistence.notification_record_layout(),
85            control,
86        )
87        .map_err(VersionError::into_storage_error)?;
88        let mut transaction = Transaction::new(&*self.persistence, false, control)
89            .map_err(VersionError::into_storage_error)?;
90        transaction
91            .stage_notification(effect)
92            .map_err(VersionError::into_storage_error)?;
93        transaction.commit(&*self.persistence, control)?;
94        transaction.acknowledge_completion(&*self.persistence, control)
95    }
96}
uqa_storage/mvcc/session/notifications.rs

uqa_storage/mvcc/session/
notifications.rs
