moqtap_client/draft16/fetch.rs
1/// Fetch lifecycle states.
2///
3/// A fetch settles two things and settles them independently: the publisher's
4/// answer, which arrives on the control stream, and the response stream that
5/// carries the objects. Section 9.17: "A publisher MAY send Objects in response
6/// to a FETCH before the FETCH_OK message is sent, but the FETCH_OK MUST NOT be
7/// sent until the End Location is known."
8///
9/// Section 9.16.3 states the other direction of the same freedom: "The FETCH_OK
10/// or REQUEST_ERROR can come at any time relative to object delivery." So the
11/// fetch is over when both have settled, in whichever order they settle.
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum FetchState {
14 /// Initial state before any FETCH message is sent.
15 Idle,
16 /// FETCH has been sent; the answer is owed and the response stream
17 /// has not ended.
18 Pending,
19 /// FETCH_OK received; data is being received on the stream.
20 Receiving,
21 /// The response stream has ended and the answer is still owed.
22 Unanswered,
23 /// Fetch has ended (error, cancel, FIN, or reset).
24 Done,
25}
26
27/// Errors that can occur during fetch state transitions.
28#[derive(Debug, thiserror::Error, PartialEq, Eq)]
29pub enum FetchError {
30 /// An event was received that is not valid for the current state.
31 #[error("invalid transition from {from:?} on event {event}")]
32 InvalidTransition {
33 /// The state the machine was in when the invalid event arrived.
34 from: FetchState,
35 /// The name of the event that was rejected.
36 event: String,
37 },
38}
39
40/// Pure state machine for a MoQT fetch request.
41/// Transitions: Idle -> Pending -> Receiving -> Done when the answer comes
42/// first, and Idle -> Pending -> Unanswered -> Done when the response stream
43/// ends first.
44pub struct FetchStateMachine {
45 state: FetchState,
46}
47
48impl Default for FetchStateMachine {
49 fn default() -> Self {
50 Self::new()
51 }
52}
53
54impl FetchStateMachine {
55 /// Creates a new state machine in the [`FetchState::Idle`] state.
56 pub fn new() -> Self {
57 Self { state: FetchState::Idle }
58 }
59
60 /// Returns the current state of the fetch request.
61 pub fn state(&self) -> FetchState {
62 self.state
63 }
64
65 /// Idle -> Pending (FETCH sent).
66 pub fn on_fetch_sent(&mut self) -> Result<(), FetchError> {
67 match self.state {
68 FetchState::Idle => {
69 self.state = FetchState::Pending;
70 Ok(())
71 }
72 _ => Err(FetchError::InvalidTransition {
73 from: self.state,
74 event: "on_fetch_sent".to_string(),
75 }),
76 }
77 }
78
79 /// Pending -> Receiving, Unanswered -> Done (FETCH_OK received).
80 ///
81 /// From Unanswered the objects have already been delivered, so the FETCH_OK
82 /// describing them is the last thing the fetch was waiting for.
83 pub fn on_fetch_ok(&mut self) -> Result<(), FetchError> {
84 match self.state {
85 FetchState::Pending => {
86 self.state = FetchState::Receiving;
87 Ok(())
88 }
89 FetchState::Unanswered => {
90 self.state = FetchState::Done;
91 Ok(())
92 }
93 _ => Err(FetchError::InvalidTransition {
94 from: self.state,
95 event: "on_fetch_ok".to_string(),
96 }),
97 }
98 }
99
100 /// Pending | Unanswered -> Done (REQUEST_ERROR received).
101 pub fn on_fetch_error(&mut self) -> Result<(), FetchError> {
102 match self.state {
103 FetchState::Pending | FetchState::Unanswered => {
104 self.state = FetchState::Done;
105 Ok(())
106 }
107 _ => Err(FetchError::InvalidTransition {
108 from: self.state,
109 event: "on_fetch_error".to_string(),
110 }),
111 }
112 }
113
114 /// Pending | Receiving | Unanswered -> Done (FETCH_CANCEL sent).
115 pub fn on_fetch_cancel(&mut self) -> Result<(), FetchError> {
116 match self.state {
117 FetchState::Pending | FetchState::Receiving | FetchState::Unanswered => {
118 self.state = FetchState::Done;
119 Ok(())
120 }
121 _ => Err(FetchError::InvalidTransition {
122 from: self.state,
123 event: "on_fetch_cancel".to_string(),
124 }),
125 }
126 }
127
128 /// Receiving -> Done, Pending -> Unanswered (stream FIN received).
129 ///
130 /// A fetch that has already ended ignores the close of its response stream,
131 /// whichever of QUIC's two endings it is, and both of the ways a fetch ends
132 /// early leave one behind. Section 9.18: the publisher of a cancelled fetch
133 /// "SHOULD promptly close the unidirectional stream, even if it is in the
134 /// middle of delivering an object". Section 9.16.3: a relay whose upstream
135 /// FETCH failed "sends a REQUEST_ERROR and can reset the unidirectional
136 /// stream", and may "wait until the cached objects have been delivered
137 /// before resetting the stream".
138 pub fn on_stream_fin(&mut self) -> Result<(), FetchError> {
139 match self.state {
140 FetchState::Receiving => {
141 self.state = FetchState::Done;
142 Ok(())
143 }
144 FetchState::Pending => {
145 self.state = FetchState::Unanswered;
146 Ok(())
147 }
148 FetchState::Done => Ok(()),
149 _ => Err(FetchError::InvalidTransition {
150 from: self.state,
151 event: "on_stream_fin".to_string(),
152 }),
153 }
154 }
155
156 /// Receiving -> Done, Pending -> Unanswered (stream RESET received).
157 ///
158 /// The same tolerance of a late close as
159 /// [`FetchStateMachine::on_stream_fin`], for the same reason.
160 pub fn on_stream_reset(&mut self) -> Result<(), FetchError> {
161 match self.state {
162 FetchState::Receiving => {
163 self.state = FetchState::Done;
164 Ok(())
165 }
166 FetchState::Pending => {
167 self.state = FetchState::Unanswered;
168 Ok(())
169 }
170 FetchState::Done => Ok(()),
171 _ => Err(FetchError::InvalidTransition {
172 from: self.state,
173 event: "on_stream_reset".to_string(),
174 }),
175 }
176 }
177}
178
179/// The same transitions, named for the end that serves the fetch.
180///
181/// A fetch this endpoint answers passes through the states in the same order
182/// as one it makes, with every message going the other way: the FETCH arrives
183/// instead of leaving, the answer leaves instead of arriving. Sharing the
184/// transitions and not the names is what lets a refusal say which event was
185/// refused, rather than naming the mirror image of it.
186impl FetchStateMachine {
187 /// Idle -> Pending (FETCH received from the peer).
188 pub fn on_fetch_received(&mut self) -> Result<(), FetchError> {
189 self.on_fetch_sent().map_err(|_| FetchError::InvalidTransition {
190 from: self.state(),
191 event: "on_fetch_received".to_string(),
192 })
193 }
194
195 /// Pending -> Receiving, Unanswered -> Done (FETCH_OK sent).
196 ///
197 /// The state is named for the requester's view; for the end answering, the
198 /// same node means the objects are being served rather than received. It is
199 /// the same node in the graph, with the same edges, so it keeps its name.
200 pub fn on_fetch_ok_sent(&mut self) -> Result<(), FetchError> {
201 self.on_fetch_ok().map_err(|_| FetchError::InvalidTransition {
202 from: self.state(),
203 event: "on_fetch_ok_sent".to_string(),
204 })
205 }
206
207 /// Pending | Unanswered -> Done (REQUEST_ERROR sent).
208 pub fn on_fetch_error_sent(&mut self) -> Result<(), FetchError> {
209 self.on_fetch_error().map_err(|_| FetchError::InvalidTransition {
210 from: self.state(),
211 event: "on_fetch_error_sent".to_string(),
212 })
213 }
214
215 /// Pending | Receiving | Unanswered -> Done (FETCH_CANCEL received).
216 pub fn on_fetch_cancel_received(&mut self) -> Result<(), FetchError> {
217 self.on_fetch_cancel().map_err(|_| FetchError::InvalidTransition {
218 from: self.state(),
219 event: "on_fetch_cancel_received".to_string(),
220 })
221 }
222
223 /// Receiving -> Done, Pending -> Unanswered (this endpoint finished the
224 /// fetch data stream).
225 pub fn on_stream_fin_sent(&mut self) -> Result<(), FetchError> {
226 self.on_stream_fin().map_err(|_| FetchError::InvalidTransition {
227 from: self.state(),
228 event: "on_stream_fin_sent".to_string(),
229 })
230 }
231}