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, SegmentFetch, SegmentRequest, SyncOutcome,
10    Transport, 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
23/// One network operation of a prepared round.
24///
25/// A host that cannot block (a browser worker with `fetch`, a single-threaded
26/// runtime) drives a round with [`PreparedSyncRound::next_operation`],
27/// performs the operation with its own I/O, and returns the result with
28/// [`PreparedSyncRound::submit`]. A host with a blocking [`Transport`] calls
29/// [`PreparedSyncRound::exchange`], which runs the same operations in order.
30#[derive(Debug)]
31pub enum RoundOperation<'a> {
32    /// `POST <mount>/blobs/{blob_id}/upload-grant` (§5.9.3).
33    UploadGrant {
34        blob_id: &'a str,
35        byte_length: u64,
36        media_type: Option<&'a str>,
37    },
38    /// `PUT` the presigned upload url the grant returned. A failure falls back
39    /// to [`RoundOperation::Upload`] with the same body.
40    UploadPut {
41        url: &'a str,
42        bytes: &'a [u8],
43        media_type: Option<&'a str>,
44    },
45    /// `PUT <mount>/blobs/{blob_id}` through the direct endpoint.
46    Upload {
47        blob_id: &'a str,
48        bytes: &'a [u8],
49        media_type: Option<&'a str>,
50    },
51    /// The combined push+pull message. `realtime` selects the socket path
52    /// (§8.7, [`Transport::realtime_sync`]) instead of `POST /sync` (§1.5).
53    Sync { request: &'a [u8], realtime: bool },
54    /// Every eligible segment body of the response, in wire order (§5.4 /
55    /// §5.5). Fetch them in parallel if the host can; the core consumes the
56    /// results in order and stops at the first failure.
57    Downloads(&'a [SegmentFetch]),
58}
59
60/// The host's answer to a [`RoundOperation`].
61#[derive(Debug)]
62pub enum RoundReply {
63    /// Answers [`RoundOperation::UploadGrant`] with the grant endpoint's body.
64    UploadGrant(Result<BlobUploadGrant, TransportError>),
65    /// Answers [`RoundOperation::UploadPut`] or [`RoundOperation::Upload`]. A
66    /// failed PUT makes `submit` ask for `Upload` next.
67    Upload(Result<(), TransportError>),
68    /// Answers [`RoundOperation::Sync`] with the assembled response bytes.
69    Sync(Result<Vec<u8>, TransportError>),
70    /// Answers [`RoundOperation::Downloads`] with one result per fetch, in
71    /// order.
72    Downloads(Vec<Result<Vec<u8>, TransportError>>),
73}
74
75/// The phase of a host-driven round. Every variant except `Downloads` names
76/// one network operation; `Downloads` carries the captured response while the
77/// host fetches its segment bodies.
78pub(crate) enum RoundPhase {
79    UploadGrant,
80    UploadPut {
81        url: String,
82    },
83    UploadDirect,
84    Sync,
85    Downloads {
86        planned: Vec<PlannedFetch>,
87        fetches: Vec<SegmentFetch>,
88        response: Message,
89        batch_started_at: i64,
90    },
91}
92
93/// Immutable request and blob bodies captured by the mutable client owner.
94/// Send this value to an I/O executor; it contains no SQLite connection.
95pub struct PreparedSyncRound {
96    pub(crate) id: uuid::Uuid,
97    pub(crate) started_at_ms: i64,
98    pub(crate) meta: RequestMeta,
99    pub(crate) message: Message,
100    pub(crate) realtime: bool,
101    pub(crate) uploads: std::collections::VecDeque<PendingUpload>,
102    pub(crate) scopes: BTreeMap<String, String>,
103    #[cfg(feature = "bench-internals")]
104    pub(crate) benchmark_phases: crate::bench::Recorder,
105    pub(crate) progress: ProgressObserver,
106    pub(crate) fixed_now: Option<i64>,
107    /// Absolute whole-round network deadline, anchored from the transport's
108    /// `round_deadline` budget on the round's first network call and carried
109    /// across `Continue`s. `exchange` scopes it onto the transport for the
110    /// duration of each network call.
111    pub(crate) round_deadline_at: Option<std::time::Instant>,
112    /// The encoded round message, filled on the first request that needs it.
113    pub(crate) encoded: Option<Vec<u8>>,
114    pub(crate) phase: RoundPhase,
115    /// The phase that follows the completed upload exchange; consumed by the
116    /// next drive of a round whose apply returned `Continue`.
117    pub(crate) continuation: Option<RoundPhase>,
118    /// Set when the current exchange has nothing left to fetch. `finish`
119    /// returns it; `apply_sync_round` reads it.
120    pub(crate) completed: Option<ExchangeResult>,
121}
122
123/// Network results returned to the same mutable owner that prepared the round.
124pub struct CompletedSyncRound {
125    pub(crate) prepared: PreparedSyncRound,
126    pub(crate) exchange: ExchangeResult,
127}
128
129pub(crate) enum ExchangeResult {
130    Upload {
131        id: String,
132        result: Result<(), TransportError>,
133    },
134    Reply {
135        response: Result<Message, TransportError>,
136        transport_failed: bool,
137        downloads: DownloadResults,
138    },
139}
140
141/// A completed upload returns to the owner to persist its pin removal before
142/// any metadata push. A completed response returns its durable apply outcome.
143// The prepared round is the continuation and stays by value so a host keeps
144// one type for both outcomes.
145#[allow(clippy::large_enum_variant)]
146pub enum AppliedSyncRound {
147    Continue(PreparedSyncRound),
148    Complete {
149        outcome: SyncOutcome,
150        controls: Vec<String>,
151        /// The same continuation predicate used by `sync_until_idle`.
152        more: bool,
153        /// A bootstrap page advanced, renewing an unbounded idle run's budget.
154        bootstrap_advanced: bool,
155    },
156}
157
158impl PreparedSyncRound {
159    /// Scope the round deadline for one network call. The round anchors its
160    /// deadline on its first network call, so local work before that call does
161    /// not consume the network budget; the anchor is absolute from then on, and
162    /// a continuation reuses the carried anchor instead of refreshing it.
163    fn scope_round_deadline(
164        &mut self,
165        transport: &mut dyn Transport,
166    ) -> Result<(), TransportError> {
167        if self.round_deadline_at.is_none() {
168            if let Some(budget) = transport.round_deadline() {
169                // A wasm32-unknown-unknown host has no `Instant`, so there is
170                // nothing to anchor the budget to. Report that instead of
171                // panicking inside `Instant::now()`.
172                #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
173                {
174                    let _ = budget;
175                    return Err(TransportError::new(
176                        "sync.invalid_request",
177                        "transport round deadline is unsupported on wasm32-unknown-unknown",
178                    ));
179                }
180                #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
181                match std::time::Instant::now().checked_add(budget) {
182                    Some(deadline) => self.round_deadline_at = Some(deadline),
183                    // Reject an unrepresentable host budget before network I/O.
184                    None => {
185                        return Err(TransportError::new(
186                            "sync.invalid_request",
187                            "transport round deadline is out of range",
188                        ))
189                    }
190                }
191            }
192        }
193        transport.set_round_deadline(self.round_deadline_at);
194        Ok(())
195    }
196
197    /// Run network I/O without accessing client state. Download errors stay at
198    /// their descriptor's position so apply retains the same completed prefix.
199    ///
200    /// This drives the same [`RoundOperation`] sequence that a non-blocking
201    /// host drives itself, so both paths share every decision the client
202    /// makes about uploads, segment fetches, and deadline state.
203    pub fn exchange(mut self, transport: &mut dyn Transport) -> CompletedSyncRound {
204        let mut scoped = false;
205        while self.completed.is_none() {
206            let reply = {
207                let Some(operation) = self.next_operation_scoped(transport, &mut scoped) else {
208                    break;
209                };
210                match operation {
211                    RoundOperation::UploadGrant {
212                        blob_id,
213                        byte_length,
214                        media_type,
215                    } => RoundReply::UploadGrant(transport.blob_upload_grant(
216                        blob_id,
217                        byte_length,
218                        media_type,
219                    )),
220                    RoundOperation::UploadPut {
221                        url,
222                        bytes,
223                        media_type,
224                    } => RoundReply::Upload(transport.blob_put_url(url, bytes, media_type)),
225                    RoundOperation::Upload {
226                        blob_id,
227                        bytes,
228                        media_type,
229                    } => RoundReply::Upload(transport.blob_upload(blob_id, bytes, media_type)),
230                    RoundOperation::Sync { request, realtime } => RoundReply::Sync(if realtime {
231                        transport.realtime_sync(request)
232                    } else {
233                        transport.sync(request)
234                    }),
235                    RoundOperation::Downloads(fetches) => {
236                        RoundReply::Downloads(transport.download_segments(fetches, &mut |_, _| {}))
237                    }
238                }
239            };
240            self.submit(reply)
241                .expect("the exchange answers its own operation");
242        }
243        transport.set_round_deadline(None);
244        self.finish().expect("the exchange completed the round")
245    }
246
247    /// The encoded round message (§1.5). Uploads listed by
248    /// [`RoundOperation::UploadGrant`] precede it; a host that drives the round
249    /// with `exchange` never needs this call.
250    pub fn request_bytes(&self) -> Vec<u8> {
251        self.encoded
252            .clone()
253            .unwrap_or_else(|| encode_message(&self.message))
254    }
255
256    /// The next network operation a host must perform, or `None` once every
257    /// operation of the current exchange is answered. Call
258    /// [`PreparedSyncRound::submit`] with its result, then repeat.
259    ///
260    /// After an upload operation completes, the round is complete for this
261    /// exchange: call [`PreparedSyncRound::finish`] and apply it, so the owner
262    /// persists the pin removal before the metadata request. The applied
263    /// round's `Continue` returns the prepared round for the next exchange.
264    pub fn next_operation(&mut self) -> Option<RoundOperation<'_>> {
265        let mut scoped = false;
266        self.next_operation_impl(None, &mut scoped)
267    }
268
269    /// Hand the result of the last [`PreparedSyncRound::next_operation`] back.
270    /// A reply of the wrong kind fails with `client.invalid_round_reply` and
271    /// leaves the round where it was.
272    pub fn submit(&mut self, reply: RoundReply) -> Result<(), TransportError> {
273        if self.completed.is_some() {
274            return Err(TransportError::new(
275                "client.invalid_round_reply",
276                "the exchange already has its reply",
277            ));
278        }
279        let phase = std::mem::replace(&mut self.phase, RoundPhase::Sync);
280        match (phase, reply) {
281            (RoundPhase::UploadGrant, RoundReply::UploadGrant(Err(error))) => {
282                let upload = self
283                    .uploads
284                    .pop_front()
285                    .expect("an upload phase owns an upload");
286                self.completed = Some(ExchangeResult::Upload {
287                    id: upload.id,
288                    result: Err(error),
289                });
290            }
291            (RoundPhase::UploadGrant, RoundReply::UploadGrant(Ok(grant))) => match grant {
292                BlobUploadGrant::Present => self.upload_complete(),
293                BlobUploadGrant::Url {
294                    url,
295                    url_expires_at_ms,
296                } => {
297                    // §5.9.3: an expiry the host clock already passed skips the
298                    // PUT; the direct endpoint still carries the body.
299                    if url_expires_at_ms.is_none_or(|expiry| expiry > now_ms(self.fixed_now)) {
300                        self.phase = RoundPhase::UploadPut { url };
301                    } else {
302                        self.phase = RoundPhase::UploadDirect;
303                    }
304                }
305                BlobUploadGrant::None => self.phase = RoundPhase::UploadDirect,
306            },
307            (RoundPhase::UploadPut { .. }, RoundReply::Upload(Ok(()))) => self.upload_complete(),
308            // A failed presigned PUT falls back to the direct endpoint with the
309            // same body, exactly like `upload_one`.
310            (RoundPhase::UploadPut { .. }, RoundReply::Upload(Err(_))) => {
311                self.phase = RoundPhase::UploadDirect;
312            }
313            (RoundPhase::UploadDirect, RoundReply::Upload(Ok(()))) => self.upload_complete(),
314            (RoundPhase::UploadDirect, RoundReply::Upload(Err(error))) => {
315                let upload = self
316                    .uploads
317                    .pop_front()
318                    .expect("an upload phase owns an upload");
319                self.completed = Some(ExchangeResult::Upload {
320                    id: upload.id,
321                    result: Err(error),
322                });
323            }
324            (RoundPhase::Sync, RoundReply::Sync(reply)) => {
325                let transport_failed = reply.is_err();
326                let response = reply.and_then(|bytes| {
327                    #[cfg(feature = "bench-internals")]
328                    let _decode_phase = self
329                        .benchmark_phases
330                        .start(crate::bench::Phase::ResponseDecode);
331                    let response = decode_message(&bytes)
332                        .map_err(|e| TransportError::new(e.code.as_str(), e.detail))?;
333                    if response.msg_kind != MsgKind::Response {
334                        return Err(TransportError::new(
335                            "sync.invalid_request",
336                            "expected a response message",
337                        ));
338                    }
339                    Ok(response)
340                });
341                let Ok(response) = response else {
342                    self.completed = Some(ExchangeResult::Reply {
343                        response,
344                        transport_failed,
345                        downloads: DownloadResults::default(),
346                    });
347                    return Ok(());
348                };
349                let mut planned: Vec<PlannedFetch> = Vec::new();
350                // A reset/floor/invalid header cannot authorize any segment
351                // fetch.
352                if matches!(response.frames.first(), Some(Frame::RespHeader { required_schema_version: None, reset_required: false, log_epoch: epoch, .. }) if self.message.frames.first().is_some_and(|f| matches!(f, Frame::ReqHeader { log_epoch: Some(request_epoch), .. } if request_epoch == epoch)))
353                {
354                    let mut scope: Option<(&str, &str)> = None;
355                    // §5.4 / §5.5: collect every eligible descriptor first so
356                    // the host can overlap the fetches. One batch start is the
357                    // fetch start for the expiry rule.
358                    let batch_started_at = now_ms(self.fixed_now);
359                    for frame in &response.frames {
360                        match frame {
361                            Frame::SubStart {
362                                id,
363                                status: SubStatus::Active,
364                                ..
365                            } => {
366                                scope = self
367                                    .scopes
368                                    .get(id)
369                                    .map(|scopes| (id.as_str(), scopes.as_str()));
370                            }
371                            Frame::SubStart { .. } | Frame::SubEnd { .. } => scope = None,
372                            Frame::Error { .. } => break,
373                            Frame::SegmentRef {
374                                segment_id,
375                                table,
376                                media_type,
377                                url,
378                                url_expires_at_ms,
379                                byte_length,
380                                row_count,
381                                ..
382                            } => {
383                                let Some((id, scopes)) = scope else { continue };
384                                let advertised = match media_type {
385                                    MediaType::Rows => self.meta.accept & 3 != 0,
386                                    MediaType::Sqlite => self.meta.accept & 4 != 0,
387                                };
388                                if !advertised || (url.is_some() && self.meta.accept & 8 == 0) {
389                                    break;
390                                }
391                                // A negative byteLength admits no body (§5.4).
392                                let cap = u64::try_from(*byte_length).unwrap_or(0);
393                                let expired = url.is_some()
394                                    && url_expires_at_ms
395                                        .is_some_and(|expiry| expiry <= batch_started_at);
396                                let (key, fetch) = match url {
397                                    Some(url) => (
398                                        DownloadKey::Url(url.clone()),
399                                        if expired {
400                                            None
401                                        } else {
402                                            Some(SegmentFetch::Url {
403                                                url: url.clone(),
404                                                byte_length: cap,
405                                            })
406                                        },
407                                    ),
408                                    None => (
409                                        DownloadKey::Segment(
410                                            segment_id.clone(),
411                                            table.clone(),
412                                            scopes.to_owned(),
413                                        ),
414                                        Some(SegmentFetch::Direct(SegmentRequest {
415                                            segment_id: segment_id.clone(),
416                                            table: table.clone(),
417                                            requested_scopes_json: scopes.to_owned(),
418                                            byte_length: cap,
419                                        })),
420                                    ),
421                                };
422                                planned.push(PlannedFetch {
423                                    key,
424                                    subscription_id: id.to_owned(),
425                                    table: table.clone(),
426                                    segment_id: segment_id.clone(),
427                                    bytes_total: u64::try_from(*byte_length).ok(),
428                                    rows_total: u64::try_from(*row_count).ok(),
429                                    fetch,
430                                });
431                                if expired {
432                                    break;
433                                }
434                            }
435                            _ => {}
436                        }
437                    }
438                    if !planned.is_empty() {
439                        let progress = self.progress.clone();
440                        let on_progress = |index: usize, bytes: u64| {
441                            let Some(planned_fetch) = planned.get(index) else {
442                                return;
443                            };
444                            progress.update(|p| {
445                                p.phase = ProgressPhase::Download;
446                                p.subscription_id = Some(planned_fetch.subscription_id.clone());
447                                p.table = Some(planned_fetch.table.clone());
448                                p.segment_id = Some(planned_fetch.segment_id.clone());
449                                p.bytes_received = bytes;
450                                p.bytes_total = planned_fetch.bytes_total;
451                                p.rows_processed = 0;
452                                p.rows_total = planned_fetch.rows_total;
453                            });
454                        };
455                        // Announce the download phase before the first byte lands.
456                        on_progress(0, 0);
457                        let fetches: Vec<SegmentFetch> = planned
458                            .iter()
459                            .filter_map(|planned_fetch| planned_fetch.fetch.clone())
460                            .collect();
461                        self.phase = RoundPhase::Downloads {
462                            planned,
463                            fetches,
464                            response,
465                            batch_started_at,
466                        };
467                        return Ok(());
468                    }
469                }
470                self.completed = Some(ExchangeResult::Reply {
471                    response: Ok(response),
472                    transport_failed,
473                    downloads: DownloadResults::default(),
474                });
475            }
476            (
477                RoundPhase::Downloads {
478                    planned,
479                    batch_started_at,
480                    response,
481                    ..
482                },
483                RoundReply::Downloads(results),
484            ) => {
485                let mut downloads = DownloadResults::default();
486                let progress = self.progress.clone();
487                let mut results = results.into_iter();
488                for planned_fetch in planned {
489                    let result = match planned_fetch.fetch {
490                        None => Err(TransportError::new(
491                            "sync.segment_expired",
492                            "signed segment URL expired before fetch",
493                        )),
494                        // A shorter result list means the batch stopped after an
495                        // earlier error, and apply aborts there.
496                        Some(_) => match results.next() {
497                            Some(result) => result,
498                            None => break,
499                        },
500                    };
501                    if let Ok(bytes) = &result {
502                        progress.update(|p| {
503                            p.phase = ProgressPhase::Download;
504                            p.subscription_id = Some(planned_fetch.subscription_id.clone());
505                            p.table = Some(planned_fetch.table.clone());
506                            p.segment_id = Some(planned_fetch.segment_id.clone());
507                            p.bytes_received = bytes.len() as u64;
508                            p.bytes_total = planned_fetch.bytes_total;
509                            p.rows_processed = 0;
510                            p.rows_total = planned_fetch.rows_total;
511                        });
512                    }
513                    let failed = result.is_err();
514                    downloads
515                        .entries
516                        .push_back((planned_fetch.key, batch_started_at, result));
517                    if failed {
518                        break;
519                    }
520                }
521                self.completed = Some(ExchangeResult::Reply {
522                    response: Ok(response),
523                    transport_failed: false,
524                    downloads,
525                });
526            }
527            (phase, _) => {
528                self.phase = phase;
529                return Err(TransportError::new(
530                    "client.invalid_round_reply",
531                    "the reply does not answer the pending operation",
532                ));
533            }
534        }
535        Ok(())
536    }
537
538    /// The completed round once every operation is answered, ready for
539    /// `SyncClient::apply_sync_round`.
540    pub fn finish(mut self) -> Option<CompletedSyncRound> {
541        let exchange = self.completed.take()?;
542        Some(CompletedSyncRound {
543            prepared: self,
544            exchange,
545        })
546    }
547
548    fn upload_complete(&mut self) {
549        let upload = self
550            .uploads
551            .pop_front()
552            .expect("an upload phase owns an upload");
553        self.completed = Some(ExchangeResult::Upload {
554            id: upload.id,
555            result: Ok(()),
556        });
557        self.continuation = Some(if self.uploads.is_empty() {
558            RoundPhase::Sync
559        } else {
560            RoundPhase::UploadGrant
561        });
562    }
563
564    fn ensure_encoded(&mut self) {
565        if self.encoded.is_none() {
566            #[cfg(feature = "bench-internals")]
567            let _phase = self
568                .benchmark_phases
569                .start(crate::bench::Phase::RequestEncode);
570            self.encoded = Some(encode_message(&self.message));
571        }
572    }
573
574    fn next_operation_scoped<'a>(
575        &'a mut self,
576        transport: &mut dyn Transport,
577        scoped: &mut bool,
578    ) -> Option<RoundOperation<'a>> {
579        self.next_operation_impl(Some(transport), scoped)
580    }
581
582    fn next_operation_impl<'a>(
583        &'a mut self,
584        mut scope: Option<&mut dyn Transport>,
585        scoped: &mut bool,
586    ) -> Option<RoundOperation<'a>> {
587        if self.completed.is_some() {
588            return None;
589        }
590        // The previous exchange ended with an upload; the owner has applied it.
591        if let Some(next) = self.continuation.take() {
592            self.phase = next;
593        }
594        if matches!(self.phase, RoundPhase::UploadGrant) && self.uploads.is_empty() {
595            self.phase = RoundPhase::Sync;
596        }
597        // A staged body that failed to read locally performs no network I/O,
598        // so it cannot be the round's first network call and must not anchor
599        // the round deadline either.
600        if matches!(self.phase, RoundPhase::UploadGrant)
601            && self
602                .uploads
603                .front()
604                .is_some_and(|upload| upload.bytes.is_err())
605        {
606            let upload = self
607                .uploads
608                .pop_front()
609                .expect("the failed upload is staged");
610            self.completed = Some(ExchangeResult::Upload {
611                id: upload.id,
612                result: Err(upload.bytes.unwrap_err()),
613            });
614            return None;
615        }
616        if matches!(self.phase, RoundPhase::Sync) {
617            self.ensure_encoded();
618        }
619        let needs_network = !matches!(self.phase, RoundPhase::Downloads { .. });
620        // The round's first network call anchors the deadline after the
621        // encode; a continuation reuses the carried anchor instead of
622        // refreshing it.
623        if needs_network && !*scoped {
624            *scoped = true;
625            if let Some(transport) = scope.as_mut() {
626                if let Err(error) = self.scope_round_deadline(&mut **transport) {
627                    if matches!(
628                        self.phase,
629                        RoundPhase::UploadGrant
630                            | RoundPhase::UploadPut { .. }
631                            | RoundPhase::UploadDirect
632                    ) {
633                        let upload = self
634                            .uploads
635                            .pop_front()
636                            .expect("an upload phase owns an upload");
637                        self.completed = Some(ExchangeResult::Upload {
638                            id: upload.id,
639                            result: Err(error),
640                        });
641                    } else {
642                        self.completed = Some(ExchangeResult::Reply {
643                            response: Err(error),
644                            transport_failed: false,
645                            downloads: DownloadResults::default(),
646                        });
647                    }
648                    return None;
649                }
650            }
651        }
652        match &self.phase {
653            RoundPhase::UploadGrant => {
654                let upload = self
655                    .uploads
656                    .front()
657                    .expect("an upload phase owns an upload");
658                let bytes = upload
659                    .bytes
660                    .as_ref()
661                    .expect("a staged body that failed to read has no operation");
662                Some(RoundOperation::UploadGrant {
663                    blob_id: &upload.id,
664                    byte_length: bytes.len() as u64,
665                    media_type: upload.media_type.as_deref(),
666                })
667            }
668            RoundPhase::UploadPut { url, .. } => {
669                let upload = self
670                    .uploads
671                    .front()
672                    .expect("an upload phase owns an upload");
673                let bytes = upload
674                    .bytes
675                    .as_ref()
676                    .expect("a staged body that failed to read has no operation");
677                Some(RoundOperation::UploadPut {
678                    url,
679                    bytes,
680                    media_type: upload.media_type.as_deref(),
681                })
682            }
683            RoundPhase::UploadDirect => {
684                let upload = self
685                    .uploads
686                    .front()
687                    .expect("an upload phase owns an upload");
688                let bytes = upload
689                    .bytes
690                    .as_ref()
691                    .expect("a staged body that failed to read has no operation");
692                Some(RoundOperation::Upload {
693                    blob_id: &upload.id,
694                    bytes,
695                    media_type: upload.media_type.as_deref(),
696                })
697            }
698            RoundPhase::Sync => Some(RoundOperation::Sync {
699                request: self.encoded.as_deref().expect("the request is encoded"),
700                realtime: self.realtime,
701            }),
702            RoundPhase::Downloads { fetches, .. } => Some(RoundOperation::Downloads(fetches)),
703        }
704    }
705}
706
707#[derive(PartialEq)]
708enum DownloadKey {
709    Url(String),
710    Segment(String, String, String),
711}
712
713/// One eligible `SEGMENT_REF` of the response, collected before the batch
714/// fetch so `download_segments` sees the whole page list at once.
715pub(crate) struct PlannedFetch {
716    key: DownloadKey,
717    subscription_id: String,
718    table: String,
719    segment_id: String,
720    bytes_total: Option<u64>,
721    rows_total: Option<u64>,
722    /// `None` when the signed URL had already expired at the batch start: the
723    /// fetch is never attempted and its descriptor fails with
724    /// `sync.segment_expired`. Collection stops there, so only the last
725    /// collected fetch can be `None`.
726    fetch: Option<SegmentFetch>,
727}
728
729#[derive(Default)]
730pub(crate) struct DownloadResults {
731    entries: std::collections::VecDeque<(DownloadKey, i64, Result<Vec<u8>, TransportError>)>,
732    pub controls: Vec<String>,
733}
734
735impl DownloadResults {
736    fn take(&mut self, key: DownloadKey) -> Result<Vec<u8>, TransportError> {
737        let position = self
738            .entries
739            .iter()
740            .position(|(candidate, ..)| *candidate == key);
741        position
742            .and_then(|i| self.entries.remove(i))
743            .map(|(_, _, bytes)| bytes)
744            .unwrap_or_else(|| {
745                Err(TransportError::new(
746                    "client.round_download_missing",
747                    "round apply requested an uncaptured segment",
748                ))
749            })
750    }
751}
752
753impl Transport for DownloadResults {
754    fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
755        Err(TransportError::new(
756            "client.round_network_on_owner",
757            "network requests must execute outside the client owner",
758        ))
759    }
760    fn realtime_sync(&mut self, request: &[u8]) -> Result<Vec<u8>, TransportError> {
761        self.sync(request)
762    }
763    fn download_segment(
764        &mut self,
765        request: &SegmentRequest,
766        _: &mut dyn FnMut(u64),
767    ) -> Result<Vec<u8>, TransportError> {
768        self.take(DownloadKey::Segment(
769            request.segment_id.clone(),
770            request.table.clone(),
771            request.requested_scopes_json.clone(),
772        ))
773    }
774    fn fetch_url(
775        &mut self,
776        url: &str,
777        _: u64,
778        _: &mut dyn FnMut(u64),
779    ) -> Result<Vec<u8>, TransportError> {
780        self.take(DownloadKey::Url(url.to_owned()))
781    }
782    fn segment_fetch_started_at(&self, url: &str) -> Option<i64> {
783        self.entries
784            .iter()
785            .find_map(|(key, time, _)| (*key == DownloadKey::Url(url.to_owned())).then_some(*time))
786    }
787    fn realtime_connect(&mut self) -> Result<(), TransportError> {
788        Err(TransportError::new(
789            "client.round_network_on_owner",
790            "network connections must execute outside the client owner",
791        ))
792    }
793    fn realtime_send(&mut self, text: &str) -> Result<(), TransportError> {
794        self.controls.push(text.to_owned());
795        Ok(())
796    }
797    fn realtime_close(&mut self) -> Result<(), TransportError> {
798        Ok(())
799    }
800}
801
802#[cfg(test)]
803pub(crate) mod test_support {
804    use super::*;
805
806    /// Drive one exchange through the split API, answering every operation
807    /// with the blocking transport under test.
808    pub(crate) fn drive_round(
809        mut prepared: PreparedSyncRound,
810        transport: &mut dyn Transport,
811    ) -> CompletedSyncRound {
812        while let Some(operation) = prepared.next_operation() {
813            let reply = match operation {
814                RoundOperation::UploadGrant {
815                    blob_id,
816                    byte_length,
817                    media_type,
818                } => RoundReply::UploadGrant(transport.blob_upload_grant(
819                    blob_id,
820                    byte_length,
821                    media_type,
822                )),
823                RoundOperation::UploadPut {
824                    url,
825                    bytes,
826                    media_type,
827                } => RoundReply::Upload(transport.blob_put_url(url, bytes, media_type)),
828                RoundOperation::Upload {
829                    blob_id,
830                    bytes,
831                    media_type,
832                } => RoundReply::Upload(transport.blob_upload(blob_id, bytes, media_type)),
833                RoundOperation::Sync { request, realtime } => RoundReply::Sync(if realtime {
834                    transport.realtime_sync(request)
835                } else {
836                    transport.sync(request)
837                }),
838                RoundOperation::Downloads(fetches) => {
839                    RoundReply::Downloads(transport.download_segments(fetches, &mut |_, _| {}))
840                }
841            };
842            prepared
843                .submit(reply)
844                .expect("the driver answers the pending operation");
845        }
846        prepared.finish().expect("the round has no work left")
847    }
848}
849
850#[cfg(test)]
851mod tests {
852    use super::*;
853    use crate::{BlobUploadGrant, SegmentRequest, SyncClient, TransportError};
854
855    /// Every upload call fails with the same typed timeout, modelling a round
856    /// whose deadline was spent before the signed upload. The fallback to the
857    /// direct endpoint must not replace the typed code with a generic one.
858    struct SpentDeadline;
859
860    impl Transport for SpentDeadline {
861        fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
862            unreachable!()
863        }
864        fn realtime_sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
865            unreachable!()
866        }
867        fn download_segment(
868            &mut self,
869            _: &SegmentRequest,
870            _: &mut dyn FnMut(u64),
871        ) -> Result<Vec<u8>, TransportError> {
872            unreachable!()
873        }
874        fn blob_upload_grant(
875            &mut self,
876            _: &str,
877            _: u64,
878            _: Option<&str>,
879        ) -> Result<BlobUploadGrant, TransportError> {
880            Ok(BlobUploadGrant::Url {
881                url: "https://storage.example/signed".into(),
882                url_expires_at_ms: None,
883            })
884        }
885        fn blob_put_url(
886            &mut self,
887            _: &str,
888            _: &[u8],
889            _: Option<&str>,
890        ) -> Result<(), TransportError> {
891            Err(TransportError::new(
892                "transport.timeout",
893                "signed upload timed out",
894            ))
895        }
896        fn blob_upload(
897            &mut self,
898            _: &str,
899            _: &[u8],
900            _: Option<&str>,
901        ) -> Result<(), TransportError> {
902            Err(TransportError::new(
903                "transport.timeout",
904                "direct upload timed out",
905            ))
906        }
907        fn realtime_connect(&mut self) -> Result<(), TransportError> {
908            Ok(())
909        }
910        fn realtime_send(&mut self, _: &str) -> Result<(), TransportError> {
911            Ok(())
912        }
913        fn realtime_close(&mut self) -> Result<(), TransportError> {
914            Ok(())
915        }
916    }
917
918    #[test]
919    fn a_failed_presigned_put_falls_back_to_the_direct_endpoint() {
920        let schema = serde_json::json!({"version":1,"tables":[{"name":"attachments","primaryKey":"id","columns":[
921            {"name":"id","type":"string","nullable":false},
922            {"name":"file","type":"blob_ref","nullable":true}],"scopes":[]}]});
923        let mut client =
924            SyncClient::new("spent-deadline".into(), &schema, Default::default()).unwrap();
925        client
926            .set_meta(crate::client::LOG_EPOCH_KEY, "epoch")
927            .unwrap();
928        client
929            .upload_blob(b"body", Some("text/plain".into()), None)
930            .unwrap();
931        let mut prepared = client.prepare_sync_round(false).unwrap();
932        let mut transport = SpentDeadline;
933        let mut steps = Vec::new();
934        while let Some(operation) = prepared.next_operation() {
935            let reply = match operation {
936                RoundOperation::UploadGrant {
937                    blob_id,
938                    byte_length,
939                    media_type,
940                } => {
941                    steps.push("grant");
942                    RoundReply::UploadGrant(transport.blob_upload_grant(
943                        blob_id,
944                        byte_length,
945                        media_type,
946                    ))
947                }
948                RoundOperation::UploadPut {
949                    url,
950                    bytes,
951                    media_type,
952                } => {
953                    steps.push("put");
954                    RoundReply::Upload(transport.blob_put_url(url, bytes, media_type))
955                }
956                RoundOperation::Upload {
957                    blob_id,
958                    bytes,
959                    media_type,
960                } => {
961                    steps.push("direct");
962                    RoundReply::Upload(transport.blob_upload(blob_id, bytes, media_type))
963                }
964                _ => panic!("only upload operations precede the round message"),
965            };
966            prepared.submit(reply).unwrap();
967        }
968        assert_eq!(steps, ["grant", "put", "direct"]);
969        let outcome = match client.apply_sync_round(prepared.finish().unwrap()) {
970            AppliedSyncRound::Complete { outcome, .. } => outcome,
971            AppliedSyncRound::Continue(_) => panic!("an upload failure does not continue"),
972        };
973        match outcome {
974            SyncOutcome::Failed { error_code, .. } => {
975                assert_eq!(error_code, "transport.timeout")
976            }
977            other => panic!("expected a typed upload failure, got {other:?}"),
978        }
979    }
980
981    /// Answers one response that references one segment and serves a body
982    /// longer than the descriptor declares.
983    struct OversizedSegment {
984        segment_id: String,
985        body: Vec<u8>,
986    }
987
988    impl Transport for OversizedSegment {
989        fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
990            Ok(encode_message(&Message {
991                wire_version: ssp2::decode::WIRE_VERSION,
992                msg_kind: MsgKind::Response,
993                frames: vec![
994                    Frame::RespHeader {
995                        required_schema_version: None,
996                        latest_schema_version: None,
997                        log_epoch: "epoch".into(),
998                        reset_required: false,
999                    },
1000                    Frame::SubStart {
1001                        id: "tasks".into(),
1002                        status: SubStatus::Active,
1003                        reason_code: String::new(),
1004                        effective_scopes: Vec::new(),
1005                        bootstrap: true,
1006                    },
1007                    Frame::SegmentRef {
1008                        segment_id: self.segment_id.clone(),
1009                        media_type: MediaType::Rows,
1010                        table: "tasks".into(),
1011                        byte_length: 4,
1012                        row_count: 1,
1013                        as_of_commit_seq: 1,
1014                        scope_digest: "digest".into(),
1015                        row_cursor: None,
1016                        next_row_cursor: None,
1017                        url: None,
1018                        url_expires_at_ms: None,
1019                    },
1020                    Frame::SubEnd {
1021                        next_cursor: 1,
1022                        bootstrap_state: None,
1023                    },
1024                ],
1025            }))
1026        }
1027        fn realtime_sync(&mut self, request: &[u8]) -> Result<Vec<u8>, TransportError> {
1028            self.sync(request)
1029        }
1030        fn download_segment(
1031            &mut self,
1032            _: &SegmentRequest,
1033            _: &mut dyn FnMut(u64),
1034        ) -> Result<Vec<u8>, TransportError> {
1035            Ok(self.body.clone())
1036        }
1037        fn realtime_connect(&mut self) -> Result<(), TransportError> {
1038            Ok(())
1039        }
1040        fn realtime_send(&mut self, _: &str) -> Result<(), TransportError> {
1041            Ok(())
1042        }
1043        fn realtime_close(&mut self) -> Result<(), TransportError> {
1044            Ok(())
1045        }
1046    }
1047
1048    #[test]
1049    fn a_segment_body_longer_than_its_byte_length_fails_before_the_content_check() {
1050        use sha2::Digest;
1051        let schema = serde_json::json!({"version":1,"tables":[{"name":"tasks","primaryKey":"id","columns":[
1052            {"name":"id","type":"string","nullable":false}],"scopes":[]}]});
1053        let mut client = SyncClient::new("oversized".into(), &schema, Default::default()).unwrap();
1054        client
1055            .set_meta(crate::client::LOG_EPOCH_KEY, "epoch")
1056            .unwrap();
1057        client
1058            .subscribe("tasks".into(), "tasks".into(), Vec::new(), None)
1059            .unwrap();
1060        // The body hashes to its content address, so only the §5.4 cap can
1061        // refuse it.
1062        let body = b"twelve bytes".to_vec();
1063        let segment_id = format!(
1064            "sha256:{}",
1065            sha2::Sha256::digest(&body)
1066                .iter()
1067                .map(|byte| format!("{byte:02x}"))
1068                .collect::<String>()
1069        );
1070        let mut transport = OversizedSegment { segment_id, body };
1071        match client.sync(&mut transport) {
1072            SyncOutcome::Failed {
1073                error_code,
1074                message,
1075                ..
1076            } => {
1077                assert_eq!(error_code, "sync.invalid_request");
1078                assert_eq!(message, "segment exceeds its byteLength");
1079            }
1080            other => panic!("expected the byteLength cap to fail the round, got {other:?}"),
1081        }
1082    }
1083
1084    /// Records each single fetch the default batch runs and fails the second.
1085    #[derive(Default)]
1086    struct CountingFetches {
1087        calls: Vec<String>,
1088    }
1089
1090    impl Transport for CountingFetches {
1091        fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
1092            unreachable!()
1093        }
1094        fn realtime_sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
1095            unreachable!()
1096        }
1097        fn download_segment(
1098            &mut self,
1099            request: &SegmentRequest,
1100            on_progress: &mut dyn FnMut(u64),
1101        ) -> Result<Vec<u8>, TransportError> {
1102            self.calls.push(request.segment_id.clone());
1103            on_progress(4);
1104            if request.segment_id == "sha256:b" {
1105                Err(TransportError::new(
1106                    "transport.timeout",
1107                    "second fetch failed",
1108                ))
1109            } else {
1110                Ok(request.segment_id.clone().into_bytes())
1111            }
1112        }
1113        fn realtime_connect(&mut self) -> Result<(), TransportError> {
1114            Ok(())
1115        }
1116        fn realtime_send(&mut self, _: &str) -> Result<(), TransportError> {
1117            Ok(())
1118        }
1119        fn realtime_close(&mut self) -> Result<(), TransportError> {
1120            Ok(())
1121        }
1122    }
1123
1124    #[test]
1125    fn the_default_batch_runs_the_single_fetches_in_order_and_stops_at_the_first_error() {
1126        let mut transport = CountingFetches::default();
1127        let fetches: Vec<SegmentFetch> = ["a", "b", "c"]
1128            .into_iter()
1129            .map(|id| {
1130                SegmentFetch::Direct(SegmentRequest {
1131                    segment_id: format!("sha256:{id}"),
1132                    table: "tasks".to_owned(),
1133                    requested_scopes_json: "{}".to_owned(),
1134                    byte_length: 4,
1135                })
1136            })
1137            .collect();
1138        let mut reported = Vec::new();
1139        let results = transport.download_segments(&fetches, &mut |index, bytes| {
1140            reported.push((index, bytes));
1141        });
1142        assert_eq!(
1143            transport.calls,
1144            vec!["sha256:a", "sha256:b"],
1145            "the third fetch was never attempted"
1146        );
1147        assert_eq!(results.len(), 2, "one result per attempted fetch");
1148        assert_eq!(results[0].as_ref().unwrap(), b"sha256:a");
1149        assert_eq!(results[1].as_ref().unwrap_err().code, "transport.timeout");
1150        assert_eq!(reported, vec![(0, 4), (1, 4)]);
1151    }
1152}
syncular_client/round.rs

syncular_client/
round.rs
