Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/str0m/src/channel.rs

rust719 lines
1//! Data channel related types.
2 
3use std::{fmt, str, time::Instant};
4 
5use crate::sctp::RtcSctp;
6use crate::util::already_happened;
7use crate::{Rtc, RtcError};
8 
9pub use crate::sctp::ChannelConfig;
10pub use crate::sctp::Reliability;
11pub use crate::sctp::SctpInitData;
12pub use crate::sctp::SctpReceiveLimits;
13 
14/// Identifier of a data channel.
15///
16/// This is NOT the SCTP stream id.
17// Deliberately not Deref or From to avoid this Id being created outside of this module.
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
19pub struct ChannelId(usize);
20 
21/// Data channel data from remote peer.
22///
23/// This is obtained via [`Event::ChannelData`][crate::Event::ChannelData].
24#[derive(PartialEq, Eq)]
25pub struct ChannelData {
26    /// Identifier of the channel this data was sent on.
27    ///
28    /// The channel would have been previously announced via
29    /// [`Event::ChannelOpen`][crate::Event::ChannelOpen].
30    pub id: ChannelId,
31 
32    /// Tells whether the sender sent this data as binary or text.
33    pub binary: bool,
34 
35    /// The actual data sent. If `binary` is false, this can be converted to text.
36    pub data: Vec<u8>,
37}
38 
39/// Channel for sending data to the remote peer.
40///
41/// Get this handle from [`Rtc::channel()`][crate::Rtc::channel()].
42pub struct Channel<'a> {
43    sctp_stream_id: u16,
44    rtc: &'a mut Rtc,
45}
46 
47impl<'a> Channel<'a> {
48    pub(crate) fn new(sctp_stream_id: u16, rtc: &'a mut Rtc) -> Self {
49        Channel {
50            rtc,
51            sctp_stream_id,
52        }
53    }
54 
55    /// Write data to the remote peer and indicate whether it's text or binary.
56    ///
57    /// Returns true or false whether the buffer was accepted or not.
58    #[must_use = "Whether the buffer was accepted by the write()"]
59    pub fn write(&mut self, binary: bool, buf: &[u8]) -> Result<bool, RtcError> {
60        // If it's not available, don't accept.
61        let available = self.rtc.sctp.available();
62        if buf.len() > available {
63            return Ok(false);
64        }
65 
66        // Try write.
67        let written = self.rtc.sctp.write(self.sctp_stream_id, binary, buf)?;
68 
69        // Invariant: if available calculation is correct, we should have accepted.
70        assert_eq!(
71            written,
72            buf.len(),
73            "Data channel write() less than entire buffer"
74        );
75 
76        Ok(true)
77    }
78 
79    /// Get the amount of buffered data.
80    ///
81    /// Returns 0 if the channel is closed or encountered some error. This is to
82    /// be similar to the [RTCPeerConnection equivalent][buff]
83    ///
84    /// [buff]: https://developer.mozilla.org/en-US/docs/Web/API/RTCDataChannel/bufferedAmount
85    pub fn buffered_amount(&mut self) -> usize {
86        self.rtc.sctp.buffered_amount(self.sctp_stream_id)
87    }
88 
89    /// Set the threshold to emit an
90    /// [`Event::ChannelBufferedAmountLow`][crate::Event::ChannelBufferedAmountLow]
91    ///
92    /// Setting this on a closed or broken channel does not show an error. This is
93    /// be similar to the [RTCPeerConnection equivalent][buff]
94    ///
95    /// [buff]: https://developer.mozilla.org/en-US/docs/Web/API/RTCDataChannel/bufferedAmountLowThreshold
96    pub fn set_buffered_amount_low_threshold(&mut self, threshold: usize) {
97        self.rtc
98            .sctp
99            .set_buffered_amount_low_threshold(self.sctp_stream_id, threshold);
100    }
101 
102    /// Get the channel config.
103    ///
104    /// The config is not available in every case depending on whether the channel was
105    /// negotiated in- or out of band.
106    ///
107    /// # In-band negotiation (the usual case)
108    ///
109    /// For (regular) in-band negotiation (DCEP, Data Channel Establishment Protocol), this
110    /// returns `None` until the DCEP handshake completes. The config is guaranteed to be
111    /// available when [`Event::ChannelOpen`][crate::Event::ChannelOpen] is emitted.
112    ///
113    /// # Out-of-band negotiation
114    ///
115    /// Returns `None` when the remote side created the data channel connection without using
116    /// DCEP. This is called out-of-band negotiation, where the remote peer opens a stream
117    /// but doesn't send the channel configuration through the DCEP protocol messages.
118    ///
119    /// For locally created out-of-band channels, the config is always available since it
120    /// was provided during channel creation.
121    ///
122    /// In str0m, DCEP is disabled by setting the `negotiated` field to `Some(stream_id)` in
123    /// [`ChannelConfig`]. This corresponds to the `negotiated: true` property in the
124    /// browser's [`createDataChannel()`][n] dictionary.
125    ///
126    /// [n]: https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/createDataChannel#negotiated
127    pub fn config(&self) -> Option<&ChannelConfig> {
128        self.rtc.sctp.config(self.sctp_stream_id)
129    }
130}
131 
132impl fmt::Debug for ChannelData {
133    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
134        let mut ds = f.debug_struct("ChannelData");
135 
136        ds.field("id", &self.id);
137        ds.field("binary", &self.binary);
138 
139        let len = &self.data.len();
140        if self.binary {
141            ds.field("data", len);
142        } else {
143            match str::from_utf8(&self.data) {
144                Ok(s) => {
145                    const MAX_LINE_WIDTH: usize = 79;
146                    const REST_OF_LINE_WIDTH: usize =
147                        "ChannelData { id: ChannelId(0), binary: false, data: \"\" }".len();
148                    const TUPLE_WIDTH: usize = "(xxx, ..)".len();
149                    const DATA_WIDTH: usize = MAX_LINE_WIDTH - REST_OF_LINE_WIDTH;
150                    const PREFIX_WIDTH: usize = DATA_WIDTH - TUPLE_WIDTH;
151                    if s.is_ascii() {
152                        if len > &DATA_WIDTH {
153                            let trunc: String = s.chars().take(PREFIX_WIDTH).collect();
154                            ds.field("data", &format_args!("({}, \"{}\"..)", len, trunc));
155                        } else {
156                            ds.field("data", &s);
157                        }
158                    } else {
159                        ds.field("data", len);
160                    }
161                }
162                Err(e) => {
163                    ds.field("data", &format_args!("{:?}", (len, &e)));
164                }
165            }
166        }
167 
168        ds.finish()
169    }
170}
171 
172#[derive(Debug, Default)]
173pub(crate) struct ChannelHandler {
174    allocations: Vec<ChannelAllocation>,
175    next_channel_id: usize,
176    /// Stream IDs of closed channels whose reset handshake is still
177    /// outstanding, excluded from allocation until it completes.
178    closed_stream_ids: Vec<u16>,
179}
180 
181#[derive(Debug)]
182struct ChannelAllocation {
183    id: ChannelId,
184 
185    /// Stream id, when it is known. This might be delayed awaiting sctp initialization to
186    /// know if we are client or server, or awaiting the reset handshake of a previous
187    /// incarnation of a negotiated id.
188    sctp_stream_id: Option<u16>,
189 
190    /// The out-of-band negotiated stream id the user asked for. Promoted to
191    /// `sctp_stream_id` once reset which was holding it completes.
192    negotiated_stream_id: Option<u16>,
193 
194    /// Holds the config until it is used in handle_timeout.
195    config: Option<ChannelConfig>,
196}
197 
198impl ChannelHandler {
199    pub fn new_channel(&mut self, config: &ChannelConfig) -> ChannelId {
200        let id = self.next_channel_id();
201 
202        // Out-of-band negotiated means the user names the stream id
203        // instead of us allocating one. We record it as a request in
204        // and leave `sctp_stream_id` unset.
205        let negotiated_stream_id = config.negotiated;
206        if let Some(sctp_stream_id) = negotiated_stream_id {
207            let exists = self.allocations.iter().any(|a| {
208                a.sctp_stream_id == Some(sctp_stream_id)
209                    || a.negotiated_stream_id == Some(sctp_stream_id)
210            });
211            assert!(
212                !exists,
213                "sctp_stream_id ({}) exists already",
214                sctp_stream_id
215            );
216        }
217 
218        let alloc = ChannelAllocation {
219            id,
220            sctp_stream_id: None,
221            negotiated_stream_id,
222            // The config is none until we confirm we definitely want this channel.
223            config: None,
224        };
225 
226        debug!("Allocate channel id: {:?}", id);
227        self.allocations.push(alloc);
228 
229        id
230    }
231 
232    pub fn confirm(&mut self, id: ChannelId, config: ChannelConfig) {
233        let a = self
234            .allocations
235            .iter_mut()
236            .find(|a| a.id == id)
237            .expect("Entry for issued channel id");
238        a.config = Some(config);
239    }
240 
241    /// For translating sctp stream id to ChannelId. Any event out of sctp goes via this.
242    pub fn channel_id_by_stream_id(&self, sctp_stream_id: u16) -> Option<ChannelId> {
243        self.allocations
244            .iter()
245            .find(|a| a.sctp_stream_id == Some(sctp_stream_id))
246            .map(|a| a.id)
247    }
248 
249    /// Look up sctp stream id for channel id.
250    pub fn stream_id_by_channel_id(&self, id: ChannelId) -> Option<u16> {
251        self.allocations
252            .iter()
253            .find(|a| a.id == id)
254            .and_then(|a| a.sctp_stream_id)
255    }
256 
257    pub(crate) fn handle_timeout(&mut self, _now: Instant, sctp: &mut RtcSctp) {
258        if !sctp.is_inited() {
259            return;
260        }
261 
262        // Allocate sctp channel ids for ones that are missing.
263        self.do_allocations(sctp);
264 
265        // After do_allocations so we get a channel for any confirmed.
266        self.open_channels(sctp);
267    }
268 
269    /// Allocate next available `ChannelId`.
270    fn next_channel_id(&mut self) -> ChannelId {
271        let id = self.next_channel_id;
272        self.next_channel_id += 1;
273 
274        ChannelId(id)
275    }
276 
277    /// Whether a stream id is held back awaiting a reset completion.
278    fn is_held(&self, sctp_stream_id: u16) -> bool {
279        self.closed_stream_ids.contains(&sctp_stream_id)
280    }
281 
282    fn need_allocation(&self) -> bool {
283        self.allocations.iter().any(|a| {
284            a.sctp_stream_id.is_none()
285                // A negotiated id that is still held cannot be allocated yet.
286                && a.negotiated_stream_id.is_none_or(|want| !self.is_held(want))
287        })
288    }
289 
290    fn need_open(&self) -> bool {
291        self.allocations
292            .iter()
293            .any(|a| a.config.is_some() && a.sctp_stream_id.is_some())
294    }
295 
296    // Do automatic allocations of sctp stream id.
297    fn do_allocations(&mut self, sctp: &RtcSctp) {
298        if !self.need_allocation() {
299            return;
300        }
301 
302        // RFC 8831
303        // Unless otherwise defined or negotiated, the
304        // streams are picked based on the DTLS role (the client picks even
305        // stream identifiers, and the server picks odd stream identifiers).
306        let base = if sctp.is_client() { 0 } else { 1 };
307 
308        let mut taken: Vec<u16> = self
309            .allocations
310            .iter()
311            .filter_map(|a| a.sctp_stream_id)
312            .chain(self.closed_stream_ids.iter().copied())
313            .collect();
314 
315        for a in &mut self.allocations {
316            if a.sctp_stream_id.is_some() {
317                continue;
318            }
319 
320            // Out-of-band negotiated. The user picked the id, all we do is wait until
321            // no closed generation of it is still awaiting its reset handshake.
322            if let Some(want) = a.negotiated_stream_id {
323                if taken.contains(&want) {
324                    debug!("Negotiated stream id {} still held, retry later", want);
325                    continue;
326                }
327 
328                debug!("Associate negotiated stream id {:?} => {}", a.id, want);
329                a.sctp_stream_id = Some(want);
330                taken.push(want);
331                continue;
332            }
333 
334            // We need to allocate. Walk this parity's ids until we find a free one.
335            //
336            // A remote peer can fill every id (sctp-proto enforces no stream-count),
337            // the next allocation could overflow, instead fail the allocation gracefully.
338            let mut proposed = base;
339            while taken.contains(&proposed) {
340                match proposed.checked_add(2) {
341                    Some(next) => proposed = next,
342                    None => break, // id space for this parity is exhausted
343                }
344            }
345 
346            if taken.contains(&proposed) {
347                // Exhausted, the loop broke on overflow, leave the channel
348                // unallocated, it is retried on the next timeout.
349                warn!("SCTP stream id space exhausted, cannot allocate {:?}", a.id);
350                continue;
351            }
352 
353            // Found the next free.
354            debug!("Associate stream id {:?} => {}", a.id, proposed);
355            a.sctp_stream_id = Some(proposed);
356            taken.push(proposed);
357        }
358    }
359 
360    // Actually open channels.
361    fn open_channels(&mut self, sctp: &mut RtcSctp) {
362        for a in &mut self.allocations {
363            // The stream id must be known before the config is taken. A negotiated
364            // channel waiting out a previous incarnation's reset has no id yet, and
365            // consuming its config here would leave nothing to open it with once the
366            // id is released.
367            let Some(sctp_stream_id) = a.sctp_stream_id else {
368                continue;
369            };
370            let Some(config) = a.config.take() else {
371                continue;
372            };
373 
374            debug!("Open stream for: {:?}", a.id);
375            sctp.open_stream(sctp_stream_id, config);
376        }
377    }
378 
379    pub fn poll_timeout(&self, sctp: &RtcSctp) -> Option<Instant> {
380        if sctp.is_inited() && (self.need_allocation() || self.need_open()) {
381            Some(already_happened())
382        } else {
383            None
384        }
385    }
386 
387    pub fn ensure_channel_id_for(&mut self, sctp_stream_id: u16) {
388        let exists = self
389            .allocations
390            .iter()
391            .any(|a| a.sctp_stream_id == Some(sctp_stream_id));
392 
393        if !exists {
394            let id = self.next_channel_id();
395            let alloc = ChannelAllocation {
396                id,
397                sctp_stream_id: Some(sctp_stream_id),
398                negotiated_stream_id: None,
399                config: None,
400            };
401            self.allocations.push(alloc);
402        }
403    }
404 
405    // NB: Maybe this should still be &mut self or even `self` to prove singular ownership
406    pub fn close_channel(&self, id: ChannelId, sctp: &mut RtcSctp) {
407        if let Some(sctp_stream_id) = self
408            .allocations
409            .iter()
410            .find(|a| a.id == id)
411            .and_then(|s| s.sctp_stream_id)
412        {
413            sctp.close_stream(sctp_stream_id);
414        }
415    }
416 
417    /// The reset handshake for a held stream ID completed, release it for
418    /// reallocation immediately.
419    ///
420    /// Completions are reported per stream generation, so the same id can be held
421    /// more than once when resets overlap. Release a single generation, the id only
422    /// becomes reusable once every one of them has completed.
423    pub fn stream_reset_complete(&mut self, stream_id: u16) {
424        if let Some(pos) = self.closed_stream_ids.iter().position(|c| *c == stream_id) {
425            self.closed_stream_ids.remove(pos);
426        }
427    }
428 
429    /// The association is gone, so no reset can ever complete on it.
430    /// The held IDs belong to an association nobody will send on again.
431    pub fn association_lost(&mut self) {
432        self.closed_stream_ids.clear();
433    }
434 
435    /// Remove a closed channel.
436    pub fn remove_channel(&mut self, id: ChannelId, reset_pending: bool) {
437        let stream_id = self
438            .allocations
439            .iter()
440            .find(|a| a.id == id)
441            .and_then(|a| a.sctp_stream_id);
442 
443        if let (true, Some(stream_id)) = (reset_pending, stream_id) {
444            self.closed_stream_ids.push(stream_id);
445        }
446 
447        self.allocations.retain(|a| a.id != id)
448    }
449}
450 
451#[cfg(test)]
452mod tests {
453    use super::*;
454 
455    #[test]
456    fn channel_id_allocation() {
457        let mut handler = ChannelHandler::default();
458 
459        // allocate first channel, get unique id
460        assert_eq!(handler.new_channel(&Default::default()), ChannelId(0));
461 
462        // allocate second channel, get unique id
463        assert_eq!(handler.new_channel(&Default::default()), ChannelId(1));
464 
465        // free channel 0, allocate two more channels and verify that the
466        // new channels have unique IDs.
467        handler.remove_channel(ChannelId(0), true);
468        assert_eq!(handler.new_channel(&Default::default()), ChannelId(2));
469        assert_eq!(handler.new_channel(&Default::default()), ChannelId(3));
470    }
471 
472    #[test]
473    fn stream_id_held_until_reset_complete() {
474        let mut handler = ChannelHandler::default();
475 
476        // Simulate two channels with known stream IDs (as if do_allocations ran
477        // for a client: even IDs 0, 2).
478        let id0 = handler.new_channel(&Default::default());
479        let _id1 = handler.new_channel(&Default::default());
480        // Manually set stream IDs as do_allocations would.
481        handler.allocations[0].sctp_stream_id = Some(0);
482        handler.allocations[1].sctp_stream_id = Some(2);
483 
484        // Close channel 0 (stream ID 0) with its reset outstanding. It should be held.
485        handler.remove_channel(id0, true);
486        assert_eq!(handler.closed_stream_ids, vec![0]);
487 
488        // Build the taken list as do_allocations does — stream 0 is held.
489        let _id2 = handler.new_channel(&Default::default());
490        let taken: Vec<u16> = handler
491            .allocations
492            .iter()
493            .filter_map(|a| a.sctp_stream_id)
494            .chain(handler.closed_stream_ids.iter().copied())
495            .collect();
496        assert!(taken.contains(&0), "stream 0 should be held");
497        assert!(taken.contains(&2), "stream 2 should be active");
498 
499        // The completion signal is the only thing that releases it.
500        handler.stream_reset_complete(0);
501        assert!(handler.closed_stream_ids.is_empty());
502 
503        let taken_after: Vec<u16> = handler
504            .allocations
505            .iter()
506            .filter_map(|a| a.sctp_stream_id)
507            .chain(handler.closed_stream_ids.iter().copied())
508            .collect();
509        assert!(
510            !taken_after.contains(&0),
511            "stream 0 should be available after reset completion"
512        );
513    }
514 
515    #[test]
516    fn negotiated_stream_id_cannot_overlap_held_generation() {
517        let sctp = RtcSctp::new(1200);
518        let mut handler = ChannelHandler::default();
519        let config = ChannelConfig {
520            negotiated: Some(4),
521            ..Default::default()
522        };
523 
524        // The old generation was granted the id it asked for.
525        let old = handler.new_channel(&config);
526        handler.do_allocations(&sctp);
527        assert_eq!(handler.stream_id_by_channel_id(old), Some(4));
528 
529        handler.remove_channel(old, true);
530        assert_eq!(handler.closed_stream_ids, vec![4]);
531 
532        // A replacement asking for the same id may be declared, but must stall
533        // without the id until the old reset completes.
534        let _replacement = handler.new_channel(&config);
535        handler.do_allocations(&sctp);
536 
537        let overlaps_held_generation = handler.allocations.iter().any(|allocation| {
538            handler
539                .closed_stream_ids
540                .iter()
541                .any(|closed| allocation.sctp_stream_id == Some(*closed))
542        });
543 
544        assert!(
545            !overlaps_held_generation,
546            "a negotiated ID must remain unavailable until its old reset completes"
547        );
548    }
549 
550    #[test]
551    fn negotiated_stream_keeps_config_while_waiting_for_reset() {
552        let mut sctp = RtcSctp::new(1200);
553        let mut handler = ChannelHandler::default();
554        let config = ChannelConfig {
555            label: "replacement".into(),
556            negotiated: Some(4),
557            ..Default::default()
558        };
559 
560        handler.closed_stream_ids.push(4);
561        let replacement = handler.new_channel(&config);
562        handler.confirm(replacement, config);
563 
564        handler.do_allocations(&sctp);
565        handler.open_channels(&mut sctp);
566 
567        let allocation = handler
568            .allocations
569            .iter()
570            .find(|allocation| allocation.id == replacement)
571            .unwrap();
572        assert!(
573            allocation.config.is_some(),
574            "a negotiated channel must retain its config while its stream id is held"
575        );
576 
577        handler.stream_reset_complete(4);
578        handler.do_allocations(&sctp);
579        handler.open_channels(&mut sctp);
580 
581        assert_eq!(
582            sctp.config(4).map(|config| config.label.as_str()),
583            Some("replacement"),
584            "the negotiated channel must open once the old reset completes"
585        );
586    }
587 
588    #[test]
589    fn negotiated_stream_waiting_for_reset_does_not_spin_timeout() {
590        let now = Instant::now();
591        let mut sctp = RtcSctp::new(1200);
592        sctp.init(true, now, None, None).unwrap();
593 
594        let mut handler = ChannelHandler::default();
595        let config = ChannelConfig {
596            negotiated: Some(4),
597            ..Default::default()
598        };
599 
600        handler.closed_stream_ids.push(4);
601        let replacement = handler.new_channel(&config);
602        handler.confirm(replacement, config);
603        handler.handle_timeout(now, &mut sctp);
604 
605        assert_eq!(
606            handler.poll_timeout(&sctp),
607            None,
608            "a channel that cannot open until network input arrives must not request an immediate timeout"
609        );
610    }
611 
612    #[test]
613    fn one_reset_completion_releases_only_one_stream_generation() {
614        let mut handler = ChannelHandler::default();
615 
616        // sctp-proto 0.10.3 reports reset completion per stream generation.
617        // Two outstanding generations of the same stream id therefore need two
618        // completion events before the id is reusable.
619        handler.closed_stream_ids.extend([4, 4]);
620 
621        handler.stream_reset_complete(4);
622 
623        assert_eq!(
624            handler.closed_stream_ids,
625            vec![4],
626            "the first completion must not release a newer pending generation"
627        );
628    }
629 
630    #[test]
631    fn stream_id_remains_held_without_reset_completion() {
632        let sctp = RtcSctp::new(1200);
633        let mut handler = ChannelHandler::default();
634 
635        let id = handler.new_channel(&Default::default());
636        handler.allocations[0].sctp_stream_id = Some(0);
637        handler.remove_channel(id, true);
638 
639        // There is no timer that can release the id.
640        for _ in 0..100 {
641            handler.do_allocations(&sctp);
642        }
643 
644        assert_eq!(
645            handler.closed_stream_ids,
646            vec![0],
647            "elapsed time alone cannot make an SCTP stream ID safe to reuse"
648        );
649    }
650 
651    #[test]
652    fn stream_id_not_held_when_no_reset_is_pending() {
653        let mut handler = ChannelHandler::default();
654 
655        let id0 = handler.new_channel(&Default::default());
656        handler.allocations[0].sctp_stream_id = Some(0);
657 
658        // The stream never made it into the association, so no reset was started and
659        // no completion will ever arrive. Holding the id would leak it forever.
660        handler.remove_channel(id0, false);
661        assert!(handler.closed_stream_ids.is_empty());
662    }
663 
664    #[test]
665    fn association_lost_releases_held_stream_ids() {
666        let mut handler = ChannelHandler::default();
667 
668        let id0 = handler.new_channel(&Default::default());
669        handler.allocations[0].sctp_stream_id = Some(0);
670        handler.remove_channel(id0, true);
671        assert_eq!(handler.closed_stream_ids, vec![0]);
672 
673        // No reset can complete on a dead association, nothing holds the id back.
674        handler.association_lost();
675        assert!(handler.closed_stream_ids.is_empty());
676    }
677 
678    /// A remote peer decides which SCTP stream ids exist on the association, and
679    /// `ensure_channel_id_for()` gives every one of them an allocation. A peer that
680    /// opens every id of our parity therefore leaves `do_allocations()` with no free
681    /// id to hand out.
682    ///
683    /// The allocator walks `proposed += 2` with no upper bound. Past 65535 that is a
684    /// debug-build overflow panic, and in release it wraps back to `base` and spins
685    /// forever inside `do_allocations()`, hanging the whole event loop.
686    ///
687    /// Failing to allocate is fine, str0m retries on the next timeout. Panicking or
688    /// hanging is not.
689    #[test]
690    fn stream_id_space_exhausted_by_remote() {
691        let sctp = RtcSctp::new(1200);
692        let base: u16 = if sctp.is_client() { 0 } else { 1 };
693 
694        let mut handler = ChannelHandler::default();
695 
696        // What the peer did: claim every stream id of our parity. Going through
697        // `ensure_channel_id_for()` for each is the realistic route but is quadratic,
698        // so seed the equivalent state directly.
699        handler.closed_stream_ids = (base..=u16::MAX).step_by(2).collect();
700        assert_eq!(handler.closed_stream_ids.len(), 32768);
701 
702        // Now the local application asks for a data channel.
703        let id = handler.new_channel(&Default::default());
704 
705        handler.do_allocations(&sctp);
706 
707        let alloc = handler
708            .allocations
709            .iter()
710            .find(|a| a.id == id)
711            .expect("the allocation entry to still be there");
712 
713        assert!(
714            alloc.sctp_stream_id.is_none(),
715            "no id can be allocated when the peer holds the whole parity"
716        );
717    }
718}