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 /// The correlated reply arrived and was released.
52 Answered,
53 /// No reply arrived before the call deadline.
54 Unanswered,
55 /// Accepted, and no subscriber matched it.
56 NoSubscriber,
57 /// Accepted; subscribers matched and none took it.
58 NoneAccepted,
59 /// The append's outcome is not known.
60 Unresolved,
61 /// Refused (admission, schema, or a library failure).
62 Refused,
63}
64
65/// Who published.
66#[derive(Clone, Copy, Debug, PartialEq, Eq)]
67pub enum TapSource {
68 /// A connection, by its id.
69 Connection(u64),
70 /// An in-process publisher: the server or an embedded host named by its station.
71 Server,
72}
73
74/// How a tapped message is named: by the server's message id when it was
75/// admitted, by the tap's observation number when it was not.
76#[derive(Clone, Copy, Debug, PartialEq, Eq)]
77pub enum TapIdent {
78 /// The transport-scoped message id an admitted publish was given.
79 Message(u64),
80 /// A connection-local call operation, separate from channel message ids.
81 Call {
82 /// Calling connection.
83 connection: u64,
84 /// Correlation operation.
85 operation: u64,
86 },
87 /// The tap's own count of the publish, for one that has no message id.
88 Observation(u64),
89}
90
91/// One publish as the hook hands it to the tap. Borrowed throughout: the tap
92/// copies only what an admitted message keeps.
93#[derive(Clone, Copy, Debug)]
94pub struct PublishFact<'a> {
95 /// The subject (channel name).
96 pub subject: &'a str,
97 /// The payload; its length is the message's size, and it is copied only
98 /// for a subscription holding a [`PayloadGrant`].
99 pub payload: &'a [u8],
100 /// What became of it.
101 pub outcome: TapOutcome,
102 /// The message id, when the publish was admitted and given one.
103 pub message_id: Option<u64>,
104 /// Who published.
105 pub source: TapSource,
106 /// The id of the message it answers (the envelope's causal parent).
107 pub reply_to: Option<&'a str>,
108 /// What the publish recorded; `None` when it recorded nothing: no tap
109 /// was live when it began, or the tap could not lend it a record. The
110 /// tap then first sees the message as it keeps it, and knows neither
111 /// whom it reached nor when.
112 pub recorded: Option<Recorded<'a>>,
113}
114
115/// What a recording publish saw, all together, so a fact can never carry
116/// recipients' queue instants without the receipt they are measured from
117/// (Hermes 7 Oct 06:55).
118#[derive(Clone, Copy, Debug)]
119pub struct Recorded<'a> {
120 /// The subscriber pid of every local subscription whose inbox queued the
121 /// message, with the instant it did, as the library's fan-out recorded
122 /// them; empty when it reached none.
123 pub recipients: &'a [Recipient],
124 /// When the server received the publish, before admission: the message's
125 /// first sight (lane brief ruling 6).
126 pub received_at: Instant,
127 /// When a durable channel's append of the message completed, before its
128 /// fan-out; `None` for one that is not durable.
129 pub appended_at: Option<Instant>,
130}
131
132/// A station a message reached: its id, shared rather than copied, and the
133/// instant the first of its subscriptions' inboxes queued the message.
134#[derive(Clone, Debug, PartialEq, Eq)]
135pub struct Reached {
136 /// The station.
137 pub station: Arc<str>,
138 /// When its first subscription's inbox queued the message.
139 pub queued_at: Instant,
140}
141
142/// One point a message crossed: the node, shared rather than copied, and how
143/// long after the message was first seen.
144#[derive(Clone, Debug, PartialEq, Eq)]
145pub struct Hop {
146 /// The node the message crossed.
147 pub node: Arc<str>,
148 /// Milliseconds after the message was first seen.
149 pub at_ms: u64,
150}
151
152/// The per-message facts a tap keeps.
153#[derive(Clone, Debug, PartialEq, Eq)]
154pub struct TapMessage {
155 /// Its id.
156 pub ident: TapIdent,
157 /// The subject (channel name).
158 pub subject: String,
159 /// Payload size in bytes.
160 pub size: u64,
161 /// What became of it.
162 pub outcome: TapOutcome,
163 /// When the server first saw it, on the monotonic clock; the record's
164 /// wall time is the tap's wall base plus the elapsed time to here.
165 pub seen_at: Instant,
166 /// Who published it.
167 pub source: TapSource,
168 /// The station that published it: the registered worker identity of
169 /// the publishing connection, shared rather than copied. `None` for a
170 /// connection that is not a registered worker, and for the server. An embedded
171 /// host carries its named publisher identity here.
172 pub station: Option<Arc<str>>,
173 /// The stations that received it: each registered worker whose
174 /// subscription's inbox queued the message, once, in the fan-out's order,
175 /// shared by every tap that kept it: empty when the recorded fan-out
176 /// reached no registered worker. `None` when its recipients were not
177 /// recorded, so whom it reached is not known.
178 pub to_stations: Option<Arc<[Reached]>>,
179 /// Where it went, in order.
180 pub hops: Vec<Hop>,
181 /// The id of the message it answers, when it has one.
182 pub reply_to: Option<String>,
183 /// Held by a rule. Always `false` until rules (R4) exist.
184 pub held: bool,
185 /// The payload, only on a subscription carrying a [`PayloadGrant`].
186 pub payload: Option<Vec<u8>>,
187 /// Terminal call facts; absent on a channel publish.
188 pub call: Option<CallDetails>,
189}
190
191/// What a drain hands the subscriber: messages that were all admitted at
192/// one divisor.
193#[derive(Clone, Debug, PartialEq, Eq)]
194pub struct TapBatch {
195 /// One more than this subscription's previous batch, from 1.
196 pub seq: u64,
197 /// N in "sampled 1 in N" for every message in this batch.
198 pub sampled: NonZeroU32,
199 /// Messages evicted from the subscription's ring since its previous
200 /// batch.
201 pub dropped: u64,
202 /// The messages, oldest first.
203 pub messages: Vec<TapMessage>,
204}
205
206/// Terminal call fields the console can read without a payload grant.
207#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
208pub struct CallDetails {
209 /// Request receipt to correlated reply receipt, in milliseconds.
210 #[serde(skip_serializing_if = "Option::is_none")]
211 pub latency_ms: Option<u64>,
212 /// The server's named refusal, when it refused the call.
213 #[serde(skip_serializing_if = "Option::is_none")]
214 pub refusal: Option<String>,
215}
216
217/// One call as seen at its terminal correlation or deadline.
218#[derive(Clone, Copy, Debug)]
219pub struct CallFact<'a> {
220 /// Called subject.
221 pub subject: &'a str,
222 /// Request payload bytes.
223 pub size: u64,
224 /// Calling connection.
225 pub connection: u64,
226 /// Pending-reply operation, when admitted.
227 pub operation: Option<u64>,
228 /// The calling station, captured at receipt.
229 pub station: Option<&'a Arc<str>>,
230 /// The serving station, captured at dispatch.
231 pub served_by: Option<&'a Arc<str>>,
232 /// Request receipt.
233 pub received_at: Instant,
234 /// Dispatch to the serving station.
235 pub forwarded_at: Instant,
236 /// Reply, refusal or deadline observation.
237 pub finished_at: Instant,
238 /// Terminal outcome.
239 pub outcome: TapOutcome,
240 /// The server's refusal words.
241 pub refusal: Option<&'a str>,
242}