1use std::collections::HashSet;
2use std::io::Write;
3use std::path::PathBuf;
4use std::time::Instant;
5
6use anyhow::{Context, Result, bail};
7use flate2::write::GzEncoder;
8
9use super::blob::{
10 MissingLocalBlob, attachment_blob_contract, contract_by_hash, missing_counts,
11 unique_blob_contracts,
12};
13use super::persistence::{ApplySyncPage, ClientSyncPage};
14use super::planner::{
15 PendingChange, TransferBudget, TransferObject, plan_change_prefix, plan_transfers,
16};
17use super::wire::{
18 BlobUploadContract, MAX_BLOB_TRANSFER_BYTES, MAX_BLOB_TRANSFER_OBJECTS, MAX_PULL_BATCH,
19 MAX_PUSH_BATCH, MissingBlobsRequest, MissingBlobsResponse, SyncResponse,
20 sync_server_url_is_valid, validate_blob_hashes,
21};
22use crate::attachments::lifecycle::LifecyclePolicy;
23use crate::db::Database;
24use crate::ids::{new_id, now};
25
26const GZIP_THRESHOLD: usize = 256;
27const MAX_HTTP_ERROR_DETAIL_BYTES: usize = 1024;
28
29#[derive(Clone, PartialEq, Eq)]
30pub struct SyncRequestContext {
31 session_id: String,
32 request: usize,
33}
34
35#[derive(Clone, PartialEq, Eq)]
36pub struct SyncHttpHeader {
37 pub name: String,
38 pub value: String,
39}
40
41#[derive(Clone)]
42pub struct PreparedSyncRequest {
43 pub method: String,
44 pub url: String,
45 pub headers: Vec<SyncHttpHeader>,
46 pub body: Vec<u8>,
47 pub context: SyncRequestContext,
48}
49
50pub struct SyncHttpResponse {
51 pub status: u16,
52 pub headers: Vec<SyncHttpHeader>,
53 pub body: Vec<u8>,
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct SyncPageOutcome {
58 pub page: usize,
59 pub pushed: usize,
60 pub pulled: usize,
61 pub blob_uploaded: usize,
62 pub blob_uploaded_bytes: u64,
63 pub blob_downloaded: usize,
64 pub blob_downloaded_bytes: u64,
65 pub cursor: i64,
66 pub complete: bool,
67 pub has_more: bool,
68 pub local_more: bool,
69 pub request_bytes: usize,
70 pub request_wire_bytes: usize,
71 pub response_decoded_bytes: usize,
72 pub response_compression: String,
73 pub apply_ms: u128,
74}
75
76#[derive(Debug, Clone, PartialEq, Eq)]
77pub struct SyncSessionSummary {
78 pub pushed: i64,
79 pub pulled: usize,
80 pub blob_uploaded: usize,
81 pub blob_uploaded_bytes: u64,
82 pub blob_downloaded: usize,
83 pub blob_downloaded_bytes: u64,
84 pub blob_upload_remaining: usize,
85 pub blob_upload_remaining_bytes: u64,
86 pub blob_download_remaining: usize,
87 pub blob_download_remaining_bytes: u64,
88 pub cursor: i64,
89 pub complete: bool,
90 pub pages: usize,
91 pub request_bytes: usize,
92 pub request_wire_bytes: usize,
93 pub response_decoded_bytes: usize,
94 pub response_compression: String,
95 pub apply_ms: u128,
96}
97
98struct ActivePage {
99 page: ClientSyncPage,
100 contracts: Vec<BlobUploadContract>,
101 probed: bool,
102 missing: HashSet<String>,
103 uploads: Vec<BlobUploadContract>,
104 upload_index: usize,
105 confirmed: bool,
106 metadata_sent: bool,
107 downloads: Vec<MissingLocalBlob>,
108 download_index: usize,
109 has_more: bool,
110 page_pushed: usize,
111 page_pulled: usize,
112 page_uploaded: usize,
113 page_uploaded_bytes: u64,
114 page_downloaded: usize,
115 page_downloaded_bytes: u64,
116 transfer_budget_blocked: bool,
117}
118
119impl ActivePage {
120 fn new(mut page: ClientSyncPage) -> Result<Self> {
121 let mut hashes = HashSet::new();
122 let mut count = page.request.changes.len();
123 for (index, change) in page.request.changes.iter().enumerate() {
124 if let Some(contract) = attachment_blob_contract(change)?
125 && hashes.insert(contract.sha256)
126 && hashes.len() > MAX_BLOB_TRANSFER_OBJECTS
127 {
128 count = index;
129 break;
130 }
131 }
132 page.request.changes.truncate(count);
133 page.pending = count;
134 let contracts = unique_blob_contracts(&page.request.changes)?;
135 Ok(Self {
136 page,
137 contracts,
138 probed: false,
139 missing: HashSet::new(),
140 uploads: Vec::new(),
141 upload_index: 0,
142 confirmed: false,
143 metadata_sent: false,
144 downloads: Vec::new(),
145 download_index: 0,
146 has_more: false,
147 page_pushed: 0,
148 page_pulled: 0,
149 page_uploaded: 0,
150 page_uploaded_bytes: 0,
151 page_downloaded: 0,
152 page_downloaded_bytes: 0,
153 transfer_budget_blocked: false,
154 })
155 }
156}
157
158#[derive(Clone)]
159enum RequestKind {
160 Missing,
161 Upload {
162 lease_id: String,
163 contract: BlobUploadContract,
164 },
165 Confirm,
166 Metadata {
167 request_bytes: usize,
168 },
169 Download {
170 blob: MissingLocalBlob,
171 },
172}
173
174struct OutstandingRequest {
175 prepared: PreparedSyncRequest,
176 kind: RequestKind,
177}
178
179pub struct SyncSession {
180 database: Database,
181 server: String,
182 auth_token: Option<String>,
183 blob_dir: PathBuf,
184 lifecycle_policy: LifecyclePolicy,
185 page_budget: Option<usize>,
186 attempted_at: String,
187 session_id: String,
188 request_number: usize,
189 active: Option<ActivePage>,
190 outstanding: Option<OutstandingRequest>,
191 known_server_blobs: HashSet<String>,
192 transfer_budget: TransferBudget,
193 summary: SyncSessionSummary,
194 last_has_more: bool,
195 last_local_more: bool,
196 stopped: bool,
197}
198
199impl SyncSession {
200 pub async fn start(
201 database: Database,
202 server: String,
203 auth_token: Option<String>,
204 page_budget: Option<usize>,
205 ) -> Result<Self> {
206 let blob_dir = crate::attachments::default_blob_dir(database.path());
207 Self::start_with_attachment_storage(
208 database,
209 server,
210 auth_token,
211 page_budget,
212 blob_dir,
213 LifecyclePolicy::default(),
214 )
215 .await
216 }
217
218 pub async fn start_with_attachment_storage(
219 database: Database,
220 server: String,
221 auth_token: Option<String>,
222 page_budget: Option<usize>,
223 blob_dir: PathBuf,
224 lifecycle_policy: LifecyclePolicy,
225 ) -> Result<Self> {
226 let attempted_at = now();
227 database.begin_sync_attempt(attempted_at.clone()).await?;
228 if !sync_server_url_is_valid(&server) {
229 let message = "invalid sync server URL";
230 database.record_sync_error(message.to_string()).await?;
231 bail!("{message}");
232 }
233 Ok(Self {
234 database,
235 server: server.trim_end_matches('/').to_string(),
236 auth_token,
237 blob_dir,
238 lifecycle_policy,
239 page_budget,
240 attempted_at,
241 session_id: new_id(),
242 request_number: 0,
243 active: None,
244 outstanding: None,
245 known_server_blobs: HashSet::new(),
246 transfer_budget: TransferBudget {
247 objects: MAX_BLOB_TRANSFER_OBJECTS,
248 bytes: MAX_BLOB_TRANSFER_BYTES,
249 completed_objects: 0,
250 },
251 summary: SyncSessionSummary {
252 pushed: 0,
253 pulled: 0,
254 blob_uploaded: 0,
255 blob_uploaded_bytes: 0,
256 blob_downloaded: 0,
257 blob_downloaded_bytes: 0,
258 blob_upload_remaining: 0,
259 blob_upload_remaining_bytes: 0,
260 blob_download_remaining: 0,
261 blob_download_remaining_bytes: 0,
262 cursor: 0,
263 complete: false,
264 pages: 0,
265 request_bytes: 0,
266 request_wire_bytes: 0,
267 response_decoded_bytes: 0,
268 response_compression: "none".to_string(),
269 apply_ms: 0,
270 },
271 last_has_more: false,
272 last_local_more: false,
273 stopped: false,
274 })
275 }
276
277 pub async fn prepare_request(&mut self) -> Result<Option<PreparedSyncRequest>> {
278 if self.stopped {
279 return Ok(None);
280 }
281 if let Some(outstanding) = &self.outstanding {
282 return Ok(Some(outstanding.prepared.clone()));
283 }
284 if self.active.is_none() {
285 let page = self
286 .database
287 .prepare_client_sync_page(self.server.clone(), MAX_PUSH_BATCH, MAX_PULL_BATCH)
288 .await?;
289 self.active = Some(ActivePage::new(page)?);
290 }
291
292 let active = self.active.as_ref().expect("active sync page");
293 if !active.contracts.is_empty() && !active.probed {
294 return self.prepare_json_request(
295 "POST",
296 "/sync/blobs/missing",
297 &MissingBlobsRequest {
298 blobs: active.contracts.clone(),
299 },
300 RequestKind::Missing,
301 );
302 }
303 if active.upload_index < active.uploads.len() {
304 let contract = active.uploads[active.upload_index].clone();
305 let upload = self
306 .database
307 .prepare_blob_upload(&self.blob_dir, &contract)
308 .await?;
309 let mut headers = vec![header("content-type", &contract.media_type)];
310 headers.push(header("x-aven-workspace-id", &contract.workspace_id));
311 headers.push(header("x-aven-byte-size", &contract.byte_size.to_string()));
312 headers.push(header("x-aven-width", &contract.width.to_string()));
313 headers.push(header("x-aven-height", &contract.height.to_string()));
314 return self.prepare_raw_request(
315 "PUT",
316 &format!("/sync/blobs/{}", contract.sha256),
317 headers,
318 upload.bytes,
319 RequestKind::Upload {
320 lease_id: upload.lease_id,
321 contract,
322 },
323 );
324 }
325 if !active.contracts.is_empty() && !active.confirmed {
326 return self.prepare_json_request(
327 "POST",
328 "/sync/blobs/missing",
329 &MissingBlobsRequest {
330 blobs: active.contracts.clone(),
331 },
332 RequestKind::Confirm,
333 );
334 }
335 if !active.metadata_sent {
336 let body = serde_json::to_vec(&active.page.request).context("encode sync request")?;
337 let request_bytes = body.len();
338 let mut headers = vec![header("content-type", "application/json")];
339 let body = if request_bytes > GZIP_THRESHOLD {
340 headers.push(header("content-encoding", "gzip"));
341 gzip_encode(&body)?
342 } else {
343 body
344 };
345 return self.prepare_raw_request(
346 "POST",
347 "/sync",
348 headers,
349 body,
350 RequestKind::Metadata { request_bytes },
351 );
352 }
353 if active.download_index < active.downloads.len() {
354 let blob = active.downloads[active.download_index].clone();
355 return self.prepare_raw_request(
356 "GET",
357 &format!("/sync/blobs/{}", blob.sha256),
358 vec![header("accept-encoding", "identity")],
359 Vec::new(),
360 RequestKind::Download { blob },
361 );
362 }
363
364 self.finish_page().await?;
365 if self.stopped {
366 Ok(None)
367 } else {
368 Box::pin(self.prepare_request()).await
369 }
370 }
371
372 fn prepare_json_request<T: serde::Serialize>(
373 &mut self,
374 method: &str,
375 path: &str,
376 value: &T,
377 kind: RequestKind,
378 ) -> Result<Option<PreparedSyncRequest>> {
379 self.prepare_raw_request(
380 method,
381 path,
382 vec![header("content-type", "application/json")],
383 serde_json::to_vec(value)?,
384 kind,
385 )
386 }
387
388 fn prepare_raw_request(
389 &mut self,
390 method: &str,
391 path: &str,
392 mut headers: Vec<SyncHttpHeader>,
393 body: Vec<u8>,
394 kind: RequestKind,
395 ) -> Result<Option<PreparedSyncRequest>> {
396 if let Some(token) = &self.auth_token {
397 headers.push(header("authorization", &format!("Bearer {token}")));
398 }
399 self.request_number += 1;
400 let prepared = PreparedSyncRequest {
401 method: method.to_string(),
402 url: format!("{}{}", self.server, path),
403 headers,
404 body,
405 context: SyncRequestContext {
406 session_id: self.session_id.clone(),
407 request: self.request_number,
408 },
409 };
410 self.outstanding = Some(OutstandingRequest {
411 prepared: prepared.clone(),
412 kind,
413 });
414 Ok(Some(prepared))
415 }
416
417 pub async fn accept_response(
418 &mut self,
419 context: &SyncRequestContext,
420 response: SyncHttpResponse,
421 ) -> Result<SyncPageOutcome> {
422 {
423 let outstanding = self
424 .outstanding
425 .as_ref()
426 .context("no outstanding sync request")?;
427 validate_context(&outstanding.prepared.context, context)?;
428 }
429 if !(200..300).contains(&response.status) {
430 bail!("{}", sync_http_error(&response));
431 }
432 let outstanding = self
433 .outstanding
434 .as_ref()
435 .expect("validated outstanding request");
436 let prepared_body_len = outstanding.prepared.body.len();
437 let request_kind = outstanding.kind.clone();
438 let response_bytes = response.body.len();
439 match request_kind {
440 RequestKind::Missing => {
441 let decoded: MissingBlobsResponse = serde_json::from_slice(&response.body)
442 .context("decode missing blobs response")?;
443 validate_blob_hashes(&decoded.missing)?;
444 let active = self.active.as_mut().expect("active sync page");
445 let requested = active
446 .contracts
447 .iter()
448 .map(|c| c.sha256.as_str())
449 .collect::<HashSet<_>>();
450 if decoded
451 .missing
452 .iter()
453 .any(|hash| !requested.contains(hash.as_str()))
454 {
455 bail!("error invalid-blob-missing-response");
456 }
457 let missing = decoded.missing.into_iter().collect::<HashSet<_>>();
458 self.known_server_blobs.extend(
459 active
460 .contracts
461 .iter()
462 .filter(|contract| !missing.contains(&contract.sha256))
463 .map(|contract| contract.sha256.clone()),
464 );
465 active.probed = true;
466 active.missing = missing;
467 let pending = active
468 .page
469 .request
470 .changes
471 .iter()
472 .map(|change| {
473 let blob = match attachment_blob_contract(change)? {
474 Some(contract) if active.missing.contains(&contract.sha256) => {
475 Some(TransferObject {
476 sha256: contract.sha256,
477 byte_size: u64::try_from(contract.byte_size)
478 .context("attachment bytes exceed u64")?,
479 })
480 }
481 _ => None,
482 };
483 Ok(PendingChange { missing_blob: blob })
484 })
485 .collect::<Result<Vec<_>>>()?;
486 let plan = plan_change_prefix(&pending, self.transfer_budget);
487 active.transfer_budget_blocked = plan.change_count < pending.len();
488 active.page.request.changes.truncate(plan.change_count);
489 active.page.pending = plan.change_count;
490 active.contracts = unique_blob_contracts(&active.page.request.changes)?;
491 active.uploads = plan
492 .transfers
493 .iter()
494 .map(|object| contract_by_hash(&active.contracts, &object.sha256))
495 .collect::<Result<Vec<_>>>()?;
496 }
497 RequestKind::Upload { lease_id, contract } => {
498 self.database.finish_blob_upload(&lease_id).await?;
499 let bytes =
500 u64::try_from(contract.byte_size).context("attachment bytes exceed u64")?;
501 self.known_server_blobs.insert(contract.sha256.clone());
502 self.transfer_budget.consume(&TransferObject {
503 sha256: contract.sha256,
504 byte_size: bytes,
505 });
506 let active = self.active.as_mut().expect("active sync page");
507 active.upload_index += 1;
508 active.page_uploaded += 1;
509 active.page_uploaded_bytes += bytes;
510 self.summary.blob_uploaded += 1;
511 self.summary.blob_uploaded_bytes += bytes;
512 }
513 RequestKind::Confirm => {
514 let decoded: MissingBlobsResponse = serde_json::from_slice(&response.body)
515 .context("decode missing blobs response")?;
516 validate_blob_hashes(&decoded.missing)?;
517 if !decoded.missing.is_empty() {
518 bail!("error attachment-blob-admission-missing");
519 }
520 self.active.as_mut().expect("active sync page").confirmed = true;
521 }
522 RequestKind::Metadata { request_bytes } => {
523 let decoded: SyncResponse =
524 serde_json::from_slice(&response.body).context("decode sync response")?;
525 let active = self.active.as_mut().expect("active sync page");
526 let request = active.page.request.clone();
527 let pending = active.page.pending;
528 let has_more = decoded.has_more;
529 let cursor = decoded.cursor;
530 let apply_started = Instant::now();
531 let pulled = self
532 .database
533 .apply_client_sync_page(ApplySyncPage {
534 request,
535 response: decoded,
536 attempted_at: self.attempted_at.clone(),
537 previous_pushed: self.summary.pushed,
538 previous_pulled: self.summary.pulled,
539 })
540 .await?;
541 let apply_ms = apply_started.elapsed().as_millis();
542 active.metadata_sent = true;
543 active.has_more = has_more;
544 active.page_pushed = pending;
545 active.page_pulled = pulled;
546 self.summary.pushed += pending as i64;
547 self.summary.pulled += pulled;
548 self.summary.cursor = cursor;
549 self.summary.request_bytes += request_bytes;
550 self.summary.request_wire_bytes += prepared_body_len;
551 self.summary.response_decoded_bytes += response_bytes;
552 self.summary.response_compression =
553 header_value(&response.headers, "content-encoding")
554 .unwrap_or("none")
555 .to_string();
556 self.summary.apply_ms += apply_ms;
557 let missing = self.database.missing_local_blobs(&self.blob_dir).await?;
558 let objects = missing
559 .iter()
560 .map(|blob| {
561 Ok(TransferObject {
562 sha256: blob.sha256.clone(),
563 byte_size: u64::try_from(blob.byte_size)
564 .context("attachment bytes exceed u64")?,
565 })
566 })
567 .collect::<Result<Vec<_>>>()?;
568 let plan = plan_transfers(&objects, self.transfer_budget);
569 active.transfer_budget_blocked |= plan.len() < objects.len();
570 let planned = plan
571 .iter()
572 .map(|o| o.sha256.as_str())
573 .collect::<HashSet<_>>();
574 active.downloads = missing
575 .into_iter()
576 .filter(|b| planned.contains(b.sha256.as_str()))
577 .collect();
578 }
579 RequestKind::Download { blob } => {
580 let expected =
581 usize::try_from(blob.byte_size).context("attachment bytes exceed usize")?;
582 let content_length_valid = header_value(&response.headers, "content-length")
583 .map(|length| length.parse::<usize>().ok() == Some(expected))
584 .unwrap_or(true);
585 if !content_length_valid || response.body.len() != expected {
586 bail!("error attachment-blob-remote-invalid");
587 }
588 self.database
589 .store_downloaded_blob(
590 &self.blob_dir,
591 self.lifecycle_policy,
592 &blob,
593 response.body,
594 )
595 .await?;
596 let bytes = u64::try_from(blob.byte_size).context("attachment bytes exceed u64")?;
597 self.transfer_budget.consume(&TransferObject {
598 sha256: blob.sha256,
599 byte_size: bytes,
600 });
601 let active = self.active.as_mut().expect("active sync page");
602 active.download_index += 1;
603 active.page_downloaded += 1;
604 active.page_downloaded_bytes += bytes;
605 self.summary.blob_downloaded += 1;
606 self.summary.blob_downloaded_bytes += bytes;
607 }
608 }
609 self.outstanding = None;
610 let page_finished = self.active.as_ref().is_some_and(|active| {
611 active.metadata_sent && active.download_index >= active.downloads.len()
612 });
613 if page_finished {
614 let mut outcome = self.outcome();
615 self.finish_page().await?;
616 outcome.page = self.summary.pages;
617 outcome.complete = self.summary.complete;
618 outcome.has_more = self.last_has_more;
619 outcome.local_more = self.last_local_more;
620 return Ok(outcome);
621 }
622 Ok(self.outcome())
623 }
624
625 async fn finish_page(&mut self) -> Result<()> {
626 let active = self.active.take().expect("active sync page");
627 self.summary.pages += 1;
628 self.last_has_more = active.has_more;
629 let local_more = self.database.pending_sync_change_count().await? > 0;
630 self.last_local_more = local_more;
631 let downloads = self.database.missing_local_blobs(&self.blob_dir).await?;
632 let download_counts = missing_counts(&downloads);
633 self.summary.blob_download_remaining = download_counts.count as usize;
634 self.summary.blob_download_remaining_bytes = download_counts.bytes;
635 let uploads = self
636 .database
637 .pending_blob_contracts()
638 .await?
639 .into_iter()
640 .filter(|contract| !self.known_server_blobs.contains(&contract.sha256))
641 .collect::<Vec<_>>();
642 self.summary.blob_upload_remaining = uploads.len();
643 self.summary.blob_upload_remaining_bytes = uploads
644 .iter()
645 .map(|c| u64::try_from(c.byte_size).unwrap_or(0))
646 .sum();
647 let complete = !local_more && !active.has_more && downloads.is_empty();
648 self.summary.complete = complete;
649 let budget_exhausted = self
650 .page_budget
651 .is_some_and(|budget| self.summary.pages >= budget);
652 let transfer_stalled = active.transfer_budget_blocked
653 || self.transfer_budget.objects == 0
654 || self.transfer_budget.bytes == 0;
655 self.stopped = complete || budget_exhausted || transfer_stalled;
656 Ok(())
657 }
658
659 fn outcome(&self) -> SyncPageOutcome {
660 let active = self.active.as_ref();
661 SyncPageOutcome {
662 page: self.summary.pages + usize::from(active.is_some_and(|page| page.metadata_sent)),
663 pushed: active.map_or(0, |page| page.page_pushed),
664 pulled: active.map_or(0, |page| page.page_pulled),
665 blob_uploaded: active.map_or(0, |page| page.page_uploaded),
666 blob_uploaded_bytes: active.map_or(0, |page| page.page_uploaded_bytes),
667 blob_downloaded: active.map_or(0, |page| page.page_downloaded),
668 blob_downloaded_bytes: active.map_or(0, |page| page.page_downloaded_bytes),
669 cursor: self.summary.cursor,
670 complete: self.summary.complete,
671 has_more: active.map_or(self.last_has_more, |page| page.has_more),
672 local_more: self.last_local_more,
673 request_bytes: self.summary.request_bytes,
674 request_wire_bytes: self.summary.request_wire_bytes,
675 response_decoded_bytes: self.summary.response_decoded_bytes,
676 response_compression: self.summary.response_compression.clone(),
677 apply_ms: self.summary.apply_ms,
678 }
679 }
680
681 pub async fn fail_request(
682 &mut self,
683 context: &SyncRequestContext,
684 error: impl Into<String>,
685 ) -> Result<()> {
686 let outstanding = self
687 .outstanding
688 .as_ref()
689 .context("no outstanding sync request")?;
690 validate_context(&outstanding.prepared.context, context)?;
691 if let RequestKind::Upload { lease_id, .. } = &outstanding.kind {
692 self.database.finish_blob_upload(lease_id).await?;
693 }
694 self.database.record_sync_error(error.into()).await?;
695 self.outstanding = None;
696 self.stopped = true;
697 Ok(())
698 }
699
700 pub fn summary(&self) -> SyncSessionSummary {
701 self.summary.clone()
702 }
703}
704
705fn header(name: &str, value: &str) -> SyncHttpHeader {
706 SyncHttpHeader {
707 name: name.to_string(),
708 value: value.to_string(),
709 }
710}
711
712fn validate_context(expected: &SyncRequestContext, actual: &SyncRequestContext) -> Result<()> {
713 if expected != actual {
714 bail!("sync request context does not match the outstanding page");
715 }
716 Ok(())
717}
718
719fn header_value<'a>(headers: &'a [SyncHttpHeader], name: &str) -> Option<&'a str> {
720 headers
721 .iter()
722 .find(|header| header.name.eq_ignore_ascii_case(name))
723 .map(|header| header.value.as_str())
724}
725
726fn sync_http_error(response: &SyncHttpResponse) -> String {
727 let detail = header_value(&response.headers, "content-type")
728 .filter(|content_type| {
729 content_type
730 .split(';')
731 .next()
732 .is_some_and(|value| value.trim().eq_ignore_ascii_case("text/plain"))
733 })
734 .and_then(|_| std::str::from_utf8(&response.body).ok())
735 .map(str::trim)
736 .filter(|detail| {
737 detail.starts_with("error ")
738 && detail.len() <= MAX_HTTP_ERROR_DETAIL_BYTES
739 && detail.is_ascii()
740 && !detail.chars().any(char::is_control)
741 });
742
743 let detail = match detail {
744 Some(detail) => actionable_sync_http_error_detail(detail).unwrap_or_else(|| detail.to_string()),
745 None => match response.status {
746 400 => {
747 "server rejected the sync request; check server logs and verify client/server version compatibility"
748 }
749 401 => "authentication failed; check sync.auth_token",
750 403 => "server denied the sync request; check server access settings",
751 404 => "sync endpoint not found; check sync.server_url",
752 429 => "server rate limit exceeded; retry later",
753 507 => "error attachment-quota-exceeded",
754 500..=599 => "sync server failed; check server logs",
755 _ => "server returned no readable error details",
756 }
757 .to_string(),
758 };
759 format!(
760 "sync request failed with HTTP status {}: {detail}",
761 response.status
762 )
763}
764
765fn actionable_sync_http_error_detail(detail: &str) -> Option<String> {
766 let fields = detail.strip_prefix("error sync-protocol-unsupported ")?;
767 let mut fields = fields.split_whitespace();
768 let client = fields
769 .next()?
770 .strip_prefix("client=")?
771 .parse::<u32>()
772 .ok()?;
773 let server = fields
774 .next()?
775 .strip_prefix("server=")?
776 .parse::<u32>()
777 .ok()?;
778 if fields.next().is_some() || client == server {
779 return None;
780 }
781 let action = if client > server {
782 "upgrade the sync server"
783 } else {
784 "upgrade the sync client"
785 };
786 Some(format!(
787 "client and server use incompatible sync protocol versions (client {client}, server {server}); {action}"
788 ))
789}
790
791fn gzip_encode(body: &[u8]) -> Result<Vec<u8>> {
792 let mut encoder = GzEncoder::new(Vec::new(), flate2::Compression::default());
793 encoder
794 .write_all(body)
795 .context("gzip encode sync request body")?;
796 encoder.finish().context("finish sync request compression")
797}