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: 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)))
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                                let expired = url.is_some()
392                                    && url_expires_at_ms
393                                        .is_some_and(|expiry| expiry <= batch_started_at);
394                                let (key, fetch) = match url {
395                                    Some(url) => (
396                                        DownloadKey::Url(url.clone()),
397                                        if expired {
398                                            None
399                                        } else {
400                                            Some(SegmentFetch::Url(url.clone()))
401                                        },
402                                    ),
403                                    None => (
404                                        DownloadKey::Segment(
405                                            segment_id.clone(),
406                                            table.clone(),
407                                            scopes.to_owned(),
408                                        ),
409                                        Some(SegmentFetch::Direct(SegmentRequest {
410                                            segment_id: segment_id.clone(),
411                                            table: table.clone(),
412                                            requested_scopes_json: scopes.to_owned(),
413                                        })),
414                                    ),
415                                };
416                                planned.push(PlannedFetch {
417                                    key,
418                                    subscription_id: id.to_owned(),
419                                    table: table.clone(),
420                                    segment_id: segment_id.clone(),
421                                    bytes_total: u64::try_from(*byte_length).ok(),
422                                    rows_total: u64::try_from(*row_count).ok(),
423                                    fetch,
424                                });
425                                if expired {
426                                    break;
427                                }
428                            }
429                            _ => {}
430                        }
431                    }
432                    if !planned.is_empty() {
433                        let progress = self.progress.clone();
434                        let on_progress = |index: usize, bytes: u64| {
435                            let Some(planned_fetch) = planned.get(index) else {
436                                return;
437                            };
438                            progress.update(|p| {
439                                p.phase = ProgressPhase::Download;
440                                p.subscription_id = Some(planned_fetch.subscription_id.clone());
441                                p.table = Some(planned_fetch.table.clone());
442                                p.segment_id = Some(planned_fetch.segment_id.clone());
443                                p.bytes_received = bytes;
444                                p.bytes_total = planned_fetch.bytes_total;
445                                p.rows_processed = 0;
446                                p.rows_total = planned_fetch.rows_total;
447                            });
448                        };
449                        // Announce the download phase before the first byte lands.
450                        on_progress(0, 0);
451                        let fetches: Vec<SegmentFetch> = planned
452                            .iter()
453                            .filter_map(|planned_fetch| planned_fetch.fetch.clone())
454                            .collect();
455                        self.phase = RoundPhase::Downloads {
456                            planned,
457                            fetches,
458                            response,
459                            batch_started_at,
460                        };
461                        return Ok(());
462                    }
463                }
464                self.completed = Some(ExchangeResult::Reply {
465                    response: Ok(response),
466                    transport_failed,
467                    downloads: DownloadResults::default(),
468                });
469            }
470            (
471                RoundPhase::Downloads {
472                    planned,
473                    batch_started_at,
474                    response,
475                    ..
476                },
477                RoundReply::Downloads(results),
478            ) => {
479                let mut downloads = DownloadResults::default();
480                let progress = self.progress.clone();
481                let mut results = results.into_iter();
482                for planned_fetch in planned {
483                    let result = match planned_fetch.fetch {
484                        None => Err(TransportError::new(
485                            "sync.segment_expired",
486                            "signed segment URL expired before fetch",
487                        )),
488                        // A shorter result list means the batch stopped after an
489                        // earlier error, and apply aborts there.
490                        Some(_) => match results.next() {
491                            Some(result) => result,
492                            None => break,
493                        },
494                    };
495                    if let Ok(bytes) = &result {
496                        progress.update(|p| {
497                            p.phase = ProgressPhase::Download;
498                            p.subscription_id = Some(planned_fetch.subscription_id.clone());
499                            p.table = Some(planned_fetch.table.clone());
500                            p.segment_id = Some(planned_fetch.segment_id.clone());
501                            p.bytes_received = bytes.len() as u64;
502                            p.bytes_total = planned_fetch.bytes_total;
503                            p.rows_processed = 0;
504                            p.rows_total = planned_fetch.rows_total;
505                        });
506                    }
507                    let failed = result.is_err();
508                    downloads
509                        .entries
510                        .push_back((planned_fetch.key, batch_started_at, result));
511                    if failed {
512                        break;
513                    }
514                }
515                self.completed = Some(ExchangeResult::Reply {
516                    response: Ok(response),
517                    transport_failed: false,
518                    downloads,
519                });
520            }
521            (phase, _) => {
522                self.phase = phase;
523                return Err(TransportError::new(
524                    "client.invalid_round_reply",
525                    "the reply does not answer the pending operation",
526                ));
527            }
528        }
529        Ok(())
530    }
531
532    /// The completed round once every operation is answered, ready for
533    /// `SyncClient::apply_sync_round`.
534    pub fn finish(mut self) -> Option<CompletedSyncRound> {
535        let exchange = self.completed.take()?;
536        Some(CompletedSyncRound {
537            prepared: self,
538            exchange,
539        })
540    }
541
542    fn upload_complete(&mut self) {
543        let upload = self
544            .uploads
545            .pop_front()
546            .expect("an upload phase owns an upload");
547        self.completed = Some(ExchangeResult::Upload {
548            id: upload.id,
549            result: Ok(()),
550        });
551        self.continuation = Some(if self.uploads.is_empty() {
552            RoundPhase::Sync
553        } else {
554            RoundPhase::UploadGrant
555        });
556    }
557
558    fn ensure_encoded(&mut self) {
559        if self.encoded.is_none() {
560            #[cfg(feature = "bench-internals")]
561            let _phase = self
562                .benchmark_phases
563                .start(crate::bench::Phase::RequestEncode);
564            self.encoded = Some(encode_message(&self.message));
565        }
566    }
567
568    fn next_operation_scoped<'a>(
569        &'a mut self,
570        transport: &mut dyn Transport,
571        scoped: &mut bool,
572    ) -> Option<RoundOperation<'a>> {
573        self.next_operation_impl(Some(transport), scoped)
574    }
575
576    fn next_operation_impl<'a>(
577        &'a mut self,
578        mut scope: Option<&mut dyn Transport>,
579        scoped: &mut bool,
580    ) -> Option<RoundOperation<'a>> {
581        if self.completed.is_some() {
582            return None;
583        }
584        // The previous exchange ended with an upload; the owner has applied it.
585        if let Some(next) = self.continuation.take() {
586            self.phase = next;
587        }
588        if matches!(self.phase, RoundPhase::UploadGrant) && self.uploads.is_empty() {
589            self.phase = RoundPhase::Sync;
590        }
591        // A staged body that failed to read locally performs no network I/O,
592        // so it cannot be the round's first network call and must not anchor
593        // the round deadline either.
594        if matches!(self.phase, RoundPhase::UploadGrant)
595            && self
596                .uploads
597                .front()
598                .is_some_and(|upload| upload.bytes.is_err())
599        {
600            let upload = self
601                .uploads
602                .pop_front()
603                .expect("the failed upload is staged");
604            self.completed = Some(ExchangeResult::Upload {
605                id: upload.id,
606                result: Err(upload.bytes.unwrap_err()),
607            });
608            return None;
609        }
610        if matches!(self.phase, RoundPhase::Sync) {
611            self.ensure_encoded();
612        }
613        let needs_network = !matches!(self.phase, RoundPhase::Downloads { .. });
614        // The round's first network call anchors the deadline after the
615        // encode; a continuation reuses the carried anchor instead of
616        // refreshing it.
617        if needs_network && !*scoped {
618            *scoped = true;
619            if let Some(transport) = scope.as_mut() {
620                if let Err(error) = self.scope_round_deadline(&mut **transport) {
621                    if matches!(
622                        self.phase,
623                        RoundPhase::UploadGrant
624                            | RoundPhase::UploadPut { .. }
625                            | RoundPhase::UploadDirect
626                    ) {
627                        let upload = self
628                            .uploads
629                            .pop_front()
630                            .expect("an upload phase owns an upload");
631                        self.completed = Some(ExchangeResult::Upload {
632                            id: upload.id,
633                            result: Err(error),
634                        });
635                    } else {
636                        self.completed = Some(ExchangeResult::Reply {
637                            response: Err(error),
638                            transport_failed: false,
639                            downloads: DownloadResults::default(),
640                        });
641                    }
642                    return None;
643                }
644            }
645        }
646        match &self.phase {
647            RoundPhase::UploadGrant => {
648                let upload = self
649                    .uploads
650                    .front()
651                    .expect("an upload phase owns an upload");
652                let bytes = upload
653                    .bytes
654                    .as_ref()
655                    .expect("a staged body that failed to read has no operation");
656                Some(RoundOperation::UploadGrant {
657                    blob_id: &upload.id,
658                    byte_length: bytes.len() as u64,
659                    media_type: upload.media_type.as_deref(),
660                })
661            }
662            RoundPhase::UploadPut { url, .. } => {
663                let upload = self
664                    .uploads
665                    .front()
666                    .expect("an upload phase owns an upload");
667                let bytes = upload
668                    .bytes
669                    .as_ref()
670                    .expect("a staged body that failed to read has no operation");
671                Some(RoundOperation::UploadPut {
672                    url,
673                    bytes,
674                    media_type: upload.media_type.as_deref(),
675                })
676            }
677            RoundPhase::UploadDirect => {
678                let upload = self
679                    .uploads
680                    .front()
681                    .expect("an upload phase owns an upload");
682                let bytes = upload
683                    .bytes
684                    .as_ref()
685                    .expect("a staged body that failed to read has no operation");
686                Some(RoundOperation::Upload {
687                    blob_id: &upload.id,
688                    bytes,
689                    media_type: upload.media_type.as_deref(),
690                })
691            }
692            RoundPhase::Sync => Some(RoundOperation::Sync {
693                request: self.encoded.as_deref().expect("the request is encoded"),
694                realtime: self.realtime,
695            }),
696            RoundPhase::Downloads { fetches, .. } => Some(RoundOperation::Downloads(fetches)),
697        }
698    }
699}
700
701#[derive(PartialEq)]
702enum DownloadKey {
703    Url(String),
704    Segment(String, String, String),
705}
706
707/// One eligible `SEGMENT_REF` of the response, collected before the batch
708/// fetch so `download_segments` sees the whole page list at once.
709pub(crate) struct PlannedFetch {
710    key: DownloadKey,
711    subscription_id: String,
712    table: String,
713    segment_id: String,
714    bytes_total: Option<u64>,
715    rows_total: Option<u64>,
716    /// `None` when the signed URL had already expired at the batch start: the
717    /// fetch is never attempted and its descriptor fails with
718    /// `sync.segment_expired`. Collection stops there, so only the last
719    /// collected fetch can be `None`.
720    fetch: Option<SegmentFetch>,
721}
722
723#[derive(Default)]
724pub(crate) struct DownloadResults {
725    entries: std::collections::VecDeque<(DownloadKey, i64, Result<Vec<u8>, TransportError>)>,
726    pub controls: Vec<String>,
727}
728
729impl DownloadResults {
730    fn take(&mut self, key: DownloadKey) -> Result<Vec<u8>, TransportError> {
731        let position = self
732            .entries
733            .iter()
734            .position(|(candidate, ..)| *candidate == key);
735        position
736            .and_then(|i| self.entries.remove(i))
737            .map(|(_, _, bytes)| bytes)
738            .unwrap_or_else(|| {
739                Err(TransportError::new(
740                    "client.round_download_missing",
741                    "round apply requested an uncaptured segment",
742                ))
743            })
744    }
745}
746
747impl Transport for DownloadResults {
748    fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
749        Err(TransportError::new(
750            "client.round_network_on_owner",
751            "network requests must execute outside the client owner",
752        ))
753    }
754    fn realtime_sync(&mut self, request: &[u8]) -> Result<Vec<u8>, TransportError> {
755        self.sync(request)
756    }
757    fn download_segment(
758        &mut self,
759        request: &SegmentRequest,
760        _: &mut dyn FnMut(u64),
761    ) -> Result<Vec<u8>, TransportError> {
762        self.take(DownloadKey::Segment(
763            request.segment_id.clone(),
764            request.table.clone(),
765            request.requested_scopes_json.clone(),
766        ))
767    }
768    fn fetch_url(&mut self, url: &str, _: &mut dyn FnMut(u64)) -> Result<Vec<u8>, TransportError> {
769        self.take(DownloadKey::Url(url.to_owned()))
770    }
771    fn segment_fetch_started_at(&self, url: &str) -> Option<i64> {
772        self.entries
773            .iter()
774            .find_map(|(key, time, _)| (*key == DownloadKey::Url(url.to_owned())).then_some(*time))
775    }
776    fn realtime_connect(&mut self) -> Result<(), TransportError> {
777        Err(TransportError::new(
778            "client.round_network_on_owner",
779            "network connections must execute outside the client owner",
780        ))
781    }
782    fn realtime_send(&mut self, text: &str) -> Result<(), TransportError> {
783        self.controls.push(text.to_owned());
784        Ok(())
785    }
786    fn realtime_close(&mut self) -> Result<(), TransportError> {
787        Ok(())
788    }
789}
790
791#[cfg(test)]
792pub(crate) mod test_support {
793    use super::*;
794
795    /// Drive one exchange through the split API, answering every operation
796    /// with the blocking transport under test.
797    pub(crate) fn drive_round(
798        mut prepared: PreparedSyncRound,
799        transport: &mut dyn Transport,
800    ) -> CompletedSyncRound {
801        while let Some(operation) = prepared.next_operation() {
802            let reply = match operation {
803                RoundOperation::UploadGrant {
804                    blob_id,
805                    byte_length,
806                    media_type,
807                } => RoundReply::UploadGrant(transport.blob_upload_grant(
808                    blob_id,
809                    byte_length,
810                    media_type,
811                )),
812                RoundOperation::UploadPut {
813                    url,
814                    bytes,
815                    media_type,
816                } => RoundReply::Upload(transport.blob_put_url(url, bytes, media_type)),
817                RoundOperation::Upload {
818                    blob_id,
819                    bytes,
820                    media_type,
821                } => RoundReply::Upload(transport.blob_upload(blob_id, bytes, media_type)),
822                RoundOperation::Sync { request, realtime } => RoundReply::Sync(if realtime {
823                    transport.realtime_sync(request)
824                } else {
825                    transport.sync(request)
826                }),
827                RoundOperation::Downloads(fetches) => {
828                    RoundReply::Downloads(transport.download_segments(fetches, &mut |_, _| {}))
829                }
830            };
831            prepared
832                .submit(reply)
833                .expect("the driver answers the pending operation");
834        }
835        prepared.finish().expect("the round has no work left")
836    }
837}
838
839#[cfg(test)]
840mod tests {
841    use super::*;
842    use crate::{BlobUploadGrant, SegmentRequest, SyncClient, TransportError};
843
844    /// Every upload call fails with the same typed timeout, modelling a round
845    /// whose deadline was spent before the signed upload. The fallback to the
846    /// direct endpoint must not replace the typed code with a generic one.
847    struct SpentDeadline;
848
849    impl Transport for SpentDeadline {
850        fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
851            unreachable!()
852        }
853        fn realtime_sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
854            unreachable!()
855        }
856        fn download_segment(
857            &mut self,
858            _: &SegmentRequest,
859            _: &mut dyn FnMut(u64),
860        ) -> Result<Vec<u8>, TransportError> {
861            unreachable!()
862        }
863        fn blob_upload_grant(
864            &mut self,
865            _: &str,
866            _: u64,
867            _: Option<&str>,
868        ) -> Result<BlobUploadGrant, TransportError> {
869            Ok(BlobUploadGrant::Url {
870                url: "https://storage.example/signed".into(),
871                url_expires_at_ms: None,
872            })
873        }
874        fn blob_put_url(
875            &mut self,
876            _: &str,
877            _: &[u8],
878            _: Option<&str>,
879        ) -> Result<(), TransportError> {
880            Err(TransportError::new(
881                "transport.timeout",
882                "signed upload timed out",
883            ))
884        }
885        fn blob_upload(
886            &mut self,
887            _: &str,
888            _: &[u8],
889            _: Option<&str>,
890        ) -> Result<(), TransportError> {
891            Err(TransportError::new(
892                "transport.timeout",
893                "direct upload timed out",
894            ))
895        }
896        fn realtime_connect(&mut self) -> Result<(), TransportError> {
897            Ok(())
898        }
899        fn realtime_send(&mut self, _: &str) -> Result<(), TransportError> {
900            Ok(())
901        }
902        fn realtime_close(&mut self) -> Result<(), TransportError> {
903            Ok(())
904        }
905    }
906
907    #[test]
908    fn a_failed_presigned_put_falls_back_to_the_direct_endpoint() {
909        let schema = serde_json::json!({"version":1,"tables":[{"name":"attachments","primaryKey":"id","columns":[
910            {"name":"id","type":"string","nullable":false},
911            {"name":"file","type":"blob_ref","nullable":true}],"scopes":[]}]});
912        let mut client =
913            SyncClient::new("spent-deadline".into(), &schema, Default::default()).unwrap();
914        client.set_meta(crate::client::LOG_EPOCH_KEY, "epoch");
915        client
916            .upload_blob(b"body", Some("text/plain".into()), None)
917            .unwrap();
918        let mut prepared = client.prepare_sync_round(false).unwrap();
919        let mut transport = SpentDeadline;
920        let mut steps = Vec::new();
921        while let Some(operation) = prepared.next_operation() {
922            let reply = match operation {
923                RoundOperation::UploadGrant {
924                    blob_id,
925                    byte_length,
926                    media_type,
927                } => {
928                    steps.push("grant");
929                    RoundReply::UploadGrant(transport.blob_upload_grant(
930                        blob_id,
931                        byte_length,
932                        media_type,
933                    ))
934                }
935                RoundOperation::UploadPut {
936                    url,
937                    bytes,
938                    media_type,
939                } => {
940                    steps.push("put");
941                    RoundReply::Upload(transport.blob_put_url(url, bytes, media_type))
942                }
943                RoundOperation::Upload {
944                    blob_id,
945                    bytes,
946                    media_type,
947                } => {
948                    steps.push("direct");
949                    RoundReply::Upload(transport.blob_upload(blob_id, bytes, media_type))
950                }
951                _ => panic!("only upload operations precede the round message"),
952            };
953            prepared.submit(reply).unwrap();
954        }
955        assert_eq!(steps, ["grant", "put", "direct"]);
956        let outcome = match client.apply_sync_round(prepared.finish().unwrap()) {
957            AppliedSyncRound::Complete { outcome, .. } => outcome,
958            AppliedSyncRound::Continue(_) => panic!("an upload failure does not continue"),
959        };
960        match outcome {
961            SyncOutcome::Failed { error_code, .. } => {
962                assert_eq!(error_code, "transport.timeout")
963            }
964            other => panic!("expected a typed upload failure, got {other:?}"),
965        }
966    }
967
968    /// Records each single fetch the default batch runs and fails the second.
969    #[derive(Default)]
970    struct CountingFetches {
971        calls: Vec<String>,
972    }
973
974    impl Transport for CountingFetches {
975        fn sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
976            unreachable!()
977        }
978        fn realtime_sync(&mut self, _: &[u8]) -> Result<Vec<u8>, TransportError> {
979            unreachable!()
980        }
981        fn download_segment(
982            &mut self,
983            request: &SegmentRequest,
984            on_progress: &mut dyn FnMut(u64),
985        ) -> Result<Vec<u8>, TransportError> {
986            self.calls.push(request.segment_id.clone());
987            on_progress(4);
988            if request.segment_id == "sha256:b" {
989                Err(TransportError::new(
990                    "transport.timeout",
991                    "second fetch failed",
992                ))
993            } else {
994                Ok(request.segment_id.clone().into_bytes())
995            }
996        }
997        fn realtime_connect(&mut self) -> Result<(), TransportError> {
998            Ok(())
999        }
1000        fn realtime_send(&mut self, _: &str) -> Result<(), TransportError> {
1001            Ok(())
1002        }
1003        fn realtime_close(&mut self) -> Result<(), TransportError> {
1004            Ok(())
1005        }
1006    }
1007
1008    #[test]
1009    fn the_default_batch_runs_the_single_fetches_in_order_and_stops_at_the_first_error() {
1010        let mut transport = CountingFetches::default();
1011        let fetches: Vec<SegmentFetch> = ["a", "b", "c"]
1012            .into_iter()
1013            .map(|id| {
1014                SegmentFetch::Direct(SegmentRequest {
1015                    segment_id: format!("sha256:{id}"),
1016                    table: "tasks".to_owned(),
1017                    requested_scopes_json: "{}".to_owned(),
1018                })
1019            })
1020            .collect();
1021        let mut reported = Vec::new();
1022        let results = transport.download_segments(&fetches, &mut |index, bytes| {
1023            reported.push((index, bytes));
1024        });
1025        assert_eq!(
1026            transport.calls,
1027            vec!["sha256:a", "sha256:b"],
1028            "the third fetch was never attempted"
1029        );
1030        assert_eq!(results.len(), 2, "one result per attempted fetch");
1031        assert_eq!(results[0].as_ref().unwrap(), b"sha256:a");
1032        assert_eq!(results[1].as_ref().unwrap_err().code, "transport.timeout");
1033        assert_eq!(reported, vec![(0, 4), (1, 4)]);
1034    }
1035}
syncular_client/round.rs

syncular_client/
round.rs
