cu29_logstream/
stream.rs

1//! Transport-independent, packet-oriented stream resource contracts.
2//!
3//! Data and feedback use the same transport guarantees: complete packets with
4//! verified integrity. Delivery, ordering, and uniqueness are not guaranteed.
5//! UDP supplies packet integrity through the network stack. Raw serial supplies
6//! neither packet boundaries nor integrity; the framing adapter between LogStream
7//! and serial adds and verifies a checksum and discards damaged frames.
8
9use core::fmt::Debug;
10
11/// Failure to submit one complete logstream packet.
12#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13pub enum CuStreamTxError {
14    /// The bounded transport cannot accept this packet immediately.
15    WouldBlock,
16    /// The transport has failed and cannot accept the packet.
17    Failed(&'static str),
18}
19
20/// Failure to receive one complete logstream packet.
21#[derive(Clone, Copy, Debug, PartialEq, Eq)]
22pub enum CuStreamRxError {
23    /// The next packet does not fit in the caller-provided buffer.
24    BufferTooSmall { needed: usize },
25    /// The transport has failed and cannot receive packets.
26    Failed(&'static str),
27}
28
29/// One-way, fire-and-forget transmission of complete logstream packets.
30///
31/// Implementations must return immediately without waiting, locking, allocating,
32/// retrying, acknowledging, or accepting a partial packet. `Ok(())` means only
33/// that the resource accepted this packet once; it does not guarantee delivery.
34///
35/// Carriers must protect complete packet boundaries and integrity. Packet hardware
36/// or the network stack may supply the checksum; adapters over byte streams must
37/// add framing, an integrity check, and receive resynchronization. The common
38/// LogStream envelope has no checksum. Carrier overhead is outside the packet MTU.
39pub trait CuStreamTx: Debug + Send + Sync {
40    /// Advance an accepted packet with bounded, nonblocking work. Returns true
41    /// while bytes remain. Drivers must keep polling until it returns false.
42    /// The default implementation reports an idle transport.
43    fn poll_pending(&mut self) -> core::result::Result<bool, CuStreamTxError> {
44        Ok(false)
45    }
46
47    fn try_send(&mut self, packet: &[u8]) -> core::result::Result<(), CuStreamTxError>;
48}
49
50impl<T: CuStreamTx + ?Sized> CuStreamTx for alloc::boxed::Box<T> {
51    fn poll_pending(&mut self) -> core::result::Result<bool, CuStreamTxError> {
52        (**self).poll_pending()
53    }
54    #[inline]
55    fn try_send(&mut self, packet: &[u8]) -> core::result::Result<(), CuStreamTxError> {
56        (**self).try_send(packet)
57    }
58}
59
60/// Nonblocking reception of complete logstream packets.
61///
62/// `Ok(None)` means no complete packet is available. `Ok(Some(len))` places one
63/// complete packet in `packet[..len]`. This is the receive half of the one-way
64/// data plane; it does not acknowledge traffic or imply a reverse channel.
65///
66/// Only deliver complete packets that passed the carrier's integrity check. Drop
67/// corrupt, truncated, or malformed frames before exposing them to FEC; corruption
68/// must become packet loss, which FEC can recover. Never deliver a partial packet.
69/// Byte-stream adapters must verify and strip their framing checksum and
70/// resynchronize after damaged frames. In-memory carriers must preserve bytes.
71/// Record digests bind reconstructed content and recovery references separately;
72/// they do not replace this carrier integrity contract.
73pub trait CuStreamRx: Debug + Send + Sync {
74    fn try_recv(
75        &mut self,
76        packet: &mut [u8],
77    ) -> core::result::Result<Option<usize>, CuStreamRxError>;
78}
79
80impl<T: CuStreamRx + ?Sized> CuStreamRx for alloc::boxed::Box<T> {
81    #[inline]
82    fn try_recv(
83        &mut self,
84        packet: &mut [u8],
85    ) -> core::result::Result<Option<usize>, CuStreamRxError> {
86        (**self).try_recv(packet)
87    }
88}
89
90/// Optional advisory receive direction. This logical interface may share the
91/// physical stream carrier. A shared endpoint must have one receive owner and
92/// route complete packet kinds; cloned competing readers cannot provide routing.
93/// Protocol decoding, capability negotiation, and feedback policy belong above it.
94///
95/// The complete-packet integrity contract of [`CuStreamRx`] applies here too.
96/// Discard damaged or truncated frames in the transport adapter before delivery.
97/// Reports may be lost, duplicated, or reordered; report sequencing belongs to
98/// the feedback controller.
99pub trait CuFeedbackRx: Debug + Send + Sync {
100    fn try_recv_feedback(
101        &mut self,
102        packet: &mut [u8],
103    ) -> core::result::Result<Option<usize>, CuStreamRxError>;
104}
105
106/// Optional advisory transmit direction; same atomic, bounded, complete-packet
107/// integrity contract as [`CuStreamTx`]. Acceptance guarantees neither delivery
108/// nor ordering. A raw byte stream needs a framing and integrity adapter.
109pub trait CuFeedbackTx: Debug + Send + Sync {
110    fn try_send_feedback(&mut self, packet: &[u8]) -> core::result::Result<(), CuStreamTxError>;
111}
112
113/// Static autonomous wiring. There is no feedback polling or capability discovery.
114#[derive(Debug)]
115pub struct OneWay<T> {
116    tx: T,
117}
118impl<T> OneWay<T> {
119    pub const fn new(tx: T) -> Self {
120        Self { tx }
121    }
122    pub fn into_inner(self) -> T {
123        self.tx
124    }
125}
126impl<T: CuStreamTx> CuStreamTx for OneWay<T> {
127    fn poll_pending(&mut self) -> core::result::Result<bool, CuStreamTxError> {
128        self.tx.poll_pending()
129    }
130
131    fn try_send(&mut self, packet: &[u8]) -> core::result::Result<(), CuStreamTxError> {
132        self.tx.try_send(packet)
133    }
134}
135impl<T: CuStreamTx> CuFeedbackRx for OneWay<T> {
136    fn try_recv_feedback(
137        &mut self,
138        _packet: &mut [u8],
139    ) -> core::result::Result<Option<usize>, CuStreamRxError> {
140        Ok(None)
141    }
142}
143
144/// Explicit feedback wiring for separate logical endpoints. The endpoints may be
145/// resource handles backed by one carrier; physical out-of-band links are optional.
146#[derive(Debug)]
147pub struct SeparateFeedback<T, R> {
148    pub tx: T,
149    pub feedback_rx: R,
150}
151impl<T: CuStreamTx, R: CuFeedbackRx> CuStreamTx for SeparateFeedback<T, R> {
152    fn poll_pending(&mut self) -> core::result::Result<bool, CuStreamTxError> {
153        self.tx.poll_pending()
154    }
155
156    fn try_send(&mut self, packet: &[u8]) -> core::result::Result<(), CuStreamTxError> {
157        self.tx.try_send(packet)
158    }
159}
160impl<T: CuStreamTx, R: CuFeedbackRx> CuFeedbackRx for SeparateFeedback<T, R> {
161    fn try_recv_feedback(
162        &mut self,
163        packet: &mut [u8],
164    ) -> core::result::Result<Option<usize>, CuStreamRxError> {
165        self.feedback_rx.try_recv_feedback(packet)
166    }
167}
cu29_logstream/stream.rs

cu29_logstream/
stream.rs
