syncular_client/
round.rs

1//! Owned round inputs and network results. SQLite never leaves its owner.
2use std::collections::BTreeMap;
3
4use ssp2::model::{Frame, MediaType, Message, MsgKind, SubStatus};
5use ssp2::{decode_message, encode_message};
6
7use crate::client::RequestMeta;
8use crate::{
9    BlobUploadGrant, ProgressObserver, ProgressPhase, SegmentRequest, SyncOutcome, Transport,
10    TransportError,
11};
12
13pub(crate) fn now_ms(fixed: Option<i64>) -> i64 {
14    fixed.unwrap_or_else(crate::wall_clock_ms)
15}
16
17pub(crate) struct PendingUpload {
18    pub id: String,
19    pub bytes: Result<Vec<u8>, TransportError>,
20    pub media_type: Option<String>,
21}
22
23pub(crate) fn upload_one(
24    transport: &mut dyn Transport,
25    id: &str,
26    bytes: &[u8],
27    media_type: Option<&str>,
28    fixed_now: Option<i64>,
29) -> Result<(), TransportError> {
30    match transport.blob_upload_grant(id, bytes.len() as u64, media_type)? {
31        BlobUploadGrant::Present => return Ok(()),
32        BlobUploadGrant::Url {
33            url,
34            url_expires_at_ms,
35        } => {
36            if url_expires_at_ms.is_none_or(|expiry| expiry > now_ms(fixed_now))
37                && transport.blob_put_url(&url, bytes, media_type).is_ok()
38            {
39                return Ok(());
40            }
41        }
42        BlobUploadGrant::None => {}
43    }
44    transport.blob_upload(id, bytes, media_type)
45}
46
47/// Immutable request and blob bodies captured by the mutable client owner.
48/// Send this value to an I/O executor; it contains no SQLite connection.
49pub struct PreparedSyncRound {
50    pub(crate) id: uuid::Uuid,
51    pub(crate) started_at_ms: i64,
52    pub(crate) meta: RequestMeta,
53    pub(crate) message: Message,
54    pub(crate) realtime: bool,
55    pub(crate) uploads: std::collections::VecDeque<PendingUpload>,
56    pub(crate) scopes: BTreeMap<String, String>,
57    #[cfg(feature = "bench-internals")]
58    pub(crate) benchmark_phases: crate::bench::Recorder,
59    pub(crate) progress: ProgressObserver,
60    pub(crate) fixed_now: Option<i64>,
61    /// Absolute whole-round network deadline, anchored from the transport's
62    /// `round_deadline` budget on the round's first network call and carried
63    /// across `Continue`s. `exchange` scopes it onto the transport for the
64    /// duration of each network call.
65    pub(crate) round_deadline_at: Option<std::time::Instant>,
66}
67
68/// Network results returned to the same mutable owner that prepared the round.
69pub struct CompletedSyncRound {
70    pub(crate) prepared: PreparedSyncRound,
71    pub(crate) exchange: ExchangeResult,
72}
73
74pub(crate) enum ExchangeResult {
75    Upload {
76        id: String,
77        result: Result<(), TransportError>,
78    },
79    Reply {
80        response: Result<Message, TransportError>,
81        transport_failed: bool,
82        downloads: DownloadResults,
83    },
84}
85
86/// A completed upload returns to the owner to persist its pin removal before
87/// any metadata push. A completed response returns its durable apply outcome.
88pub enum AppliedSyncRound {
89    Continue(PreparedSyncRound),
90    Complete {
91        outcome: SyncOutcome,
92        controls: Vec<String>,
93        /// The same continuation predicate used by `sync_until_idle`.
94        more: bool,
95        /// A bootstrap page advanced, renewing an unbounded idle run's budget.
96        bootstrap_advanced: bool,
97    },
98}
99
100impl PreparedSyncRound {
101    /// Scope the round deadline for one network call. The round anchors its
102    /// deadline on its first network call, so local work before that call does
103    /// not consume the network budget; the anchor is absolute from then on, and
104    /// a continuation reuses the carried anchor instead of refreshing it.
105    fn scope_round_deadline(
106        &mut self,
107        transport: &mut dyn Transport,
108    ) -> Result<(), TransportError> {
109        if self.round_deadline_at.is_none() {
110            if let Some(budget) = transport.round_deadline() {
111                // A wasm32-unknown-unknown host has no `Instant`, so there is
112                // nothing to anchor the budget to. Report that instead of
113                // panicking inside `Instant::now()`.
114                #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
115                {
116                    let _ = budget;
117                    return Err(TransportError::new(
118                        "sync.invalid_request",
119                        "transport round deadline is unsupported on wasm32-unknown-unknown",
120                    ));
121                }
122                #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
123                match std::time::Instant::now().checked_add(budget) {
124                    Some(deadline) => self.round_deadline_at = Some(deadline),
125                    // Reject an unrepresentable host budget before network I/O.
126                    None => {
127                        return Err(TransportError::new(
128                            "sync.invalid_request",
129                            "transport round deadline is out of range",
130                        ))
131                    }
132                }
133            }
134        }
135        transport.set_round_deadline(self.round_deadline_at);
136        Ok(())
137    }
138
139    /// Run network I/O without accessing client state. Download errors stay at
140    /// their descriptor's position so apply retains the same completed prefix.
141    pub fn exchange(mut self, transport: &mut dyn Transport) -> CompletedSyncRound {
142        if let Some(upload) = self.uploads.pop_front() {
143            // A staged body that failed to read locally performs no network
144            // I/O, so it cannot be the round's first network call and must not
145            // anchor the round deadline either.
146            let result = match upload.bytes {
147                Ok(bytes) => self.scope_round_deadline(transport).and_then(|()| {
148                    upload_one(
149                        transport,
150                        &upload.id,
151                        &bytes,
152                        upload.media_type.as_deref(),
153                        self.fixed_now,
154                    )
155                }),
156                Err(error) => Err(error),
157            };
158            transport.set_round_deadline(None);
159            return CompletedSyncRound {
160                prepared: self,
161                exchange: ExchangeResult::Upload {
162                    id: upload.id,
163                    result,
164                },
165            };
166        }
167        let mut transport_failed = false;
168        let response = (|| {
169            #[cfg(feature = "bench-internals")]
170            let encode_phase = self
171                .benchmark_phases
172                .start(crate::bench::Phase::RequestEncode);
173            let request = encode_message(&self.message);
174            #[cfg(feature = "bench-internals")]
175            drop(encode_phase);
176            // The round's first network call anchors the deadline here, after
177            // the encode; a continuation's encode runs against the already
178            // anchored deadline like the rest of the round (§1.8).
179            self.scope_round_deadline(transport)?;
180            let round = if self.realtime {
181                transport.realtime_sync(&request)
182            } else {
183                transport.sync(&request)
184            };
185            transport_failed = round.is_err();
186            let bytes = round?;
187            #[cfg(feature = "bench-internals")]
188            let _decode_phase = self
189                .benchmark_phases
190                .start(crate::bench::Phase::ResponseDecode);
191            let response = decode_message(&bytes)
192                .map_err(|e| TransportError::new(e.code.as_str(), e.detail))?;
193            if response.msg_kind != MsgKind::Response {
194                return Err(TransportError::new(
195                    "sync.invalid_request",
196                    "expected a response message",
197                ));
198            }
199            Ok(response)
200        })();
201        let mut downloads = DownloadResults::default();
202        if let Ok(response) = &response {
203            // A reset/floor/invalid header cannot authorize any segment fetch.
204            if matches!(response.frames.first(), Some(Frame::RespHeader { required_schema_version: None, reset_required: Some(false), log_epoch: Some(epoch), .. }) if self.message.frames.first().is_some_and(|f| matches!(f, Frame::ReqHeader { log_epoch: Some(request_epoch), .. } if request_epoch == epoch)))
205            {
206                let mut scope: Option<(&str, &str)> = None;
207                for frame in &response.frames {
208                    match frame {
209                        Frame::SubStart {
210                            id,
211                            status: SubStatus::Active,
212                            ..
213                        } => {
214                            scope = self
215                                .scopes
216                                .get(id)
217                                .map(|scopes| (id.as_str(), scopes.as_str()));
218                        }
219                        Frame::SubStart { .. } | Frame::SubEnd { .. } => scope = None,
220                        Frame::Error { .. } => break,
221                        Frame::SegmentRef {
222                            segment_id,
223                            table,
224                            media_type,
225                            url,
226                            url_expires_at_ms,
227                            byte_length,
228                            row_count,
229                            ..
230                        } => {
231                            let Some((id, scopes)) = scope else { continue };
232                            let advertised = match media_type {
233                                MediaType::Rows => self.meta.accept & 3 != 0,
234                                MediaType::Sqlite => self.meta.accept & 4 != 0,
235                            };
236                            if !advertised || (url.is_some() && self.meta.accept & 8 == 0) {
237                                break;
238                            }
239                            let started_at = now_ms(self.fixed_now);
240                            let key = match url {
241                                Some(url) => DownloadKey::Url(url.clone()),
242                                None => DownloadKey::Segment(
243                                    segment_id.clone(),
244                                    table.clone(),
245                                    scopes.to_owned(),
246                                ),
247                            };
248                            self.progress.update(|p| {
249                                p.phase = ProgressPhase::Download;
250                                p.subscription_id = Some(id.to_owned());
251                                p.table = Some(table.clone());
252                                p.segment_id = Some(segment_id.clone());
253                                p.bytes_received = 0;
254                                p.bytes_total = u64::try_from(*byte_length).ok();
255                                p.rows_processed = 0;
256                                p.rows_total = u64::try_from(*row_count).ok();
257                            });
258                            let mut on_progress =
259                                |bytes| self.progress.update(|p| p.bytes_received = bytes);
260                            let bytes = if let Some(url) = url {
261                                if url_expires_at_ms.is_some_and(|expiry| expiry <= started_at) {
262                                    Err(TransportError::new(
263                                        "sync.segment_expired",
264                                        "signed segment URL expired before fetch",
265                                    ))
266                                } else {
267                                    transport.fetch_url(url, &mut on_progress)
268                                }
269                            } else {
270                                transport.download_segment(
271                                    &SegmentRequest {
272                                        segment_id: segment_id.clone(),
273                                        table: table.clone(),
274                                        requested_scopes_json: scopes.to_owned(),
275                                    },
276                                    &mut on_progress,
277                                )
278                            };
279                            let failed = bytes.is_err();
280                            downloads.entries.push_back((key, started_at, bytes));
281                            if failed {
282                                break;
283                            }
284                        }
285                        _ => {}
286                    }
287                }
288            }
289        }
290        transport.set_round_deadline(None);
291        CompletedSyncRound {
292            prepared: self,
293            exchange: ExchangeResult::Reply {
294                response,
295                downloads,
296                transport_failed,
297            },
298        }
299    }
300}
301
302#[derive(PartialEq)]
303enum DownloadKey {
304    Url(String),
305    Segment(String, String, String),
306}
307
308#[derive(Default)]
309pub(crate) struct DownloadResults {
310    entries: std::collections::VecDeque<(DownloadKey, i64, Result<Vec<u8>, TransportError>)>,
311    pub controls: Vec<String>,
312}
313
314impl DownloadResults {
315    fn take(&mut self, key: DownloadKey) -> Result<Vec<u8>, TransportError> {
316        let position = self
317            .entries
318            .iter()
319            .position(|(candidate, ..)| *candidate == key);
320        position
321            .and_then(|i| self.entries.remove(i))
322            .map(|(_, _, bytes)| bytes)
323            .unwrap_or_else(|| {
324                Err(TransportError::new(
325                    "client.round_download_missing",
326                    "round apply requested an uncaptured segment",
327                ))
328            })
329    }
330}
331
332impl Transport for DownloadResults {
333    fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
334        Err(TransportError::new(
335            "client.round_network_on_owner",
336            "network requests must execute outside the client owner",
337        ))
338    }
339    fn realtime_sync(&mut self, request: &[u8]) -> Result<Vec<u8>, TransportError> {
340        self.sync(request)
341    }
342    fn download_segment(
343        &mut self,
344        request: &SegmentRequest,
345        _: &mut dyn FnMut(u64),
346    ) -> Result<Vec<u8>, TransportError> {
347        self.take(DownloadKey::Segment(
348            request.segment_id.clone(),
349            request.table.clone(),
350            request.requested_scopes_json.clone(),
351        ))
352    }
353    fn fetch_url(&mut self, url: &str, _: &mut dyn FnMut(u64)) -> Result<Vec<u8>, TransportError> {
354        self.take(DownloadKey::Url(url.to_owned()))
355    }
356    fn segment_fetch_started_at(&self, url: &str) -> Option<i64> {
357        self.entries
358            .iter()
359            .find_map(|(key, time, _)| (*key == DownloadKey::Url(url.to_owned())).then_some(*time))
360    }
361    fn realtime_connect(&mut self) -> Result<(), TransportError> {
362        Err(TransportError::new(
363            "client.round_network_on_owner",
364            "network connections must execute outside the client owner",
365        ))
366    }
367    fn realtime_send(&mut self, text: &str) -> Result<(), TransportError> {
368        self.controls.push(text.to_owned());
369        Ok(())
370    }
371    fn realtime_close(&mut self) -> Result<(), TransportError> {
372        Ok(())
373    }
374}
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379    use crate::{BlobUploadGrant, SegmentRequest, TransportError};
380
381    /// Every upload call fails with the same typed timeout, modelling a round
382    /// whose deadline was spent before the signed upload. The fallback to the
383    /// direct endpoint must not replace the typed code with a generic one.
384    struct SpentDeadline;
385
386    impl Transport for SpentDeadline {
387        fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
388            unreachable!()
389        }
390        fn realtime_sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
391            unreachable!()
392        }
393        fn download_segment(
394            &mut self,
395            _: &SegmentRequest,
396            _: &mut dyn FnMut(u64),
397        ) -> Result<Vec<u8>, TransportError> {
398            unreachable!()
399        }
400        fn blob_upload_grant(
401            &mut self,
402            _: &str,
403            _: u64,
404            _: Option<&str>,
405        ) -> Result<BlobUploadGrant, TransportError> {
406            Ok(BlobUploadGrant::Url {
407                url: "https://storage.example/signed".into(),
408                url_expires_at_ms: None,
409            })
410        }
411        fn blob_put_url(
412            &mut self,
413            _: &str,
414            _: &[u8],
415            _: Option<&str>,
416        ) -> Result<(), TransportError> {
417            Err(TransportError::new(
418                "transport.timeout",
419                "signed upload timed out",
420            ))
421        }
422        fn blob_upload(
423            &mut self,
424            _: &str,
425            _: &[u8],
426            _: Option<&str>,
427        ) -> Result<(), TransportError> {
428            Err(TransportError::new(
429                "transport.timeout",
430                "direct upload timed out",
431            ))
432        }
433        fn realtime_connect(&mut self) -> Result<(), TransportError> {
434            Ok(())
435        }
436        fn realtime_send(&mut self, _: &str) -> Result<(), TransportError> {
437            Ok(())
438        }
439        fn realtime_close(&mut self) -> Result<(), TransportError> {
440            Ok(())
441        }
442    }
443
444    #[test]
445    fn a_spent_round_deadline_survives_the_signed_upload_fallback() {
446        let mut transport = SpentDeadline;
447        let error = upload_one(&mut transport, "sha256:test", &[0u8; 4], None, None).unwrap_err();
448        assert_eq!(error.code, "transport.timeout");
449    }
450}
syncular_client/round.rs

syncular_client/
round.rs
