use std::fs::{File, OpenOptions};
use std::io::Write;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Weak};
use std::thread::JoinHandle;
use std::time::Duration;
use super::NotificationHub;
use fs2::FileExt;
use parking_lot::Mutex;
use uqa_core::CancellationToken;
use uqa_sql::SQLError;
use uqa_storage::StorageEncryptionKey;
use uqa_storage::{
notifications::{
NotificationListenerKey, NotificationListenerMetadata, NotificationPublication,
PendingNotification,
},
read_control::StorageReadControl,
PersistentStorageBackend,
};
use uqa_storage_sqlite::notifications::NotificationRegistry;
pub(super) use uqa_storage::notifications::{
NotificationListenerRow as CrossProcessListenerRow,
NotificationQueueEntry as CrossProcessQueueEntry,
NotificationQueueState as CrossProcessQueueState,
};
pub(super) struct ListenerLease {
owner_id: [u8; 16],
path: PathBuf,
file: Option<File>,
}
impl ListenerLease {
pub(super) const fn owner_id(&self) -> [u8; 16] {
self.owner_id
}
}
impl Drop for ListenerLease {
fn drop(&mut self) {
if let Some(file) = self.file.take() {
let _ = FileExt::unlock(&file);
drop(file);
}
let _ = std::fs::remove_file(&self.path);
}
}
pub(super) struct CrossProcessRegistryTransaction {
transaction: uqa_storage_sqlite::notifications::NotificationRegistryTransaction,
}
impl CrossProcessRegistryTransaction {
pub(super) const fn pending_acknowledgement(&self) -> Option<[u8; 32]> {
self.transaction.pending_acknowledgement()
}
pub(super) fn prepare_publication(
&mut self,
process_id: i32,
pending: &[PendingNotification],
listener: Option<&CrossProcessListenerRow>,
control: &StorageReadControl,
) -> Result<NotificationPublication, SQLError> {
self.transaction
.prepare_publication(process_id, pending, listener, control)
.map_err(registry_error)
}
pub(super) fn suspend_publication(
&mut self,
publication: &NotificationPublication,
owner: [u8; 16],
) -> Result<(), SQLError> {
self.transaction
.suspend_publication(publication, owner)
.map_err(registry_error)
}
pub(super) fn resume_publication(
&mut self,
publication: &NotificationPublication,
owner: [u8; 16],
control: &StorageReadControl,
) -> Result<bool, SQLError> {
self.transaction
.resume_publication(publication, owner, control)
.map_err(registry_error)
}
pub(super) fn allocate_backend_process_id(&self) -> Result<i32, SQLError> {
self.transaction
.allocate_backend_process_id()
.map_err(registry_error)
}
pub(super) fn queue_state(&self) -> Result<CrossProcessQueueState, SQLError> {
self.transaction.queue_state().map_err(registry_error)
}
pub(super) fn entries_from(
&self,
from_sequence: u64,
) -> Result<Vec<CrossProcessQueueEntry>, SQLError> {
self.transaction
.entries_from(from_sequence)
.map_err(registry_error)
}
pub(super) fn delete_entries_before(&self, sequence: u64) -> Result<(), SQLError> {
self.transaction
.delete_entries_before(sequence)
.map_err(registry_error)
}
pub(super) fn visit_entries_from(
&self,
from_sequence: u64,
max_entries: std::num::NonZeroUsize,
control: &StorageReadControl,
visit: &mut dyn FnMut(
uqa_storage::notifications::NotificationQueueEntryRef<'_>,
)
-> uqa_storage::StorageBackendResult<std::ops::ControlFlow<()>>,
) -> Result<uqa_storage::notifications::NotificationQueueScan, SQLError> {
self.transaction
.visit_entries_from(from_sequence, max_entries, control, visit)
.map_err(registry_error)
}
pub(super) fn listener_metadata_after(
&self,
after: Option<NotificationListenerKey>,
) -> Result<Option<NotificationListenerMetadata>, SQLError> {
self.transaction
.listener_metadata_after(after)
.map_err(registry_error)
}
pub(super) fn listener(
&self,
key: NotificationListenerKey,
max_channels: Option<usize>,
) -> Result<Option<CrossProcessListenerRow>, SQLError> {
self.transaction
.listener(key, max_channels)
.map_err(registry_error)
}
pub(super) fn save_listener(&self, listener: &CrossProcessListenerRow) -> Result<(), SQLError> {
self.transaction
.save_listener(listener)
.map_err(registry_error)
}
pub(super) fn drop_listener(
&self,
owner_id: [u8; 16],
session_id: u64,
) -> Result<(), SQLError> {
self.transaction
.drop_listener(owner_id, session_id)
.map_err(registry_error)
}
pub(super) fn commit(self) -> Result<(), SQLError> {
self.transaction.commit().map_err(registry_error)
}
pub(super) fn commit_with_control(self, control: &StorageReadControl) -> Result<(), SQLError> {
self.transaction
.commit_with_control(control)
.map_err(registry_error)
}
}
fn registry_error(error: uqa_storage::StorageBackendError) -> SQLError {
match error {
uqa_storage::StorageBackendError::Other(message) => SQLError::Internal(message),
error => uqa_execution::storage_errors::storage_error(
"asynchronous notification registry",
&error,
),
}
}
fn open_registry_transaction(
registry: &NotificationRegistry,
) -> Result<CrossProcessRegistryTransaction, SQLError> {
registry
.begin()
.map(|transaction| CrossProcessRegistryTransaction { transaction })
.map_err(registry_error)
}
pub(super) fn open_registry(
database_path: &Path,
key: Option<&StorageEncryptionKey>,
) -> Result<NotificationRegistry, String> {
NotificationRegistry::open(database_path, key).map_err(|error| error.to_string())
}
pub(super) struct CrossProcessCoordinator {
database_path: PathBuf,
registry: NotificationRegistry,
wake_port: u16,
shutdown: CancellationToken,
worker: Mutex<Option<JoinHandle<()>>>,
recovery: Mutex<Option<RecoverySession>>,
#[cfg(test)]
gate_wait: Arc<Mutex<Option<std::sync::mpsc::SyncSender<()>>>>,
}
#[derive(Clone)]
struct RecoverySession {
backend: Arc<dyn PersistentStorageBackend>,
control: StorageReadControl,
}
impl CrossProcessCoordinator {
pub(super) fn allocate_backend_process_id(
registry: &NotificationRegistry,
) -> Result<i32, SQLError> {
let transaction = open_registry_transaction(registry)?;
let process_id = transaction.allocate_backend_process_id()?;
transaction.commit()?;
Ok(process_id)
}
pub(super) fn open(
database_path: &Path,
registry: NotificationRegistry,
) -> Result<(Self, TcpListener), String> {
let listener = TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0))
.map_err(|error| format!("bind asynchronous notification wake listener: {error}"))?;
let wake_port = listener
.local_addr()
.map_err(|error| format!("read asynchronous notification wake address: {error}"))?
.port();
Ok((
Self {
database_path: database_path.to_path_buf(),
registry,
wake_port,
shutdown: CancellationToken::new(),
worker: Mutex::new(None),
recovery: Mutex::new(None),
#[cfg(test)]
gate_wait: Arc::new(Mutex::new(None)),
},
listener,
))
}
pub(super) fn start_worker(
&self,
listener: TcpListener,
hub: Weak<NotificationHub>,
) -> Result<(), String> {
listener.set_nonblocking(true).map_err(|error| {
format!("configure asynchronous notification recovery polling: {error}")
})?;
let shutdown = self.shutdown.clone();
#[cfg(test)]
let gate_wait = Arc::clone(&self.gate_wait);
let worker = std::thread::Builder::new()
.name("uqa-notification-wake".into())
.spawn(move || loop {
if shutdown.is_cancelled() {
break;
}
match listener.accept() {
Ok((stream, _)) => drop(stream),
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::park_timeout(Duration::from_millis(100));
}
Err(error) => {
if let Some(hub) = hub.upgrade() {
hub.record_cross_error(
SQLError::Internal(format!(
"accept asynchronous notification wake connection: {error}"
)),
&shutdown,
);
}
break;
}
}
if shutdown.is_cancelled() {
break;
}
let Some(hub) = hub.upgrade() else { break };
#[cfg(test)]
if let Some(observer) = gate_wait.lock().clone() {
super::registration::observe_gate_wait(observer);
}
hub.synchronize_cross_process_notifications(&shutdown);
})
.map_err(|error| format!("spawn asynchronous notification wake worker: {error}"))?;
*self.worker.lock() = Some(worker);
Ok(())
}
pub(super) const fn wake_port(&self) -> u16 {
self.wake_port
}
pub(super) fn begin_registry_transaction(
&self,
) -> Result<CrossProcessRegistryTransaction, SQLError> {
self.begin_registry_transaction_with_cancellation(None)
}
pub(super) fn begin_registry_transaction_with_cancellation(
&self,
cancellation: Option<&uqa_core::CancellationToken>,
) -> Result<CrossProcessRegistryTransaction, SQLError> {
let recovery = super::registration::lock(&self.recovery, cancellation)?
.clone()
.ok_or_else(|| {
SQLError::Internal(
"notification recovery has no independent storage session".into(),
)
})?;
let store = recovery
.backend
.notification_publications()
.ok_or_else(|| {
SQLError::Internal(
"notification recovery session omitted atomic publication".into(),
)
})?;
let control = cancellation.map_or_else(
|| recovery.control.clone(),
|cancellation| StorageReadControl::new(recovery.control.memory(), cancellation),
);
self.registry
.begin_recovered(store, &control)
.map(|transaction| CrossProcessRegistryTransaction { transaction })
.map_err(registry_error)
}
pub(super) fn initialize_recovery_with_cancellation(
&self,
backend: &Arc<dyn PersistentStorageBackend>,
control: &StorageReadControl,
cancellation: Option<&uqa_core::CancellationToken>,
) -> Result<(), SQLError> {
let mut recovery = super::registration::lock(&self.recovery, cancellation)?;
if recovery.is_none() {
let session = backend.open_session().map_err(registry_error)?;
if let Some(cancellation) = cancellation {
cancellation.check()?;
}
if session.backend.notification_publications().is_none() {
return Err(SQLError::Internal(
"persistent backend does not support atomic notification publication".into(),
));
}
let allowance = session
.backend
.retention_control()
.unwrap_or_else(|| control.clone());
*recovery = Some(RecoverySession {
backend: session.backend,
control: StorageReadControl::new(allowance.memory(), &self.shutdown),
});
}
Ok(())
}
pub(super) fn begin_publication_transaction(
&self,
control: &StorageReadControl,
resume: Option<(&NotificationPublication, [u8; 16])>,
) -> Result<CrossProcessRegistryTransaction, SQLError> {
let recovery = self.recovery.lock().clone().ok_or_else(|| {
SQLError::Internal("notification recovery has no independent storage session".into())
})?;
let store = recovery
.backend
.notification_publications()
.ok_or_else(|| {
SQLError::Internal(
"notification recovery session omitted atomic publication".into(),
)
})?;
self.registry
.begin_publishing(store, control, resume, &mut |owner| {
self.listener_is_alive(owner, &[])
.map_err(|error| uqa_storage::StorageBackendError::Other(error.to_string()))
})
.map(|transaction| CrossProcessRegistryTransaction { transaction })
.map_err(registry_error)
}
pub(super) fn recovery_control(&self) -> Result<StorageReadControl, SQLError> {
self.recovery_control_with_cancellation(None)
}
pub(super) fn recovery_control_with_cancellation(
&self,
cancellation: Option<&uqa_core::CancellationToken>,
) -> Result<StorageReadControl, SQLError> {
super::registration::lock(&self.recovery, cancellation)?
.as_ref()
.map(|session| {
cancellation.map_or_else(
|| session.control.clone(),
|cancellation| StorageReadControl::new(session.control.memory(), cancellation),
)
})
.ok_or_else(|| {
SQLError::Internal(
"notification recovery session omitted its retention allowance".into(),
)
})
}
pub(super) fn recovery_initialized(&self) -> bool {
self.recovery.lock().is_some()
}
pub(super) fn poll_needed(&self, local_owners: &[[u8; 16]]) -> Result<bool, SQLError> {
let Some(recovery) = self.recovery.lock().clone() else {
return Ok(false);
};
let store = recovery
.backend
.notification_publications()
.ok_or_else(|| {
SQLError::Internal(
"notification recovery session omitted atomic publication".into(),
)
})?;
self.registry
.poll_needed(store, local_owners, &recovery.control)
.map_err(registry_error)
}
pub(super) fn create_listener_lease(&self) -> Result<ListenerLease, SQLError> {
for _ in 0..16 {
let mut owner_id = [0_u8; 16];
getrandom::fill(&mut owner_id).map_err(|error| {
SQLError::Internal(format!(
"allocate asynchronous notification listener identity: {error}"
))
})?;
let path = lease_path(&self.database_path, owner_id);
match OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&path)
{
Ok(file) => {
let lease = ListenerLease {
owner_id,
path,
file: Some(file),
};
lease
.file
.as_ref()
.expect("newly owned lease file")
.try_lock_exclusive()
.map_err(|error| {
SQLError::Internal(format!(
"lock asynchronous notification listener lease `{}`: {error}",
lease.path.display()
))
})?;
return Ok(lease);
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => {
return Err(SQLError::Internal(format!(
"create asynchronous notification listener lease `{}`: {error}",
path.display()
)));
}
}
}
Err(SQLError::Internal(
"could not allocate a unique asynchronous notification listener lease".into(),
))
}
pub(super) fn listener_is_alive(
&self,
owner_id: [u8; 16],
local_owner_ids: &[[u8; 16]],
) -> Result<bool, SQLError> {
if local_owner_ids.contains(&owner_id) {
return Ok(true);
}
let path = lease_path(&self.database_path, owner_id);
let file = match OpenOptions::new().read(true).write(true).open(&path) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(error) => {
return Err(SQLError::Internal(format!(
"open asynchronous notification listener lease `{}`: {error}",
path.display()
)));
}
};
match file.try_lock_exclusive() {
Ok(()) => {
FileExt::unlock(&file).map_err(|error| {
SQLError::Internal(format!(
"unlock stale asynchronous notification listener lease `{}`: {error}",
path.display()
))
})?;
drop(file);
let _ = std::fs::remove_file(path);
Ok(false)
}
Err(error) if error.kind() == fs2::lock_contended_error().kind() => Ok(true),
Err(error) => Err(SQLError::Internal(format!(
"probe asynchronous notification listener lease `{}`: {error}",
path.display()
))),
}
}
pub(super) fn wake(ports: impl IntoIterator<Item = u16>) {
for port in ports {
let address = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, port));
if let Ok(mut stream) = TcpStream::connect_timeout(&address, Duration::from_millis(100))
{
let _ = stream.write_all(&[1]);
}
}
}
}
impl CrossProcessCoordinator {
#[cfg(test)]
pub(super) fn worker_is_retained(&self) -> bool {
self.worker.lock().is_some()
}
#[cfg(test)]
pub(super) fn observe_worker_gate_wait(&self, observer: std::sync::mpsc::SyncSender<()>) {
*self.gate_wait.lock() = Some(observer);
}
pub(super) fn shutdown(&self) {
self.shutdown.cancel();
if let Some(worker) = self.worker.lock().take() {
worker.thread().unpark();
if worker.thread().id() != std::thread::current().id() {
let _ = worker.join();
}
}
}
}
impl Drop for CrossProcessCoordinator {
fn drop(&mut self) {
self.shutdown();
}
}
fn lease_path(database_path: &Path, owner_id: [u8; 16]) -> PathBuf {
let mut suffix = String::with_capacity(2 * owner_id.len() + 24);
suffix.push_str(".uqa-notification-");
for byte in owner_id {
use std::fmt::Write as _;
write!(suffix, "{byte:02x}").expect("write listener lease suffix");
}
suffix.push_str(".lease");
suffixed_path(database_path, &suffix)
}
fn suffixed_path(database_path: &Path, suffix: &str) -> PathBuf {
let mut path = database_path.as_os_str().to_owned();
path.push(suffix);
PathBuf::from(path)
}