moqtap_client/draft16/session/request_id.rs
1use std::collections::BTreeSet;
2
3use moqtap_codec::varint::VarInt;
4
5/// Which end of the session this endpoint is, which fixes the parity of the
6/// request IDs it may allocate.
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum Role {
9 /// Client: even request IDs, 0, 2, 4, ...
10 Client,
11 /// Server: odd request IDs, 1, 3, 5, ...
12 Server,
13}
14
15/// Errors from request ID allocation or validation.
16#[derive(Debug, thiserror::Error, PartialEq, Eq)]
17pub enum RequestIdError {
18 /// The request ID is not below the current MAX_REQUEST_ID.
19 #[error("request ID {0} exceeds max {1}")]
20 ExceedsMax(u64, u64),
21 /// The request ID has the wrong parity for the endpoint that sent it.
22 #[error("request ID {0} has wrong parity for {1:?}")]
23 WrongParity(u64, Role),
24 /// MAX_REQUEST_ID must only increase; it decreased.
25 #[error("max request ID can only increase: was {0}, got {1}")]
26 Decreased(u64, u64),
27 /// No request IDs are available (max is 0 or exhausted).
28 #[error("no request IDs available (blocked)")]
29 Blocked,
30 /// A new request carried an ID the peer has already used.
31 #[error("request ID {0} was already used by the peer")]
32 Duplicate(u64),
33}
34
35/// How many skipped peer IDs are remembered as still unused. A skipped ID that
36/// falls out of this window is treated as used.
37const MAX_SKIPPED_PEER_IDS: usize = 1024;
38
39/// Allocates this endpoint's request IDs and checks the parity of the peer's.
40///
41/// Draft-16 Section 9.1: "The client's Request ID starts at 0 and are even and
42/// the server's Request ID starts at 1 and are odd. The Request ID increments
43/// by 2 with each FETCH, SUBSCRIBE, REQUEST_UPDATE, SUBSCRIBE_NAMESPACE,
44/// PUBLISH, PUBLISH_NAMESPACE or TRACK_STATUS request." The step of two is
45/// what keeps the two endpoints' ID spaces disjoint, so it is not a spacing
46/// convention an implementation may tighten: one that steps by one starts
47/// issuing the IDs reserved for its peer on its second request, and the same
48/// section makes that a session close with INVALID_REQUEST_ID.
49///
50/// The ceiling is exclusive. Section 9.5 describes the MAX_REQUEST_ID
51/// message's Max Request ID field as "The new Maximum Request ID for the
52/// session plus 1", and closes the session with TOO_MANY_REQUESTS on a request
53/// ID "equal to or larger than this". Section 9.3.1.3 gives the setup
54/// parameter of the same name a default of 0 and reads that as "the peer MUST
55/// NOT send requests", which only holds if a ceiling of 0 forbids the ID 0 as
56/// well.
57pub struct RequestIdAllocator {
58 role: Role,
59 next_id: u64,
60 max_id: u64,
61 peer_next_id: u64,
62 peer_skipped: BTreeSet<u64>,
63}
64
65impl RequestIdAllocator {
66 /// Create an allocator for the given role, starting at ID 0 or 1 with a
67 /// ceiling of 0 - blocked until the peer raises it.
68 pub fn new(role: Role) -> Self {
69 let next_id = match role {
70 Role::Client => 0,
71 Role::Server => 1,
72 };
73 // The peer's sequence is the other half of the space, and starts at
74 // the other end for the same reason this one does.
75 let peer_next_id = match role {
76 Role::Client => 1,
77 Role::Server => 0,
78 };
79 Self { role, next_id, max_id: 0, peer_next_id, peer_skipped: BTreeSet::new() }
80 }
81
82 /// The role this allocator allocates for.
83 pub fn role(&self) -> Role {
84 self.role
85 }
86
87 /// Allocate the next request ID.
88 ///
89 /// # Errors
90 ///
91 /// [`RequestIdError::Blocked`] when the next ID would reach the ceiling.
92 pub fn allocate(&mut self) -> Result<VarInt, RequestIdError> {
93 if self.is_blocked() {
94 return Err(RequestIdError::Blocked);
95 }
96 let id = VarInt::from_u64(self.next_id).unwrap();
97 self.next_id += 2;
98 Ok(id)
99 }
100
101 /// Update the maximum allowed request ID (can only increase).
102 ///
103 /// # Errors
104 ///
105 /// [`RequestIdError::Decreased`] if the new value is not strictly greater
106 /// than the current one, which Section 9.5 calls a protocol
107 /// violation.
108 pub fn update_max(&mut self, new_max: u64) -> Result<(), RequestIdError> {
109 if new_max <= self.max_id {
110 return Err(RequestIdError::Decreased(self.max_id, new_max));
111 }
112 self.max_id = new_max;
113 Ok(())
114 }
115
116 /// Check the parity of a request ID the peer put on the wire.
117 ///
118 /// The parity checked here is the **peer's**, the opposite of this
119 /// allocator's own, so a client accepts only odd IDs.
120 ///
121 /// This deliberately does not check the ID against a ceiling. The ceiling
122 /// that applies to a peer's request ID is the MAX_REQUEST_ID *this*
123 /// endpoint advertised, which the allocator does not hold - `max_id` here
124 /// is the budget the peer granted us, a different number that may be
125 /// larger or smaller. The endpoint owns that check.
126 ///
127 /// Parity is the half that needs no history. The duplicate half is
128 /// [`Self::record_peer_id`], which does.
129 ///
130 /// # Errors
131 ///
132 /// [`RequestIdError::WrongParity`] carrying the ID and the **peer's**
133 /// role, so the message names the endpoint that broke the rule rather than
134 /// the one that caught it.
135 pub fn validate_peer_id(&self, id: u64) -> Result<(), RequestIdError> {
136 let peer_role = match self.role {
137 Role::Client => Role::Server,
138 Role::Server => Role::Client,
139 };
140 let peer_sends_even = peer_role == Role::Client;
141 if id.is_multiple_of(2) != peer_sends_even {
142 return Err(RequestIdError::WrongParity(id, peer_role));
143 }
144 Ok(())
145 }
146
147 /// Whether the next allocation would reach the ceiling.
148 ///
149 /// A ceiling of 0 needs no special case: the client's first ID is 0 and
150 /// the server's is 1, and neither is below 0.
151 pub fn is_blocked(&self) -> bool {
152 self.next_id >= self.max_id
153 }
154
155 /// The ID after the highest one the peer has used: its first ID until it
156 /// sends a request, then the highest seen plus 2.
157 pub fn peer_next_id(&self) -> u64 {
158 self.peer_next_id
159 }
160
161 /// Take the Request ID of a new request the peer sent, refusing one it has
162 /// already used.
163 ///
164 /// The sequence is not enforced. Draft-16 opens SUBSCRIBE_NAMESPACE on its
165 /// own stream, so requests can arrive out of order across streams; the
166 /// editors' guidance for draft-16 (moq-wg/moq-transport#1459) is to check
167 /// parity, the ceiling and duplicates only. IDs jumped over are remembered
168 /// as unused, up to a fixed number of the highest.
169 ///
170 /// Only a **new** request is passed here. A message that names a request
171 /// already open reuses its ID on purpose.
172 ///
173 /// # Errors
174 ///
175 /// [`RequestIdError::WrongParity`] for an ID from this endpoint's half,
176 /// [`RequestIdError::Duplicate`] for one already used or no longer
177 /// remembered as skipped.
178 pub fn record_peer_id(&mut self, id: u64) -> Result<(), RequestIdError> {
179 self.validate_peer_id(id)?;
180 if id < self.peer_next_id {
181 return if self.peer_skipped.remove(&id) {
182 Ok(())
183 } else {
184 Err(RequestIdError::Duplicate(id))
185 };
186 }
187 let window = 2 * MAX_SKIPPED_PEER_IDS as u64;
188 let mut skipped = id.saturating_sub(window).max(self.peer_next_id);
189 while skipped < id {
190 self.peer_skipped.insert(skipped);
191 skipped += 2;
192 }
193 while self.peer_skipped.len() > MAX_SKIPPED_PEER_IDS {
194 self.peer_skipped.pop_first();
195 }
196 self.peer_next_id = id.saturating_add(2);
197 Ok(())
198 }
199
200 /// Get the current maximum request ID.
201 pub fn max_id(&self) -> u64 {
202 self.max_id
203 }
204}
205
206#[cfg(test)]
207mod tests {
208 use super::*;
209
210 #[test]
211 fn a_skip_past_the_window_forgets_the_oldest_skipped_ids() {
212 let mut ids = RequestIdAllocator::new(Role::Server);
213 let far = 2 * (MAX_SKIPPED_PEER_IDS as u64 + 10);
214 ids.record_peer_id(far).unwrap();
215 assert_eq!(ids.peer_skipped.len(), MAX_SKIPPED_PEER_IDS);
216 assert_eq!(ids.record_peer_id(0), Err(RequestIdError::Duplicate(0)));
217 ids.record_peer_id(far - 2).unwrap();
218 assert_eq!(ids.peer_next_id(), far + 2);
219 }
220}