weir_client/client.rs
1//! Transport-agnostic client core: the error type, the `WeirClient` struct, and
2//! every method that only needs a blocking `Read + Write`.
3//!
4//! This lives apart from `unix.rs` for one reason: none of it depends on Unix.
5//! While it sat inside the Unix module, `tls.rs` — which uses `TcpStream` and
6//! rustls and nothing else — had to be gated `#[cfg(all(unix, feature = "tls"))]`
7//! purely because it imports these two types. That made the mTLS client
8//! Unix-only by file layout rather than by any real constraint.
9
10use std::io::{self, Read, Write};
11
12use weir_core::{
13 Durability, Envelope, HEADER_LEN, Header, MAX_PAYLOAD_HARD_CAP, MessageType, NackReason,
14};
15
16/// The largest payload any weir daemon response carries: `Ack` and
17/// `HealthCheckResponse` are empty, `Nack` is one reason byte, and
18/// `VersionMismatch` adds the daemon's wire version — 2 bytes.
19///
20/// A larger declared length on a *response* is a desync or a non-weir peer. A
21/// future wire version that grew this would arrive with a version bump the
22/// client already rejects before reaching here.
23const MAX_RESPONSE_PAYLOAD_LEN: usize = 2;
24
25/// The transport [`WeirClient`] uses when no type parameter is given.
26///
27/// Unix keeps the Unix-domain socket it has always defaulted to, so
28/// `WeirClient` continues to mean `WeirClient<UnixStream>` there and no
29/// existing signature changes. Elsewhere there is no such socket, so the
30/// default is the TCP stream the TLS connector wraps.
31#[cfg(unix)]
32pub type DefaultTransport = std::os::unix::net::UnixStream;
33/// The transport [`WeirClient`] uses when no type parameter is given.
34#[cfg(not(unix))]
35pub type DefaultTransport = std::net::TcpStream;
36
37/// All errors that can be returned by [`WeirClient`] methods.
38///
39/// `#[non_exhaustive]`: the client error set may grow post-1.0 (it already gained
40/// `VersionMismatch` during the 1.0 hardening), so downstream matches must carry
41/// a wildcard arm and adding a variant later is not a breaking change.
42#[derive(Debug)]
43#[non_exhaustive]
44pub enum ClientError {
45 /// An I/O error on the underlying socket.
46 Io(io::Error),
47 /// The daemon sent a Nack with a recognised reason.
48 Nack(NackReason),
49 /// The daemon rejected the frame because its wire version differs from the
50 /// client's. Carries the daemon's `WIRE_VERSION` (the second Nack-payload
51 /// byte) so the caller can report both sides and decide whether to upgrade
52 /// the daemon or downgrade the client.
53 VersionMismatch {
54 /// The wire-protocol version the daemon speaks.
55 daemon_version: u8,
56 },
57 /// The daemon sent a Nack with an unrecognised reason byte.
58 UnknownNack(u8),
59 /// The payload exceeds the protocol hard cap and was rejected locally,
60 /// before any bytes were sent. The daemon would always reject a payload this
61 /// large (and a large frame can race the daemon's Nack-and-close, surfacing as
62 /// a bare broken-pipe), so the client rejects it up front with no round-trip.
63 /// A deployment's configured `max_payload_bytes` may be lower than this cap; a
64 /// payload over that lower limit is reported by the daemon as
65 /// [`ClientError::Nack`]`(`[`NackReason::PayloadTooLarge`]`)`.
66 PayloadTooLarge {
67 /// The rejected payload's length in bytes.
68 len: usize,
69 /// The protocol hard cap ([`weir_core::MAX_PAYLOAD_HARD_CAP`]).
70 limit: usize,
71 },
72 /// The daemon's response violated the wire protocol.
73 Protocol(String),
74 /// `push_default` was called but no default durability was set.
75 NoDefaultDurability,
76 /// `push` was called with an empty payload, rejected locally before any bytes
77 /// are sent. An empty payload IS the WAB end-of-records sentinel, so the daemon
78 /// Nacks it ([`NackReason::EmptyPayload`]) and closes the connection; rejecting
79 /// up front keeps the connection usable, mirroring the local
80 /// [`PayloadTooLarge`][Self::PayloadTooLarge] reject.
81 EmptyPayload,
82}
83
84impl ClientError {
85 /// Whether the connection is still usable after this error.
86 ///
87 /// - **Recoverable (`true`)** — the connection is still usable: the error is a
88 /// *local* or *per-record* rejection that does **not** poison the client.
89 /// Retry a different payload, or continue, on the **same** client.
90 /// - **Non-recoverable (`false`)** — the connection is dead or in an
91 /// indeterminate state: the error poisoned the client (set its internal
92 /// `poisoned`/`closed_after_nack` flag). **Drop this client and reconnect**;
93 /// every subsequent call on it will fail fast (see [`WeirClient::is_poisoned`]).
94 ///
95 /// This is the stable way to branch on whether to reconnect: [`ClientError`] is
96 /// `#[non_exhaustive]`, so an exhaustive `match` on its variants will not compile
97 /// downstream, and the variant set may grow post-1.0. Calling `is_recoverable`
98 /// (or checking [`WeirClient::is_poisoned`] after a failed call) keeps working
99 /// across additions.
100 ///
101 /// The classification is kept in lock-step with the client's poison logic: a
102 /// variant is recoverable **iff** producing it does not set the client's
103 /// `poisoned` or `closed_after_nack` flag.
104 ///
105 /// | Variant | Recoverable? | Why |
106 /// |---|---|---|
107 /// | [`PayloadTooLarge`][Self::PayloadTooLarge] | yes | local pre-send rejection; no bytes sent, connection untouched |
108 /// | [`EmptyPayload`][Self::EmptyPayload] | yes | local pre-send rejection; no bytes sent, connection untouched |
109 /// | [`NoDefaultDurability`][Self::NoDefaultDurability] | yes | local misconfiguration; returned before any I/O |
110 /// | [`Nack`][Self::Nack]`(`[`InternalError`][NackReason::InternalError]`)` | yes | transient daemon condition; the daemon keeps the connection open |
111 /// | [`Nack`][Self::Nack]`(`any other reason`)` | no | the daemon closes the connection after the Nack |
112 /// | [`VersionMismatch`][Self::VersionMismatch] | no | a Nack reason; the daemon closes the connection |
113 /// | [`UnknownNack`][Self::UnknownNack] | no | a Nack reason; the daemon closes the connection |
114 /// | [`Io`][Self::Io] | no | a socket read/write failure poisons the client |
115 /// | [`Protocol`][Self::Protocol] | no | a malformed/aborted response poisons the client (also the "reconnect" guard error) |
116 ///
117 /// Note: a *server* `Nack(PayloadTooLarge)` (a payload over the daemon's
118 /// configured cap) is **not** recoverable — the daemon Nacks then closes the
119 /// connection — whereas the *local* [`PayloadTooLarge`][Self::PayloadTooLarge]
120 /// error (a payload over the protocol hard cap, rejected before any bytes are
121 /// sent) **is**. They are distinct variants despite the shared name.
122 ///
123 /// New `#[non_exhaustive]` variants default to **non-recoverable** (`false`):
124 /// "drop and reconnect" is the safe assumption for an error this build does not
125 /// yet understand.
126 #[must_use]
127 pub fn is_recoverable(&self) -> bool {
128 match self {
129 // Local, pre-I/O rejections: the connection was never touched.
130 Self::PayloadTooLarge { .. } | Self::NoDefaultDurability | Self::EmptyPayload => true,
131 // The only Nack the daemon keeps the connection open for (transient);
132 // every other Nack reason closes it (see `WeirClient::note_nack`).
133 Self::Nack(NackReason::InternalError) => true,
134 // Every other Nack reason closes the connection.
135 Self::Nack(_) => false,
136 // VersionMismatch / UnknownNack are Nack reasons too — the daemon
137 // closes the connection after them (they route through `note_nack`).
138 Self::VersionMismatch { .. } | Self::UnknownNack(_) => false,
139 // A socket failure poisons the client; a protocol violation either
140 // poisons it (bad response) or IS the "drop and reconnect" guard error.
141 Self::Io(_) | Self::Protocol(_) => false,
142 // `#[non_exhaustive]`: a future variant we don't recognise — assume the
143 // worst (reconnect) rather than risk reusing a poisoned connection.
144 // Unreachable within this crate (all current variants are matched
145 // above), but required so the match stays total if a variant is added.
146 #[allow(unreachable_patterns)]
147 _ => false,
148 }
149 }
150}
151
152impl std::fmt::Display for ClientError {
153 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
154 match self {
155 Self::Io(e) => write!(f, "socket I/O error: {e}"),
156 Self::Nack(r) => write!(f, "server nack: {r:?}"),
157 Self::VersionMismatch { daemon_version } => write!(
158 f,
159 "wire version mismatch: daemon speaks v{daemon_version}, this client speaks v{}",
160 weir_core::WIRE_VERSION
161 ),
162 Self::UnknownNack(b) => write!(f, "server nack with unknown reason {b:#04x}"),
163 Self::PayloadTooLarge { len, limit } => write!(
164 f,
165 "payload too large: {len} bytes exceeds the {limit}-byte protocol hard cap"
166 ),
167 Self::Protocol(msg) => write!(f, "protocol error: {msg}"),
168 Self::NoDefaultDurability => {
169 write!(f, "push_default called but no default durability was set")
170 }
171 Self::EmptyPayload => write!(
172 f,
173 "payload is empty; rejected locally (an empty payload is the WAB end-of-records sentinel)"
174 ),
175 }
176 }
177}
178
179impl std::error::Error for ClientError {
180 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
181 match self {
182 Self::Io(e) => Some(e),
183 _ => None,
184 }
185 }
186}
187
188impl From<io::Error> for ClientError {
189 fn from(e: io::Error) -> Self {
190 Self::Io(e)
191 }
192}
193
194/// A synchronous blocking client for one connection to the weir daemon.
195///
196/// The type parameter `S` is the underlying transport. The default is
197/// [`DefaultTransport`] — a Unix domain socket on Unix. When the `tls` feature
198/// is enabled, `S` can also be `TlsStream` (TCP + mutual TLS).
199///
200/// The underlying stream runs in blocking mode. Each method sends one request
201/// frame and reads one response frame before returning. Requests are not
202/// pipelined; for concurrent producers, create one `WeirClient` per thread.
203///
204/// Drop the client to close the connection. The daemon treats EOF as a clean
205/// disconnect.
206pub struct WeirClient<S = DefaultTransport> {
207 // pub(crate) so the TLS connector (a sibling module) can reach the
208 // underlying TcpStream to apply socket timeouts (F43).
209 pub(crate) stream: S,
210 pub(crate) default_durability: Option<Durability>,
211 /// Set once a response read fails (a timeout mid-frame, a partial/aborted
212 /// read, or a protocol violation). The stream is then in an indeterminate
213 /// state — leftover bytes from the aborted response could be mis-read as the
214 /// NEXT request's reply, acking a record the daemon never confirmed (a false
215 /// ack). Once poisoned, every method fails fast; the caller must reconnect
216 /// (G04).
217 pub(crate) poisoned: bool,
218 /// Set once the daemon Nacked a frame for a reason that closes the connection
219 /// (every Nack except the transient `InternalError`). The connection is then
220 /// cleanly dead — the next call would otherwise just hit a broken pipe — so we
221 /// fail fast with a clear "reconnect" error instead (QA#2).
222 pub(crate) closed_after_nack: bool,
223}
224
225// Manual Debug (not derived): a `#[derive(Debug)]` would require `S: Debug`, which
226// excludes `WeirClient<TlsStream>` (rustls's `StreamOwned` is not `Debug`). Print
227// the non-secret fields and omit the transport so every client type — Unix and
228// TLS — implements `Debug` (C-DEBUG, S40).
229impl<S> std::fmt::Debug for WeirClient<S> {
230 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
231 f.debug_struct("WeirClient")
232 .field("default_durability", &self.default_durability)
233 .field("poisoned", &self.poisoned)
234 .field("closed_after_nack", &self.closed_after_nack)
235 .finish_non_exhaustive()
236 }
237}
238
239// ── Shared methods over any blocking Read+Write transport ──────────────────────
240
241impl<S: Read + Write> WeirClient<S> {
242 /// Sets the default durability tier used by [`push_default`][Self::push_default].
243 pub fn set_default_durability(&mut self, durability: Durability) {
244 self.default_durability = Some(durability);
245 }
246
247 /// Pushes `payload` to the daemon with the given `durability` tier.
248 ///
249 /// Blocks until the daemon replies. On `Ack` the record is durably stored
250 /// according to the requested tier. On `Nack` returns the reason as
251 /// [`ClientError::Nack`].
252 ///
253 /// An `Ack` means the record is durably **buffered** at the requested tier —
254 /// **not** that it has reached the downstream sink yet (the daemon drains in
255 /// batches once a segment seals). See the crate-level "Ack vs. delivery" note.
256 ///
257 /// The ack is **per record, not a liveness probe for the next one.** If the
258 /// daemon dies (or the socket is severed) between two pushes, the *first*
259 /// push after the failure can still return `Ok` — its frame was written and
260 /// acked, or the breakage isn't observed until the next read/write — and only
261 /// the *following* push surfaces the broken pipe. So a single `Ok` does not
262 /// guarantee the connection is healthy for subsequent pushes. After any push
263 /// error, branch on [`ClientError::is_recoverable`] /
264 /// [`is_poisoned`][Self::is_poisoned] to decide whether to keep using this
265 /// client or drop and reconnect, rather than assuming a prior `Ok` proved the
266 /// connection good.
267 #[must_use = "a dropped push result hides whether the record was acked; an \
268 unhandled Nack also closes the connection"]
269 pub fn push(
270 &mut self,
271 payload: impl AsRef<[u8]>,
272 durability: Durability,
273 ) -> Result<(), ClientError> {
274 self.ensure_usable()?;
275 let bytes = payload.as_ref();
276 // Local guard against an empty payload. An empty payload IS the WAB
277 // end-of-records sentinel, so the daemon Nacks it (NackReason::EmptyPayload)
278 // and closes the connection — which would poison this client. Reject it up
279 // front, before any bytes are sent, so the connection stays usable
280 // (recoverable), mirroring the hard-cap guard below.
281 if bytes.is_empty() {
282 return Err(ClientError::EmptyPayload);
283 }
284 // Local guard against the protocol hard cap. The daemon always rejects a
285 // payload this large, and for a big frame it Nacks then closes the
286 // connection before we finish streaming — so the Nack races our write and
287 // would surface as a bare broken-pipe. Reject up front, no round-trip, with
288 // a clear typed error. (A deployment's configured cap may be lower than the
289 // hard cap; a payload over that lower limit still goes to the daemon and is
290 // handled by the write-error recovery below.)
291 if bytes.len() > MAX_PAYLOAD_HARD_CAP {
292 return Err(ClientError::PayloadTooLarge {
293 len: bytes.len(),
294 limit: MAX_PAYLOAD_HARD_CAP,
295 });
296 }
297 // copy_from_slice at the API boundary so downstream handling is zero-copy.
298 let payload = weir_core::Payload::copy_from_slice(bytes);
299 let header = Header::new(MessageType::Push, durability, 0);
300 let frame = Envelope::new(header, payload).encode();
301
302 if let Err(write_err) = self.stream.write_all(&frame) {
303 // The write failed partway. The daemon may have Nacked (e.g. a payload
304 // over its configured cap) and closed the connection before we finished
305 // streaming — its Nack can already be sitting in our receive buffer.
306 // Read it so the caller sees the real reason instead of a broken-pipe.
307 // Either way the connection is now dead.
308 if let Ok(resp) = self.read_response()
309 && resp.header().message_type() == MessageType::Nack
310 {
311 self.closed_after_nack = true;
312 return Err(nack_error(resp.payload()));
313 }
314 self.poisoned = true;
315 return Err(write_err.into());
316 }
317
318 let resp = self.read_response()?;
319 match resp.header().message_type() {
320 MessageType::Ack => Ok(()),
321 // A Nack may carry an empty payload, which `nack_error` reports as a
322 // `Protocol` desync; poison in that case (see `surface_nack`).
323 MessageType::Nack => Err(self.surface_nack(resp.payload())),
324 // Any other frame type is a stream desync: the daemon sent something we
325 // never expect here, so leftover/unexpected bytes could be mis-read as a
326 // later reply (a false ack). Poison the connection (Protocol → not
327 // recoverable, matching `ensure_usable`).
328 other => {
329 self.poisoned = true;
330 Err(ClientError::Protocol(format!(
331 "expected Ack or Nack, got {other:?}"
332 )))
333 }
334 }
335 }
336
337 /// Records that a Nack closed the connection (every reason except the
338 /// transient `InternalError`, which the daemon keeps open) so the next call
339 /// fails fast with a clear reconnect error rather than a broken pipe (QA#2).
340 /// Returns the error unchanged for convenient `Err(self.note_nack(e))` use.
341 fn note_nack(&mut self, err: ClientError) -> ClientError {
342 if !matches!(err, ClientError::Nack(NackReason::InternalError)) {
343 self.closed_after_nack = true;
344 }
345 err
346 }
347
348 /// Decodes a Nack frame's payload and records its connection effect.
349 ///
350 /// `nack_error` maps a well-formed Nack to a `Nack`/`VersionMismatch`/`UnknownNack`
351 /// error (routed through [`note_nack`][Self::note_nack] to mark a closing Nack),
352 /// but an **empty** Nack payload is a stream desync that it reports as
353 /// [`ClientError::Protocol`]. A desync means leftover/unexpected bytes could be
354 /// mis-read as a later reply (a false ack), so we poison the connection in that
355 /// case — keeping `Protocol` non-recoverable and `is_poisoned()` in agreement.
356 fn surface_nack(&mut self, payload: &[u8]) -> ClientError {
357 match nack_error(payload) {
358 ClientError::Protocol(msg) => {
359 self.poisoned = true;
360 ClientError::Protocol(msg)
361 }
362 other => self.note_nack(other),
363 }
364 }
365
366 /// Pushes at the connection's default durability tier.
367 ///
368 /// Returns [`ClientError::NoDefaultDurability`] if no default was set via
369 /// [`set_default_durability`][Self::set_default_durability] (or via a
370 /// `connect_with_default` / `connect_tls` constructor that accepted a
371 /// `default_durability`).
372 pub fn push_default(&mut self, payload: impl AsRef<[u8]>) -> Result<(), ClientError> {
373 let d = self
374 .default_durability
375 .ok_or(ClientError::NoDefaultDurability)?;
376 self.push(payload, d)
377 }
378
379 /// Sends a `HealthCheck` frame and returns `Ok(())` on a valid
380 /// `HealthCheckResponse`. Returns an error if the daemon is unreachable
381 /// or responds with a Nack.
382 pub fn health_check(&mut self) -> Result<(), ClientError> {
383 self.ensure_usable()?;
384 let header = Header::new(MessageType::HealthCheck, Durability::Durable, 0);
385 let frame = Envelope::new(header, vec![]).encode();
386 // Poison on a write failure, exactly as `push` does. A bare `?` here
387 // left the client in a state the crate documents as impossible: the
388 // error is non-recoverable while `is_poisoned()` returns false, and
389 // those two are promised to be equivalent (see `is_poisoned`). A partial
390 // write of the frame also desyncs the stream, which is precisely what
391 // poisoning exists to stop the caller from ignoring.
392 if let Err(e) = self.stream.write_all(&frame) {
393 self.poisoned = true;
394 return Err(e.into());
395 }
396
397 let resp = self.read_response()?;
398 match resp.header().message_type() {
399 MessageType::HealthCheckResponse => Ok(()),
400 // An empty Nack payload is a desync `nack_error` reports as `Protocol`;
401 // poison in that case (see `surface_nack`).
402 MessageType::Nack => Err(self.surface_nack(resp.payload())),
403 // Any other frame type is a stream desync: poison the connection so a
404 // later call can't mis-read a stale frame as its reply (Protocol → not
405 // recoverable, matching `ensure_usable`).
406 other => {
407 self.poisoned = true;
408 Err(ClientError::Protocol(format!(
409 "expected HealthCheckResponse, got {other:?}"
410 )))
411 }
412 }
413 }
414
415 /// Whether this client's connection is dead and must be rebuilt.
416 ///
417 /// Reports the union of the two ways a connection becomes unusable, so it is the
418 /// exact complement of [`is_recoverable`][ClientError::is_recoverable] for the
419 /// error that caused it:
420 ///
421 /// - an **indeterminate stream** (the internal `poisoned` flag) — a
422 /// response-read failure/timeout, a protocol violation in the reply, or any
423 /// stream desync (an unexpected daemon message type, an empty-payload Nack);
424 /// leftover/unexpected bytes could be mis-read as a later reply (a false ack);
425 /// - a **connection-closing Nack** (the internal `closed_after_nack` flag) —
426 /// every Nack reason except the transient
427 /// [`InternalError`][NackReason::InternalError], after which the daemon closes
428 /// the connection.
429 ///
430 /// In either state **every** method fails fast (it will not touch the socket
431 /// again); the caller must drop this client and open a fresh connection.
432 ///
433 /// A long-lived producer thread (e.g. one driven from an async runtime) should
434 /// check this after a failed call — or, equivalently, branch on
435 /// [`ClientError::is_recoverable`] — and rebuild the connection when it reports
436 /// poisoned / non-recoverable, rather than retrying on the dead client:
437 ///
438 /// ```no_run
439 /// # use weir_client::{WeirClient, ClientError};
440 /// # use weir_core::Durability;
441 /// # fn reconnect() -> WeirClient { unreachable!() }
442 /// # let mut client = reconnect();
443 /// if let Err(e) = client.push(b"record", Durability::Durable) {
444 /// if !e.is_recoverable() || client.is_poisoned() {
445 /// client = reconnect(); // drop the dead client, rebuild
446 /// }
447 /// }
448 /// ```
449 ///
450 /// See [`ClientError::is_recoverable`] for the per-error view of the same
451 /// distinction (recoverable = connection still usable; non-recoverable = drop
452 /// and reconnect). The two are kept equivalent: `is_poisoned()` is true after a
453 /// failed call exactly when that call's error is non-recoverable.
454 #[must_use]
455 pub fn is_poisoned(&self) -> bool {
456 self.poisoned || self.closed_after_nack
457 }
458
459 /// Returns the poisoned-connection error if a prior response read failed.
460 /// Once poisoned the stream may hold leftover bytes from an aborted response
461 /// that would be mis-read as this call's reply — a false ack — so we refuse
462 /// to use it (G04). The caller must drop the client and reconnect.
463 pub(crate) fn ensure_usable(&self) -> Result<(), ClientError> {
464 if self.closed_after_nack {
465 return Err(ClientError::Protocol(
466 "connection closed by the daemon after a Nack; \
467 drop this client and reconnect before sending again"
468 .into(),
469 ));
470 }
471 if self.poisoned {
472 return Err(ClientError::Protocol(
473 "client connection poisoned by a prior response-read error/timeout; \
474 drop this client and reconnect before sending again"
475 .into(),
476 ));
477 }
478 Ok(())
479 }
480
481 fn read_response(&mut self) -> Result<Envelope, ClientError> {
482 // Any failure here leaves the stream in an indeterminate state (a timeout
483 // mid-frame, a partial/aborted read, or a protocol violation): the unread
484 // tail of this response could be mis-read as the NEXT request's reply.
485 // Poison the client so every subsequent call fails fast (G04).
486 let result = self.read_response_inner();
487 if result.is_err() {
488 self.poisoned = true;
489 }
490 result
491 }
492
493 fn read_response_inner(&mut self) -> Result<Envelope, ClientError> {
494 let mut header_buf = [0u8; HEADER_LEN];
495 self.stream.read_exact(&mut header_buf)?;
496
497 let header =
498 Header::decode(&header_buf).map_err(|e| ClientError::Protocol(e.to_string()))?;
499
500 let payload_len = header.payload_len() as usize;
501 // Cap before allocating. The bound is the RESPONSE bound, not the record
502 // one: every weir response payload is at most 2 bytes (`Ack` and
503 // `HealthCheckResponse` are 0, `Nack` is 1, `VersionMismatch` is 2) —
504 // see the producer checklist in docs/wire_protocol.md, which tells every
505 // other implementer to "cap the response payload_len at a few bytes".
506 //
507 // This used to compare against MAX_PAYLOAD_HARD_CAP (16 MiB), the cap on
508 // a RECORD being sent the other way. So the reference client violated the
509 // checklist it publishes: a desynced or hostile peer could make it
510 // allocate 16 MiB and — with the default `read_timeout` of None — block
511 // forever waiting for bytes that never come.
512 if payload_len > MAX_RESPONSE_PAYLOAD_LEN {
513 return Err(ClientError::Protocol(format!(
514 "response payload_len {payload_len} exceeds the {MAX_RESPONSE_PAYLOAD_LEN}-byte \
515 response cap; refusing to allocate"
516 )));
517 }
518 let mut payload_buf = vec![0u8; payload_len];
519 if payload_len > 0 {
520 self.stream.read_exact(&mut payload_buf)?;
521 }
522
523 let mut crc_buf = [0u8; 4];
524 self.stream.read_exact(&mut crc_buf)?;
525 let expected = u32::from_le_bytes(crc_buf);
526 let computed = crc32fast::hash(&payload_buf);
527 if expected != computed {
528 return Err(ClientError::Protocol(format!(
529 "response payload CRC mismatch: expected {expected:#010x}, computed {computed:#010x}"
530 )));
531 }
532
533 Ok(Envelope::new(header, payload_buf))
534 }
535
536 /// Crate-internal constructor so `tls.rs` can build the struct without
537 /// going through a Unix-socket constructor.
538 #[cfg_attr(not(feature = "tls"), allow(dead_code))]
539 pub(crate) fn from_parts(stream: S, default_durability: Option<Durability>) -> Self {
540 Self {
541 stream,
542 default_durability,
543 poisoned: false,
544 closed_after_nack: false,
545 }
546 }
547}
548
549pub(crate) fn nack_error(payload: &[u8]) -> ClientError {
550 match payload.first().copied() {
551 Some(b) => match NackReason::try_from(b) {
552 // A VersionMismatch Nack carries the daemon's WIRE_VERSION as a
553 // second payload byte (`[0x02, daemon_version]`). Surface it so the
554 // caller can report both sides; fall back to the bare reason if the
555 // daemon (somehow) omitted it.
556 Ok(NackReason::VersionMismatch) => match payload.get(1).copied() {
557 Some(daemon_version) => ClientError::VersionMismatch { daemon_version },
558 None => ClientError::Nack(NackReason::VersionMismatch),
559 },
560 Ok(r) => ClientError::Nack(r),
561 Err(_) => ClientError::UnknownNack(b),
562 },
563 None => ClientError::Protocol("Nack frame had empty payload".into()),
564 }
565}