Skip to main content

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}