mod completion;
#[cfg(any(windows, all(unix, not(target_os = "emscripten"))))]
mod cross_process;
mod hub;
#[cfg(not(any(windows, all(unix, not(target_os = "emscripten")))))]
mod cross_process {
use std::sync::Arc;
use uqa_sql::SQLError;
use uqa_storage::{
notifications::{NotificationPublication, PendingNotification},
read_control::StorageReadControl,
PersistentStorageBackend,
};
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct CrossProcessQueueState {
pub(super) next_sequence: u64,
pub(super) head_position: u64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct CrossProcessQueueEntry {
pub(super) sequence: u64,
pub(super) process_id: i32,
pub(super) channel: String,
pub(super) payload: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct CrossProcessListenerRow {
pub(super) owner_id: [u8; 16],
pub(super) session_id: u64,
pub(super) process_id: i32,
pub(super) wake_port: u16,
pub(super) channels: Vec<String>,
pub(super) transaction_open: bool,
pub(super) next_sequence: u64,
pub(super) position: u64,
}
pub(super) struct ListenerLease {
owner_id: [u8; 16],
}
impl ListenerLease {
pub(super) const fn owner_id(&self) -> [u8; 16] {
self.owner_id
}
}
pub(super) struct CrossProcessRegistryTransaction;
impl CrossProcessRegistryTransaction {
pub(super) fn suspend_publication(
&mut self,
_publication: &uqa_storage::notifications::NotificationPublication,
_owner: [u8; 16],
) -> Result<(), SQLError> {
Err(unsupported())
}
pub(super) fn resume_publication(
&mut self,
_publication: &uqa_storage::notifications::NotificationPublication,
_owner: [u8; 16],
_control: &StorageReadControl,
) -> Result<bool, SQLError> {
Err(unsupported())
}
pub(super) fn prepare_publication(
&mut self,
_process_id: i32,
_pending: &[PendingNotification],
_listener: Option<&CrossProcessListenerRow>,
_control: &StorageReadControl,
) -> Result<NotificationPublication, SQLError> {
Err(unsupported())
}
pub(super) const fn pending_acknowledgement(&self) -> Option<[u8; 32]> {
None
}
pub(super) fn queue_state(&self) -> Result<CrossProcessQueueState, SQLError> {
Err(unsupported())
}
pub(super) fn entries_from(
&self,
_from_sequence: u64,
) -> Result<Vec<CrossProcessQueueEntry>, SQLError> {
Err(unsupported())
}
pub(super) fn delete_entries_before(&self, _sequence: u64) -> Result<(), SQLError> {
Err(unsupported())
}
pub(super) fn listeners(&self) -> Result<Vec<CrossProcessListenerRow>, SQLError> {
Err(unsupported())
}
pub(super) fn save_listener(
&self,
_listener: &CrossProcessListenerRow,
) -> Result<(), SQLError> {
Err(unsupported())
}
pub(super) fn drop_listener(
&self,
_owner_id: [u8; 16],
_session_id: u64,
) -> Result<(), SQLError> {
Err(unsupported())
}
pub(super) fn commit(self) -> Result<(), SQLError> {
Err(unsupported())
}
}
pub(super) struct CrossProcessCoordinator;
impl CrossProcessCoordinator {
pub(super) fn begin_publication_transaction(
&self,
_control: &StorageReadControl,
_resume: Option<(
&uqa_storage::notifications::NotificationPublication,
[u8; 16],
)>,
) -> Result<CrossProcessRegistryTransaction, SQLError> {
Err(unsupported())
}
pub(super) fn initialize_recovery(
&self,
_backend: &Arc<dyn PersistentStorageBackend>,
_control: &StorageReadControl,
) -> Result<(), SQLError> {
Err(unsupported())
}
pub(super) fn recovery_control(&self) -> Result<StorageReadControl, SQLError> {
Err(unsupported())
}
pub(super) const fn recovery_initialized(&self) -> bool {
false
}
pub(super) fn poll_needed(&self, _local_owners: &[[u8; 16]]) -> Result<bool, SQLError> {
Err(unsupported())
}
pub(super) fn begin_registry_transaction(
&self,
) -> Result<CrossProcessRegistryTransaction, SQLError> {
Err(unsupported())
}
pub(super) fn create_listener_lease(&self) -> Result<ListenerLease, SQLError> {
Err(unsupported())
}
pub(super) fn listener_is_alive(
&self,
_owner_id: [u8; 16],
_local_owner_ids: &[[u8; 16]],
) -> Result<bool, SQLError> {
Err(unsupported())
}
pub(super) const fn wake_port(&self) -> u16 {
0
}
pub(super) fn wake(_ports: &[u16]) {}
}
fn unsupported() -> SQLError {
SQLError::Internal(
"cross-process asynchronous notifications are unavailable on this target".into(),
)
}
}
use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
use std::sync::{Arc, OnceLock, Weak};
use std::time::{Duration, Instant};
use crate::Engine;
use cross_process::{
CrossProcessCoordinator, CrossProcessListenerRow, CrossProcessQueueState,
CrossProcessRegistryTransaction, ListenerLease,
};
use parking_lot::{Condvar, Mutex, MutexGuard};
use uqa_sql::SQLError;
pub(crate) use uqa_storage::notifications::PendingNotification;
#[cfg(test)]
use uqa_storage::notifications::NOTIFICATION_QUEUE_PAGE_BYTES;
use uqa_storage::notifications::{
notification_end_position, notifications_fit_queue, queue_page, MAX_NOTIFICATION_CHANNEL_BYTES,
MAX_NOTIFICATION_PAYLOAD_BYTES, MAX_NOTIFICATION_QUEUE_PAGES,
};
const NOTIFICATION_QUEUE_WARNING_INTERVAL: Duration = Duration::from_secs(5);
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SQLNotification {
pub process_id: i32,
pub channel: String,
pub payload: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum PendingListenAction {
Listen(String),
Unlisten(String),
UnlistenAll,
}
impl PendingListenAction {
fn apply(&self, channels: &mut Vec<String>) {
match self {
Self::Listen(channel) if !channels.contains(channel) => channels.push(channel.clone()),
Self::Listen(_) => {}
Self::Unlisten(channel) => channels.retain(|candidate| candidate != channel),
Self::UnlistenAll => channels.clear(),
}
}
}
#[derive(Clone)]
struct CommittedNotification {
sequence: u64,
process_id: i32,
channel: String,
payload: String,
}
struct NotificationListener {
process_id: i32,
channels: Vec<String>,
queue: Weak<Mutex<VecDeque<SQLNotification>>>,
wake: Weak<Condvar>,
next_sequence: u64,
position: u64,
transaction_open: bool,
lease: Option<ListenerLease>,
}
struct NotificationSessionCommit<'a> {
session_id: u64,
process_id: i32,
channels: Vec<String>,
queue: &'a Arc<Mutex<VecDeque<SQLNotification>>>,
wake: &'a Arc<Condvar>,
notices: &'a Arc<Mutex<Vec<(String, String)>>>,
pending: &'a [PendingNotification],
}
struct PreparedDelivery {
session_id: u64,
notifications: Vec<SQLNotification>,
}
pub(super) struct CrossNotificationCommit {
registry: Option<CrossProcessRegistryTransaction>,
new_lease: Option<ListenerLease>,
publisher_lease: Option<ListenerLease>,
publication_applied: bool,
publication: Option<uqa_storage::notifications::NotificationPublication>,
previous_publication: Option<[u8; 32]>,
wake_ports: Vec<u16>,
warning: Option<String>,
}
#[derive(Clone, Copy)]
struct CrossNotificationRequest<'a> {
session_id: u64,
process_id: i32,
channels: &'a [String],
pending: &'a [PendingNotification],
control: &'a uqa_storage::read_control::StorageReadControl,
durable_publication: bool,
}
struct PreparedCrossSubscription {
new_lease: Option<ListenerLease>,
listeners: Vec<CrossProcessListenerRow>,
subscription: Option<CrossProcessListenerRow>,
}
#[cfg(any(windows, all(unix, not(target_os = "emscripten"))))]
struct CrossProcessState {
database_path: std::path::PathBuf,
encryption_key: Option<uqa_storage::StorageEncryptionKey>,
registry: Mutex<Option<uqa_storage_sqlite::notifications::NotificationRegistry>>,
hub: Weak<NotificationHub>,
coordinator: Mutex<Option<Arc<CrossProcessCoordinator>>>,
}
#[cfg(not(any(windows, all(unix, not(target_os = "emscripten")))))]
struct CrossProcessState;
impl CrossProcessState {
#[cfg(any(windows, all(unix, not(target_os = "emscripten"))))]
fn initialized_coordinator(&self) -> Option<Arc<CrossProcessCoordinator>> {
self.coordinator.lock().clone()
}
#[cfg(not(any(windows, all(unix, not(target_os = "emscripten")))))]
fn initialized_coordinator(&self) -> Option<Arc<CrossProcessCoordinator>> {
None
}
#[cfg(any(windows, all(unix, not(target_os = "emscripten"))))]
fn registry(
&self,
) -> Result<uqa_storage_sqlite::notifications::NotificationRegistry, SQLError> {
let mut initialized = self.registry.lock();
if let Some(registry) = initialized.as_ref() {
return Ok(registry.clone());
}
let registry =
cross_process::open_registry(&self.database_path, self.encryption_key.as_ref())
.map_err(SQLError::Internal)?;
*initialized = Some(registry.clone());
Ok(registry)
}
#[cfg(any(windows, all(unix, not(target_os = "emscripten"))))]
fn allocate_backend_process_id(&self) -> Result<i32, SQLError> {
CrossProcessCoordinator::allocate_backend_process_id(&self.registry()?)
}
#[cfg(not(any(windows, all(unix, not(target_os = "emscripten")))))]
fn allocate_backend_process_id(&self) -> Result<i32, SQLError> {
Err(SQLError::Internal(
"cross-process backend process identifiers are unavailable on this target".into(),
))
}
#[cfg(any(windows, all(unix, not(target_os = "emscripten"))))]
fn coordinator(&self) -> Result<Arc<CrossProcessCoordinator>, SQLError> {
let mut initialized = self.coordinator.lock();
if let Some(coordinator) = initialized.as_ref() {
return Ok(Arc::clone(coordinator));
}
let (coordinator, listener) =
CrossProcessCoordinator::open(&self.database_path, self.registry()?)
.map_err(SQLError::Internal)?;
let coordinator = Arc::new(coordinator);
coordinator
.start_worker(listener, self.hub.clone())
.map_err(SQLError::Internal)?;
*initialized = Some(Arc::clone(&coordinator));
Ok(coordinator)
}
#[cfg(not(any(windows, all(unix, not(target_os = "emscripten")))))]
fn coordinator(&self) -> Result<Arc<CrossProcessCoordinator>, SQLError> {
Err(SQLError::Internal(
"cross-process asynchronous notifications are unavailable on this target".into(),
))
}
}
pub(super) struct NotificationCommitGuard<'a> {
_gate: MutexGuard<'a, ()>,
cross: Option<Box<CrossNotificationCommit>>,
}
impl NotificationCommitGuard<'_> {
pub(super) fn retain(mut self) -> (Option<Box<CrossNotificationCommit>>, Result<(), SQLError>) {
let suspended = self.cross.as_mut().map_or(Ok(()), |prepared| {
if !prepared.publication_applied {
if let (Some(registry), Some(publication)) =
(prepared.registry.as_mut(), prepared.publication.as_ref())
{
let owner = prepared
.publisher_lease
.as_ref()
.ok_or_else(|| {
SQLError::Internal(
"retained notification has no publication lease".into(),
)
})?
.owner_id();
registry.suspend_publication(publication, owner)?;
}
}
prepared.registry.take();
Ok(())
});
(self.cross, suspended)
}
}
#[derive(Default)]
struct NotificationHubState {
listeners: BTreeMap<u64, NotificationListener>,
entries: VecDeque<CommittedNotification>,
next_sequence: u64,
head_position: u64,
last_queue_warning: Option<Instant>,
}
pub(crate) struct NotificationHub {
commit_gate: Mutex<()>,
state: Mutex<NotificationHubState>,
max_queue_pages: u64,
cross: Option<CrossProcessState>,
cross_error: Mutex<Option<String>>,
}
impl Default for NotificationHub {
fn default() -> Self {
Self {
commit_gate: Mutex::new(()),
state: Mutex::new(NotificationHubState::default()),
max_queue_pages: MAX_NOTIFICATION_QUEUE_PAGES,
cross: None,
cross_error: Mutex::new(None),
}
}
}
fn append_notification(
state: &mut NotificationHubState,
process_id: i32,
notification: &PendingNotification,
) {
let end_position = notification_end_position(state.head_position, notification);
let sequence = state.next_sequence;
state.entries.push_back(CommittedNotification {
sequence,
process_id,
channel: notification.channel.clone(),
payload: notification.payload.clone(),
});
state.next_sequence = state.next_sequence.saturating_add(1);
state.head_position = end_position;
}
fn projected_tail_position(
state: &NotificationHubState,
session_id: u64,
final_channels: &[String],
) -> u64 {
state
.listeners
.iter()
.filter_map(|(listener_id, listener)| {
(*listener_id != session_id || !final_channels.is_empty()).then_some(listener.position)
})
.min()
.unwrap_or(state.head_position)
}
fn queue_usage(state: &NotificationHubState, max_queue_pages: u64) -> f64 {
let tail = state
.listeners
.values()
.map(|listener| listener.position)
.min()
.unwrap_or(state.head_position);
let pages = queue_page(state.head_position).saturating_sub(queue_page(tail));
pages as f64 / max_queue_pages as f64
}
fn validate_channel(channel: &str) -> Result<(), SQLError> {
if channel.is_empty() {
return Err(SQLError::Routine {
sqlstate: "22023".into(),
message: "channel name cannot be empty".into(),
});
}
if channel.len() >= MAX_NOTIFICATION_CHANNEL_BYTES {
return Err(SQLError::Routine {
sqlstate: "22023".into(),
message: "channel name too long".into(),
});
}
Ok(())
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum NotificationHubIdentity {
Durable(uqa_storage::PersistentStorageIdentity),
Provider(usize),
}
static DATABASE_NOTIFICATION_HUBS: OnceLock<
Mutex<HashMap<NotificationHubIdentity, Weak<NotificationHub>>>,
> = OnceLock::new();
#[derive(Default)]
struct BackendProcessIdRegistry {
active: BTreeSet<i32>,
next: i32,
}
static BACKEND_PROCESS_IDS: OnceLock<Mutex<BackendProcessIdRegistry>> = OnceLock::new();
pub(crate) fn allocate_backend_process_id() -> i32 {
let registry = BACKEND_PROCESS_IDS.get_or_init(|| {
Mutex::new(BackendProcessIdRegistry {
active: BTreeSet::new(),
next: 1,
})
});
let mut registry = registry.lock();
for _ in 0..i32::MAX {
let candidate = registry.next;
registry.next = registry.next.checked_add(1).unwrap_or(1);
if registry.active.insert(candidate) {
return candidate;
}
}
panic!("exhausted positive backend process identifiers");
}
pub(crate) fn release_backend_process_id(process_id: i32) {
if let Some(registry) = BACKEND_PROCESS_IDS.get() {
registry.lock().active.remove(&process_id);
}
}
pub(crate) fn shared_provider_notification_hub(
identity: Option<uqa_storage::PersistentStorageIdentity>,
provider: &Arc<dyn uqa_storage::PersistentStorageProvider>,
) -> Arc<NotificationHub> {
shared_notification_hub(
identity,
NotificationHubIdentity::Provider(Arc::as_ptr(provider).cast::<()>() as usize),
provider.auxiliary_encryption_key(),
)
}
pub(crate) fn shared_backend_notification_hub(
identity: Option<uqa_storage::PersistentStorageIdentity>,
backend: &Arc<dyn uqa_storage::PersistentStorageBackend>,
) -> Arc<NotificationHub> {
shared_notification_hub(
identity,
NotificationHubIdentity::Provider(Arc::as_ptr(backend).cast::<()>() as usize),
backend.auxiliary_encryption_key(),
)
}
fn shared_notification_hub(
identity: Option<uqa_storage::PersistentStorageIdentity>,
fallback: NotificationHubIdentity,
encryption_key: Option<uqa_storage::StorageEncryptionKey>,
) -> Arc<NotificationHub> {
let identity = identity.map_or(fallback, NotificationHubIdentity::Durable);
let registry = DATABASE_NOTIFICATION_HUBS.get_or_init(|| Mutex::new(HashMap::new()));
let mut registry = registry.lock();
registry.retain(|_, hub| hub.strong_count() != 0);
if let Some(hub) = registry.get(&identity).and_then(Weak::upgrade) {
return hub;
}
let hub = match &identity {
NotificationHubIdentity::Durable(uqa_storage::PersistentStorageIdentity::File(path)) => {
NotificationHub::for_database_file(path, encryption_key)
}
_ => Arc::new(NotificationHub::default()),
};
registry.insert(identity, Arc::downgrade(&hub));
hub
}
impl Engine {
pub(crate) fn listen(&self, channel: &str) -> Result<(), SQLError> {
validate_channel(channel)?;
self.pending_listen_action(PendingListenAction::Listen(channel.to_string()))
}
pub(crate) fn unlisten(&self, channel: Option<&str>) -> Result<(), SQLError> {
let action = match channel {
Some(channel) => {
validate_channel(channel)?;
PendingListenAction::Unlisten(channel.to_string())
}
None => PendingListenAction::UnlistenAll,
};
self.pending_listen_action(action)
}
fn pending_listen_action(&self, action: PendingListenAction) -> Result<(), SQLError> {
let mut stack = self.session.transactions.lock();
let frame = stack.last_mut().ok_or_else(|| {
SQLError::Internal("LISTEN state changed without a transaction frame".into())
})?;
frame.pending_listen_actions.push(action);
Ok(())
}
pub(crate) fn notify(&self, channel: &str, payload: &str) -> Result<(), SQLError> {
validate_channel(channel)?;
if payload.len() >= MAX_NOTIFICATION_PAYLOAD_BYTES {
return Err(SQLError::Routine {
sqlstate: "22023".into(),
message: "payload string too long".into(),
});
}
let notification = PendingNotification {
channel: channel.to_string(),
payload: payload.to_string(),
};
let mut stack = self.session.transactions.lock();
let frame = stack.last_mut().ok_or_else(|| {
SQLError::Internal("NOTIFY executed without a transaction frame".into())
})?;
if !frame.pending_notifications.contains(¬ification) {
frame.pending_notifications.push(notification);
}
Ok(())
}
pub(super) fn begin_notification_transaction(&self) -> Result<(), SQLError> {
self.notification_hub.begin_transaction(self.session_id)
}
pub(super) fn rollback_notification_state(&self) -> Result<(), SQLError> {
self.notification_hub.rollback_session(self.session_id)
}
pub(crate) fn clear_notification_listener_without_transaction(&self) -> Result<(), SQLError> {
self.notification_hub.replace_idle_session(
self.session_id,
self.backend_process_id(),
Vec::new(),
&self.runtime.notifications,
&self.runtime.notification_wake,
&self.runtime.notices,
)?;
self.session.state.write().listened_channels.clear();
Ok(())
}
#[must_use]
pub fn backend_process_id(&self) -> i32 {
self.session
.backend_process_id
.load(std::sync::atomic::Ordering::Acquire)
}
pub(crate) fn listening_channels(&self) -> Vec<String> {
self.session.state.read().listened_channels.clone()
}
pub(crate) fn notification_queue_usage(&self) -> Result<f64, SQLError> {
self.prepare_notification_recovery()?;
self.notification_hub.usage()
}
pub fn poll_sql_notifications(&self) -> Result<usize, SQLError> {
if self.transaction_depth() != 0 {
return Ok(0);
}
self.notification_hub
.try_synchronize_cross_process_session(Some((self.session_id, false)))?;
if let Some(error) = self.notification_hub.cross_error.lock().take() {
return Err(SQLError::Internal(error));
}
Ok(self.runtime.notifications.lock().len())
}
pub fn wait_for_sql_notifications(&self, timeout: Duration) -> Result<bool, SQLError> {
if self.transaction_depth() != 0 {
return Ok(false);
}
let started = Instant::now();
loop {
if self.poll_sql_notifications()? != 0 {
return Ok(true);
}
let elapsed = started.elapsed();
if elapsed >= timeout {
return Ok(false);
}
let mut notifications = self.runtime.notifications.lock();
if !notifications.is_empty() {
return Ok(true);
}
self.runtime
.notification_wake
.wait_for(&mut notifications, timeout.saturating_sub(elapsed));
}
}
pub fn take_sql_notifications(&self) -> Vec<SQLNotification> {
if self.transaction_depth() != 0 {
return Vec::new();
}
self.runtime.notifications.lock().drain(..).collect()
}
}
impl crate::TransactionFrame {
fn final_listened_channels(&self, current: &[String]) -> Vec<String> {
let mut channels = current.to_vec();
for action in &self.pending_listen_actions {
action.apply(&mut channels);
}
channels
}
pub(super) fn merge_pending_listen_actions(&mut self, pending: Vec<PendingListenAction>) {
self.pending_listen_actions.extend(pending);
}
pub(super) fn merge_pending_notifications(&mut self, pending: Vec<PendingNotification>) {
for notification in pending {
if !self.pending_notifications.contains(¬ification) {
self.pending_notifications.push(notification);
}
}
}
pub(super) fn restore_pending_notification_savepoint(&mut self, position: usize) {
self.pending_listen_actions = self.savepoints[position].pending_listen_actions.clone();
self.pending_notifications = self.savepoints[position].pending_notifications.clone();
}
}
#[cfg(test)]
mod recovery_tests;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn queue_warning_identifies_the_oldest_transaction_and_is_throttled() {
let queue = Arc::new(Mutex::new(VecDeque::new()));
let wake = Arc::new(Condvar::new());
let mut state = NotificationHubState {
head_position: NOTIFICATION_QUEUE_PAGE_BYTES,
..NotificationHubState::default()
};
state.listeners.insert(
7,
NotificationListener {
process_id: 42,
channels: vec!["events".into()],
queue: Arc::downgrade(&queue),
wake: Arc::downgrade(&wake),
next_sequence: 0,
position: 0,
transaction_open: true,
lease: None,
},
);
let hub = NotificationHub {
commit_gate: Mutex::new(()),
state: Mutex::new(NotificationHubState::default()),
max_queue_pages: 1,
cross: None,
cross_error: Mutex::new(None),
};
assert_eq!(
hub.queue_warning(&mut state).as_deref(),
Some("NOTIFY queue is 100% full\nDETAIL: The server process with PID 42 is among those with the oldest transactions.\nHINT: The NOTIFY queue cannot be emptied until that process ends its current transaction.")
);
assert!(hub.queue_warning(&mut state).is_none());
}
}