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}
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    /// Run network I/O without accessing client state. Download errors stay at
102    /// their descriptor's position so apply retains the same completed prefix.
103    pub fn exchange(mut self, transport: &mut dyn Transport) -> CompletedSyncRound {
104        if let Some(upload) = self.uploads.pop_front() {
105            let result = upload.bytes.and_then(|bytes| {
106                upload_one(
107                    transport,
108                    &upload.id,
109                    &bytes,
110                    upload.media_type.as_deref(),
111                    self.fixed_now,
112                )
113            });
114            return CompletedSyncRound {
115                prepared: self,
116                exchange: ExchangeResult::Upload {
117                    id: upload.id,
118                    result,
119                },
120            };
121        }
122        let mut transport_failed = false;
123        let response = (|| {
124            #[cfg(feature = "bench-internals")]
125            let encode_phase = self
126                .benchmark_phases
127                .start(crate::bench::Phase::RequestEncode);
128            let request = encode_message(&self.message);
129            #[cfg(feature = "bench-internals")]
130            drop(encode_phase);
131            let round = if self.realtime {
132                transport.realtime_sync(&request)
133            } else {
134                transport.sync(&request)
135            };
136            transport_failed = round.is_err();
137            let bytes = round?;
138            #[cfg(feature = "bench-internals")]
139            let _decode_phase = self
140                .benchmark_phases
141                .start(crate::bench::Phase::ResponseDecode);
142            let response = decode_message(&bytes)
143                .map_err(|e| TransportError::new(e.code.as_str(), e.detail))?;
144            if response.msg_kind != MsgKind::Response {
145                return Err(TransportError::new(
146                    "sync.invalid_request",
147                    "expected a response message",
148                ));
149            }
150            Ok(response)
151        })();
152        let mut downloads = DownloadResults::default();
153        if let Ok(response) = &response {
154            // A reset/floor/invalid header cannot authorize any segment fetch.
155            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)))
156            {
157                let mut scope: Option<(&str, &str)> = None;
158                for frame in &response.frames {
159                    match frame {
160                        Frame::SubStart {
161                            id,
162                            status: SubStatus::Active,
163                            ..
164                        } => {
165                            scope = self
166                                .scopes
167                                .get(id)
168                                .map(|scopes| (id.as_str(), scopes.as_str()));
169                        }
170                        Frame::SubStart { .. } | Frame::SubEnd { .. } => scope = None,
171                        Frame::Error { .. } => break,
172                        Frame::SegmentRef {
173                            segment_id,
174                            table,
175                            media_type,
176                            url,
177                            url_expires_at_ms,
178                            byte_length,
179                            row_count,
180                            ..
181                        } => {
182                            let Some((id, scopes)) = scope else { continue };
183                            let advertised = match media_type {
184                                MediaType::Rows => self.meta.accept & 3 != 0,
185                                MediaType::Sqlite => self.meta.accept & 4 != 0,
186                            };
187                            if !advertised || (url.is_some() && self.meta.accept & 8 == 0) {
188                                break;
189                            }
190                            let started_at = now_ms(self.fixed_now);
191                            let key = match url {
192                                Some(url) => DownloadKey::Url(url.clone()),
193                                None => DownloadKey::Segment(
194                                    segment_id.clone(),
195                                    table.clone(),
196                                    scopes.to_owned(),
197                                ),
198                            };
199                            self.progress.update(|p| {
200                                p.phase = ProgressPhase::Download;
201                                p.subscription_id = Some(id.to_owned());
202                                p.table = Some(table.clone());
203                                p.segment_id = Some(segment_id.clone());
204                                p.bytes_received = 0;
205                                p.bytes_total = u64::try_from(*byte_length).ok();
206                                p.rows_processed = 0;
207                                p.rows_total = u64::try_from(*row_count).ok();
208                            });
209                            let mut on_progress =
210                                |bytes| self.progress.update(|p| p.bytes_received = bytes);
211                            let bytes = if let Some(url) = url {
212                                if url_expires_at_ms.is_some_and(|expiry| expiry <= started_at) {
213                                    Err(TransportError::new(
214                                        "sync.segment_expired",
215                                        "signed segment URL expired before fetch",
216                                    ))
217                                } else {
218                                    transport.fetch_url(url, &mut on_progress)
219                                }
220                            } else {
221                                transport.download_segment(
222                                    &SegmentRequest {
223                                        segment_id: segment_id.clone(),
224                                        table: table.clone(),
225                                        requested_scopes_json: scopes.to_owned(),
226                                    },
227                                    &mut on_progress,
228                                )
229                            };
230                            let failed = bytes.is_err();
231                            downloads.entries.push_back((key, started_at, bytes));
232                            if failed {
233                                break;
234                            }
235                        }
236                        _ => {}
237                    }
238                }
239            }
240        }
241        CompletedSyncRound {
242            prepared: self,
243            exchange: ExchangeResult::Reply {
244                response,
245                downloads,
246                transport_failed,
247            },
248        }
249    }
250}
251
252#[derive(PartialEq)]
253enum DownloadKey {
254    Url(String),
255    Segment(String, String, String),
256}
257
258#[derive(Default)]
259pub(crate) struct DownloadResults {
260    entries: std::collections::VecDeque<(DownloadKey, i64, Result<Vec<u8>, TransportError>)>,
261    pub controls: Vec<String>,
262}
263
264impl DownloadResults {
265    fn take(&mut self, key: DownloadKey) -> Result<Vec<u8>, TransportError> {
266        let position = self
267            .entries
268            .iter()
269            .position(|(candidate, ..)| *candidate == key);
270        position
271            .and_then(|i| self.entries.remove(i))
272            .map(|(_, _, bytes)| bytes)
273            .unwrap_or_else(|| {
274                Err(TransportError::new(
275                    "client.round_download_missing",
276                    "round apply requested an uncaptured segment",
277                ))
278            })
279    }
280}
281
282impl Transport for DownloadResults {
283    fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
284        Err(TransportError::new(
285            "client.round_network_on_owner",
286            "network requests must execute outside the client owner",
287        ))
288    }
289    fn realtime_sync(&mut self, request: &[u8]) -> Result<Vec<u8>, TransportError> {
290        self.sync(request)
291    }
292    fn download_segment(
293        &mut self,
294        request: &SegmentRequest,
295        _: &mut dyn FnMut(u64),
296    ) -> Result<Vec<u8>, TransportError> {
297        self.take(DownloadKey::Segment(
298            request.segment_id.clone(),
299            request.table.clone(),
300            request.requested_scopes_json.clone(),
301        ))
302    }
303    fn fetch_url(&mut self, url: &str, _: &mut dyn FnMut(u64)) -> Result<Vec<u8>, TransportError> {
304        self.take(DownloadKey::Url(url.to_owned()))
305    }
306    fn segment_fetch_started_at(&self, url: &str) -> Option<i64> {
307        self.entries
308            .iter()
309            .find_map(|(key, time, _)| (*key == DownloadKey::Url(url.to_owned())).then_some(*time))
310    }
311    fn realtime_connect(&mut self) -> Result<(), TransportError> {
312        Err(TransportError::new(
313            "client.round_network_on_owner",
314            "network connections must execute outside the client owner",
315        ))
316    }
317    fn realtime_send(&mut self, text: &str) -> Result<(), TransportError> {
318        self.controls.push(text.to_owned());
319        Ok(())
320    }
321    fn realtime_close(&mut self) -> Result<(), TransportError> {
322        Ok(())
323    }
324}
syncular_client/round.rs

syncular_client/
round.rs
