1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
//! Chain subscription: replay-from-sequence plus live events (ADR-103 D).
//!
//! [`ChainManager::subscribe`](crate::chain::ChainManager::subscribe)
//! returns a [`ChainSubscription`]. It yields the retained events from
//! `from_seq`, then live events, with no gap and no duplicate between the
//! two:
//!
//! 1. under the chain lock, `subscribe` creates the live receiver and
//! records the snapshot end (the chain's next sequence); every append
//! publishes to the hub under that same lock, so they cannot interleave;
//! 2. the replay is then paged in short lock holds (`REPLAY_PAGE` events,
//! at most `REPLAY_SCAN` scanned per hold) over `[from_seq, end)`;
//! 3. live events below `end` are dropped (belt and braces).
//!
//! A deep `from_seq` therefore never clones the chain under the lock, and a
//! `from_seq` older than the replay window is refused with a typed error.
//!
//! Live delivery uses a bounded `tokio::sync::broadcast`. `send` never
//! blocks, so a stalled subscriber cannot slow an append; it falls behind,
//! the oldest items are overwritten, and its next read yields
//! [`ChainItem::Lagged`] carrying `resume_from`, the sequence to
//! resubscribe from to recover. The subscription holds only a `Weak`
//! reference to the manager and ends when the manager is dropped.
use crate::chain::ChainEvent;
/// Live buffer per subscriber (events). A subscriber slower than this
/// many events behind sees [`ChainItem::Lagged`].
pub const CHAIN_SUBSCRIBE_CAPACITY: usize = 1024;
/// Default replay window: the furthest back (in events behind the head) a
/// subscriber may start. Adjust with `ChainManager::set_max_replay_window`.
pub const DEFAULT_MAX_REPLAY_WINDOW: u64 = 100_000;
/// Events returned per replay lock hold.
#[cfg(feature = "native")]
const REPLAY_PAGE: usize = 256;
/// Retained events examined per replay lock hold (bounds hold time when
/// `kind_prefix` rejects most events).
#[cfg(feature = "native")]
const REPLAY_SCAN: usize = 4096;
/// Which events a subscription yields.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ChainFilter {
/// Only events whose `kind` starts with this prefix (`None` = all).
pub kind_prefix: Option<String>,
/// Replay retained events with `sequence >= from_seq` before going
/// live. `None` = live only.
pub from_seq: Option<u64>,
}
impl ChainFilter {
#[cfg(feature = "native")]
fn matches(&self, e: &ChainEvent) -> bool {
self.kind_prefix
.as_deref()
.is_none_or(|p| e.kind.starts_with(p))
}
}
/// Why a subscription was refused.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum SubscribeError {
/// `from_seq` is further back than the replay window allows.
#[error("from_seq {requested} is older than the replay window (oldest allowed {oldest_allowed})")]
ReplayWindowExceeded { requested: u64, oldest_allowed: u64 },
}
/// One item of a subscription.
#[derive(Debug, Clone)]
// Public fan-out item cloned per subscriber; boxing the event would change the public API and add an allocation per event.
#[allow(clippy::large_enum_variant)]
pub enum ChainItem {
/// A chain event matching the filter.
Event(ChainEvent),
/// The subscriber fell behind and `missed` events were dropped.
/// Resubscribe with `from_seq = resume_from` to recover them.
Lagged { missed: u64, resume_from: u64 },
}
/// Live fan-out owned by the chain. No-op without the `native` feature.
#[derive(Debug)]
pub(crate) struct EventHub {
#[cfg(feature = "native")]
tx: tokio::sync::broadcast::Sender<ChainEvent>,
}
impl Default for EventHub {
fn default() -> Self {
Self {
#[cfg(feature = "native")]
tx: tokio::sync::broadcast::channel(CHAIN_SUBSCRIBE_CAPACITY).0,
}
}
}
impl EventHub {
/// Publish an appended event. Call under the chain lock, before the
/// next append. Never blocks; clones only when someone is listening.
#[cfg_attr(not(feature = "native"), allow(unused_variables))]
pub(crate) fn publish(&self, event: &ChainEvent) {
#[cfg(feature = "native")]
if self.tx.receiver_count() > 0 {
let _ = self.tx.send(event.clone());
}
}
#[cfg(feature = "native")]
pub(crate) fn subscribe(&self) -> tokio::sync::broadcast::Receiver<ChainEvent> {
self.tx.subscribe()
}
}
/// A replay-then-live stream of chain events.
#[cfg(feature = "native")]
#[derive(Debug)]
pub struct ChainSubscription {
filter: ChainFilter,
chain: std::sync::Weak<crate::chain::ChainManager>,
/// Next sequence to page from the replay; replay is done at `end`.
cursor: u64,
/// Snapshot end: replay covers `[from_seq, end)`, live starts at `end`.
end: u64,
page: std::collections::VecDeque<ChainEvent>,
rx: tokio::sync::broadcast::Receiver<ChainEvent>,
/// Sequence after the last event this subscription has passed.
resume_from: u64,
}
#[cfg(feature = "native")]
impl ChainSubscription {
pub(crate) fn new(
filter: ChainFilter,
chain: std::sync::Weak<crate::chain::ChainManager>,
rx: tokio::sync::broadcast::Receiver<ChainEvent>,
end: u64,
) -> Self {
let cursor = filter.from_seq.unwrap_or(end).min(end);
Self {
filter,
chain,
cursor,
end,
page: Default::default(),
rx,
resume_from: cursor,
}
}
/// Next item, or `None` when the chain is gone.
pub async fn next(&mut self) -> Option<ChainItem> {
use tokio::sync::broadcast::error::RecvError;
loop {
if let Some(e) = self.page.pop_front() {
self.resume_from = e.sequence + 1;
return Some(ChainItem::Event(e));
}
if self.cursor >= self.end {
break;
}
let chain = self.chain.upgrade()?;
let (page, next) = chain.replay_page(
self.cursor,
self.end,
self.filter.kind_prefix.as_deref(),
REPLAY_PAGE,
REPLAY_SCAN,
);
// Always make progress; a page that did not advance means
// nothing retained is left below `end`.
self.cursor = if next > self.cursor { next } else { self.end };
self.page.extend(page);
}
loop {
match self.rx.recv().await {
Ok(e) if e.sequence < self.end => continue,
Ok(e) => {
self.resume_from = e.sequence + 1;
if !self.filter.matches(&e) {
continue;
}
return Some(ChainItem::Event(e));
}
Err(RecvError::Lagged(n)) => {
return Some(ChainItem::Lagged {
missed: n,
resume_from: self.resume_from,
});
}
Err(RecvError::Closed) => return None,
}
}
}
}