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

syncular_client/
round.rs
