Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/str0m/src/sctp/mod.rs

rust1893 lines
1#![allow(clippy::new_without_default)]
2 
3use std::collections::{HashSet, VecDeque};
4use std::fmt;
5use std::net::SocketAddr;
6use std::panic::UnwindSafe;
7use std::sync::Arc;
8use std::time::{Duration, Instant};
9 
10use bytes::Bytes;
11pub use sctp_proto::ReceiveLimits as SctpReceiveLimits;
12use sctp_proto::{Association, AssociationHandle, DatagramEvent};
13use sctp_proto::{Endpoint, EndpointConfig, Stream, StreamEvent, Transmit};
14use sctp_proto::{Event, Payload, PayloadProtocolIdentifier, ServerConfig, TransportConfig};
15 
16use snap::{b64_encode, webrtc_transport_config};
17 
18pub use sctp_proto::Error as ProtoError;
19use sctp_proto::ReliabilityType;
20 
21mod snap;
22pub use snap::SctpInitData;
23 
24mod dcep;
25use dcep::DcepAck;
26use dcep::DcepOpen;
27 
28mod error;
29pub use error::SctpError;
30 
31/// Bytes that can be buffered inside str0m across all streams.
32const MAX_BUFFERED_ACROSS_STREAMS: usize = 128 * 1024;
33 
34/// Maximum message size we advertise in SDP (what we can receive)
35pub const LOCAL_MAX_MESSAGE_SIZE: u32 = 256 * 1024;
36 
37/// Default max message size if remote doesn't advertise
38pub const DEFAULT_REMOTE_MAX_MESSAGE_SIZE: u32 = 64 * 1024;
39 
40pub(crate) struct RtcSctp {
41    state: RtcSctpState,
42    endpoint: Endpoint,
43    fake_addr: SocketAddr,
44    handle: AssociationHandle,
45    assoc: Option<Association>,
46    // Sorted by `id` so lookups can binary search. Keep it that way.
47    entries: Vec<StreamEntry>,
48    // Stream ids that still owe a `StreamEvent::ResetComplete`,
49    // thos ids needs to be held back from reuse.
50    reset_pending: HashSet<u16>,
51    // Completed resets awaiting delivery to the caller.
52    // Used to guarantee emission ordering, ResetComplete must
53    // be sent after Close.
54    reset_complete: VecDeque<u16>,
55    pushed_back_transmit: Option<VecDeque<Bytes>>,
56    receive_limits: Option<SctpReceiveLimits>,
57    last_now: Instant,
58    client: bool,
59    remote_max_message_size: u32,
60    snap_enabled: bool,
61    snap_init: Option<SctpInitData>,
62    #[cfg(test)]
63    max_payload_size: usize,
64}
65 
66/// This is okay because there is no way for a user of Rtc to interact with the Sctp subsystem
67/// in a way that would allow them to observe a potentially broken invariant when catching a panic.
68impl UnwindSafe for RtcSctp {}
69 
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
71enum RtcSctpState {
72    Uninited,
73    AwaitRemoteAssociation,
74    AwaitAssociationEstablished,
75    Established,
76}
77 
78impl RtcSctpState {
79    pub fn propagate_endpoint_to_assoc(&self) -> bool {
80        matches!(
81            self,
82            RtcSctpState::AwaitAssociationEstablished | RtcSctpState::Established
83        )
84    }
85}
86 
87#[derive(Debug)]
88struct StreamEntry {
89    /// Config as provided when opening the channel. This is None if we discover
90    /// the channel from the remote peer before getting a DcepOpen or local open_stream.
91    config: Option<ChannelConfig>,
92    /// Current state
93    state: StreamEntryState,
94    /// Actual stream id. Negotiated or automatically allocated.
95    id: u16,
96    /// If we are to close this entry.
97    do_close: bool,
98    /// Deadline for retrying `open_stream` when it fails.
99    open_deadline: Option<Instant>,
100    /// If the queued outgoing data drops below this threshold, Rtc is to emit an
101    /// event to the user.
102    buffered_threshold: BufferedThresholdConfig,
103}
104 
105#[derive(Debug)]
106/// Tracks the `buffered_amount_low_threshold` for a stream.
107///
108/// Lets us defer applying user changes to the underlying SCTP
109/// stream until the next poll cycle, without first querying the current
110/// configured value.
111enum BufferedThresholdConfig {
112    /// No threshold has been set or it was cleared after an error.
113    Unconfigured,
114    /// A user-requested threshold to apply on the next poll.
115    Desired(usize),
116    /// The threshold value currently configured in the SCTP stream.
117    Configured(usize),
118}
119 
120impl BufferedThresholdConfig {
121    pub fn set(&mut self, v: usize) {
122        let is_change = match self {
123            BufferedThresholdConfig::Unconfigured => true,
124            BufferedThresholdConfig::Desired(w) if v != *w => true,
125            BufferedThresholdConfig::Configured(x) if v != *x => true,
126            _ => false,
127        };
128        if is_change {
129            *self = BufferedThresholdConfig::Desired(v);
130        }
131    }
132}
133 
134pub(crate) enum SctpEvent {
135    Transmit {
136        packets: VecDeque<Bytes>,
137    },
138    Open {
139        id: u16,
140        label: String,
141    },
142    Close {
143        id: u16,
144        /// Whether a reset handshake is outstanding for this stream id.
145        reset_pending: bool,
146    },
147    /// The reset handshake for a closed stream completed and sctp-proto emitted
148    /// StreamEvent::ResetComplete. The stream id is now safe to be used again.
149    ///
150    /// Always reported after the `Close` for the same id.
151    StreamResetComplete {
152        id: u16,
153    },
154    Data {
155        id: u16,
156        binary: bool,
157        data: Vec<u8>,
158    },
159    BufferedAmountLow {
160        id: u16,
161    },
162    AssociationLost,
163}
164 
165/// These are the possible paths:
166/// ```text
167/// local inited, in-band                                     AwaitOpen -> AwaitDcepAck -> Open
168/// local inited, out-of-band                                 AwaitOpen                 -> Open
169/// remote inited, in-band     AwaitConfig -> (receive dcep)                            -> Open
170/// remote inited, out-of-band AwaitConfig -> (open_stream)                             -> Open
171/// ```
172#[derive(Debug, Clone, Copy, PartialEq, Eq)]
173enum StreamEntryState {
174    /// A new stream declared locally, not discovered from remote.
175    AwaitOpen,
176    /// A new stream, discovered from remote. It can either be in-band or out-of band
177    /// We will either receive DcepOpen in-band, or a open_stream() call out-of-band.
178    AwaitConfig,
179    /// If we have sent DcepOpen and are waiting for the ack.
180    AwaitDcepAck,
181    /// Stream is open, ready to send data.
182    Open,
183    /// If some error occurs.
184    Closed,
185}
186 
187/// (Low level) configuration for a data channel.
188#[derive(Debug, Clone, PartialEq, Eq, Hash)]
189pub struct ChannelConfig {
190    /// The label to use for the user to identify the channel.
191    pub label: String,
192    /// Whether channel is guaranteed ordered delivery of messages.
193    pub ordered: bool,
194    /// The reliability setting, which can allow to drop messages.
195    pub reliability: Reliability,
196    /// Whether channel is negotiated in-band (DCEP) or out-of-band.
197    /// None means in-band negotiated. Some(stream_id) means out-of-band.
198    pub negotiated: Option<u16>,
199    /// Protocol name.
200    ///
201    /// Defaults to ""
202    pub protocol: String,
203}
204 
205impl Default for ChannelConfig {
206    fn default() -> Self {
207        Self {
208            label: Default::default(),
209            ordered: true,
210            reliability: Default::default(),
211            negotiated: Default::default(),
212            protocol: Default::default(),
213        }
214    }
215}
216 
217/// Reliability setting of a data channel.
218#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
219pub enum Reliability {
220    /// Packets are delivered in order, with retransmits.
221    #[default]
222    Reliable,
223    /// Packets delivered out of order with a max lifetime.
224    MaxPacketLifetime {
225        /// The lifetime of a packet in milliseconds.
226        lifetime: u16,
227    },
228    /// Packets delivered out of order with a max number of retransmits.
229    MaxRetransmits {
230        /// Number of retransmits before giving up.
231        retransmits: u16,
232    },
233}
234 
235impl StreamEntry {
236    fn set_state(&mut self, state: StreamEntryState) -> bool {
237        if self.state == state {
238            return false;
239        }
240        debug!("Stream {:?} -> {:?}", self.state, state);
241        self.state = state;
242        true
243    }
244 
245    #[must_use]
246    fn configure_reliability(&mut self, stream: &mut Stream) -> bool {
247        let dcep: DcepOpen = self.config.as_ref().expect("config to be set").into();
248 
249        let ret = stream.set_reliability_params(
250            dcep.unordered,
251            dcep.channel_type,
252            dcep.reliability_parameter,
253        );
254 
255        if let Err(e) = ret {
256            warn!(
257                "Failed to set reliability params on stream {}: {:?}",
258                self.id, e
259            );
260            self.do_close = true;
261            return false;
262        }
263 
264        true
265    }
266}
267 
268/// SCTP framing per outgoing datagram: 12-byte common header + ~16-byte DATA
269/// chunk header + some extra.
270const SCTP_OVERHEAD: usize = 40;
271 
272/// How long `open_stream` keeps retrying when blocked by a transient error
273/// which could happen during reset handshake (RFC 6525).
274/// Past this the peer is considered broken and the open gives up.
275const STREAM_OPEN_TIMEOUT: Duration = Duration::from_secs(10);
276 
277/// Backstop poll interval while an open is retrying. Retries are normally
278/// driven by the network input that unblocks them,
279/// either the reciprocal reset or RECONFIG-RESPONSE arriving.
280/// This interval guarantees progress in case of absence of the latter messages.
281const STREAM_OPEN_RETRY_INTERVAL: Duration = Duration::from_millis(500);
282 
283/// Empirical: SCTP `max_payload_size` of 1200 has produced 1277-byte DTLS-wrapped
284/// datagrams (77 bytes combined SCTP + DTLS overhead).
285const _: () = assert!(
286    crate::io::MAX_DTLS_OVERHEAD + SCTP_OVERHEAD >= 80,
287    "MAX_DTLS_OVERHEAD + SCTP_OVERHEAD must cover observed 77-byte SCTP-over-DTLS overhead"
288);
289 
290impl RtcSctp {
291    #[cfg(test)]
292    pub fn new(mtu: usize) -> Self {
293        Self::with_receive_limits(mtu, None)
294    }
295 
296    pub fn with_receive_limits(mtu: usize, receive_limits: Option<SctpReceiveLimits>) -> Self {
297        let mut config = EndpointConfig::default();
298        let max_payload = mtu
299            .saturating_sub(crate::io::MAX_DTLS_OVERHEAD)
300            .saturating_sub(SCTP_OVERHEAD);
301        config.max_payload_size(max_payload as u32);
302        #[cfg(test)]
303        let max_payload_size = max_payload;
304        let mut server_config = ServerConfig::default();
305        server_config.transport = webrtc_transport_config(receive_limits);
306        let endpoint = Endpoint::new(Arc::new(config), Some(Arc::new(server_config)));
307        let fake_addr = "1.1.1.1:5000".parse().unwrap();
308 
309        RtcSctp {
310            state: RtcSctpState::Uninited,
311            endpoint,
312            fake_addr,
313            handle: AssociationHandle(0), // temporary
314            assoc: None,
315            entries: Vec::new(),
316            reset_pending: HashSet::new(),
317            reset_complete: VecDeque::new(),
318            pushed_back_transmit: None,
319            receive_limits,
320            last_now: Instant::now(), // placeholder until init()
321            client: false,
322            remote_max_message_size: DEFAULT_REMOTE_MAX_MESSAGE_SIZE,
323            snap_enabled: false,
324            snap_init: None,
325            #[cfg(test)]
326            max_payload_size,
327        }
328    }
329 
330    #[cfg(test)]
331    pub(crate) fn max_payload_size(&self) -> usize {
332        self.max_payload_size
333    }
334 
335    pub fn is_inited(&self) -> bool {
336        self.state != RtcSctpState::Uninited
337    }
338 
339    pub fn is_closing(&self) -> bool {
340        self.assoc.as_ref().is_some_and(|a| a.is_closing())
341    }
342 
343    pub fn is_closed(&self) -> bool {
344        self.assoc.as_ref().is_none_or(|a| a.is_closed())
345    }
346 
347    pub fn init(
348        &mut self,
349        client: bool,
350        now: Instant,
351        sctp_init_data: Option<SctpInitData>,
352        remote_max_message_size: Option<u32>,
353    ) -> Result<(), SctpError> {
354        if self.state != RtcSctpState::Uninited {
355            return Err(SctpError::Proto(ProtoError::Other(
356                "SCTP already initialized".into(),
357            )));
358        }
359 
360        self.client = client;
361        self.last_now = now;
362 
363        if let Some(max_msg_size) = remote_max_message_size {
364            self.remote_max_message_size = max_msg_size;
365        }
366 
367        if let Some(mut snap_data) = sctp_init_data {
368            // SNAP path: both local and remote INIT chunks must be present.
369            if snap_data.local_init.is_none() || snap_data.remote_init.is_none() {
370                return Err(SctpError::Proto(ProtoError::Other(
371                    "SNAP requires both local and remote SCTP INIT chunks".into(),
372                )));
373            }
374 
375            // Enforce the local resource policy for both SDP and direct SNAP.
376            if self.receive_limits.is_some() {
377                snap_data.transport = webrtc_transport_config(self.receive_limits);
378            }
379            let config = snap_data.into_client_config();
380            debug!(
381                "New {} association (out-of-band: true)",
382                if client { "local" } else { "server" },
383            );
384            let (handle, mut assoc) = self
385                .endpoint
386                .connect(config, self.fake_addr)
387                .map_err(|e| SctpError::Proto(ProtoError::Other(e.to_string())))?;
388            assoc.set_max_send_message_size(self.remote_max_message_size);
389            self.handle = handle;
390            self.assoc = Some(assoc);
391 
392            // With SNAP, both sides exchanged INIT chunks out-of-band. The
393            // sctp-proto association is already in established state (via
394            // `with_snap`). We set our state to Established immediately
395            // even though DTLS may not be connected yet. This is safe
396            // because the `dtls_connected` guard in `do_poll_output`
397            // prevents any SCTP packets from flowing until the DTLS
398            // handshake completes.
399            set_state(&mut self.state, RtcSctpState::Established);
400        } else if client {
401            // Normal client path: initiate the SCTP association.
402            let mut config = SctpInitData::default().into_client_config();
403            let mut transport = TransportConfig::default()
404                .with_max_init_retransmits(None)
405                .with_max_data_retransmits(None)
406                .with_max_receive_message_size(LOCAL_MAX_MESSAGE_SIZE)
407                .with_max_send_message_size(self.remote_max_message_size);
408            if let Some(limits) = self.receive_limits {
409                transport = transport.with_receive_limits(limits);
410            }
411            config.transport = Arc::new(transport);
412 
413            debug!("New local association (out-of-band: false)");
414            let (handle, assoc) = self
415                .endpoint
416                .connect(config, self.fake_addr)
417                .map_err(|e| SctpError::Proto(ProtoError::Other(e.to_string())))?;
418            self.handle = handle;
419            self.assoc = Some(assoc);
420            set_state(&mut self.state, RtcSctpState::AwaitAssociationEstablished);
421        } else {
422            // Normal server path: wait for the remote to initiate.
423            set_state(&mut self.state, RtcSctpState::AwaitRemoteAssociation);
424        }
425 
426        Ok(())
427    }
428 
429    pub fn local_max_message_size(&self) -> u32 {
430        self.receive_limits
431            .map_or(LOCAL_MAX_MESSAGE_SIZE, SctpReceiveLimits::max_message_size)
432    }
433 
434    pub fn is_client(&self) -> bool {
435        self.client
436    }
437 
438    /// Enable SNAP by pre-populating the init data.
439    pub fn enable_snap(&mut self) {
440        self.snap_enabled = true;
441        self.snap_init
442            .get_or_insert_with(|| SctpInitData::with_optional_receive_limits(self.receive_limits));
443    }
444 
445    /// Whether local offers should opt in to SNAP.
446    pub fn snap_enabled(&self) -> bool {
447        self.snap_enabled
448    }
449 
450    /// Ensure the local SNAP INIT chunk is generated. Returns `false` if
451    /// generation failed (degrades to non-SNAP).
452    pub fn ensure_local_snap_init(&mut self) -> bool {
453        let init_data = self
454            .snap_init
455            .get_or_insert_with(|| SctpInitData::with_optional_receive_limits(self.receive_limits));
456        if init_data.local_init_chunk().is_err() {
457            self.snap_init = None;
458            false
459        } else {
460            true
461        }
462    }
463 
464    /// Discard pending SNAP negotiation state before SCTP starts.
465    ///
466    /// This preserves local opt-in for future offers.
467    pub fn disable_pending_snap(&mut self) {
468        if !self.is_inited() {
469            self.snap_init = None;
470        }
471    }
472 
473    /// Get the local INIT chunk as a base64 string for SDP, if applicable.
474    ///
475    /// Returns `None` when:
476    /// - SNAP is not active
477    /// - SCTP is established without SNAP (non-SNAP association)
478    pub fn local_sctp_init_for_sdp(&self) -> Option<String> {
479        let d = self.snap_init.as_ref()?;
480        if self.is_inited() && d.remote_init.is_none() {
481            // Established non-SNAP association - MUST NOT inject sctp-init.
482            return None;
483        }
484        d.local_init.as_ref().map(|b| b64_encode(b))
485    }
486 
487    /// Get the cached remote INIT string, if set.
488    pub fn snap_remote_init_string(&self) -> Option<String> {
489        self.snap_init.as_ref().and_then(|d| d.remote_init_string())
490    }
491 
492    /// Whether this is an established SNAP association (has remote init).
493    pub fn is_snap_established(&self) -> bool {
494        self.is_inited()
495            && self
496                .snap_init
497                .as_ref()
498                .and_then(|d| d.remote_init.as_ref())
499                .is_some()
500    }
501 
502    /// Set the remote SNAP INIT from a base64 string. Returns `Ok(true)` if
503    /// accepted, `Ok(false)` on decode error (degrades to non-SNAP).
504    pub fn set_remote_snap_init_string(&mut self, value: &str) -> bool {
505        let init_data = self
506            .snap_init
507            .get_or_insert_with(|| SctpInitData::with_optional_receive_limits(self.receive_limits));
508        match init_data.set_remote_init_string(value) {
509            Ok(()) => true,
510            Err(_) => {
511                self.disable_pending_snap();
512                false
513            }
514        }
515    }
516 
517    /// Build a cloned `SctpInitData` for passing to `init()`, if both local
518    /// and remote INIT chunks are present.
519    pub fn build_snap_init_data(&self) -> Option<SctpInitData> {
520        let d = self.snap_init.as_ref()?;
521        if d.local_init.is_none() || d.remote_init.is_none() {
522            return None;
523        }
524        Some(d.clone())
525    }
526 
527    /// Opens a new stream.
528    pub fn open_stream(&mut self, id: u16, config: ChannelConfig) {
529        // The channel might already have arrived via SCTP, and if it is negotiated out-of-band
530        // we are waiting for the configuration.
531        let entry = stream_entry(
532            &mut self.entries,
533            id,
534            StreamEntryState::AwaitOpen,
535            "open_stream",
536        );
537 
538        let in_band = config.negotiated.is_none();
539 
540        // Stream should not already have a config, we are either waiting for DcepOpen, or this is
541        // out-of-band configuration, in which case this call is setting the config.
542        if entry.config.is_some() {
543            warn!("Stream is already configured: {}", id);
544            entry.do_close = true;
545            return;
546        } else {
547            entry.config = Some(config);
548        }
549 
550        // If we are in AwaitConfig, the stream was discovered from the remote peer before
551        // we got to do open_stream. This means we _must_ be in the out-of-band track,
552        // since we shouldn't call open_stream on remotely started in-band.
553        if entry.state == StreamEntryState::AwaitConfig {
554            if in_band {
555                warn!("open_stream in-band negotiation for remote stream: {}", id);
556                entry.do_close = true;
557            } else {
558                // out-of-band where remote started. We can go to Open, but must configure the local
559                // stream for it first.
560 
561                // The stream can already be gone even though our entry is
562                // still AwaitConfig. Close the channel gracefully instead of panicking.
563                let Some(assoc) = self.assoc.as_mut() else {
564                    entry.do_close = true;
565                    return;
566                };
567                let Ok(mut stream) = assoc.stream(entry.id) else {
568                    entry.do_close = true;
569                    return;
570                };
571 
572                if !entry.configure_reliability(&mut stream) {
573                    return;
574                }
575 
576                entry.set_state(StreamEntryState::Open);
577            }
578        }
579    }
580 
581    /// Close stream.
582    pub fn close_stream(&mut self, id: u16) {
583        if let Some(entry) = entry_by_id_mut(&mut self.entries, id) {
584            entry.do_close = true;
585 
586            // Explicitly close the sctp stream to allow re-use of the same id.
587            let _ = self.sctp_propagate_close(id);
588        }
589    }
590 
591    pub fn close(&mut self) -> Result<(), SctpError> {
592        let Some(assoc) = &mut self.assoc else {
593            return Ok(());
594        };
595 
596        if assoc.is_closing() || assoc.is_closed() {
597            return Ok(());
598        }
599 
600        Ok(assoc.shutdown()?)
601    }
602 
603    pub fn is_open(&self, id: u16) -> bool {
604        if self.state != RtcSctpState::Established {
605            return false;
606        }
607 
608        let Some(rec) = entry_by_id(&self.entries, id) else {
609            return false;
610        };
611 
612        rec.state == StreamEntryState::Open
613    }
614 
615    // TODO: fix sctp-proto so we don't need &mut here.
616    pub fn available(&mut self) -> usize {
617        let Some(assoc) = &mut self.assoc else {
618            return 0;
619        };
620 
621        // The amount currently buffered.
622        let total: usize = self
623            .entries
624            .iter()
625            .filter_map(|e| {
626                assoc
627                    .stream(e.id)
628                    .ok()
629                    .and_then(|s| s.buffered_amount().ok())
630            })
631            .sum();
632 
633        MAX_BUFFERED_ACROSS_STREAMS.saturating_sub(total)
634    }
635 
636    pub fn write(&mut self, id: u16, binary: bool, buf: &[u8]) -> Result<usize, SctpError> {
637        if self.state != RtcSctpState::Established || self.is_closing() || self.is_closed() {
638            return Err(SctpError::WriteBeforeEstablished);
639        }
640 
641        let assoc = self
642            .assoc
643            .as_mut()
644            .ok_or(SctpError::WriteBeforeEstablished)?;
645 
646        let rec = entry_by_id(&self.entries, id).expect("stream entry for write");
647 
648        if rec.state != StreamEntryState::Open {
649            return Err(SctpError::WriteBeforeEstablished);
650        }
651 
652        let mut stream = assoc.stream(id)?;
653 
654        let ppi = if binary {
655            if buf.is_empty() {
656                PayloadProtocolIdentifier::BinaryEmpty
657            } else {
658                PayloadProtocolIdentifier::Binary
659            }
660        } else if buf.is_empty() {
661            PayloadProtocolIdentifier::StringEmpty
662        } else {
663            PayloadProtocolIdentifier::String
664        };
665 
666        Ok(stream.write_with_ppi(buf, ppi)?)
667    }
668 
669    pub fn buffered_amount(&mut self, id: u16) -> usize {
670        let Some(assoc) = self.assoc.as_mut() else {
671            return 0;
672        };
673 
674        let Ok(stream) = assoc.stream(id) else {
675            return 0;
676        };
677 
678        stream.buffered_amount().unwrap_or(0)
679    }
680 
681    pub fn set_buffered_amount_low_threshold(&mut self, id: u16, threshold: usize) {
682        let entry =
683            entry_by_id_mut(&mut self.entries, id).expect("stream entry for valid channel id");
684 
685        // This update will be propagated on next poll.
686        entry.buffered_threshold.set(threshold);
687    }
688 
689    pub fn handle_input(&mut self, now: Instant, data: &[u8]) {
690        trace!("Handle input: {}", data.len());
691 
692        // TODO, remove Bytes in sctp and just use &[u8].
693        let data = data.to_vec().into();
694        let r = self.endpoint.handle(now, self.fake_addr, None, None, data);
695 
696        let Some((handle, event)) = r else {
697            return;
698        };
699 
700        match event {
701            DatagramEvent::NewAssociation(a) => {
702                // In slow or unreliable networks from browsers (use 3g or slow 4g) settings.
703                // The browser resends a new associations and str0m would override the previously
704                // acked association. Webrtc should use only 1 association.
705                if self.assoc.is_some() {
706                    return;
707                }
708                debug!("New remote association");
709                // Remote side initiated the association
710                self.assoc = Some(a);
711                self.handle = handle;
712                set_state(&mut self.state, RtcSctpState::AwaitAssociationEstablished);
713            }
714            DatagramEvent::AssociationEvent(event) => {
715                self.assoc
716                    .as_mut()
717                    .expect("association for event")
718                    .handle_event(event);
719            }
720        }
721    }
722 
723    pub fn handle_timeout(&mut self, now: Instant) {
724        if self.state == RtcSctpState::Uninited {
725            // Need to call `init()` before any timeouts are accepted.
726            return;
727        }
728 
729        self.last_now = now;
730 
731        // Remove closed entries.
732        self.entries.retain(|e| e.state != StreamEntryState::Closed);
733 
734        let Some(assoc) = &mut self.assoc else {
735            return;
736        };
737 
738        assoc.handle_timeout(now);
739 
740        // propagate events between endpoint and association.
741        while let Some(e) = assoc.poll_endpoint_event() {
742            if let Some(ae) = self.endpoint.handle_event(self.handle, e) {
743                assoc.handle_event(ae);
744            }
745        }
746    }
747 
748    pub fn poll(&mut self) -> Option<SctpEvent> {
749        let r = self.do_poll();
750 
751        if let Some(r) = &r {
752            trace!("Poll {:?}", r);
753        }
754 
755        r
756    }
757 
758    pub fn do_poll(&mut self) -> Option<SctpEvent> {
759        // A completed handshake or a written DCEP open can produce more to send, so
760        // those start over from the top instead of falling through. There is one
761        // restart per channel waiting to open, which is why this is a loop and not
762        // recursion.
763        'restart: loop {
764            if self.state == RtcSctpState::Uninited {
765                // Need to call `init()` before any polling starts.
766                return None;
767            }
768 
769            // Remove closed entries. handle_timeout() also does this, but the
770            // remote can reuse a stream id (its reset handshake completed) before
771            // our next timeout.
772            self.entries.retain(|e| e.state != StreamEntryState::Closed);
773 
774            if let Some(t) = self.pushed_back_transmit.take() {
775                return Some(SctpEvent::Transmit { packets: t });
776            }
777 
778            while let Some(t) = self.poll_transmit() {
779                let Some(buf) = transmit_to_vec(t) else {
780                    continue;
781                };
782 
783                return Some(SctpEvent::Transmit { packets: buf });
784            }
785 
786            // Don't progress to move data between association and endpoint until we have an
787            // association we want to drive forward.
788            if !self.state.propagate_endpoint_to_assoc() {
789                return None;
790            }
791 
792            let assoc = self.assoc.as_mut()?;
793 
794            while let Some(e) = assoc.poll() {
795                if let Event::Connected = e {
796                    assoc.set_max_send_message_size(self.remote_max_message_size);
797                    set_state(&mut self.state, RtcSctpState::Established);
798                    continue 'restart;
799                }
800 
801                if let Event::AssociationLost { ref reason } = e {
802                    debug!("Association lost, reason: {}", reason);
803                    // No reset can complete on a dead association.
804                    self.reset_pending.clear();
805                    self.reset_complete.clear();
806                    return Some(SctpEvent::AssociationLost);
807                }
808 
809                if let Event::Stream(se) = e {
810                    match se {
811                        StreamEvent::Readable { id } | StreamEvent::Writable { id } => {
812                            stream_entry(
813                                &mut self.entries,
814                                id,
815                                StreamEntryState::AwaitConfig,
816                                "readable/writable",
817                            );
818                        }
819                        StreamEvent::Finished { id } | StreamEvent::Stopped { id, .. } => {
820                            // sctp-proto unregistered it when a reset arrived.
821                            // Id reuse is signalled separately by StreamEvent::ResetComplete.
822                            //
823                            // sctp-proto arms that completion here, so from now on a reset
824                            // is outstanding for the id even if we never closed it locally.
825                            self.reset_pending.insert(id);
826 
827                            // Only a live entry has anything to drop. Closed entries and
828                            // missing ones already went through Close, and an AwaitOpen
829                            // entry is a new incarnation waiting on the same id, it must
830                            // not be killed by the old pending teardown.
831                            if let Some(entry) = entry_by_id_mut(&mut self.entries, id) {
832                                if entry.state != StreamEntryState::Closed
833                                    && entry.state != StreamEntryState::AwaitOpen
834                                {
835                                    debug!("Stream {} finished", id);
836                                    entry.do_close = true;
837                                }
838                            }
839                        }
840                        StreamEvent::ResetComplete { id } => {
841                            // The reset handshake for this id has fully completed.
842                            debug!("Stream {} reset complete", id);
843                            self.reset_pending.remove(&id);
844                            self.reset_complete.push_back(id);
845                        }
846                        StreamEvent::BufferedAmountLow { id } => {
847                            return Some(SctpEvent::BufferedAmountLow { id });
848                        }
849                        _ => {}
850                    }
851                }
852            }
853 
854            // Must wait for association state to be established before opening streams.
855            if self.state != RtcSctpState::Established {
856                return None;
857            }
858 
859            for entry in self.entries.iter_mut() {
860                let want_open = entry.state == StreamEntryState::AwaitOpen;
861 
862                if want_open {
863                    debug!("Open stream {}", entry.id);
864                    match assoc.open_stream(entry.id, PayloadProtocolIdentifier::Unknown) {
865                        Ok(mut s) => {
866                            entry.open_deadline = None;
867 
868                            if !entry.configure_reliability(&mut s) {
869                                entry.set_state(StreamEntryState::Closed);
870                                let stream_id = entry.id;
871                                let reset_pending = self.sctp_propagate_close(stream_id);
872                                return Some(SctpEvent::Close {
873                                    id: stream_id,
874                                    reset_pending,
875                                });
876                            }
877 
878                            let config = entry.config.as_ref().expect("config if AwaitOpen");
879                            let in_band = config.negotiated.is_none();
880 
881                            if in_band {
882                                let dcep: DcepOpen = config.into();
883                                let mut buf = vec![0; 1500];
884                                let n = dcep.marshal_to(&mut buf);
885                                buf.truncate(n);
886 
887                                match s.write_with_ppi(&buf, PayloadProtocolIdentifier::Dcep) {
888                                    Ok(l) => {
889                                        assert!(n == l);
890                                        entry.set_state(StreamEntryState::AwaitDcepAck);
891 
892                                        // Start over with polling, since we might have caused
893                                        // some network traffic by writing the DcepOpen.
894                                        continue 'restart;
895                                    }
896                                    Err(e) => {
897                                        warn!(
898                                            "Failed to write DCEP open on stream {}: {:?}",
899                                            entry.id, e
900                                        );
901                                        entry.do_close = true;
902                                        entry.set_state(StreamEntryState::Closed);
903                                        let stream_id = entry.id;
904                                        let reset_pending = self.sctp_propagate_close(stream_id);
905                                        return Some(SctpEvent::Close {
906                                            id: stream_id,
907                                            reset_pending,
908                                        });
909                                    }
910                                }
911                            }
912 
913                            // Continuing means we are opening the stream out-of-band.
914                        }
915                        Err(
916                            e @ (ProtoError::ErrStreamAlreadyExist
917                            | ProtoError::ErrStreamResetPending),
918                        ) => {
919                            let config = entry.config.as_ref().expect("config if AwaitOpen");
920                            let in_band = config.negotiated.is_none();
921 
922                            // ErrStreamAlreadyExist has two causes:
923                            // - the remote created it by sending on it first
924                            // - a previous incarnation of the id is still registered,
925                            //   reset handshake hasn't finished
926                            let stale_incarnation = matches!(e, ProtoError::ErrStreamAlreadyExist)
927                                && assoc
928                                    .stream(entry.id)
929                                    .map(|s| !s.is_readable() && !s.is_writable())
930                                    .unwrap_or(false);
931 
932                            if in_band
933                                || stale_incarnation
934                                || matches!(e, ProtoError::ErrStreamResetPending)
935                            {
936                                // RFC 6525 reset handshake for a previous
937                                // incarnation of this stream id hasn't finished yet
938                                //
939                                // - AlreadyExist clears when the remote's reciprocal reset arrives
940                                // - ResetPending when the RECONFIG-RESPONSE arrives (silently)
941                                //
942                                // In both cases,  stay in AwaitOpen and retry until past the deadline.
943                                let deadline = *entry
944                                    .open_deadline
945                                    .get_or_insert(self.last_now + STREAM_OPEN_TIMEOUT);
946 
947                                if self.last_now < deadline {
948                                    debug!(
949                                        "Stream {} open blocked ({:?}), will retry",
950                                        entry.id, e
951                                    );
952                                    continue;
953                                }
954 
955                                debug!("Opening stream {} failed after retries: {:?}", entry.id, e);
956                                entry.do_close = true;
957                                entry.set_state(StreamEntryState::Closed);
958                                let stream_id = entry.id;
959                                let reset_pending = self.sctp_propagate_close(stream_id);
960                                return Some(SctpEvent::Close {
961                                    id: stream_id,
962                                    reset_pending,
963                                });
964                            }
965 
966                            // Continuing means we are adopting the live out-of-band stream the
967                            // remote created. It skipped the Ok branch above, so reliability
968                            // params haven't been applied to it yet.
969                            let mut stream = assoc.stream(entry.id).expect("stream that exists");
970                            if !entry.configure_reliability(&mut stream) {
971                                entry.set_state(StreamEntryState::Closed);
972                                let stream_id = entry.id;
973                                let reset_pending = self.sctp_propagate_close(stream_id);
974                                return Some(SctpEvent::Close {
975                                    id: stream_id,
976                                    reset_pending,
977                                });
978                            }
979                        }
980                        Err(e) => {
981                            warn!("Opening stream {} failed: {:?}", entry.id, e);
982                            entry.do_close = true;
983                            entry.set_state(StreamEntryState::Closed);
984                            let stream_id = entry.id;
985                            let reset_pending = self.sctp_propagate_close(stream_id);
986                            return Some(SctpEvent::Close {
987                                id: stream_id,
988                                reset_pending,
989                            });
990                        }
991                    }
992 
993                    // Consider out-of-band stream open.
994                    let config = entry.config.as_ref().expect("config if AwaitOpen");
995                    let in_band = config.negotiated.is_none();
996                    assert!(!in_band);
997 
998                    let label = config.label.clone();
999                    entry.set_state(StreamEntryState::Open);
1000 
1001                    return Some(SctpEvent::Open {
1002                        id: entry.id,
1003                        label,
1004                    });
1005                }
1006 
1007                if entry.do_close && entry.state != StreamEntryState::Closed {
1008                    entry.set_state(StreamEntryState::Closed);
1009                    let stream_id = entry.id;
1010                    let reset_pending = self.sctp_propagate_close(stream_id);
1011                    return Some(SctpEvent::Close {
1012                        id: stream_id,
1013                        reset_pending,
1014                    });
1015                }
1016 
1017                let mut stream = match assoc.stream(entry.id) {
1018                    Ok(v) => v,
1019                    Err(e) => {
1020                        // This is expected on browser refresh or similar abrupt shutdown.
1021                        debug!("Getting stream {} failed: {:?}", entry.id, e);
1022                        entry.do_close = true;
1023                        continue;
1024                    }
1025                };
1026 
1027                // Propagate the desired buffered threshold.
1028                // The idea is to only do this if the user has changed the value for it without
1029                // incurring the cost of looking up the currently confifured value.
1030                if let BufferedThresholdConfig::Desired(x) = entry.buffered_threshold {
1031                    if let Err(e) = stream.set_buffered_amount_low_threshold(x) {
1032                        debug!("Setting buffered_amount_low_threshold failed: {:?}", e);
1033                        entry.do_close = true;
1034                        entry.buffered_threshold = BufferedThresholdConfig::Unconfigured;
1035                        continue;
1036                    }
1037 
1038                    entry.buffered_threshold = BufferedThresholdConfig::Configured(x);
1039                }
1040                match stream_read_data(&mut stream) {
1041                    Ok(Some((buf, ppi))) => {
1042                        if ppi != PayloadProtocolIdentifier::Dcep {
1043                            // This is the normal path for incoming data.
1044                            let buf = ppi_adjust_buf(buf, ppi);
1045                            let binary = matches!(
1046                                ppi,
1047                                PayloadProtocolIdentifier::Binary
1048                                    | PayloadProtocolIdentifier::BinaryEmpty
1049                            );
1050                            return Some(SctpEvent::Data {
1051                                id: entry.id,
1052                                binary,
1053                                data: buf,
1054                            });
1055                        }
1056 
1057                        // It's Dcep, either a DcepOpen or DcepAck.
1058                        match entry.state {
1059                            // We are in AwaitConfig state which means we are either going to get it via
1060                            // the DcepOpen, or by an out-of-band configuration via open_stream.
1061                            // This indicates we are doing in-band.
1062                            StreamEntryState::AwaitConfig => {
1063                                let dcep: DcepOpen = match buf.as_slice().try_into() {
1064                                    Ok(v) => v,
1065                                    Err(e) => {
1066                                        warn!("Failed to read incoming DCEP {}: {:?}", entry.id, e);
1067                                        entry.do_close = true;
1068                                        continue;
1069                                    }
1070                                };
1071 
1072                                if entry.config.is_none() {
1073                                    entry.config = Some((&dcep).into());
1074                                } else {
1075                                    warn!("Received DcepOpen for configured stream: {}", entry.id);
1076                                }
1077 
1078                                // Apply DcepOpen's reliability parameters to the sctp-proto stream.
1079                                // Without this, the DCEP-receiving side of an in-band channel sends
1080                                // with stream defaults: ordered and fully reliable.
1081                                if !entry.configure_reliability(&mut stream) {
1082                                    continue;
1083                                }
1084 
1085                                let mut obuf = [0];
1086                                DcepAck.marshal_to(&mut obuf);
1087                                match stream.write_with_ppi(&obuf, PayloadProtocolIdentifier::Dcep)
1088                                {
1089                                    Ok(l) => {
1090                                        assert!(obuf.len() == l);
1091                                        entry.set_state(StreamEntryState::Open);
1092 
1093                                        return Some(SctpEvent::Open {
1094                                            id: entry.id,
1095                                            label: dcep.label,
1096                                        });
1097                                    }
1098                                    Err(e) => {
1099                                        warn!(
1100                                            "Failed to write DCEP ack on stream {}: {:?}",
1101                                            entry.id, e
1102                                        );
1103                                        entry.do_close = true;
1104                                        entry.set_state(StreamEntryState::Closed);
1105                                        let stream_id = entry.id;
1106                                        let reset_pending = self.sctp_propagate_close(stream_id);
1107                                        return Some(SctpEvent::Close {
1108                                            id: stream_id,
1109                                            reset_pending,
1110                                        });
1111                                    }
1112                                }
1113                            }
1114                            StreamEntryState::AwaitDcepAck => {
1115                                let res: Result<DcepAck, _> = buf.as_slice().try_into();
1116 
1117                                if let Err(e) = res {
1118                                    warn!("Failed to read incoming DCEP ACK {}: {:?}", entry.id, e);
1119                                    entry.do_close = true;
1120                                    continue;
1121                                }
1122 
1123                                entry.set_state(StreamEntryState::Open);
1124                                let config = entry.config.as_ref().expect("config when DcepAck");
1125 
1126                                return Some(SctpEvent::Open {
1127                                    id: entry.id,
1128                                    label: config.label.clone(),
1129                                });
1130                            }
1131                            _ => {
1132                                warn!(
1133                                    "Stream {} in wrong state when receiving DCEP: {:?}",
1134                                    entry.id, entry.state
1135                                );
1136                                entry.do_close = true;
1137                                continue;
1138                            }
1139                        }
1140                    }
1141                    Ok(None) => continue,
1142                    Err(_) => entry.do_close = true,
1143                }
1144            }
1145 
1146            // Reset completions are reported last. Reaching here means the entry loop had
1147            // no `Close` left to emit, so the close for a released id has been delivered.
1148            if let Some(id) = self.reset_complete.pop_front() {
1149                return Some(SctpEvent::StreamResetComplete { id });
1150            }
1151 
1152            return None;
1153        }
1154    }
1155 
1156    pub fn poll_timeout(&mut self) -> Option<Instant> {
1157        let assoc_timeout = self.assoc.as_mut().and_then(|a| a.poll_timeout());
1158 
1159        // Wakeup backstop for entries whose open_stream() is being retried.
1160        //
1161        // Normally no wakeup is needed: a blocked open re-attempts on the
1162        // next do_poll(), and the packet that unblocks it (the peer's
1163        // RECONFIG-RESPONSE or reset) itself triggers handle_input() and that poll.
1164        //
1165        // Returning a wakeup while any entry retries guarantees the open
1166        // either resolves or fails within STREAM_OPEN_TIMEOUT.
1167        let retry_timeout = self
1168            .entries
1169            .iter()
1170            .any(|e| e.state == StreamEntryState::AwaitOpen && e.open_deadline.is_some())
1171            .then(|| self.last_now + STREAM_OPEN_RETRY_INTERVAL);
1172 
1173        match (assoc_timeout, retry_timeout) {
1174            (Some(a), Some(r)) => Some(a.min(r)),
1175            (a, r) => a.or(r),
1176        }
1177    }
1178 
1179    pub fn push_back_transmit(&mut self, data: VecDeque<Bytes>) {
1180        trace!("Push back transmit: {}", data.len());
1181        assert!(self.pushed_back_transmit.is_none());
1182        self.pushed_back_transmit = Some(data);
1183    }
1184 
1185    fn poll_transmit(&mut self) -> Option<Transmit> {
1186        if let Some(t) = self.endpoint.poll_transmit() {
1187            return Some(t);
1188        }
1189 
1190        if let Some(t) = self.assoc.as_mut()?.poll_transmit(self.last_now) {
1191            return Some(t);
1192        }
1193 
1194        None
1195    }
1196 
1197    pub fn config(&self, sctp_stream_id: u16) -> Option<&ChannelConfig> {
1198        entry_by_id(&self.entries, sctp_stream_id).and_then(|s| s.config.as_ref())
1199    }
1200 
1201    /// Close the sctp stream to allow re-use of the same id.
1202    ///
1203    /// Returns whether a reset handshake is outstanding for `stream_id`.
1204    fn sctp_propagate_close(&mut self, stream_id: u16) -> bool {
1205        let did_reset = match self.assoc.as_mut().map(|assoc| assoc.stream(stream_id)) {
1206            // `close()` only fails on the reset, which needs an established
1207            // association. Closing a channel while the association is shutting down
1208            // therefore queues nothing and no completion will ever arrive.
1209            Some(Ok(mut stream)) => stream.close().is_ok(),
1210            // No stream to reset.
1211            _ => false,
1212        };
1213 
1214        if did_reset {
1215            self.reset_pending.insert(stream_id);
1216        }
1217 
1218        self.reset_pending.contains(&stream_id)
1219    }
1220 
1221    #[cfg(test)]
1222    pub(crate) fn remote_max_message_size(&self) -> u32 {
1223        self.remote_max_message_size
1224    }
1225}
1226 
1227fn transmit_to_vec(t: Transmit) -> Option<VecDeque<Bytes>> {
1228    let Payload::RawEncode(v) = t.payload else {
1229        return None;
1230    };
1231 
1232    Some(v.into())
1233}
1234 
1235fn set_state(current_state: &mut RtcSctpState, state: RtcSctpState) {
1236    if *current_state != state {
1237        debug!("{:?} => {:?}", current_state, state);
1238        *current_state = state;
1239    }
1240}
1241 
1242fn entry_index(entries: &[StreamEntry], id: u16) -> Result<usize, usize> {
1243    entries.binary_search_by_key(&id, |e| e.id)
1244}
1245 
1246fn entry_by_id(entries: &[StreamEntry], id: u16) -> Option<&StreamEntry> {
1247    entry_index(entries, id).ok().map(|i| &entries[i])
1248}
1249 
1250fn entry_by_id_mut(entries: &mut [StreamEntry], id: u16) -> Option<&mut StreamEntry> {
1251    entry_index(entries, id).ok().map(|i| &mut entries[i])
1252}
1253 
1254fn stream_entry<'a>(
1255    entries: &'a mut Vec<StreamEntry>,
1256    id: u16,
1257    initial_state: StreamEntryState,
1258    reason: &'static str,
1259) -> &'a mut StreamEntry {
1260    let idx = match entry_index(entries, id) {
1261        Ok(idx) => idx,
1262        Err(idx) => {
1263            debug!("New stream {} ({:?}): {}", id, initial_state, reason);
1264            entries.insert(
1265                idx,
1266                StreamEntry {
1267                    config: None,
1268                    state: initial_state,
1269                    id,
1270                    do_close: false,
1271                    open_deadline: None,
1272                    buffered_threshold: BufferedThresholdConfig::Unconfigured,
1273                },
1274            );
1275            idx
1276        }
1277    };
1278 
1279    &mut entries[idx]
1280}
1281 
1282fn stream_read_data(
1283    stream: &mut Stream,
1284) -> Result<Option<(Vec<u8>, PayloadProtocolIdentifier)>, SctpError> {
1285    let Some(chunks) = stream.read()? else {
1286        return Ok(None);
1287    };
1288 
1289    let n = chunks.len();
1290    let mut buf = vec![0; n];
1291 
1292    let l = chunks.read(&mut buf)?;
1293    assert!(l == n);
1294 
1295    use PayloadProtocolIdentifier::*;
1296    match chunks.ppi {
1297        Dcep | String | Binary => {} // keep as is
1298        StringEmpty | BinaryEmpty => buf.clear(),
1299        _ => {
1300            return Err(SctpError::Proto(ProtoError::Other(
1301                "Unknown PayloadProtocolIdentifier".into(),
1302            )));
1303        }
1304    }
1305 
1306    Ok(Some((buf, chunks.ppi)))
1307}
1308 
1309fn ppi_adjust_buf(mut buf: Vec<u8>, ppi: PayloadProtocolIdentifier) -> Vec<u8> {
1310    match ppi {
1311        PayloadProtocolIdentifier::StringEmpty | PayloadProtocolIdentifier::BinaryEmpty => {
1312            buf.clear();
1313            buf
1314        }
1315        _ => buf,
1316    }
1317}
1318 
1319impl fmt::Debug for SctpEvent {
1320    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1321        match self {
1322            Self::Transmit { packets } => f
1323                .debug_struct("Transmit")
1324                .field("packets", &packets.len())
1325                .finish(),
1326            Self::Open { id, label } => f
1327                .debug_struct("Open")
1328                .field("id", id)
1329                .field("label", label)
1330                .finish(),
1331            Self::Close { id, reset_pending } => f
1332                .debug_struct("Close")
1333                .field("id", id)
1334                .field("reset_pending", reset_pending)
1335                .finish(),
1336            Self::StreamResetComplete { id } => f
1337                .debug_struct("StreamResetComplete")
1338                .field("id", id)
1339                .finish(),
1340            Self::Data { id, binary, data } => f
1341                .debug_struct("Data")
1342                .field("id", id)
1343                .field("binary", binary)
1344                .field("data", &data.len())
1345                .finish(),
1346            Self::BufferedAmountLow { id } => {
1347                f.debug_struct("BufferedAmountLow").field("id", id).finish()
1348            }
1349            Self::AssociationLost => f.debug_struct("AssociationLost").finish(),
1350        }
1351    }
1352}
1353 
1354impl From<&ChannelConfig> for DcepOpen {
1355    fn from(v: &ChannelConfig) -> Self {
1356        let (channel_type, reliability_parameter) = (&v.reliability).into();
1357        DcepOpen {
1358            unordered: !v.ordered,
1359            channel_type,
1360            reliability_parameter,
1361            priority: 0,
1362            label: v.label.clone(),
1363            protocol: v.protocol.clone(),
1364        }
1365    }
1366}
1367 
1368impl From<&Reliability> for (ReliabilityType, u32) {
1369    fn from(v: &Reliability) -> Self {
1370        match v {
1371            Reliability::Reliable => (ReliabilityType::Reliable, 0),
1372            Reliability::MaxPacketLifetime { lifetime } => {
1373                (ReliabilityType::Timed, *lifetime as u32)
1374            }
1375            Reliability::MaxRetransmits { retransmits } => {
1376                (ReliabilityType::Rexmit, *retransmits as u32)
1377            }
1378        }
1379    }
1380}
1381 
1382impl From<&DcepOpen> for ChannelConfig {
1383    fn from(v: &DcepOpen) -> Self {
1384        ChannelConfig {
1385            label: v.label.clone(),
1386            ordered: !v.unordered,
1387            reliability: (v.channel_type, v.reliability_parameter).into(),
1388            negotiated: None,
1389            protocol: v.protocol.clone(),
1390        }
1391    }
1392}
1393 
1394impl From<(ReliabilityType, u32)> for Reliability {
1395    fn from((r, p): (ReliabilityType, u32)) -> Self {
1396        match r {
1397            ReliabilityType::Reliable => Reliability::Reliable,
1398            ReliabilityType::Rexmit => Reliability::MaxRetransmits {
1399                retransmits: p as u16,
1400            },
1401            ReliabilityType::Timed => Reliability::MaxPacketLifetime { lifetime: p as u16 },
1402        }
1403    }
1404}
1405 
1406#[cfg(test)]
1407mod tests {
1408    use super::*;
1409 
1410    /// Insert an entry directly, keeping `entries` sorted by id.
1411    fn insert_entry(entries: &mut Vec<StreamEntry>, entry: StreamEntry) {
1412        let idx = entry_index(entries, entry.id).expect_err("entry should not already exist");
1413        entries.insert(idx, entry);
1414    }
1415    use str0m_proto::DATAGRAM_MTU_TARGET;
1416 
1417    #[test]
1418    fn partial_snap_init_requires_both_chunks() {
1419        let now = Instant::now();
1420        let mut sctp = RtcSctp::new(DATAGRAM_MTU_TARGET);
1421        let mut init_data = SctpInitData::new();
1422 
1423        init_data.local_init_chunk().unwrap();
1424 
1425        let err = sctp.init(true, now, Some(init_data), None).unwrap_err();
1426        assert!(
1427            err.to_string()
1428                .contains("SNAP requires both local and remote SCTP INIT chunks")
1429        );
1430    }
1431 
1432    #[test]
1433    fn malformed_remote_snap_does_not_disable_local_opt_in() {
1434        let mut sctp = RtcSctp::new(DATAGRAM_MTU_TARGET);
1435        sctp.enable_snap();
1436 
1437        assert!(!sctp.set_remote_snap_init_string("!!!not-valid-base64!!!"));
1438        assert!(sctp.snap_enabled());
1439        assert!(sctp.ensure_local_snap_init());
1440        assert!(sctp.local_sctp_init_for_sdp().is_some());
1441    }
1442 
1443    /// Helper to connect a client and server RtcSctp pair to Established state.
1444    fn connect_client_server() -> (RtcSctp, RtcSctp) {
1445        connect_client_server_with_limits(None, None)
1446    }
1447 
1448    fn connect_client_server_with_limits(
1449        client_limits: Option<SctpReceiveLimits>,
1450        server_limits: Option<SctpReceiveLimits>,
1451    ) -> (RtcSctp, RtcSctp) {
1452        let now = Instant::now();
1453        let mut client = RtcSctp::with_receive_limits(DATAGRAM_MTU_TARGET, client_limits);
1454        let mut server = RtcSctp::with_receive_limits(DATAGRAM_MTU_TARGET, server_limits);
1455 
1456        client.init(true, now, None, None).unwrap();
1457        server.init(false, now, None, None).unwrap();
1458 
1459        // Exchange packets until both are Established.
1460        for _ in 0..20 {
1461            // Drain client transmits -> feed to server
1462            while let Some(t) = client.poll_transmit() {
1463                if let Some(bufs) = transmit_to_vec(t) {
1464                    for buf in bufs {
1465                        server.handle_input(now, &buf);
1466                    }
1467                }
1468            }
1469            // Process server events
1470            while let Some(e) = server.do_poll() {
1471                if let SctpEvent::Transmit { packets } = e {
1472                    for buf in packets {
1473                        client.handle_input(now, &buf);
1474                    }
1475                }
1476            }
1477 
1478            // Drain server transmits -> feed to client
1479            while let Some(t) = server.poll_transmit() {
1480                if let Some(bufs) = transmit_to_vec(t) {
1481                    for buf in bufs {
1482                        client.handle_input(now, &buf);
1483                    }
1484                }
1485            }
1486            // Process client events
1487            while let Some(e) = client.do_poll() {
1488                if let SctpEvent::Transmit { packets } = e {
1489                    for buf in packets {
1490                        server.handle_input(now, &buf);
1491                    }
1492                }
1493            }
1494 
1495            // Check if both established
1496            if client.state == RtcSctpState::Established
1497                && server.state == RtcSctpState::Established
1498            {
1499                break;
1500            }
1501        }
1502 
1503        assert_eq!(client.state, RtcSctpState::Established);
1504        assert_eq!(server.state, RtcSctpState::Established);
1505 
1506        (client, server)
1507    }
1508 
1509    #[test]
1510    fn receive_limits_apply_to_both_association_roles() {
1511        fn pump(from: &mut RtcSctp, to: &mut RtcSctp) -> Vec<SctpEvent> {
1512            let mut output = Vec::new();
1513            while let Some(event) = from.do_poll() {
1514                if let SctpEvent::Transmit { packets } = event {
1515                    for packet in packets {
1516                        to.handle_input(from.last_now, &packet);
1517                    }
1518                } else {
1519                    output.push(event);
1520                }
1521            }
1522            output
1523        }
1524        let limits = SctpReceiveLimits::new(8192, 32768, 64, 8);
1525        for receiver_is_client in [false, true] {
1526            let (mut client, mut server) = connect_client_server_with_limits(
1527                receiver_is_client.then_some(limits),
1528                (!receiver_is_client).then_some(limits),
1529            );
1530            let (sender, receiver) = if receiver_is_client {
1531                (&mut server, &mut client)
1532            } else {
1533                (&mut client, &mut server)
1534            };
1535            for id in [0, 65000] {
1536                let config = ChannelConfig {
1537                    negotiated: Some(id),
1538                    ..Default::default()
1539                };
1540                sender.open_stream(id, config.clone());
1541                receiver.open_stream(id, config);
1542            }
1543            pump(sender, receiver);
1544            pump(receiver, sender);
1545            let mut now = Instant::now();
1546            for (id, size) in [(0, 8192), (65000, 512), (65000, 8193)] {
1547                sender.write(id, true, &vec![42; size]).unwrap();
1548                let mut received = None;
1549                let mut lost = false;
1550                for _ in 0..200 {
1551                    now += Duration::from_millis(10);
1552                    sender.handle_timeout(now);
1553                    receiver.handle_timeout(now);
1554                    pump(sender, receiver);
1555                    for event in pump(receiver, sender) {
1556                        match event {
1557                            SctpEvent::Data {
1558                                id: stream, data, ..
1559                            } => received = Some((stream, data)),
1560                            SctpEvent::AssociationLost => lost = true,
1561                            _ => {}
1562                        }
1563                    }
1564                    if received.is_some() || lost {
1565                        break;
1566                    }
1567                }
1568                if size <= 8192 {
1569                    assert!(!lost);
1570                    assert_eq!(received, Some((id, vec![42; size])));
1571                } else {
1572                    assert!(lost);
1573                    assert!(received.is_none());
1574                }
1575            }
1576        }
1577    }
1578 
1579    #[test]
1580    fn snap_advertises_the_configured_receive_window() {
1581        let limits = SctpReceiveLimits::new(8192, 32768, 64, 8);
1582        let mut direct = SctpInitData::with_receive_limits(limits);
1583        let init = direct.local_init_chunk().unwrap();
1584        assert_eq!(u32::from_be_bytes(init[8..12].try_into().unwrap()), 32768);
1585        let mut sctp = RtcSctp::with_receive_limits(DATAGRAM_MTU_TARGET, Some(limits));
1586        sctp.enable_snap();
1587        assert!(sctp.ensure_local_snap_init());
1588        let init = sctp
1589            .snap_init
1590            .as_ref()
1591            .unwrap()
1592            .local_init
1593            .as_ref()
1594            .unwrap();
1595        assert_eq!(u32::from_be_bytes(init[8..12].try_into().unwrap()), 32768);
1596    }
1597 
1598    #[test]
1599    fn transmit_retains_packet_ownership_and_order() {
1600        let packets = vec![Bytes::from(vec![1; 48]), Bytes::from(vec![2; 512])];
1601        let pointers = [packets[0].as_ptr(), packets[1].as_ptr()];
1602        let transmit = Transmit {
1603            now: Instant::now(),
1604            remote: "127.0.0.1:5000".parse().unwrap(),
1605            ecn: None,
1606            local_ip: None,
1607            payload: Payload::RawEncode(packets),
1608        };
1609        let output = transmit_to_vec(transmit).unwrap();
1610        assert_eq!(output.len(), 2);
1611        for (n, packet) in output.iter().enumerate() {
1612            assert_eq!(packet.as_ptr(), pointers[n]);
1613            assert!(packet.iter().all(|byte| *byte == n as u8 + 1));
1614        }
1615    }
1616 
1617    /// A stream the remote opened can be gone from the association by the time the
1618    /// application configures it.
1619    ///
1620    /// `StreamEntryState::AwaitConfig` is set when a `Readable`/`Writable` event
1621    /// arrives
1622    /// for an id we have no config for. If the peer then resets that stream, sctp-proto
1623    /// unregisters it during `handle_input()`, while our entry stays `AwaitConfig` until
1624    /// the reset event is drained in a later `do_poll()`.
1625    #[test]
1626    fn open_stream_out_of_band_after_remote_reset() {
1627        let (mut client, _server) = connect_client_server();
1628 
1629        let stream_id: u16 = 3;
1630        assert!(
1631            client.assoc.as_mut().unwrap().stream(stream_id).is_err(),
1632            "the association must not have this stream for the test to mean anything"
1633        );
1634 
1635        insert_entry(
1636            &mut client.entries,
1637            StreamEntry {
1638                config: None,
1639                state: StreamEntryState::AwaitConfig,
1640                id: stream_id,
1641                do_close: false,
1642                open_deadline: None,
1643                buffered_threshold: BufferedThresholdConfig::Unconfigured,
1644            },
1645        );
1646 
1647        client.open_stream(
1648            stream_id,
1649            ChannelConfig {
1650                label: "negotiated".to_string(),
1651                ordered: true,
1652                reliability: Reliability::Reliable,
1653                negotiated: Some(stream_id),
1654                protocol: String::new(),
1655            },
1656        );
1657 
1658        let entry = client
1659            .entries
1660            .iter()
1661            .find(|e| e.id == stream_id)
1662            .expect("entry to still exist");
1663 
1664        assert!(
1665            entry.do_close,
1666            "a stream that vanished from the association should be marked for close"
1667        );
1668        assert_ne!(
1669            entry.state,
1670            StreamEntryState::Open,
1671            "must not go Open when there is no underlying stream"
1672        );
1673    }
1674 
1675    /// Regression test: when `assoc.open_stream()` returns `ErrStreamAlreadyExist`
1676    /// for an in-band (DCEP) data channel, the entry must eventually transition to
1677    /// Closed and emit `SctpEvent::Close`. The error is transient (a reset
1678    /// handshake could clear it), so the entry retries first — but bounded by
1679    /// `STREAM_OPEN_TIMEOUT`, not the infinite loop this once was.
1680    #[test]
1681    fn err_stream_already_exist_in_band_returns_close() {
1682        let (mut client, _server) = connect_client_server();
1683 
1684        let stream_id: u16 = 0;
1685 
1686        // Pre-create the stream in the association so the next open_stream() with the
1687        // same ID will return ErrStreamAlreadyExist.
1688        let assoc = client.assoc.as_mut().unwrap();
1689        assoc
1690            .open_stream(stream_id, PayloadProtocolIdentifier::Unknown)
1691            .expect("first open_stream should succeed");
1692 
1693        // Manually add an entry in AwaitOpen state with in-band config (negotiated: None).
1694        // This simulates a locally-initiated in-band channel whose stream ID conflicts
1695        // with one already opened by the remote peer.
1696        insert_entry(
1697            &mut client.entries,
1698            StreamEntry {
1699                config: Some(ChannelConfig {
1700                    label: "test".to_string(),
1701                    ordered: true,
1702                    reliability: Reliability::Reliable,
1703                    negotiated: None, // in-band
1704                    protocol: String::new(),
1705                }),
1706                state: StreamEntryState::AwaitOpen,
1707                id: stream_id,
1708                do_close: false,
1709                open_deadline: None,
1710                buffered_threshold: BufferedThresholdConfig::Unconfigured,
1711            },
1712        );
1713 
1714        // The first poll retries instead of failing (arms the deadline).
1715        let event = client.do_poll();
1716        assert!(
1717            event.is_none(),
1718            "expected retry (no event) for stream {stream_id}, got {event:?}"
1719        );
1720 
1721        // Past the deadline the open gives up and closes.
1722        let later = Instant::now() + STREAM_OPEN_TIMEOUT + Duration::from_secs(1);
1723        client.handle_timeout(later);
1724        let event = client.do_poll();
1725 
1726        assert!(
1727            matches!(&event, Some(SctpEvent::Close { id, .. }) if *id == stream_id),
1728            "expected SctpEvent::Close for stream {stream_id}, got {event:?}"
1729        );
1730 
1731        // Verify entry transitioned to Closed.
1732        let entry = entry_by_id(&client.entries, stream_id).unwrap();
1733        assert_eq!(entry.state, StreamEntryState::Closed);
1734    }
1735 
1736    #[test]
1737    fn negotiated_reuse_does_not_report_open_for_old_closed_stream() {
1738        let (mut client, _server) = connect_client_server();
1739        let stream_id = 0;
1740 
1741        let assoc = client.assoc.as_mut().unwrap();
1742        let mut old = assoc
1743            .open_stream(stream_id, PayloadProtocolIdentifier::Unknown)
1744            .expect("old stream should open");
1745        old.close().expect("old stream should start closing");
1746 
1747        insert_entry(
1748            &mut client.entries,
1749            StreamEntry {
1750                config: Some(ChannelConfig {
1751                    label: "replacement".to_string(),
1752                    ordered: true,
1753                    reliability: Reliability::Reliable,
1754                    negotiated: Some(stream_id),
1755                    protocol: String::new(),
1756                }),
1757                state: StreamEntryState::AwaitOpen,
1758                id: stream_id,
1759                do_close: false,
1760                open_deadline: None,
1761                buffered_threshold: BufferedThresholdConfig::Unconfigured,
1762            },
1763        );
1764 
1765        for _ in 0..10 {
1766            match client.do_poll() {
1767                Some(SctpEvent::Open { id, .. }) if id == stream_id => {
1768                    panic!("replacement was reported open while the old stream still exists")
1769                }
1770                Some(_) => {}
1771                None => break,
1772            }
1773        }
1774    }
1775 
1776    #[test]
1777    fn close_is_reported_before_reset_complete_when_events_are_batched() {
1778        let now = Instant::now();
1779        let (mut client, mut server) = connect_client_server();
1780        let stream_id = 0;
1781        let config = ChannelConfig {
1782            label: "batched-close".to_string(),
1783            negotiated: Some(stream_id),
1784            ..Default::default()
1785        };
1786 
1787        client.open_stream(stream_id, config.clone());
1788        server.open_stream(stream_id, config);
1789        assert!(matches!(
1790            client.do_poll(),
1791            Some(SctpEvent::Open { id, .. }) if id == stream_id
1792        ));
1793        assert!(matches!(
1794            server.do_poll(),
1795            Some(SctpEvent::Open { id, .. }) if id == stream_id
1796        ));
1797 
1798        client.close_stream(stream_id);
1799 
1800        // Move all reset packets in both directions without polling client
1801        // application events. This batches Finished and ResetComplete in the
1802        // association event queue, which is valid for a sans-I/O caller.
1803        for _ in 0..4 {
1804            while let Some(transmit) = client.poll_transmit() {
1805                for packet in transmit_to_vec(transmit).unwrap() {
1806                    server.handle_input(now, &packet);
1807                }
1808            }
1809            while let Some(transmit) = server.poll_transmit() {
1810                for packet in transmit_to_vec(transmit).unwrap() {
1811                    client.handle_input(now, &packet);
1812                }
1813            }
1814        }
1815 
1816        for _ in 0..10 {
1817            match client.do_poll() {
1818                Some(SctpEvent::Close { id, .. }) if id == stream_id => return,
1819                Some(SctpEvent::StreamResetComplete { id }) if id == stream_id => {
1820                    panic!("ResetComplete was reported before Close")
1821                }
1822                Some(_) => {}
1823                None => break,
1824            }
1825        }
1826 
1827        panic!("stream close was not reported");
1828    }
1829 
1830    /// Regression test: the DCEP-receiving side of an in-band channel must apply
1831    /// the DcepOpen reliability parameters (here: unordered) to its sctp-proto
1832    /// stream, instead of sending with stream defaults (ordered, fully reliable).
1833    #[test]
1834    fn dcep_receiver_applies_reliability_params() {
1835        let now = Instant::now();
1836        let (mut client, mut server) = connect_client_server();
1837 
1838        // Drain `from`, feeding packets to `to` and returning the raw bytes.
1839        let pump = |from: &mut RtcSctp, to: &mut RtcSctp| {
1840            let mut wire = vec![];
1841            while let Some(e) = from.do_poll() {
1842                if let SctpEvent::Transmit { packets } = e {
1843                    for p in packets {
1844                        to.handle_input(now, &p);
1845                        wire.extend(p);
1846                    }
1847                }
1848            }
1849            wire
1850        };
1851 
1852        // Client opens an unordered in-band (DCEP) channel; once the handshake is
1853        // pumped through, the server (DCEP receiver) sends data back.
1854        client.open_stream(
1855            0,
1856            ChannelConfig {
1857                ordered: false,
1858                ..Default::default()
1859            },
1860        );
1861        pump(&mut client, &mut server); // DcepOpen
1862        pump(&mut server, &mut client); // DcepAck, server side now open
1863        let payload = b"from dcep receiver";
1864        server.write(0, true, payload).unwrap();
1865 
1866        // The 16-byte DATA chunk header puts the flags byte 15 bytes before the
1867        // payload it carries. 0x04 is the U (unordered) flag.
1868        let wire = pump(&mut server, &mut client);
1869        let pos = wire
1870            .windows(payload.len())
1871            .position(|w| w == payload)
1872            .unwrap();
1873        assert!(
1874            wire[pos - 15] & 0x04 != 0,
1875            "DCEP receiver should send unordered"
1876        );
1877    }
1878 
1879    #[test]
1880    fn max_payload_size_matches_mtu_minus_overhead() {
1881        let overhead = crate::io::MAX_DTLS_OVERHEAD + SCTP_OVERHEAD;
1882 
1883        let default_sctp = RtcSctp::new(DATAGRAM_MTU_TARGET);
1884        assert_eq!(
1885            default_sctp.max_payload_size(),
1886            DATAGRAM_MTU_TARGET - overhead
1887        );
1888 
1889        let small_sctp = RtcSctp::new(900);
1890        assert_eq!(small_sctp.max_payload_size(), 900 - overhead);
1891    }
1892}