liminal/protocol/
worker_refusal.rs

1//! A worker's closed refusal vocabulary and its canonical wire record.
2
3use core::fmt;
4use serde_json::{Value, json};
5
6use super::{Frame, ProtocolError};
7
8/// A refused reply carries a canonical record instead of application bytes.
9pub const PUSH_REPLY_REFUSED_FLAG: u8 = 0x01;
10
11/// Wire-configured UTF-8 byte ceiling, keeping refusal allocation bounded while
12/// allowing a useful explanation. Peers must share this protocol value.
13pub const WORKER_REFUSAL_WORDS_MAX: usize = 1024;
14
15/// Worst canonical record: every words byte requires a six-byte JSON escape,
16/// plus the fixed keys, delimiters and longest code. Checked before parsing so
17/// a worker cannot force an unbounded JSON allocation.
18const RECORD_BYTES_MAX: usize =
19    6 * WORKER_REFUSAL_WORDS_MAX + "{\"code\":\"unavailable\",\"words\":\"\"}".len();
20
21/// The worker's reason, distinct from a route's admission refusal.
22#[derive(Clone, Copy, Debug, PartialEq, Eq)]
23pub enum WorkerRefusalCode {
24    /// The worker cannot admit more work now.
25    Busy,
26    /// The request cannot be served as supplied.
27    Invalid,
28    /// A dependency or the worker is unavailable.
29    Unavailable,
30    /// The worker encountered an internal failure.
31    Internal,
32}
33
34impl WorkerRefusalCode {
35    /// The complete vocabulary, in wire-code order.
36    pub const ALL: [Self; 4] = [Self::Busy, Self::Invalid, Self::Unavailable, Self::Internal];
37
38    /// The name carried in the canonical worker record.
39    #[must_use]
40    pub const fn name(self) -> &'static str {
41        match self {
42            Self::Busy => "busy",
43            Self::Invalid => "invalid",
44            Self::Unavailable => "unavailable",
45            Self::Internal => "internal",
46        }
47    }
48
49    /// The caller's stable named refusal.
50    #[must_use]
51    pub const fn caller_name(self) -> &'static str {
52        match self {
53            Self::Busy => "worker_busy",
54            Self::Invalid => "worker_invalid",
55            Self::Unavailable => "worker_unavailable",
56            Self::Internal => "worker_internal",
57        }
58    }
59
60    /// Numeric representation on the existing conversation-error wire. The
61    /// words remain entirely application supplied; classification uses this
62    /// registry, never a prefix guessed from those words.
63    #[must_use]
64    pub const fn reason_code(self) -> u16 {
65        match self {
66            Self::Busy => 0x0401,
67            Self::Invalid => 0x0402,
68            Self::Unavailable => 0x0403,
69            Self::Internal => 0x0404,
70        }
71    }
72
73    /// Recognises only the four allocated conversation-error reasons.
74    #[must_use]
75    pub fn from_reason_code(reason: u16) -> Option<Self> {
76        Self::ALL
77            .into_iter()
78            .find(|code| code.reason_code() == reason)
79    }
80
81    fn parse(name: &str) -> Result<Self, ProtocolError> {
82        Self::ALL
83            .into_iter()
84            .find(|code| code.name() == name)
85            .ok_or_else(|| ProtocolError::codec("worker_refusal_unknown_code"))
86    }
87}
88
89/// A validated refusal. Words are data and are never included in `Debug`.
90#[derive(Clone, PartialEq, Eq)]
91pub struct WorkerRefusal {
92    code: WorkerRefusalCode,
93    words: String,
94}
95
96impl WorkerRefusal {
97    /// Validates the closed code and the byte bound before retaining words.
98    ///
99    /// # Errors
100    /// An unknown code or words exceeding the configured wire ceiling.
101    pub fn new(code: &str, words: impl AsRef<str>) -> Result<Self, ProtocolError> {
102        let code = WorkerRefusalCode::parse(code)?;
103        let words = words.as_ref();
104        if words.len() > WORKER_REFUSAL_WORDS_MAX {
105            return Err(ProtocolError::codec(format!(
106                "worker_refusal_words_too_long: words exceed {WORKER_REFUSAL_WORDS_MAX} UTF-8 bytes"
107            )));
108        }
109        Ok(Self {
110            code,
111            words: words.to_owned(),
112        })
113    }
114
115    /// The validated worker reason.
116    #[must_use]
117    pub const fn code(&self) -> WorkerRefusalCode {
118        self.code
119    }
120
121    /// The worker's words, unchanged.
122    #[must_use]
123    pub fn words(&self) -> &str {
124        &self.words
125    }
126
127    /// Transfers the words to the caller without another allocation.
128    #[must_use]
129    pub fn into_words(self) -> String {
130        self.words
131    }
132
133    /// Compact JSON through a sorted value, the console's canonical form.
134    ///
135    /// # Errors
136    /// The canonical JSON encoder failed; its input is never quoted.
137    pub fn canonical_bytes(&self) -> Result<Vec<u8>, ProtocolError> {
138        serde_json::to_vec(&json!({"code": self.code.name(), "words": self.words}))
139            .map_err(|_| ProtocolError::codec("worker_refusal_encode_failed"))
140    }
141
142    /// Accepts only the exact canonical record, including its member order and
143    /// spelling. Comparing the re-encoding also rejects duplicate members and
144    /// alternate escapes without ever copying their values into an error.
145    ///
146    /// # Errors
147    /// An oversized, malformed, noncanonical or out-of-vocabulary record.
148    pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
149        if payload.len() > RECORD_BYTES_MAX {
150            return Err(ProtocolError::codec("worker_refusal_record_too_large"));
151        }
152        let value: Value = serde_json::from_slice(payload)
153            .map_err(|_| ProtocolError::codec("worker_refusal_malformed_record"))?;
154        let object = value
155            .as_object()
156            .filter(|object| object.len() == 2)
157            .ok_or_else(|| ProtocolError::codec("worker_refusal_invalid_fields"))?;
158        let code = object
159            .get("code")
160            .and_then(Value::as_str)
161            .ok_or_else(|| ProtocolError::codec("worker_refusal_invalid_code"))?;
162        let words = object
163            .get("words")
164            .and_then(Value::as_str)
165            .ok_or_else(|| ProtocolError::codec("worker_refusal_invalid_words"))?;
166        let refusal = Self::new(code, words)?;
167        if refusal.canonical_bytes()? != payload {
168            return Err(ProtocolError::codec("worker_refusal_noncanonical_record"));
169        }
170        Ok(refusal)
171    }
172}
173
174impl fmt::Debug for WorkerRefusal {
175    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
176        formatter
177            .debug_struct("WorkerRefusal")
178            .field("code", &self.code)
179            .field("words_bytes", &self.words.len())
180            .finish()
181    }
182}
183
184// Debug is kept beside the refusal so its opaque bytes share the redaction boundary.
185mod frame_debug {
186    use super::Frame;
187    use core::fmt;
188
189    /// An opaque field: its length, never its bytes.
190    struct ByteLength(usize);
191
192    impl fmt::Debug for ByteLength {
193        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
194            write!(formatter, "{} bytes", self.0)
195        }
196    }
197
198    macro_rules! frame_debug_field {
199        ($out:ident, envelope, $value:expr) => {
200            $out.field("envelope_payload", &ByteLength($value.payload.len()));
201        };
202        ($out:ident, payload, $value:expr) => {
203            $out.field("payload", &ByteLength($value.len()));
204        };
205        ($out:ident, sealed, $value:expr) => {
206            $out.field("sealed", &ByteLength($value.len()));
207        };
208        ($out:ident, $field:ident, $value:expr) => {
209            $out.field(stringify!($field), $value);
210        };
211    }
212
213    macro_rules! frame_debug {
214    ($($variant:ident { $($field:ident),* }),* $(,)?) => {
215        impl fmt::Debug for Frame {
216            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
217                match self {
218                    Self::Connect {
219                        flags,
220                        min_version,
221                        max_version,
222                        auth_token,
223                    } => formatter
224                        .debug_struct("Connect")
225                        .field("flags", flags)
226                        .field("min_version", min_version)
227                        .field("max_version", max_version)
228                        .field("auth_token", &ByteLength(auth_token.len()))
229                        .finish(),
230                    Self::Push { flags, stream_id, correlation_id, payload }
231                    | Self::PushReply { flags, stream_id, correlation_id, payload } => formatter
232                        .debug_struct(if matches!(self, Self::Push { .. }) { "Push" } else { "PushReply" })
233                        .field("flags", flags).field("stream_id", stream_id)
234                        .field("correlation_id", correlation_id)
235                        .field("payload", &ByteLength(payload.len())).finish(),
236                    Self::ConversationError { flags, stream_id, conversation_id, reason_code, message } => formatter
237                        .debug_struct("ConversationError").field("flags", flags)
238                        .field("stream_id", stream_id).field("conversation_id", conversation_id)
239                        .field("reason_code", reason_code)
240                        .field("message_bytes", &message.as_ref().map(String::len)).finish(),
241                    $(Self::$variant { $($field),* } => {
242                        let mut out = formatter.debug_struct(stringify!($variant));
243                        $(frame_debug_field!(out, $field, $field);)*
244                        out.finish()
245                    })*
246                }
247            }
248        }
249    };
250}
251
252    frame_debug! {
253        ConnectAck { flags, selected_version, capabilities },
254        ConnectError { flags, reason_code, message },
255        Disconnect { flags },
256        DisconnectReason { flags, reason },
257        Subscribe { flags, stream_id, channel, accepted_schemas, max_in_flight },
258        SubscribeAck { flags, stream_id, subscription_id, selected_schema },
259        SubscribeError { flags, stream_id, reason_code, message },
260        Unsubscribe { flags, stream_id, subscription_id },
261        Publish { flags, stream_id, channel, envelope, idempotency_key },
262        PublishAck { flags, stream_id, message_id, durable_position },
263        PublishError { flags, stream_id, reason_code, message },
264        ConversationOpen { flags, stream_id, conversation_id, subject },
265        ConversationMessage { flags, stream_id, conversation_id, envelope },
266        ConversationClose { flags, stream_id, conversation_id, reason_code, message },
267        Accept { flags, stream_id, referenced_message_id },
268        Defer { flags, stream_id, referenced_message_id, reason },
269        Reject { flags, stream_id, referenced_message_id, reason },
270        Ping { flags },
271        Pong { flags },
272        WorkerRegister { flags, registration },
273        WorkerRegisterAck { flags, outcome },
274        Deliver { flags, stream_id, delivery_seq, envelope, message },
275        LinkOpen { flags, stream_id, offer },
276        LinkOpenAck { flags, stream_id, outcome },
277        LinkMessage { flags, stream_id, subject, sealed },
278        Unknown { type_id, flags, stream_id, payload },
279    }
280}
liminal/protocol/worker_refusal.rs

liminal/protocol/
worker_refusal.rs
