use super::stored::{Reader, StoredSubscription};
use super::{Observation, Wake, consumer_id, failure, lock, reached};
use crate::ServerError;
use crate::server::connection::services::ConnectionSubscription;
use liminal::channel::{ConnectionInboxBudget, InboxInstall};
use liminal::protocol::SchemaId;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
pub struct Binding {
pub(crate) prefix: usize,
pub(crate) consumer: String,
pub(crate) install: InboxInstall,
}
pub struct PreparedSubscription {
prefix: usize,
subject: String,
subscription: ConnectionSubscription,
offset: Option<u64>,
}
impl PreparedSubscription {
pub(crate) fn new(
prefix: usize,
subject: String,
subscription: ConnectionSubscription,
) -> Result<Self, ServerError> {
let offset = subscription.consumer_position()?;
if subscription.is_overflowed() {
return Err(failure(format!(
"export subscription overflowed on '{subject}'"
)));
}
Ok(Self {
prefix,
subject,
subscription,
offset,
})
}
}
#[derive(Clone, Debug)]
struct SubjectSubscription {
subject: String,
schema: SchemaId,
reader: Reader,
}
#[derive(Debug, Default)]
struct PrefixSubscription {
subjects: Vec<SubjectSubscription>,
next: usize,
}
#[derive(Debug)]
struct FeedState {
prefixes: Vec<PrefixSubscription>,
next: usize,
}
pub(super) struct PendingDelivery {
pub(super) subject: String,
pub(super) schema: SchemaId,
pub(super) payload: Vec<u8>,
pub(super) position: Option<u64>,
reader: Reader,
}
impl PendingDelivery {
pub(super) fn acknowledge(&self) -> Result<(), ServerError> {
if let Some(position) = self.position {
self.reader.acknowledge(position)?;
}
Ok(())
}
}
#[derive(Debug)]
pub struct ExportSubscriptions {
link: String,
prefixes: Vec<String>,
feed: Mutex<FeedState>,
wake: Arc<Wake>,
observation: Arc<Observation>,
budget: Arc<ConnectionInboxBudget>,
heads: [Arc<AtomicU64>; 2],
}
impl ExportSubscriptions {
pub(super) fn new(
link: &str,
prefixes: &[String],
wake: Arc<Wake>,
observation: Arc<Observation>,
) -> Result<Self, ServerError> {
let mut prefixes = prefixes.to_vec();
prefixes.sort_by(|left, right| left.len().cmp(&right.len()).then_with(|| left.cmp(right)));
prefixes.dedup();
for prefix in &prefixes {
crate::auth_lys::SubjectBinding::new("wire".to_owned(), prefix.clone(), prefix.clone())
.map_err(|error| failure(error.to_string()))?;
}
let feed = FeedState {
prefixes: prefixes
.iter()
.map(|_| PrefixSubscription::default())
.collect(),
next: 0,
};
Ok(Self {
link: link.to_owned(),
prefixes,
feed: Mutex::new(feed),
wake,
observation,
heads: std::array::from_fn(|_| Arc::new(AtomicU64::new(0))),
budget: ConnectionInboxBudget::new(
crate::config::LimitsConfig::DEFAULT_MAX_CONNECTION_INBOX_BYTES,
),
})
}
pub(crate) fn link(&self) -> &str {
&self.link
}
pub(crate) fn binding(&self, subject: &str) -> Option<Binding> {
if matches!(subject, "liminal.cost" | "liminal.events") || self.wake.check().is_err() {
return None;
}
let prefix = self
.prefixes
.iter()
.position(|prefix| reached(prefix, subject))?;
let feed = match lock(&self.feed, "subscription binding") {
Ok(feed) => feed,
Err(error) => {
self.wake.refuse(&error);
return None;
}
};
if feed.prefixes[prefix]
.subjects
.iter()
.any(|source| source.subject == subject)
{
return None;
}
drop(feed);
let wake = Arc::clone(&self.wake);
Some(Binding {
prefix,
consumer: consumer_id(&self.link, &self.prefixes[prefix]),
install: InboxInstall {
budget: Arc::clone(&self.budget),
depth_cap: crate::config::LimitsConfig::DEFAULT_MAX_SUBSCRIPTION_INBOX_DEPTH,
notifier: Some(Arc::new(move || wake.notify())),
capacity: None,
},
})
}
pub(crate) fn attach(&self, prepared: PreparedSubscription) {
if let Err(error) = self.attach_checked(prepared) {
self.wake.refuse(&error);
}
}
fn attach_checked(&self, prepared: PreparedSubscription) -> Result<(), ServerError> {
self.wake.check()?;
let PreparedSubscription {
prefix,
subject,
subscription,
offset,
} = prepared;
let source = SubjectSubscription {
subject: subject.clone(),
schema: subscription.selected_schema(),
reader: Reader::Live(Arc::new(Mutex::new(subscription))),
};
let reader = source.reader.clone();
lock(&self.feed, "subscription attachment")?.prefixes[prefix]
.subjects
.push(source);
if let Some(offset) = offset {
lock(&self.observation.counters, "resumed subscription position")?
.resumed_from
.insert(subject, offset);
}
let pending = reader.has_pending()?;
if pending {
self.wake.notify();
}
self.wake.check()
}
pub(crate) fn install_stored(
&self,
store: &Arc<dyn liminal::durability::DurableStore>,
) -> Result<bool, ServerError> {
let mut installed = false;
for (index, subject) in ["liminal.cost", "liminal.events"].into_iter().enumerate() {
let Some(prefix) = self
.prefixes
.iter()
.position(|prefix| reached(prefix, subject))
else {
continue;
};
let subscription = StoredSubscription::resume(
Arc::clone(store),
subject,
&consumer_id(&self.link, &self.prefixes[prefix]),
Arc::clone(&self.heads[index]),
)?;
let offset = subscription.offset();
lock(&self.feed, "stored subscription attachment")?.prefixes[prefix]
.subjects
.push(SubjectSubscription {
subject: subject.to_owned(),
schema: SchemaId::from_schema_bytes(b"{}"),
reader: Reader::Stored(Arc::new(Mutex::new(subscription))),
});
lock(&self.observation.counters, "stored subscription position")?
.resumed_from
.insert(subject.to_owned(), offset);
installed = true;
}
Ok(installed)
}
pub(crate) fn stored_published(&self, subject: &str, head: u64) {
let index = match subject {
"liminal.cost" => 0,
"liminal.events" => 1,
_ => return,
};
self.heads[index].fetch_max(head, Ordering::Release);
if self.prefixes.iter().any(|prefix| reached(prefix, subject)) {
self.wake.notify();
}
}
pub(super) fn next_ready(&self) -> Result<Option<PendingDelivery>, ServerError> {
self.wake.check()?;
let mut feed = lock(&self.feed, "subscription delivery")?;
for _ in 0..feed.prefixes.len() {
let index = feed.next;
feed.next = (feed.next + 1) % feed.prefixes.len();
let prefix = &mut feed.prefixes[index];
for _ in 0..prefix.subjects.len() {
let index = prefix.next;
prefix.next = (prefix.next + 1) % prefix.subjects.len();
let source = &prefix.subjects[index];
if let Some((payload, position)) = source.reader.next(&source.subject)? {
let pending = PendingDelivery {
subject: source.subject.clone(),
schema: source.schema,
payload,
position,
reader: source.reader.clone(),
};
drop(feed);
return Ok(Some(pending));
}
}
}
drop(feed);
Ok(None)
}
}
#[cfg(test)]
mod tests {
use super::{Arc, ExportSubscriptions, Observation, Wake};
#[test]
fn poisoned_feed_refuses_delivery_and_stops_binding() -> Result<(), String> {
let wake = Arc::new(Wake::default());
let feeds = Arc::new(
ExportSubscriptions::new(
"orders-link",
&["orders".to_owned()],
Arc::clone(&wake),
Arc::new(Observation::default()),
)
.map_err(|error| error.to_string())?,
);
let worker = Arc::clone(&feeds);
assert!(
std::thread::spawn(move || {
let guard = worker.feed.lock().map_err(|_| "already poisoned")?;
assert!(std::hint::black_box(false), "poison fixture");
drop(guard);
Ok::<(), &str>(())
})
.join()
.is_err()
);
assert!(
feeds.next_ready().is_err(),
"poisoned feed must refuse delivery"
);
assert!(feeds.binding("orders").is_none());
assert!(
wake.stopping(),
"infallible binding must stop a faulty bridge"
);
Ok(())
}
}