//! Durable state for an interrupted multipart blob upload. //! //! A multi-gigabyte blob can be most of an hour of transfer. If the process //! dies partway, every byte already at S3 is still there, but nothing on this //! side remembers the session, so the next run starts from zero. This module is //! the memory that closes that: an `upload_id`, the ETags of the parts that //! completed, and the per-chunk nonces needed to reproduce the ciphertext from //! a part boundary. //! //! ## Why the nonces //! //! Part boundaries are server-supplied and do not align to //! [`crypto::BLOB_CHUNK_SIZE`](crate::crypto::BLOB_CHUNK_SIZE), so a resume //! generally restarts in the middle of a sealed chunk whose leading bytes are //! already uploaded. Sealing draws a fresh random nonce per chunk, so re-sealing //! that chunk would produce different bytes and the assembled object would fail //! to open. Persisting the nonce lets the boundary chunk be reproduced exactly. //! Nonces are public, they ride in the clear at the head of every sealed chunk, //! so nothing secret is at rest here. //! //! Each nonce is stored with a digest of the plaintext it sealed, and the //! resume path re-checks that digest before re-using the nonce. Re-using a //! nonce over *different* plaintext under the same key would be catastrophic //! rather than merely wrong (see [`crypto::reseal_blob_chunk`](crate::crypto::reseal_blob_chunk)); //! the digest is what makes that unreachable. //! //! ## Where it lives //! //! Nowhere, by default. The trait below is the seam; the SyncStore engine //! implements it over the app's SQLite database //! ([`store::resume`](crate::store::resume)) and installs it on the client at //! the start of each blob pass. A client used directly, without the engine, //! simply has no resume store and behaves exactly as before. That keeps the //! transport SDK free of a persistence dependency and leaves //! [`blob_upload_streaming`](crate::client::SyncKitClient::blob_upload_streaming) //! with the signature it always had: the resume key is the content hash, which //! is already its first argument, so a caller retrying after a crash calls what //! it always called and gets a resume instead of a restart. use crate::crypto::BLOB_NONCE_LEN; /// One sealed chunk's reproducibility record. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ResumeChunk { /// Index of the chunk within the blob. pub index: u32, /// The nonce this chunk was sealed with. pub nonce: [u8; BLOB_NONCE_LEN], /// SHA-256 of the chunk's plaintext, checked before the nonce is re-used. pub plain_sha: [u8; 32], } /// One completed part of a multipart session. #[derive(Debug, Clone)] pub struct ResumePart { /// 1-based part number, as S3 numbers them. pub part_number: u32, /// The ETag S3 returned for the part, required to assemble the object. pub etag: String, } /// The geometry a session was opened with. Recorded so a resume can reuse the /// session instead of opening a second one, and so a record that no longer /// describes the file at hand is recognised and dropped. #[derive(Debug, Clone)] pub struct ResumeSession { /// The S3 multipart upload id. pub upload_id: String, /// Bytes per part (every part but the last). pub part_size: u64, /// Total number of parts the plan calls for. pub part_count: u32, /// Ciphertext length of the whole blob. pub size_bytes: u64, } /// A session plus everything recorded against it. #[derive(Debug, Clone)] pub struct ResumeRecord { /// The session this record resumes. pub session: ResumeSession, /// How long ago the session was opened, in seconds. pub age_secs: i64, /// Completed parts, ascending by part number. pub parts: Vec, /// Chunk records, ascending by index. pub chunks: Vec, } impl ResumeRecord { /// The chunk record for `index`, if one was kept. pub fn chunk(&self, index: u32) -> Option<&ResumeChunk> { self.chunks .binary_search_by_key(&index, |c| c.index) .ok() .map(|i| &self.chunks[i]) } /// The lowest part number not yet completed. /// /// Parts must be contiguous from 1 to be usable: S3 assembles by part /// number, so a gap means the object cannot be completed from what is /// recorded. A record with a gap resumes from the first hole and re-uploads /// the rest, which is correct if wasteful, and gaps do not arise from the /// uploader (it completes parts in order). pub fn first_missing_part(&self) -> u32 { let mut expected = 1u32; for p in &self.parts { if p.part_number != expected { break; } expected += 1; } expected } /// The contiguous run of completed parts, which is what a resume may keep. pub fn usable_parts(&self) -> &[ResumePart] { let n = (self.first_missing_part() - 1) as usize; &self.parts[..n] } } /// Somewhere durable to record an in-flight multipart upload. /// /// Implementations are called from async code but are synchronous: every /// operation is a handful of short indexed statements against a local database, /// and the cadence is one call per completed part (parts are megabytes), not /// per chunk. /// /// **Nothing here may be load-bearing.** A resume store that errors, or that /// returns a record which turns out not to fit, must only cost a restart from /// zero. The upload path treats every method as best-effort for that reason. pub trait BlobResumeStore: Send + Sync { /// The record for `hash`, if a session is on file. fn load(&self, hash: &str) -> crate::Result>; /// Record a newly opened session, replacing any record already held for /// `hash` (its session is dead the moment a new one is opened). fn begin(&self, hash: &str, session: &ResumeSession) -> crate::Result<()>; /// Record one completed part, together with the chunk records sealed on the /// way to it, as a single atomic step. /// /// Called *after* the part is durable at S3, so a crash between the PUT and /// this call costs one part rather than corrupting the record. fn record_part( &self, hash: &str, part: &ResumePart, chunks: &[ResumeChunk], ) -> crate::Result<()>; /// Forget `hash` entirely: the upload finished, or its session is gone. fn clear(&self, hash: &str) -> crate::Result<()>; } #[cfg(test)] mod tests { use super::*; fn rec(parts: &[u32]) -> ResumeRecord { ResumeRecord { session: ResumeSession { upload_id: "u".into(), part_size: 8, part_count: 4, size_bytes: 32, }, age_secs: 0, parts: parts .iter() .map(|n| ResumePart { part_number: *n, etag: format!("e{n}"), }) .collect(), chunks: vec![], } } #[test] fn contiguous_parts_resume_after_the_last_one() { assert_eq!(rec(&[1, 2, 3]).first_missing_part(), 4); assert_eq!(rec(&[1, 2, 3]).usable_parts().len(), 3); } #[test] fn no_parts_resumes_from_the_first() { assert_eq!(rec(&[]).first_missing_part(), 1); assert!(rec(&[]).usable_parts().is_empty()); } #[test] fn a_gap_truncates_the_usable_run() { // 3 is present but unreachable: S3 cannot assemble past the hole at 2. let r = rec(&[1, 3]); assert_eq!(r.first_missing_part(), 2); assert_eq!(r.usable_parts().len(), 1); } #[test] fn chunk_lookup_finds_by_index_not_position() { let mut r = rec(&[1]); r.chunks = vec![ ResumeChunk { index: 4, nonce: [4u8; BLOB_NONCE_LEN], plain_sha: [0; 32], }, ResumeChunk { index: 9, nonce: [9u8; BLOB_NONCE_LEN], plain_sha: [0; 32], }, ]; assert_eq!(r.chunk(9).unwrap().nonce[0], 9); assert!(r.chunk(5).is_none()); } }