Skip to main content

max / synckit

Resume an interrupted multipart blob upload blob_upload_streaming takes over a session a killed process left behind and sends only the parts that never landed. No change to the call: the resume key is the content hash it already takes. The boundary chunk is the whole difficulty. Part boundaries are server-supplied and do not align to the 1 MiB sealed chunks, so a resume restarts inside a chunk whose head is already at S3, and sealing draws a random nonce per chunk. Store the nonce and the chunk plaintext's digest, verify the digest, then reseal byte-identically. The digest is not optional: resealing different plaintext under a used nonce is keystream reuse, which leaks the XOR of the two chunks. A transient failure now keeps the session rather than aborting it, since that is what the next attempt reuses; the server's 24h reaper stays the backstop. A failed resume drops the record so a reaped session cannot wedge a blob forever. BlobResumeStore lives in client::resume with no rusqlite, so the no-store build is unaffected; SqliteResumeStore implements it and the blob pass installs it through a defaulted BlobTransport method. Nothing outside this crate implements that trait, so no consumer changes.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-08 20:03 UTC
Signed with PGP, not checked
Commit: 6fe7c78a09c3c46a540fab6671347277bc3af0ac
Parent: 9caecf6
9 files changed, +1439 insertions, -18 deletions
M synckit-client/src/crypto.rs +84
@@ -417,7 +417,24 @@
417 417 pub(crate) fn seal(plaintext: &[u8], master_key: &[u8; KEY_SIZE], aad: &[u8]) -> Result<Vec<u8>> {
418 418 let mut nonce_bytes = [0u8; NONCE_SIZE];
419 419 rand::rng().fill_bytes(&mut nonce_bytes);
420 + seal_with_nonce(plaintext, master_key, aad, &nonce_bytes)
421 + }
420 422
423 + /// [`seal`] with the nonce supplied rather than drawn.
424 + ///
425 + /// The ONLY legitimate caller is one reproducing a ciphertext it already
426 + /// produced, byte for byte, from the identical plaintext, key and AAD (see
427 + /// [`reseal_blob_chunk`]). Sealing *different* plaintext under a nonce this key
428 + /// has already used reuses the XChaCha20 keystream, which leaks the XOR of the
429 + /// two plaintexts and forges the Poly1305 key. Private for that reason: nothing
430 + /// outside this module may choose a nonce.
431 + fn seal_with_nonce(
432 + plaintext: &[u8],
433 + master_key: &[u8; KEY_SIZE],
434 + aad: &[u8],
435 + nonce_bytes: &[u8; NONCE_SIZE],
436 + ) -> Result<Vec<u8>> {
437 + let nonce_bytes = *nonce_bytes;
421 438 let cipher = XChaCha20Poly1305::new(master_key.into());
422 439 let nonce = XNonce::from(nonce_bytes);
423 440
@@ -674,6 +691,73 @@
674 691 seal(chunk, master_key, &blob_chunk_aad(hash, index, chunk_count))
675 692 }
676 693
694 + /// Length of the per-chunk nonce that opens every sealed chunk.
695 + ///
696 + /// Public because a resumable uploader has to persist one nonce per chunk to
697 + /// reproduce ciphertext across a process restart. Nonces are not secret: they
698 + /// ride in the clear at the head of every sealed chunk already.
699 + pub const BLOB_NONCE_LEN: usize = NONCE_SIZE;
700 +
701 + /// The nonce a sealed chunk was sealed with, read off its head.
702 + ///
703 + /// The wire layout is `nonce || ciphertext || tag`, so this is the front of
704 + /// [`seal_blob_chunk`]'s output. Named rather than sliced at the call site so
705 + /// the layout stays knowledge of this module.
706 + pub fn blob_chunk_nonce(sealed: &[u8]) -> Result<[u8; BLOB_NONCE_LEN]> {
707 + sealed
708 + .get(..BLOB_NONCE_LEN)
709 + .and_then(|n| <[u8; BLOB_NONCE_LEN]>::try_from(n).ok())
710 + .ok_or_else(|| SyncKitError::Crypto("sealed chunk is shorter than its nonce".into()))
711 + }
712 +
713 + /// Re-seal a chunk to the exact bytes [`seal_blob_chunk`] produced for it, by
714 + /// supplying the nonce that call drew.
715 + ///
716 + /// This exists for one job: resuming an interrupted multipart upload. Part
717 + /// boundaries are server-supplied and do not align to [`BLOB_CHUNK_SIZE`], so a
718 + /// resume generally restarts mid-chunk, with the chunk's leading bytes already
719 + /// uploaded under the old nonce. Sealing that chunk again with a fresh nonce
720 + /// would splice two keystreams into one chunk and the assembled object would
721 + /// never open.
722 + ///
723 + /// # Correctness and safety
724 + ///
725 + /// `chunk` MUST be byte-identical to the plaintext originally sealed under
726 + /// `nonce`. Re-using a nonce across *different* plaintext under the same key
727 + /// destroys confidentiality and authenticity outright, so the caller is
728 + /// obliged to verify the plaintext before calling (the upload path stores a
729 + /// per-chunk plaintext digest alongside the nonce and checks it). `index` and
730 + /// `chunk_count` must also match, or the AAD differs and the result is simply a
731 + /// chunk that fails to open.
732 + pub fn reseal_blob_chunk(
733 + chunk: &[u8],
734 + master_key: &[u8; KEY_SIZE],
735 + hash: &str,
736 + index: u32,
737 + chunk_count: u32,
738 + nonce: &[u8; BLOB_NONCE_LEN],
739 + ) -> Result<Vec<u8>> {
740 + seal_with_nonce(
741 + chunk,
742 + master_key,
743 + &blob_chunk_aad(hash, index, chunk_count),
744 + nonce,
745 + )
746 + }
747 +
748 + /// Sealed length of chunk `index` of a `plaintext_len`-byte blob.
749 + ///
750 + /// The uploader needs this to map a ciphertext byte offset (a part boundary)
751 + /// back to the chunk that spans it, without a header to parse.
752 + pub fn sealed_blob_chunk_len(plaintext_len: usize, index: u32) -> usize {
753 + BlobHeader {
754 + chunk_size: BLOB_CHUNK_SIZE,
755 + total_len: plaintext_len,
756 + chunk_count: blob_chunk_count_for(plaintext_len),
757 + }
758 + .sealed_chunk_len(index)
759 + }
760 +
677 761 /// Parse the v3 header, returning it plus the number of bytes consumed (tag +
678 762 /// header). The remaining input is the sealed-chunk stream.
679 763 pub fn parse_blob_header(data: &[u8]) -> Result<(BlobHeader, usize)> {
M synckit-client/src/client/blob.rs +375 -15
@@ -28,12 +28,75 @@
28 28
29 29 use super::SyncKitClient;
30 30 use super::helpers::{Idempotency, check_response};
31 + use super::resume::{BlobResumeStore, ResumeChunk, ResumePart, ResumeRecord, ResumeSession};
31 32
32 33 /// Upper bound on a single decrypted blob held in memory, guarding against a
33 34 /// hostile server returning an absurdly large body that would OOM the client.
34 35 /// Generous (4 GiB) so legitimate large media still flow.
35 36 const MAX_BLOB_BYTES: usize = 4 * 1024 * 1024 * 1024;
36 37
38 + /// How old a recorded multipart session may be before a resume is not worth
39 + /// attempting.
40 + ///
41 + /// The server's orphan reaper aborts abandoned sessions at 24 hours, after
42 + /// which the `upload_id` is gone and every part URL request against it fails.
43 + /// Half that leaves room for a slow transfer to finish inside the window while
44 + /// keeping a record that is certainly dead from costing a round trip.
45 + const RESUME_MAX_AGE_SECS: i64 = 12 * 60 * 60;
46 +
47 + /// Whether an interrupted transfer is worth keeping the session and its resume
48 + /// record for.
49 + ///
50 + /// Anything that will fail the same way next time is not: a file that no longer
51 + /// hashes to the address it is being stored under, a plan the two sides
52 + /// disagree about, a missing key. Those abort the session exactly as they did
53 + /// before resume existed. What is left is network and session-lifetime
54 + /// failure, which is the whole of what resuming is for.
55 + fn is_resumable_failure(e: &SyncKitError) -> bool {
56 + !matches!(
57 + e,
58 + SyncKitError::IntegrityFailed { .. }
59 + | SyncKitError::InvalidArgument(_)
60 + | SyncKitError::Internal(_)
61 + | SyncKitError::Crypto(_)
62 + | SyncKitError::NoMasterKey
63 + )
64 + }
65 +
66 + /// Run a resume-store write, logging rather than propagating a failure.
67 + ///
68 + /// The store is an optimisation: losing it costs a restart from zero, which is
69 + /// what every upload did before it existed. Nothing about it may be able to
70 + /// fail an upload.
71 + fn best_effort(what: &'static str, r: Result<()>) {
72 + if let Err(e) = r {
73 + tracing::warn!("blob resume store: {what} failed, uploads will not resume: {e}");
74 + }
75 + }
76 +
77 + /// The chunk a ciphertext byte offset lands in, and how far into that chunk's
78 + /// sealed bytes it falls.
79 + ///
80 + /// A resume restarts at a part boundary, and part boundaries are chosen by the
81 + /// server with no reference to the 1 MiB sealed-chunk geometry, so the boundary
82 + /// generally falls mid-chunk. Returns `(chunk_count, 0)` when the offset is past
83 + /// the last chunk, i.e. every part was already uploaded.
84 + fn resume_boundary(plaintext_len: usize, header_len: usize, skip: usize) -> (u32, usize) {
85 + let chunk_count = crypto::blob_chunk_count_for(plaintext_len);
86 + if skip <= header_len {
87 + return (0, 0);
88 + }
89 + let mut pos = header_len;
90 + for index in 0..chunk_count {
91 + let len = crypto::sealed_blob_chunk_len(plaintext_len, index);
92 + if pos + len > skip {
93 + return (index, skip - pos);
94 + }
95 + pos += len;
96 + }
97 + (chunk_count, 0)
98 + }
99 +
37 100 /// What [`SyncKitClient::blob_upload_streaming`] did.
38 101 ///
39 102 /// The caller needs to tell the two apart: only an upload that sent bytes has
@@ -167,6 +230,18 @@
167 230 /// already held the content, nothing was read or sent, and there is nothing
168 231 /// to confirm. The dedup check happens before any file read or sealing, so
169 232 /// a re-sync of content the server already has costs one round trip.
233 + ///
234 + /// ## Resuming
235 + ///
236 + /// If a resume store is installed
237 + /// ([`set_resume_store`](Self::set_resume_store), which the SyncStore
238 + /// engine does for you) and a previous attempt on this same `hash` died
239 + /// partway, this call takes that session over and sends only the parts that
240 + /// never landed. Nothing about the call changes: the resume key is `hash`,
241 + /// which the caller already passes, so retrying a killed multi-gigabyte
242 + /// upload is the same line of code as starting one. Without a resume store,
243 + /// or when the recorded session is too old to still exist server-side, the
244 + /// upload starts from the beginning as it always has.
170 245 #[instrument(skip(self, path), fields(path = %path.display()))]
171 246 pub async fn blob_upload_streaming(
172 247 &self,
@@ -207,33 +282,112 @@
207 282 })
208 283 .await?;
209 284
285 + let resume_store = self.resume_store();
286 +
210 287 if start.already_exists {
288 + // Another device got there first. Whatever we were part-way through
289 + // uploading is moot, and its session is not ours to keep.
290 + if let Some(store) = &resume_store {
291 + best_effort("clear", store.clear(hash));
292 + }
211 293 return Ok(BlobUploadOutcome::AlreadyPresent);
212 294 }
213 295
296 + // Take over an earlier session if one is on file and still describes
297 + // this blob, otherwise open the book on the one just started.
298 + let record = resume_store
299 + .as_ref()
300 + .and_then(|store| Self::load_resume(store.as_ref(), hash, size_bytes));
301 + let session = match &record {
302 + Some(record) => {
303 + // `start` is unconditional (it carries the dedup answer), so a
304 + // resume has just opened a second session against the same key.
305 + // Release it rather than leave S3 holding two.
306 + if let Err(e) = self.blob_multipart_abort(hash, &start.upload_id).await {
307 + tracing::warn!(error = %e, "failed to release the redundant multipart session");
308 + }
309 + tracing::info!(
310 + parts_done = record.usable_parts().len(),
311 + of = record.session.part_count,
312 + "resuming an interrupted blob upload"
313 + );
314 + BlobMultipartStartResponse {
315 + upload_id: record.session.upload_id.clone(),
316 + part_size: record.session.part_size,
317 + part_count: record.session.part_count,
318 + already_exists: false,
319 + }
320 + }
321 + None => {
322 + if let Some(store) = &resume_store {
323 + best_effort(
324 + "begin",
325 + store.begin(
326 + hash,
327 + &ResumeSession {
328 + upload_id: start.upload_id.clone(),
329 + part_size: start.part_size,
330 + part_count: start.part_count,
331 + size_bytes: size_bytes as u64,
332 + },
333 + ),
334 + );
335 + }
336 + start
337 + }
338 + };
339 +
214 340 // Every failure past this point leaves uploaded parts that S3 bills for
215 341 // until the session is aborted, so the abort lives in exactly one place:
216 342 // here, around the whole transfer. The server-side orphan reaper is the
217 343 // backstop for a client that dies outright.
218 344 let transfer = self
219 - .stream_blob_parts(hash, &start, path, plaintext_len, size_bytes, &master_key)
345 + .stream_blob_parts(
346 + hash,
347 + &session,
348 + path,
349 + plaintext_len,
350 + size_bytes,
351 + &master_key,
352 + record.as_ref(),
353 + resume_store.as_deref(),
354 + )
220 355 .await;
221 356 let parts = match transfer {
222 357 Ok(parts) => parts,
223 358 Err(e) => {
224 - if let Err(abort_err) = self.blob_multipart_abort(hash, &start.upload_id).await {
359 + // With a resume store the session is the asset, not the leak:
360 + // holding it is what lets the next attempt send only what is
361 + // missing, and the 24h reaper still frees it if there is no
362 + // next attempt. Only a failure that will recur is aborted.
363 + //
364 + // A resume that itself failed is such a failure. The session may
365 + // be one the server has already reaped, in which case every
366 + // future attempt would resume into the same wall; so a second
367 + // failure drops the record and the next attempt starts clean.
368 + if record.is_none() && resume_store.is_some() && is_resumable_failure(&e) {
369 + tracing::info!(
370 + error = %e,
371 + "keeping the multipart session open so the next attempt resumes"
372 + );
373 + return Err(e);
374 + }
375 + if let Err(abort_err) = self.blob_multipart_abort(hash, &session.upload_id).await {
225 376 tracing::warn!(
226 377 error = %abort_err,
227 378 "failed to abort multipart blob session; the server reaper will release it"
228 379 );
229 380 }
381 + if let Some(store) = &resume_store {
382 + best_effort("clear", store.clear(hash));
383 + }
230 384 return Err(e);
231 385 }
232 386 };
233 387
234 388 let complete_body = Bytes::from(serde_json::to_vec(&BlobMultipartCompleteRequest {
235 389 hash: hash.to_string(),
236 - upload_id: start.upload_id.clone(),
390 + upload_id: session.upload_id.clone(),
237 391 parts,
238 392 })?);
239 393 self.retry_request(
@@ -252,9 +406,52 @@
252 406 )
253 407 .await?;
254 408
409 + // Assembled: the session no longer exists and the record describes
410 + // nothing.
411 + if let Some(store) = &resume_store {
412 + best_effort("clear", store.clear(hash));
413 + }
255 414 Ok(BlobUploadOutcome::Uploaded)
256 415 }
257 416
417 + /// The resume record for `hash`, if there is one worth acting on.
418 + ///
419 + /// Rejects, and forgets, a record that cannot describe this upload: a
420 + /// different ciphertext length (the file changed), a plan whose parts do
421 + /// not tile it, or a session old enough that the server has reaped it. Each
422 + /// of those would otherwise cost a doomed transfer before failing.
423 + fn load_resume(
424 + store: &dyn BlobResumeStore,
425 + hash: &str,
426 + size_bytes: i64,
427 + ) -> Option<ResumeRecord> {
428 + let record = match store.load(hash) {
429 + Ok(record) => record?,
430 + Err(e) => {
431 + tracing::warn!("blob resume store: load failed, starting from zero: {e}");
432 + return None;
433 + }
434 + };
435 +
436 + let session = &record.session;
437 + let stale = record.age_secs > RESUME_MAX_AGE_SECS;
438 + let fits = session.size_bytes == size_bytes as u64
439 + && session.part_size > 0
440 + && session.part_count > 0
441 + && (session.size_bytes.div_ceil(session.part_size)) == u64::from(session.part_count);
442 + if stale || !fits || record.usable_parts().is_empty() {
443 + // An empty run is not a failure, it just has nothing to save: the
444 + // upload starts at part 1 either way, and dropping the record here
445 + // means the fresh session is the one recorded.
446 + if stale || !fits {
447 + tracing::debug!(stale, fits, "discarding an unusable blob resume record");
448 + }
449 + best_effort("clear", store.clear(hash));
450 + return None;
451 + }
452 + Some(record)
453 + }
454 +
258 455 /// Seal the file chunk by chunk, cutting the ciphertext at the server's
259 456 /// signed part boundaries and PUTting each part. Returns the completed
260 457 /// `(part_number, etag)` pairs.
@@ -262,6 +459,13 @@
262 459 /// Peak memory is one part plus one chunk: the staging buffer never holds
263 460 /// more than a part's worth, because a full part is drained and sent the
264 461 /// moment it is complete.
462 + ///
463 + /// With a `resume` record the parts it names are taken as already at S3 and
464 + /// the stream picks up at the first missing one. The whole file is still
465 + /// read and hashed: the content-address check is over the plaintext, and a
466 + /// file that changed between attempts must fail it rather than be assembled
467 + /// out of two different files.
468 + #[allow(clippy::too_many_arguments)]
265 469 async fn stream_blob_parts(
266 470 &self,
267 471 hash: &str,
@@ -270,6 +474,8 @@
270 474 plaintext_len: usize,
271 475 size_bytes: i64,
272 476 master_key: &[u8; 32],
477 + resume: Option<&ResumeRecord>,
478 + resume_store: Option<&dyn BlobResumeStore>,
273 479 ) -> Result<Vec<BlobMultipartCompletedPart>> {
274 480 let part_size = usize::try_from(start.part_size).map_err(|_| {
275 481 SyncKitError::Internal(format!(
@@ -319,15 +525,36 @@
319 525 SyncKitError::Internal(format!("open blob file {}: {e}", path.display()))
320 526 })?;
321 527
528 + // Where the resume picks up, in parts and then in ciphertext bytes.
529 + let kept: Vec<BlobMultipartCompletedPart> = resume
530 + .map(|r| {
531 + r.usable_parts()
532 + .iter()
533 + .map(|p| BlobMultipartCompletedPart {
534 + part_number: p.part_number as i32,
535 + etag: p.etag.clone(),
536 + })
537 + .collect()
538 + })
539 + .unwrap_or_default();
540 + let mut next_part = kept.len() as u32 + 1;
541 + let skip = (next_part as usize - 1) * part_size;
542 + let header = crypto::blob_header_bytes(plaintext_len);
543 + let (first_chunk, within) = resume_boundary(plaintext_len, header.len(), skip);
544 +
322 545 let chunk_count = crypto::blob_chunk_count_for(plaintext_len);
323 546 let mut hasher = Sha256::new();
324 547 // Plaintext scratch: zeroized on drop so a decrypted chunk does not linger.
325 548 let mut plain = zeroize::Zeroizing::new(vec![0u8; crypto::BLOB_CHUNK_SIZE]);
326 549 let mut staged = BytesMut::with_capacity(part_size + crypto::BLOB_CHUNK_SIZE);
327 - staged.extend_from_slice(&crypto::blob_header_bytes(plaintext_len));
550 + if skip == 0 {
551 + staged.extend_from_slice(&header);
552 + }
328 553
329 - let mut completed = Vec::with_capacity(start.part_count as usize);
330 - let mut next_part = 1u32;
554 + let mut completed = kept;
555 + // Chunk records sealed since the last completed part, flushed with it so
556 + // a nonce is only ever on file for ciphertext that is durable at S3.
557 + let mut pending_chunks: Vec<ResumeChunk> = Vec::new();
331 558
332 559 for index in 0..chunk_count {
333 560 let offset = index as usize * crypto::BLOB_CHUNK_SIZE;
@@ -339,13 +566,55 @@
339 566 ))
340 567 })?;
341 568 hasher.update(&plain[..want]);
342 - staged.extend_from_slice(&crypto::seal_blob_chunk(
343 - &plain[..want],
344 - master_key,
345 - hash,
346 - index,
347 - chunk_count,
348 - )?);
569 + // Chunks below the boundary are read only to keep the content-address
570 + // check over the whole file; their ciphertext is already at S3.
571 + if index < first_chunk {
572 + continue;
573 + }
574 + // Only paid for when there is a store to record it in: without one
575 + // nothing resumes, so the per-chunk digest buys nothing.
576 + let plain_sha: Option<[u8; 32]> = resume_store
577 + .is_some()
578 + .then(|| Sha256::digest(&plain[..want]).into());
579 +
580 + let sealed = if index == first_chunk && within > 0 {
581 + // This chunk straddles the boundary: its first `within` bytes
582 + // are inside the last completed part. Sealing draws a random
583 + // nonce, so a fresh seal here would splice two keystreams into
584 + // one chunk and the assembled object would never open. Reproduce
585 + // it from the recorded nonce instead.
586 + let chunk = resume.and_then(|r| r.chunk(index)).ok_or_else(|| {
587 + SyncKitError::Internal(format!(
588 + "resume needs the nonce for chunk {index} and none was recorded"
589 + ))
590 + })?;
591 + // Re-using a nonce over different plaintext would be far worse
592 + // than a failed upload, so this is a hard gate rather than a
593 + // best guess: the file must be the one that was sealed.
594 + if Some(chunk.plain_sha) != plain_sha {
595 + return Err(SyncKitError::Internal(format!(
596 + "blob file changed under an in-flight upload (chunk {index})"
597 + )));
598 + }
599 + crypto::reseal_blob_chunk(
600 + &plain[..want],
601 + master_key,
602 + hash,
603 + index,
604 + chunk_count,
605 + &chunk.nonce,
606 + )?
607 + } else {
608 + crypto::seal_blob_chunk(&plain[..want], master_key, hash, index, chunk_count)?
609 + };
610 + if let Some(plain_sha) = plain_sha {
611 + pending_chunks.push(ResumeChunk {
612 + index,
613 + nonce: crypto::blob_chunk_nonce(&sealed)?,
614 + plain_sha,
615 + });
616 + }
617 + staged.extend_from_slice(&sealed[if index == first_chunk { within } else { 0 }..]);
349 618
350 619 // Every part but the last is exactly `part_size`; the remainder is
351 620 // the final part, so it is never sent from inside this loop.
@@ -356,8 +625,22 @@
356 625 .await?;
357 626 completed.push(BlobMultipartCompletedPart {
358 627 part_number: next_part as i32,
359 - etag,
628 + etag: etag.clone(),
360 629 });
630 + if let Some(store) = resume_store {
631 + best_effort(
632 + "record_part",
633 + store.record_part(
634 + hash,
635 + &ResumePart {
636 + part_number: next_part,
637 + etag,
638 + },
639 + &pending_chunks,
640 + ),
641 + );
642 + pending_chunks.clear();
643 + }
361 644 next_part += 1;
362 645 }
363 646 }
@@ -373,6 +656,19 @@
373 656 });
374 657 }
375 658
659 + if next_part > start.part_count {
660 + // A resume that found every part already uploaded: the previous
661 + // attempt died between the last part and the assemble call, so
662 + // there is nothing left to send.
663 + if completed.len() != start.part_count as usize {
664 + return Err(SyncKitError::Internal(format!(
665 + "resume holds {} parts, server planned {}",
666 + completed.len(),
667 + start.part_count
668 + )));
669 + }
670 + return Ok(completed);
671 + }
376 672 if next_part != start.part_count {
377 673 return Err(SyncKitError::Internal(format!(
378 674 "sealed stream produced {} parts, server planned {}",
@@ -385,8 +681,21 @@
385 681 .await?;
386 682 completed.push(BlobMultipartCompletedPart {
387 683 part_number: next_part as i32,
388 - etag,
684 + etag: etag.clone(),
389 685 });
686 + if let Some(store) = resume_store {
687 + best_effort(
688 + "record_part",
689 + store.record_part(
690 + hash,
691 + &ResumePart {
692 + part_number: next_part,
693 + etag,
694 + },
695 + &pending_chunks,
696 + ),
697 + );
698 + }
390 699
391 700 Ok(completed)
392 701 }
@@ -664,6 +973,225 @@
664 973 mod tests {
665 974 use crate::types::*;
666 975
976 + mod resume {
977 + use super::super::*;
978 + use std::sync::Mutex;
979 +
980 + /// A store that answers with whatever the test put in it.
981 + #[derive(Default)]
982 + struct Fake {
983 + record: Mutex<Option<ResumeRecord>>,
984 + cleared: Mutex<bool>,
985 + }
986 + impl BlobResumeStore for Fake {
987 + fn load(&self, _hash: &str) -> Result<Option<ResumeRecord>> {
988 + Ok(self.record.lock().unwrap().clone())
989 + }
990 + fn begin(&self, _hash: &str, _session: &ResumeSession) -> Result<()> {
991 + Ok(())
992 + }
993 + fn record_part(
994 + &self,
995 + _hash: &str,
996 + _part: &ResumePart,
997 + _chunks: &[ResumeChunk],
998 + ) -> Result<()> {
999 + Ok(())
1000 + }
1001 + fn clear(&self, _hash: &str) -> Result<()> {
1002 + *self.cleared.lock().unwrap() = true;
1003 + Ok(())
1004 + }
1005 + }
1006 +
1007 + /// A plausible session: 3 parts of 8 bytes over a 24-byte ciphertext,
1008 + /// with the first part done.
1009 + fn fake(age_secs: i64) -> Fake {
1010 + Fake {
1011 + record: Mutex::new(Some(ResumeRecord {
1012 + session: ResumeSession {
1013 + upload_id: "u".into(),
1014 + part_size: 8,
1015 + part_count: 3,
1016 + size_bytes: 24,
1017 + },
1018 + age_secs,
1019 + parts: vec![ResumePart {
1020 + part_number: 1,
1021 + etag: "e".into(),
1022 + }],
1023 + chunks: vec![],
1024 + })),
1025 + cleared: Mutex::new(false),
1026 + }
Lines truncated
M synckit-client/src/client/mod.rs +29
@@ -59,6 +59,7 @@
59 59 mod encryption;
60 60 pub(crate) mod helpers;
61 61 mod ota;
62 + pub mod resume;
62 63 mod rotation;
63 64 mod subscribe;
64 65 pub mod subscription;
@@ -383,6 +384,13 @@
383 384 /// once, and each needs its own key. A decrypt failure drops every generation
384 385 /// for the group, forcing a re-fetch.
385 386 gck_cache: RwLock<std::collections::HashMap<(GroupId, i32), crypto::ZeroizeOnDrop>>,
387 + /// Where an interrupted multipart blob upload is remembered, if anywhere.
388 + ///
389 + /// `None` unless something installs one (the SyncStore engine does, at the
390 + /// start of each blob pass). Without it a killed upload restarts from zero,
391 + /// which is the behaviour this client always had. See
392 + /// [`resume`](crate::client::resume).
393 + resume_store: RwLock<Option<Arc<dyn resume::BlobResumeStore>>>,
386 394 }
387 395
388 396 impl SyncKitClient {
@@ -441,6 +449,7 @@
441 449 master_key_id: RwLock::new(1),
442 450 pending_key: RwLock::new(None),
443 451 gck_cache: RwLock::new(std::collections::HashMap::new()),
452 + resume_store: RwLock::new(None),
444 453 }
445 454 }
446 455
@@ -462,9 +471,29 @@
462 471 master_key_id: RwLock::new(1),
463 472 pending_key: RwLock::new(None),
464 473 gck_cache: RwLock::new(std::collections::HashMap::new()),
474 + resume_store: RwLock::new(None),
465 475 }
466 476 }
467 477
478 + /// Give the client somewhere durable to remember an interrupted multipart
479 + /// blob upload, so the next attempt resumes instead of restarting.
480 + ///
481 + /// Idempotent and replaceable; installing `None` is not offered because a
482 + /// store that is present is only ever a saving. The SyncStore engine calls
483 + /// this at the start of each blob pass, which is why an app driving
484 + /// `SyncStore` gets resumable uploads without asking. Nothing else about
485 + /// the upload API changes: the resume key is the content hash, already the
486 + /// first argument to
487 + /// [`blob_upload_streaming`](Self::blob_upload_streaming).
488 + pub fn set_resume_store(&self, store: Arc<dyn resume::BlobResumeStore>) {
489 + *self.resume_store.write() = Some(store);
490 + }
491 +
492 + /// The installed resume store, if any.
493 + pub(crate) fn resume_store(&self) -> Option<Arc<dyn resume::BlobResumeStore>> {
494 + self.resume_store.read().clone()
495 + }
496 +
468 497 /// The server URL and API key this client was built with. Both are fixed at
469 498 /// construction; changing either means building a new client.
470 499 pub fn config(&self) -> &SyncKitConfig {
M synckit-client/src/store/blob.rs +21
@@ -19,6 +19,8 @@
19 19 use sha2::{Digest, Sha256};
20 20
21 21 use super::db::DbSource;
22 + use super::resume::SqliteResumeStore;
23 + use crate::client::resume::BlobResumeStore;
22 24 use crate::client::{BlobUploadOutcome, SyncKitClient};
23 25 use crate::error::{Result, SyncKitError};
24 26
@@ -53,6 +55,16 @@
53 55 fn download_url(&self, hash: &str) -> impl Future<Output = Result<String>> + Send;
54 56 /// Download the blob at `hash` from `url`, returning its ciphertext bytes.
55 57 fn download(&self, hash: &str, url: &str) -> impl Future<Output = Result<Vec<u8>>> + Send;
58 +
59 + /// Offer the transport somewhere durable to remember an upload it is part
60 + /// way through, so a process killed mid-transfer resumes rather than
61 + /// restarts.
62 + ///
63 + /// Defaulted to a no-op: a transport with no multipart session to lose (an
64 + /// in-memory test double) has nothing to record, and one that wants the
65 + /// behaviour opts in by overriding. The engine calls this once per blob
66 + /// pass, before any upload.
67 + fn install_resume_store(&self, _store: Arc<dyn BlobResumeStore>) {}
56 68 }
57 69
58 70 // The trait declares `-> impl Future + Send`; implementing with `async fn` is
@@ -70,6 +82,9 @@
70 82 async fn download(&self, hash: &str, url: &str) -> Result<Vec<u8>> {
71 83 SyncKitClient::blob_download(self, hash, url).await
72 84 }
85 + fn install_resume_store(&self, store: Arc<dyn BlobResumeStore>) {
86 + SyncKitClient::set_resume_store(self, store);
87 + }
73 88 }
74 89
75 90 /// Which rows own blobs, where they live, and how to reflect their presence.
@@ -158,6 +173,12 @@
158 173 client: &T,
159 174 policy: &Policy,
160 175 ) -> Result<u64> {
176 + // An upload interrupted here is worth resuming, so give the transport the
177 + // app's database to remember it in. Installed per pass rather than at build
178 + // time: this is the only place that knows both the transport and the DB, and
179 + // it costs one pointer write.
180 + client.install_resume_store(SqliteResumeStore::shared(db.clone()));
181 +
161 182 let pending = on_conn(db, policy, |p, c| p.pending_uploads(c)).await?;
162 183 let mut uploaded = 0u64;
163 184 for blob in pending {
M synckit-client/src/store/db.rs +4
@@ -77,6 +77,10 @@
77 77 conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
78 78 register_hash_row_id(conn)?;
79 79 ensure_scope_schema(conn)?;
80 + // Engine-owned blob-resume bookkeeping. Created here rather than in
81 + // `migration_sql` so it exists even for an app that snapshotted the
82 + // generated migration into a versioned file before these tables existed.
83 + conn.execute_batch(super::resume::RESUME_DDL)?;
80 84 Ok(())
81 85 }
82 86
M synckit-client/src/store/mod.rs +2
@@ -25,6 +25,7 @@
25 25 pub mod facade;
26 26 pub mod hlc;
27 27 pub mod migrate;
28 + pub mod resume;
28 29 pub mod scheduler;
29 30 pub mod schema;
30 31 pub(crate) mod snapshot;
@@ -47,6 +48,7 @@
47 48 ResolvedChanges, committed_hlc, load_clock, observe, record_committed, resolve_pull,
48 49 set_committed, stamp_pending,
49 50 };
51 + pub use resume::SqliteResumeStore;
50 52 pub use scheduler::{NoopObserver, SyncObserver, SyncState};
51 53 pub use schema::{ConflictStrategy, DeleteMode, RowIdScheme, SyncMode, SyncSchema, SyncTable};
52 54 pub use sync::{
M synckit-client/tests/integration/blob_multipart.rs +291 -3
@@ -67,6 +67,16 @@
67 67 /// Mount the whole session: start (with the given plan), part-URL minting,
68 68 /// the PUT target, and complete.
69 69 async fn mount_session(kit: &MockKit, cipher_len: usize, part_size: usize) -> u32 {
70 + let part_count = mount_session_without_put(kit, cipher_len, part_size).await;
71 + kit.put(PART_PUT_PATH)
72 + .reply(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
73 + .await;
74 + part_count
75 + }
76 +
77 + /// The session without the part PUT, for a test that mounts its own (one that
78 + /// fails part way, say).
79 + async fn mount_session_without_put(kit: &MockKit, cipher_len: usize, part_size: usize) -> u32 {
70 80 let part_count = cipher_len.div_ceil(part_size) as u32;
71 81
72 82 kit.post(START_PATH)
@@ -84,9 +94,6 @@
84 94 base: kit.uri(),
85 95 })
86 96 .await;
87 - kit.put(PART_PUT_PATH)
88 - .reply(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
89 - .await;
90 97 kit.post(COMPLETE_PATH).code(204).empty().await;
91 98
92 99 part_count
@@ -376,3 +383,284 @@
376 383
377 384 std::fs::remove_file(&file).ok();
378 385 }
386 +
387 + // ── Resuming an interrupted session ──
388 + //
389 + // A large blob is a long transfer, and a process killed part way used to throw
390 + // all of it away: the parts were still at S3, but nothing on this side
391 + // remembered the session. With a resume store installed the next attempt takes
392 + // the session over and sends only what is missing.
393 + //
394 + // The hard part is not the bookkeeping, it is the crypto. Part boundaries come
395 + // from the server and have nothing to do with the 1 MiB sealed-chunk geometry,
396 + // so a resume almost always restarts inside a chunk whose leading bytes are
397 + // already uploaded. Sealing draws a random nonce per chunk, so re-sealing that
398 + // chunk with a new one would splice two keystreams together and the assembled
399 + // object would never open. These tests are about that boundary.
400 +
401 + use synckit_client::client::resume::BlobResumeStore;
402 +
403 + /// PUTs that succeed for the first `ok` parts and then refuse, standing in for
404 + /// a transfer that dies part way. 403 rather than 500 so the client treats it
405 + /// as permanent and the test does not sit through the retry backoff.
406 + struct DiesAfter {
407 + ok: usize,
408 + seen: std::sync::atomic::AtomicUsize,
409 + }
410 +
411 + impl wiremock::Respond for DiesAfter {
412 + fn respond(&self, _req: &wiremock::Request) -> ResponseTemplate {
413 + let n = self.seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
414 + if n < self.ok {
415 + ResponseTemplate::new(200).append_header("ETag", format!("\"etag-{}\"", n + 1))
416 + } else {
417 + ResponseTemplate::new(403)
418 + }
419 + }
420 + }
421 +
422 + /// A resume store on its own scratch database, as the engine would install.
423 + fn resume_store(name: &str) -> Arc<dyn BlobResumeStore> {
424 + use std::sync::atomic::{AtomicU64, Ordering};
425 + static N: AtomicU64 = AtomicU64::new(0);
426 + let mut p = std::env::temp_dir();
427 + p.push(format!(
428 + "synckit_resume_{}_{}_{name}",
429 + std::process::id(),
430 + N.fetch_add(1, Ordering::Relaxed)
431 + ));
432 + std::fs::create_dir_all(&p).unwrap();
433 + synckit_client::store::SqliteResumeStore::shared(synckit_client::store::DbSource::path(
434 + p.join("app.db"),
435 + ))
436 + }
437 +
438 + async fn put_bodies(kit: &MockKit) -> Vec<Vec<u8>> {
439 + kit.requests_to(PART_PUT_PATH)
440 + .await
441 + .into_iter()
442 + .map(|r| r.body)
443 + .collect()
444 + }
445 +
446 + #[tokio::test]
447 + async fn a_killed_upload_resumes_and_the_assembled_blob_still_opens() {
448 + let kit = MockKit::start().await;
449 + let key = synckit_client::crypto::generate_master_key();
450 + let store = resume_store("kill");
451 +
452 + // Four 1 MiB chunks against 700 KiB parts: no part boundary can land on a
453 + // chunk boundary, so the resume is guaranteed to restart mid-chunk. That is
454 + // the case the stored nonces exist for.
455 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
456 + .map(|i| i as u8)
457 + .collect();
458 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
459 + let file = temp_blob("resume.bin", &plaintext);
460 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
461 + let part_size = 700 * 1024;
462 + let part_count = cipher_len.div_ceil(part_size);
463 + assert!(part_count > 3, "the fixture must have parts to resume from");
464 +
465 + // ── First attempt: dies after two parts ──
466 + let client = kit.authed();
467 + client.set_master_key_raw(key);
468 + client.set_resume_store(Arc::clone(&store));
469 +
470 + mount_session_without_put(&kit, cipher_len, part_size).await;
471 + kit.put(PART_PUT_PATH)
472 + .responder(DiesAfter {
473 + ok: 2,
474 + seen: std::sync::atomic::AtomicUsize::new(0),
475 + })
476 + .await;
477 + kit.post(ABORT_PATH).code(204).empty().await;
478 +
479 + let err = client
480 + .blob_upload_streaming(&hash, &file)
481 + .await
482 + .unwrap_err();
483 + assert!(
484 + matches!(err, SyncKitError::Server { status: 403, .. }),
485 + "got {err:?}"
486 + );
487 + // The session is the asset now: aborting it would throw away exactly what
488 + // the next attempt is going to reuse.
489 + assert_eq!(
490 + kit.hits(ABORT_PATH).await,
491 + 0,
492 + "a resumable failure must keep the session"
493 + );
494 + let first_two: Vec<Vec<u8>> = put_bodies(&kit).await.into_iter().take(2).collect();
495 +
496 + let record = store.load(&hash).unwrap().expect("a session was recorded");
497 + assert_eq!(record.usable_parts().len(), 2);
498 + assert_eq!(record.session.upload_id, "test-upload-id");
499 +
500 + // ── Second attempt: a fresh client, as a restarted process would have ──
501 + kit.reset().await;
502 + mount_session(&kit, cipher_len, part_size).await;
503 + kit.post(ABORT_PATH).code(204).empty().await;
504 +
505 + let restarted = kit.authed();
506 + restarted.set_master_key_raw(key);
507 + restarted.set_resume_store(Arc::clone(&store));
508 + restarted.blob_upload_streaming(&hash, &file).await.unwrap();
509 +
510 + let resumed = put_bodies(&kit).await;
511 + assert_eq!(
512 + resumed.len(),
513 + part_count - 2,
514 + "a resume must not re-send the parts already at S3"
515 + );
516 + // `start` is unconditional (it carries the dedup answer), so the redundant
517 + // session it opens has to be released rather than left to the reaper.
518 + assert_eq!(kit.hits(ABORT_PATH).await, 1);
519 +
520 + // The whole point: the two runs' parts concatenate into one valid v3 blob,
521 + // which means the chunk straddling the boundary came back byte-identical.
522 + let assembled: Vec<u8> = first_two
523 + .iter()
524 + .chain(resumed.iter())
525 + .flat_map(Clone::clone)
526 + .collect();
527 + assert_eq!(assembled.len(), cipher_len);
528 + assert_eq!(
529 + synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap(),
530 + plaintext,
531 + "a resumed blob must decrypt to the original"
532 + );
533 +
534 + // Complete named every part, and the kept ones carry the first run's ETags.
535 + let complete = kit.body(COMPLETE_PATH).await;
536 + let named = complete["parts"].as_array().unwrap();
537 + assert_eq!(named.len(), part_count);
538 + assert_eq!(named[0]["etag"].as_str().unwrap(), "\"etag-1\"");
539 + assert_eq!(named[1]["etag"].as_str().unwrap(), "\"etag-2\"");
540 + assert_eq!(complete["upload_id"].as_str().unwrap(), "test-upload-id");
541 +
542 + // Assembled means the record describes nothing.
543 + assert!(store.load(&hash).unwrap().is_none());
544 +
545 + std::fs::remove_file(&file).ok();
546 + }
547 +
548 + #[tokio::test]
549 + async fn a_resume_that_fails_again_gives_up_the_session_rather_than_wedging() {
550 + // A session the server has already reaped would fail identically on every
551 + // future pass. One resume attempt, then a clean slate.
552 + let kit = MockKit::start().await;
553 + let key = synckit_client::crypto::generate_master_key();
554 + let store = resume_store("wedge");
555 +
556 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 2 + 3))
557 + .map(|i| i as u8)
558 + .collect();
559 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
560 + let file = temp_blob("wedge.bin", &plaintext);
561 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
562 + let part_size = 700 * 1024;
563 +
564 + let client = kit.authed();
565 + client.set_master_key_raw(key);
566 + client.set_resume_store(Arc::clone(&store));
567 +
568 + // Attempt one: two parts land, then the transfer dies. The record survives.
569 + mount_session_without_put(&kit, cipher_len, part_size).await;
570 + kit.put(PART_PUT_PATH)
571 + .responder(DiesAfter {
572 + ok: 2,
573 + seen: std::sync::atomic::AtomicUsize::new(0),
574 + })
575 + .await;
576 + kit.post(ABORT_PATH).code(204).empty().await;
577 + client
578 + .blob_upload_streaming(&hash, &file)
579 + .await
580 + .unwrap_err();
581 + assert!(store.load(&hash).unwrap().is_some());
582 +
583 + // Attempt two resumes into a session that refuses everything.
584 + kit.reset().await;
585 + mount_session_without_put(&kit, cipher_len, part_size).await;
586 + kit.put(PART_PUT_PATH).code(403).empty().await;
587 + kit.post(ABORT_PATH).code(204).empty().await;
588 + client
589 + .blob_upload_streaming(&hash, &file)
590 + .await
591 + .unwrap_err();
592 +
593 + assert!(
594 + store.load(&hash).unwrap().is_none(),
595 + "a failed resume must drop the record so the next pass starts clean"
596 + );
597 + assert!(
598 + kit.hits(ABORT_PATH).await >= 1,
599 + "and release the parts it is giving up on"
600 + );
601 +
602 + std::fs::remove_file(&file).ok();
603 + }
604 +
605 + #[tokio::test]
606 + async fn a_file_that_changed_under_the_session_is_refused_rather_than_re_sealed() {
607 + // The nonce is the danger. Re-sealing different plaintext under a nonce this
608 + // key has already used would leak the XOR of the two chunks, so the resume
609 + // path checks a recorded plaintext digest before it re-uses one. A file
610 + // edited between attempts must stop the upload, not quietly seal.
611 + let kit = MockKit::start().await;
612 + let key = synckit_client::crypto::generate_master_key();
613 + let store = resume_store("changed");
614 +
615 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
616 + .map(|i| i as u8)
617 + .collect();
618 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
619 + let file = temp_blob("changed.bin", &plaintext);
620 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
621 + let part_size = 700 * 1024;
622 +
623 + let client = kit.authed();
624 + client.set_master_key_raw(key);
625 + client.set_resume_store(Arc::clone(&store));
626 + mount_session_without_put(&kit, cipher_len, part_size).await;
627 + kit.put(PART_PUT_PATH)
628 + .responder(DiesAfter {
629 + ok: 2,
630 + seen: std::sync::atomic::AtomicUsize::new(0),
631 + })
632 + .await;
633 + kit.post(ABORT_PATH).code(204).empty().await;
634 + client
635 + .blob_upload_streaming(&hash, &file)
636 + .await
637 + .unwrap_err();
638 + assert_eq!(store.load(&hash).unwrap().unwrap().usable_parts().len(), 2);
639 +
640 + // Same length, different bytes, and the byte is inside the chunk the resume
641 + // has to re-seal under the stored nonce (two 700 KiB parts land the boundary
642 + // in chunk 1). Every length check still passes, so the digest is the only
643 + // thing between this and a nonce re-use.
644 + let mut edited = plaintext.clone();
645 + edited[synckit_client::crypto::BLOB_CHUNK_SIZE + 5] ^= 0xff;
646 + std::fs::write(&file, &edited).unwrap();
647 +
648 + kit.reset().await;
649 + mount_session(&kit, cipher_len, part_size).await;
650 + kit.post(ABORT_PATH).code(204).empty().await;
651 + let err = client
652 + .blob_upload_streaming(&hash, &file)
653 + .await
654 + .unwrap_err();
655 + assert!(
656 + matches!(err, SyncKitError::Internal(ref m) if m.contains("changed under an in-flight upload")),
657 + "got {err:?}"
658 + );
659 + assert_eq!(
660 + kit.hits(COMPLETE_PATH).await,
661 + 0,
662 + "nothing may be assembled from two different files"
663 + );
664 +
665 + std::fs::remove_file(&file).ok();
666 + }
A synckit-client/src/client/resume.rs +221
@@ -1,0 +1,221 @@
1 + //! Durable state for an interrupted multipart blob upload.
2 + //!
3 + //! A multi-gigabyte blob can be most of an hour of transfer. If the process
4 + //! dies partway, every byte already at S3 is still there, but nothing on this
5 + //! side remembers the session, so the next run starts from zero. This module is
6 + //! the memory that closes that: an `upload_id`, the ETags of the parts that
7 + //! completed, and the per-chunk nonces needed to reproduce the ciphertext from
8 + //! a part boundary.
9 + //!
10 + //! ## Why the nonces
11 + //!
12 + //! Part boundaries are server-supplied and do not align to
13 + //! [`crypto::BLOB_CHUNK_SIZE`](crate::crypto::BLOB_CHUNK_SIZE), so a resume
14 + //! generally restarts in the middle of a sealed chunk whose leading bytes are
15 + //! already uploaded. Sealing draws a fresh random nonce per chunk, so re-sealing
16 + //! that chunk would produce different bytes and the assembled object would fail
17 + //! to open. Persisting the nonce lets the boundary chunk be reproduced exactly.
18 + //! Nonces are public, they ride in the clear at the head of every sealed chunk,
19 + //! so nothing secret is at rest here.
20 + //!
21 + //! Each nonce is stored with a digest of the plaintext it sealed, and the
22 + //! resume path re-checks that digest before re-using the nonce. Re-using a
23 + //! nonce over *different* plaintext under the same key would be catastrophic
24 + //! rather than merely wrong (see [`crypto::reseal_blob_chunk`](crate::crypto::reseal_blob_chunk));
25 + //! the digest is what makes that unreachable.
26 + //!
27 + //! ## Where it lives
28 + //!
29 + //! Nowhere, by default. The trait below is the seam; the SyncStore engine
30 + //! implements it over the app's SQLite database
31 + //! ([`store::resume`](crate::store::resume)) and installs it on the client at
32 + //! the start of each blob pass. A client used directly, without the engine,
33 + //! simply has no resume store and behaves exactly as before. That keeps the
34 + //! transport SDK free of a persistence dependency and leaves
35 + //! [`blob_upload_streaming`](crate::client::SyncKitClient::blob_upload_streaming)
36 + //! with the signature it always had: the resume key is the content hash, which
37 + //! is already its first argument, so a caller retrying after a crash calls what
38 + //! it always called and gets a resume instead of a restart.
39 +
40 + use crate::crypto::BLOB_NONCE_LEN;
41 +
42 + /// One sealed chunk's reproducibility record.
43 + #[derive(Debug, Clone, PartialEq, Eq)]
44 + pub struct ResumeChunk {
45 + /// Index of the chunk within the blob.
46 + pub index: u32,
47 + /// The nonce this chunk was sealed with.
48 + pub nonce: [u8; BLOB_NONCE_LEN],
49 + /// SHA-256 of the chunk's plaintext, checked before the nonce is re-used.
50 + pub plain_sha: [u8; 32],
51 + }
52 +
53 + /// One completed part of a multipart session.
54 + #[derive(Debug, Clone)]
55 + pub struct ResumePart {
56 + /// 1-based part number, as S3 numbers them.
57 + pub part_number: u32,
58 + /// The ETag S3 returned for the part, required to assemble the object.
59 + pub etag: String,
60 + }
61 +
62 + /// The geometry a session was opened with. Recorded so a resume can reuse the
63 + /// session instead of opening a second one, and so a record that no longer
64 + /// describes the file at hand is recognised and dropped.
65 + #[derive(Debug, Clone)]
66 + pub struct ResumeSession {
67 + /// The S3 multipart upload id.
68 + pub upload_id: String,
69 + /// Bytes per part (every part but the last).
70 + pub part_size: u64,
71 + /// Total number of parts the plan calls for.
72 + pub part_count: u32,
73 + /// Ciphertext length of the whole blob.
74 + pub size_bytes: u64,
75 + }
76 +
77 + /// A session plus everything recorded against it.
78 + #[derive(Debug, Clone)]
79 + pub struct ResumeRecord {
80 + /// The session this record resumes.
81 + pub session: ResumeSession,
82 + /// How long ago the session was opened, in seconds.
83 + pub age_secs: i64,
84 + /// Completed parts, ascending by part number.
85 + pub parts: Vec<ResumePart>,
86 + /// Chunk records, ascending by index.
87 + pub chunks: Vec<ResumeChunk>,
88 + }
89 +
90 + impl ResumeRecord {
91 + /// The chunk record for `index`, if one was kept.
92 + pub fn chunk(&self, index: u32) -> Option<&ResumeChunk> {
93 + self.chunks
94 + .binary_search_by_key(&index, |c| c.index)
95 + .ok()
96 + .map(|i| &self.chunks[i])
97 + }
98 +
99 + /// The lowest part number not yet completed.
100 + ///
101 + /// Parts must be contiguous from 1 to be usable: S3 assembles by part
102 + /// number, so a gap means the object cannot be completed from what is
103 + /// recorded. A record with a gap resumes from the first hole and re-uploads
104 + /// the rest, which is correct if wasteful, and gaps do not arise from the
105 + /// uploader (it completes parts in order).
106 + pub fn first_missing_part(&self) -> u32 {
107 + let mut expected = 1u32;
108 + for p in &self.parts {
109 + if p.part_number != expected {
110 + break;
111 + }
112 + expected += 1;
113 + }
114 + expected
115 + }
116 +
117 + /// The contiguous run of completed parts, which is what a resume may keep.
118 + pub fn usable_parts(&self) -> &[ResumePart] {
119 + let n = (self.first_missing_part() - 1) as usize;
120 + &self.parts[..n]
121 + }
122 + }
123 +
124 + /// Somewhere durable to record an in-flight multipart upload.
125 + ///
126 + /// Implementations are called from async code but are synchronous: every
127 + /// operation is a handful of short indexed statements against a local database,
128 + /// and the cadence is one call per completed part (parts are megabytes), not
129 + /// per chunk.
130 + ///
131 + /// **Nothing here may be load-bearing.** A resume store that errors, or that
132 + /// returns a record which turns out not to fit, must only cost a restart from
133 + /// zero, which is what the caller did before this existed. The upload path
134 + /// treats every method as best-effort for that reason.
135 + pub trait BlobResumeStore: Send + Sync {
136 + /// The record for `hash`, if a session is on file.
137 + fn load(&self, hash: &str) -> crate::Result<Option<ResumeRecord>>;
138 +
139 + /// Record a newly opened session, replacing any record already held for
140 + /// `hash` (its session is dead the moment a new one is opened).
141 + fn begin(&self, hash: &str, session: &ResumeSession) -> crate::Result<()>;
142 +
143 + /// Record one completed part, together with the chunk records sealed on the
144 + /// way to it, as a single atomic step.
145 + ///
146 + /// Called *after* the part is durable at S3, so a crash between the PUT and
147 + /// this call costs one part rather than corrupting the record.
148 + fn record_part(
149 + &self,
150 + hash: &str,
151 + part: &ResumePart,
152 + chunks: &[ResumeChunk],
153 + ) -> crate::Result<()>;
154 +
155 + /// Forget `hash` entirely: the upload finished, or its session is gone.
156 + fn clear(&self, hash: &str) -> crate::Result<()>;
157 + }
158 +
159 + #[cfg(test)]
160 + mod tests {
161 + use super::*;
162 +
163 + fn rec(parts: &[u32]) -> ResumeRecord {
164 + ResumeRecord {
165 + session: ResumeSession {
166 + upload_id: "u".into(),
167 + part_size: 8,
168 + part_count: 4,
169 + size_bytes: 32,
170 + },
171 + age_secs: 0,
172 + parts: parts
173 + .iter()
174 + .map(|n| ResumePart {
175 + part_number: *n,
176 + etag: format!("e{n}"),
177 + })
178 + .collect(),
179 + chunks: vec![],
180 + }
181 + }
182 +
183 + #[test]
184 + fn contiguous_parts_resume_after_the_last_one() {
185 + assert_eq!(rec(&[1, 2, 3]).first_missing_part(), 4);
186 + assert_eq!(rec(&[1, 2, 3]).usable_parts().len(), 3);
187 + }
188 +
189 + #[test]
190 + fn no_parts_resumes_from_the_first() {
191 + assert_eq!(rec(&[]).first_missing_part(), 1);
192 + assert!(rec(&[]).usable_parts().is_empty());
193 + }
194 +
195 + #[test]
196 + fn a_gap_truncates_the_usable_run() {
197 + // 3 is present but unreachable: S3 cannot assemble past the hole at 2.
198 + let r = rec(&[1, 3]);
199 + assert_eq!(r.first_missing_part(), 2);
200 + assert_eq!(r.usable_parts().len(), 1);
201 + }
202 +
203 + #[test]
204 + fn chunk_lookup_finds_by_index_not_position() {
205 + let mut r = rec(&[1]);
206 + r.chunks = vec![
207 + ResumeChunk {
208 + index: 4,
209 + nonce: [4u8; BLOB_NONCE_LEN],
210 + plain_sha: [0; 32],
211 + },
212 + ResumeChunk {
213 + index: 9,
214 + nonce: [9u8; BLOB_NONCE_LEN],
215 + plain_sha: [0; 32],
216 + },
217 + ];
218 + assert_eq!(r.chunk(9).unwrap().nonce[0], 9);
219 + assert!(r.chunk(5).is_none());
220 + }
221 + }
A synckit-client/src/store/resume.rs +412
@@ -1,0 +1,412 @@
1 + //! The engine's resume store: interrupted blob uploads, on disk.
2 + //!
3 + //! Implements [`BlobResumeStore`] over the app's own SQLite database, so a
4 + //! process killed mid-upload finds the session again on restart. The tables are
5 + //! engine-owned bookkeeping like the changelog and the conflict stash: local
6 + //! only, absent from every sync manifest, never pushed.
7 + //!
8 + //! Nothing here is secret. A nonce rides in the clear at the head of the sealed
9 + //! chunk it belongs to, an ETag is an S3 identifier, and the plaintext digests
10 + //! are of content the app already holds in the file being uploaded. Losing the
11 + //! whole table costs a restart from zero and nothing else, which is what the
12 + //! [`best_effort`](crate::client::resume) contract on the trait is about.
13 + //!
14 + //! See [`crate::client::resume`] for why the nonces have to be here at all.
15 +
16 + use std::sync::Arc;
17 +
18 + use rusqlite::{OptionalExtension, params};
19 +
20 + use super::db::DbSource;
21 + use crate::client::resume::{
22 + BlobResumeStore, ResumeChunk, ResumePart, ResumeRecord, ResumeSession,
23 + };
24 + use crate::crypto::BLOB_NONCE_LEN;
25 + use crate::error::{Result, SyncKitError};
26 +
27 + /// DDL for the resume tables.
28 + ///
29 + /// Applied from [`configure_connection`](super::db::configure_connection)
30 + /// rather than from `SyncSchema::migration_sql`, because an app that snapshotted
31 + /// the generated migration into a versioned file would never see a table added
32 + /// later. These are pure engine bookkeeping with no app-visible shape, so
33 + /// creating them on connection open is both safe and the only way to guarantee
34 + /// they exist wherever the engine runs.
35 + pub(crate) const RESUME_DDL: &str = "\
36 + -- An in-flight multipart blob upload, so a killed process resumes it.
37 + CREATE TABLE IF NOT EXISTS sync_blob_resume (
38 + hash TEXT PRIMARY KEY NOT NULL,
39 + upload_id TEXT NOT NULL,
40 + part_size INTEGER NOT NULL,
41 + part_count INTEGER NOT NULL,
42 + size_bytes INTEGER NOT NULL,
43 + created_at INTEGER NOT NULL
44 + ) WITHOUT ROWID;
45 +
46 + -- One completed part, with the ETag S3 needs to assemble the object.
47 + CREATE TABLE IF NOT EXISTS sync_blob_resume_part (
48 + hash TEXT NOT NULL,
49 + part_number INTEGER NOT NULL,
50 + etag TEXT NOT NULL,
51 + PRIMARY KEY (hash, part_number),
52 + FOREIGN KEY (hash) REFERENCES sync_blob_resume(hash) ON DELETE CASCADE
53 + ) WITHOUT ROWID;
54 +
55 + -- The nonce each sealed chunk was sealed with, so the chunk spanning a part
56 + -- boundary can be reproduced byte for byte. `plain_sha` is checked before the
57 + -- nonce is re-used: sealing different plaintext under a used nonce would break
58 + -- the cipher outright, so the resume path must be able to prove the file has
59 + -- not changed underneath it.
60 + CREATE TABLE IF NOT EXISTS sync_blob_resume_chunk (
61 + hash TEXT NOT NULL,
62 + chunk_index INTEGER NOT NULL,
63 + nonce BLOB NOT NULL,
64 + plain_sha BLOB NOT NULL,
65 + PRIMARY KEY (hash, chunk_index),
66 + FOREIGN KEY (hash) REFERENCES sync_blob_resume(hash) ON DELETE CASCADE
67 + ) WITHOUT ROWID;
68 + ";
69 +
70 + /// A [`BlobResumeStore`] over the engine's database.
71 + ///
72 + /// Opens a connection per call rather than holding one: the call rate is one
73 + /// per completed multipart part, which is megabytes of transfer apart, and a
74 + /// long-lived second writer on the app's file would be a worse trade than the
75 + /// open.
76 + pub struct SqliteResumeStore {
77 + db: DbSource,
78 + }
79 +
80 + impl SqliteResumeStore {
81 + /// A resume store backed by `db`.
82 + pub fn new(db: DbSource) -> Self {
83 + Self { db }
84 + }
85 +
86 + /// A resume store as the client wants it.
87 + pub fn shared(db: DbSource) -> Arc<dyn BlobResumeStore> {
88 + Arc::new(Self::new(db))
89 + }
90 +
91 + fn conn(&self) -> Result<rusqlite::Connection> {
92 + let conn = self.db.open()?;
93 + // The app's own connections are on the same file. A blob pass runs
94 + // alongside whatever the app is doing, so wait rather than fail on a
95 + // held write lock; every statement here is short.
96 + conn.busy_timeout(std::time::Duration::from_secs(5))?;
97 + Ok(conn)
98 + }
99 + }
100 +
101 + /// Read a fixed-width blob column, rejecting a wrong-length value rather than
102 + /// padding or truncating it into something that would seal wrongly.
103 + fn fixed<const N: usize>(bytes: &[u8], what: &str) -> Result<[u8; N]> {
104 + <[u8; N]>::try_from(bytes)
105 + .map_err(|_| SyncKitError::Database(format!("{what} is {} bytes, want {N}", bytes.len())))
106 + }
107 +
108 + impl BlobResumeStore for SqliteResumeStore {
109 + fn load(&self, hash: &str) -> Result<Option<ResumeRecord>> {
110 + let conn = self.conn()?;
111 + let Some((upload_id, part_size, part_count, size_bytes, created_at)) = conn
112 + .query_row(
113 + "SELECT upload_id, part_size, part_count, size_bytes, created_at
114 + FROM sync_blob_resume WHERE hash = ?1",
115 + params![hash],
116 + |r| {
117 + Ok((
118 + r.get::<_, String>(0)?,
119 + r.get::<_, i64>(1)?,
120 + r.get::<_, i64>(2)?,
121 + r.get::<_, i64>(3)?,
122 + r.get::<_, i64>(4)?,
123 + ))
124 + },
125 + )
126 + .optional()?
127 + else {
128 + return Ok(None);
129 + };
130 +
131 + let mut parts_stmt = conn.prepare(
132 + "SELECT part_number, etag FROM sync_blob_resume_part
133 + WHERE hash = ?1 ORDER BY part_number",
134 + )?;
135 + let parts = parts_stmt
136 + .query_map(params![hash], |r| {
137 + Ok(ResumePart {
138 + part_number: r.get::<_, i64>(0)? as u32,
139 + etag: r.get(1)?,
140 + })
141 + })?
142 + .collect::<rusqlite::Result<Vec<_>>>()?;
143 +
144 + let mut chunks_stmt = conn.prepare(
145 + "SELECT chunk_index, nonce, plain_sha FROM sync_blob_resume_chunk
146 + WHERE hash = ?1 ORDER BY chunk_index",
147 + )?;
148 + let chunks = chunks_stmt
149 + .query_map(params![hash], |r| {
150 + Ok((
151 + r.get::<_, i64>(0)? as u32,
152 + r.get::<_, Vec<u8>>(1)?,
153 + r.get::<_, Vec<u8>>(2)?,
154 + ))
155 + })?
156 + .collect::<rusqlite::Result<Vec<_>>>()?
157 + .into_iter()
158 + .map(|(index, nonce, plain_sha)| {
159 + Ok(ResumeChunk {
160 + index,
161 + nonce: fixed::<BLOB_NONCE_LEN>(&nonce, "resume nonce")?,
162 + plain_sha: fixed::<32>(&plain_sha, "resume plaintext digest")?,
163 + })
164 + })
165 + .collect::<Result<Vec<_>>>()?;
166 +
167 + Ok(Some(ResumeRecord {
168 + session: ResumeSession {
169 + upload_id,
170 + part_size: part_size as u64,
171 + part_count: part_count as u32,
172 + size_bytes: size_bytes as u64,
173 + },
174 + age_secs: (chrono::Utc::now().timestamp() - created_at).max(0),
175 + parts,
176 + chunks,
177 + }))
178 + }
179 +
180 + fn begin(&self, hash: &str, session: &ResumeSession) -> Result<()> {
181 + let mut conn = self.conn()?;
182 + let tx = conn.transaction()?;
183 + // Replace rather than merge: a new session means the parts and nonces
184 + // recorded against the old one describe an upload that no longer exists.
185 + tx.execute(
186 + "DELETE FROM sync_blob_resume WHERE hash = ?1",
187 + params![hash],
188 + )?;
189 + tx.execute(
190 + "INSERT INTO sync_blob_resume
191 + (hash, upload_id, part_size, part_count, size_bytes, created_at)
192 + VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
193 + params![
194 + hash,
195 + session.upload_id,
196 + session.part_size as i64,
197 + i64::from(session.part_count),
198 + session.size_bytes as i64,
199 + chrono::Utc::now().timestamp(),
200 + ],
201 + )?;
202 + tx.commit()?;
203 + Ok(())
204 + }
205 +
206 + fn record_part(&self, hash: &str, part: &ResumePart, chunks: &[ResumeChunk]) -> Result<()> {
207 + let mut conn = self.conn()?;
208 + let tx = conn.transaction()?;
209 + // If the session row is gone the record was cleared under us; the
210 + // foreign keys would reject these anyway, so say so rather than write
211 + // orphans.
212 + let live: bool = tx
213 + .query_row(
214 + "SELECT 1 FROM sync_blob_resume WHERE hash = ?1",
215 + params![hash],
216 + |_| Ok(true),
217 + )
218 + .optional()?
219 + .unwrap_or(false);
220 + if !live {
221 + return Ok(());
222 + }
223 + for chunk in chunks {
224 + // REPLACE: a second attempt re-seals its chunks with fresh nonces,
225 + // and the newest is the one that describes what is at S3.
226 + tx.execute(
227 + "INSERT OR REPLACE INTO sync_blob_resume_chunk
228 + (hash, chunk_index, nonce, plain_sha) VALUES (?1, ?2, ?3, ?4)",
229 + params![
230 + hash,
231 + i64::from(chunk.index),
232 + chunk.nonce.as_slice(),
233 + chunk.plain_sha.as_slice()
234 + ],
235 + )?;
236 + }
237 + tx.execute(
238 + "INSERT OR REPLACE INTO sync_blob_resume_part (hash, part_number, etag)
239 + VALUES (?1, ?2, ?3)",
240 + params![hash, i64::from(part.part_number), part.etag],
241 + )?;
242 + tx.commit()?;
243 + Ok(())
244 + }
245 +
246 + fn clear(&self, hash: &str) -> Result<()> {
247 + // ON DELETE CASCADE takes the parts and chunks; `foreign_keys` is ON for
248 + // every engine connection (see `configure_connection`).
249 + self.conn()?.execute(
250 + "DELETE FROM sync_blob_resume WHERE hash = ?1",
251 + params![hash],
252 + )?;
253 + Ok(())
254 + }
255 + }
256 +
257 + #[cfg(test)]
258 + mod tests {
259 + use super::*;
260 +
261 + fn store() -> SqliteResumeStore {
262 + use std::sync::atomic::{AtomicU64, Ordering};
263 + static N: AtomicU64 = AtomicU64::new(0);
264 + let mut p = std::env::temp_dir();
265 + p.push(format!(
266 + "synckit_resume_{}_{}",
267 + std::process::id(),
268 + N.fetch_add(1, Ordering::Relaxed)
269 + ));
270 + std::fs::create_dir_all(&p).unwrap();
271 + let store = SqliteResumeStore::new(DbSource::path(p.join("app.db")));
272 + // Opening is what applies the DDL.
273 + drop(store.conn().unwrap());
274 + store
275 + }
276 +
277 + fn session() -> ResumeSession {
278 + ResumeSession {
279 + upload_id: "upload-1".into(),
280 + part_size: 5 * 1024 * 1024,
281 + part_count: 3,
282 + size_bytes: 11 * 1024 * 1024,
283 + }
284 + }
285 +
286 + #[test]
287 + fn a_session_round_trips_with_its_parts_and_chunks() {
288 + let store = store();
289 + store.begin("aa", &session()).unwrap();
290 + store
291 + .record_part(
292 + "aa",
293 + &ResumePart {
294 + part_number: 1,
295 + etag: "\"etag-1\"".into(),
296 + },
297 + &[ResumeChunk {
298 + index: 0,
299 + nonce: [7u8; BLOB_NONCE_LEN],
300 + plain_sha: [9u8; 32],
301 + }],
302 + )
303 + .unwrap();
304 +
305 + let record = store.load("aa").unwrap().unwrap();
306 + assert_eq!(record.session.upload_id, "upload-1");
307 + assert_eq!(record.session.part_count, 3);
308 + assert_eq!(record.first_missing_part(), 2);
309 + assert_eq!(record.parts[0].etag, "\"etag-1\"");
310 + assert_eq!(record.chunk(0).unwrap().nonce, [7u8; BLOB_NONCE_LEN]);
311 + assert!(record.age_secs >= 0 && record.age_secs < 60);
312 + }
313 +
314 + #[test]
315 + fn beginning_again_discards_the_old_session_entirely() {
316 + let store = store();
317 + store.begin("aa", &session()).unwrap();
318 + store
319 + .record_part(
320 + "aa",
321 + &ResumePart {
322 + part_number: 1,
323 + etag: "old".into(),
324 + },
325 + &[ResumeChunk {
326 + index: 0,
327 + nonce: [1u8; BLOB_NONCE_LEN],
328 + plain_sha: [1u8; 32],
329 + }],
330 + )
331 + .unwrap();
332 +
333 + let mut next = session();
334 + next.upload_id = "upload-2".into();
335 + store.begin("aa", &next).unwrap();
336 +
337 + let record = store.load("aa").unwrap().unwrap();
338 + assert_eq!(record.session.upload_id, "upload-2");
339 + // Parts and nonces belong to the dead session; keeping them would
340 + // resume a session S3 no longer has.
341 + assert!(record.parts.is_empty());
342 + assert!(record.chunks.is_empty());
343 + }
344 +
345 + #[test]
346 + fn clearing_takes_the_children_with_it() {
347 + let store = store();
348 + store.begin("aa", &session()).unwrap();
349 + store
350 + .record_part(
351 + "aa",
352 + &ResumePart {
353 + part_number: 1,
354 + etag: "e".into(),
355 + },
356 + &[ResumeChunk {
357 + index: 0,
358 + nonce: [1u8; BLOB_NONCE_LEN],
359 + plain_sha: [1u8; 32],
360 + }],
361 + )
362 + .unwrap();
363 + store.clear("aa").unwrap();
364 + assert!(store.load("aa").unwrap().is_none());
365 +
366 + let conn = store.conn().unwrap();
367 + let parts: i64 = conn
368 + .query_row("SELECT count(*) FROM sync_blob_resume_part", [], |r| {
369 + r.get(0)
370 + })
371 + .unwrap();
372 + let chunks: i64 = conn
373 + .query_row("SELECT count(*) FROM sync_blob_resume_chunk", [], |r| {
374 + r.get(0)
375 + })
376 + .unwrap();
377 + assert_eq!((parts, chunks), (0, 0));
378 + }
379 +
380 + #[test]
381 + fn recording_against_a_cleared_session_writes_nothing() {
382 + let store = store();
383 + store
384 + .record_part(
385 + "gone",
386 + &ResumePart {
387 + part_number: 1,
388 + etag: "e".into(),
389 + },
390 + &[],
391 + )
392 + .unwrap();
393 + assert!(store.load("gone").unwrap().is_none());
394 + }
395 +
396 + #[test]
397 + fn a_wrong_length_nonce_is_rejected_rather_than_reshaped() {
398 + let store = store();
399 + store.begin("aa", &session()).unwrap();
400 + store
401 + .conn()
402 + .unwrap()
403 + .execute(
404 + "INSERT INTO sync_blob_resume_chunk (hash, chunk_index, nonce, plain_sha)
405 + VALUES ('aa', 0, X'0102', ?1)",
406 + params![[0u8; 32].as_slice()],
407 + )
408 + .unwrap();
409 + let err = store.load("aa").unwrap_err();
410 + assert!(err.to_string().contains("resume nonce"), "{err}");
411 + }
412 + }