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
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
//! The existing channel publish hook and its measured fan-out.
use super::{
Admitted, Arc, Hop, Instant, Ordering, PublishFact, Reached, Recipient, Tap, TapError,
TapIdent, TapMessage, TapSource, elapsed_ms, gather, lock,
};
impl Tap {
/// The hook: one publish return. With no subscription it is one atomic
/// load and nothing else. Otherwise each matching subscription decides
/// under its own lock whether to keep the message before anything is
/// allocated for it; a kept message allocates its subject and its hops
/// (and its reply-to, and a payload copy under a grant), and the drainer
/// is woken once, after every tap lock is released. Answers whether any
/// subscription kept it.
///
/// # Errors
/// Returns [`TapError::Poisoned`] when a tap lock is poisoned; any
/// subscription that kept the message before it still wakes the drainer.
/// Returns [`TapError::StationsDisagree`] when its station list was
/// refused: the message is kept, its receivers not recorded. Returns
/// [`TapError::HopsOutOfOrder`] when a hop was measured before the one
/// ahead of it: the message is kept with its hops as measured. Returns
/// [`TapError::HopBeforeFirstSight`] or [`TapError::HopOverflow`] when a
/// hop could not be carried: the message is kept without it. More than
/// one of these is [`TapError::Several`], each named.
pub fn observe(&self, fact: &PublishFact<'_>) -> Result<bool, TapError> {
self.observe_from(fact, None)
}
pub(crate) fn observe_embedded(
&self,
fact: &PublishFact<'_>,
host: &str,
) -> Result<bool, TapError> {
self.observe_from(fact, Some(host))
}
fn observe_from(&self, fact: &PublishFact<'_>, host: Option<&str>) -> Result<bool, TapError> {
if self.shared.active.load(Ordering::Acquire) == 0 {
return Ok(false);
}
let observation = self.shared.observed.fetch_add(1, Ordering::Relaxed);
#[cfg(test)]
if fact.source == TapSource::Server && host.is_none() {
self.shared.observed_server.fetch_add(1, Ordering::Relaxed);
}
let mut kept = false;
let result = self.keep(fact, observation, &mut kept, host);
if kept && let Some(waker) = self.shared.waker.get() {
waker();
}
result.map(|()| kept)
}
/// The station `source` publishes as, when it is a registered worker.
pub(super) fn station_of(&self, source: TapSource) -> Result<Option<Arc<str>>, TapError> {
Ok(match source {
TapSource::Connection(connection) => lock(&self.shared.stations, "tap stations")?
.get(&connection)
.map(Arc::clone),
TapSource::Server => None,
})
}
/// The stations `recipients` belong to: each registered worker once, in
/// order, with the instant its first recipient was queued, in one
/// allocation; the shared empty list when none is one, which
/// allocates nothing. A first pass counts them; the second fills a slice
/// of exactly that length, so the shared list is built in place.
fn stations_of(&self, recipients: &[Recipient]) -> Result<Arc<[Reached]>, TapError> {
if recipients.is_empty() {
return Ok(Arc::clone(&self.shared.no_stations));
}
let receivers = lock(&self.shared.receivers, "tap receivers")?;
let stations = lock(&self.shared.stations, "tap stations")?;
let station_at = |index: usize| {
recipients
.get(index)
.and_then(|(subscriber, _)| receivers.get(subscriber))
.and_then(|connection| stations.get(connection))
};
// The recipient at `index` names a station no earlier one named, and
// was queued at the instant it carries.
let first_at = |index: usize| {
station_at(index)
.filter(|station| !(0..index).any(|earlier| station_at(earlier) == Some(*station)))
.zip(recipients.get(index).map(|(_, queued_at)| *queued_at))
};
let named = (0..recipients.len())
.filter(|index| first_at(*index).is_some())
.count();
let list = gather(recipients.len(), named, first_at)?
.unwrap_or_else(|| Arc::clone(&self.shared.no_stations));
drop(stations);
drop(receivers);
Ok(list)
}
/// A kept message's hops, in one allocation of no more than their
/// number (lane brief rulings 3 and 8): its publishing station at 0;
/// this server's depot as a durable append completed, when it keeps a
/// persistent store; the exchange (this node) once the message was
/// appended, so an append to an in-memory store sits inside its time;
/// then each station it reached as its first inbox queued it. No exchange
/// hop: nothing measured the fan-out. A publish that appended nothing
/// took no exchange instant, which is normal and not a failure (Jules's
/// F1, Hermes 07:22), so its exchange is left out, never written at 0.
/// Each is measured from `seen_at` and kept as measured, never rewritten
/// (Hermes 04:09 and 04:27). Into `failures`, when given, go the first
/// hop measured before first sight and the first beyond u64
/// milliseconds, each left out of the hops, and the first hop measured
/// before the hop kept ahead of it, kept; each names its place in the
/// chain.
fn hops(
&self,
fact: &PublishFact<'_>,
seen_at: Instant,
station: Option<&Arc<str>>,
reached: Option<&[Reached]>,
failures: Option<&mut Vec<TapError>>,
) -> Vec<Hop> {
let reached = reached.unwrap_or_default();
let appended_at = fact.recorded.and_then(|recorded| recorded.appended_at);
let depot = self.depot.as_ref().zip(appended_at);
let mut hops = Vec::with_capacity(
usize::from(station.is_some())
+ usize::from(depot.is_some())
+ usize::from(appended_at.is_some())
+ reached.len(),
);
let mut early = None;
let mut overflow = None;
let mut out_of_order = None;
let mut place = 0;
let mut pass = |node: &Arc<str>, at: Option<Instant>| {
let hop = place;
place += 1;
let at_ms = match at {
None => 0,
Some(at) if at < seen_at => {
early.get_or_insert_with(|| {
elapsed_ms(seen_at.duration_since(at))
.map_or(TapError::HopOverflow { hop }, |early_ms| {
TapError::HopBeforeFirstSight { hop, early_ms }
})
});
return;
}
Some(at) => {
let Some(at_ms) = elapsed_ms(at.duration_since(seen_at)) else {
overflow.get_or_insert(TapError::HopOverflow { hop });
return;
};
at_ms
}
};
if let Some(before) = hops.last().map(|kept: &Hop| kept.at_ms)
&& at_ms < before
{
out_of_order.get_or_insert(TapError::HopsOutOfOrder {
hop,
at_ms,
before_ms: before,
});
}
hops.push(Hop {
node: Arc::clone(node),
at_ms,
});
};
if let Some(station) = station {
pass(station, None);
}
if let Some((depot, appended_at)) = depot {
pass(depot, Some(appended_at));
}
if let Some(appended_at) = appended_at {
pass(&self.node, Some(appended_at));
}
for reached in reached {
pass(&reached.station, Some(reached.queued_at));
}
if let Some(failures) = failures {
failures.extend([early, overflow, out_of_order].into_iter().flatten());
}
hops
}
fn keep(
&self,
fact: &PublishFact<'_>,
observation: u64,
kept: &mut bool,
host: Option<&str>,
) -> Result<(), TapError> {
let mut seen_at = None;
// Looked up once, and only when a subscription keeps the message.
let mut station = None;
let mut to_stations = None;
// A station list refused, and hops that could not be kept as
// measured, are each reported once the message is kept (Buttercup's
// V3): empty, this allocates nothing.
let mut failures = Vec::new();
let mut hops_reported = false;
let subscribers = lock(&self.shared.subscribers, "tap subscribers")?;
for subscriber in subscribers.iter() {
if !subscriber.filter.matches(fact.subject) {
continue;
}
let mut state = lock(&subscriber.state, "tap subscriber")?;
if !state.admits() {
continue;
}
let one_in = state.one_in;
if station.is_none() {
station = Some(match host {
Some(host) => Some(Arc::from(format!("liminal.{host}"))),
None => self.station_of(fact.source)?,
});
to_stations = Some(
match fact
.recorded
.map(|recorded| self.stations_of(recorded.recipients))
{
Some(Ok(list)) => Some(list),
// Kept as not recorded, and the refusal reported.
Some(Err(error @ TapError::StationsDisagree { .. })) => {
failures.push(error);
None
}
Some(Err(error)) => return Err(error),
None => None,
},
);
}
// First sight is receipt (lane brief ruling 6), or now when the
// publish recorded nothing.
let first_sight = *seen_at.get_or_insert_with(|| {
fact.recorded
.map_or_else(Instant::now, |recorded| recorded.received_at)
});
// Every keeping subscription's hops are the same; their failures
// are taken once.
let hops = self.hops(
fact,
first_sight,
station.as_ref().and_then(Option::as_ref),
to_stations.as_ref().and_then(Option::as_deref),
(!hops_reported).then_some(&mut failures),
);
hops_reported = true;
let message = TapMessage {
ident: fact
.message_id
.map_or(TapIdent::Observation(observation), TapIdent::Message),
subject: fact.subject.to_owned(),
size: u64::try_from(fact.payload.len()).unwrap_or(u64::MAX),
outcome: fact.outcome,
seen_at: first_sight,
source: fact.source,
station: station.clone().flatten(),
to_stations: to_stations.clone().flatten(),
hops,
reply_to: fact.reply_to.map(str::to_owned),
held: false,
call: None,
payload: subscriber.grant.map(|_| fact.payload.to_vec()),
};
if state.ring.push(Admitted { one_in, message }) {
self.shared.dropped_total.fetch_add(1, Ordering::Relaxed);
}
state.after_push();
drop(state);
*kept = true;
}
drop(subscribers);
match failures.len() {
0 => Ok(()),
1 => failures.pop().map_or(Ok(()), Err),
_ => Err(TapError::Several(failures)),
}
}
}