liminal_server/tap/message.rs
1//! What a tap carries: the filter, the grant, the publish it is shown, and the
2//! per-message facts it keeps.
3
4use liminal::channel::Recipient;
5use std::num::NonZeroU32;
6use std::sync::Arc;
7use std::time::Instant;
8
9/// Which messages a subscription wants: those whose subject (the channel
10/// name) equals `subject`, or starts with its prefix when it ends in `*`.
11/// An empty subject or a lone `*` matches everything.
12#[derive(Clone, Debug, Default, PartialEq, Eq)]
13pub struct TapFilter {
14 /// Exact subject, or a prefix followed by `*`.
15 pub subject: String,
16}
17
18impl TapFilter {
19 /// Whether `subject` is inside this filter.
20 #[must_use]
21 pub fn matches(&self, subject: &str) -> bool {
22 self.subject.strip_suffix('*').map_or_else(
23 || self.subject.is_empty() || self.subject == subject,
24 |prefix| subject.starts_with(prefix),
25 )
26 }
27}
28
29/// The right to see payloads on a tap. Nothing in the server mints one until
30/// grants (SRV-011 R4) exist, so no payload leaves through a tap today.
31#[derive(Clone, Copy, Debug, PartialEq, Eq)]
32pub struct PayloadGrant {
33 _sealed: (),
34}
35
36impl PayloadGrant {
37 /// Mints a grant. Test-only until R4.
38 #[cfg(test)]
39 #[must_use]
40 pub(crate) const fn server_issued() -> Self {
41 Self { _sealed: () }
42 }
43}
44
45/// What became of a tapped publish, by the cause the server knows. A
46/// dedup-suppressed re-publish is never tapped: its original was.
47#[derive(Clone, Copy, Debug, PartialEq, Eq)]
48pub enum TapOutcome {
49 /// Accepted and taken by at least one subscriber.
50 Delivered,
51 /// Accepted, and no subscriber matched it.
52 NoSubscriber,
53 /// Accepted; subscribers matched and none took it.
54 NoneAccepted,
55 /// The append's outcome is not known.
56 Unresolved,
57 /// Refused (admission, schema, or a library failure).
58 Refused,
59}
60
61/// Who published.
62#[derive(Clone, Copy, Debug, PartialEq, Eq)]
63pub enum TapSource {
64 /// A connection, by its id.
65 Connection(u64),
66 /// The server itself (`liminal.<device>`).
67 Server,
68}
69
70/// How a tapped message is named: by the server's message id when it was
71/// admitted, by the tap's observation number when it was not.
72#[derive(Clone, Copy, Debug, PartialEq, Eq)]
73pub enum TapIdent {
74 /// The transport-scoped message id an admitted publish was given.
75 Message(u64),
76 /// The tap's own count of the publish, for one that has no message id.
77 Observation(u64),
78}
79
80/// One publish as the hook hands it to the tap. Borrowed throughout: the tap
81/// copies only what an admitted message keeps.
82#[derive(Clone, Copy, Debug)]
83pub struct PublishFact<'a> {
84 /// The subject (channel name).
85 pub subject: &'a str,
86 /// The payload; its length is the message's size, and it is copied only
87 /// for a subscription holding a [`PayloadGrant`].
88 pub payload: &'a [u8],
89 /// What became of it.
90 pub outcome: TapOutcome,
91 /// The message id, when the publish was admitted and given one.
92 pub message_id: Option<u64>,
93 /// Who published.
94 pub source: TapSource,
95 /// The id of the message it answers (the envelope's causal parent).
96 pub reply_to: Option<&'a str>,
97 /// What the publish recorded; `None` when it recorded nothing: no tap
98 /// was live when it began, or the tap could not lend it a record. The
99 /// tap then first sees the message as it keeps it, and knows neither
100 /// whom it reached nor when.
101 pub recorded: Option<Recorded<'a>>,
102}
103
104/// What a recording publish saw, all together, so a fact can never carry
105/// recipients' queue instants without the receipt they are measured from
106/// (Hermes 7 Oct 06:55).
107#[derive(Clone, Copy, Debug)]
108pub struct Recorded<'a> {
109 /// The subscriber pid of every local subscription whose inbox queued the
110 /// message, with the instant it did, as the library's fan-out recorded
111 /// them; empty when it reached none.
112 pub recipients: &'a [Recipient],
113 /// When the server received the publish, before admission: the message's
114 /// first sight (lane brief ruling 6).
115 pub received_at: Instant,
116 /// When a durable channel's append of the message completed, before its
117 /// fan-out; `None` for one that is not durable.
118 pub appended_at: Option<Instant>,
119}
120
121/// A station a message reached: its id, shared rather than copied, and the
122/// instant the first of its subscriptions' inboxes queued the message.
123#[derive(Clone, Debug, PartialEq, Eq)]
124pub struct Reached {
125 /// The station.
126 pub station: Arc<str>,
127 /// When its first subscription's inbox queued the message.
128 pub queued_at: Instant,
129}
130
131/// One point a message crossed: the node, shared rather than copied, and how
132/// long after the message was first seen.
133#[derive(Clone, Debug, PartialEq, Eq)]
134pub struct Hop {
135 /// The node the message crossed.
136 pub node: Arc<str>,
137 /// Milliseconds after the message was first seen.
138 pub at_ms: u64,
139}
140
141/// The per-message facts a tap keeps.
142#[derive(Clone, Debug, PartialEq, Eq)]
143pub struct TapMessage {
144 /// Its id.
145 pub ident: TapIdent,
146 /// The subject (channel name).
147 pub subject: String,
148 /// Payload size in bytes.
149 pub size: u64,
150 /// What became of it.
151 pub outcome: TapOutcome,
152 /// When the server first saw it, on the monotonic clock; the record's
153 /// wall time is the tap's wall base plus the elapsed time to here.
154 pub seen_at: Instant,
155 /// Who published it.
156 pub source: TapSource,
157 /// The station that published it: the registered worker identity of
158 /// the publishing connection, shared rather than copied. `None` for a
159 /// connection that is not a registered worker, and for the server.
160 pub station: Option<Arc<str>>,
161 /// The stations that received it: each registered worker whose
162 /// subscription's inbox queued the message, once, in the fan-out's order,
163 /// shared by every tap that kept it: empty when the recorded fan-out
164 /// reached no registered worker. `None` when its recipients were not
165 /// recorded, so whom it reached is not known.
166 pub to_stations: Option<Arc<[Reached]>>,
167 /// Where it went, in order.
168 pub hops: Vec<Hop>,
169 /// The id of the message it answers, when it has one.
170 pub reply_to: Option<String>,
171 /// Held by a rule. Always `false` until rules (R4) exist.
172 pub held: bool,
173 /// The payload, only on a subscription carrying a [`PayloadGrant`].
174 pub payload: Option<Vec<u8>>,
175}
176
177/// What a drain hands the subscriber: messages that were all admitted at
178/// one divisor.
179#[derive(Clone, Debug, PartialEq, Eq)]
180pub struct TapBatch {
181 /// One more than this subscription's previous batch, from 1.
182 pub seq: u64,
183 /// N in "sampled 1 in N" for every message in this batch.
184 pub sampled: NonZeroU32,
185 /// Messages evicted from the subscription's ring since its previous
186 /// batch.
187 pub dropped: u64,
188 /// The messages, oldest first.
189 pub messages: Vec<TapMessage>,
190}