max / synckit

8.1 KB · 221 lines History Blame Raw

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. The upload path treats every method as best-effort for that reason.
134	pub trait BlobResumeStore: Send + Sync {
135	/// The record for `hash`, if a session is on file.
136	fn load(&self, hash: &str) -> crate::Result<Option<ResumeRecord>>;
137
138	/// Record a newly opened session, replacing any record already held for
139	/// `hash` (its session is dead the moment a new one is opened).
140	fn begin(&self, hash: &str, session: &ResumeSession) -> crate::Result<()>;
141
142	/// Record one completed part, together with the chunk records sealed on the
143	/// way to it, as a single atomic step.
144	///
145	/// Called after the part is durable at S3, so a crash between the PUT and
146	/// this call costs one part rather than corrupting the record.
147	fn record_part(
148	&self,
149	hash: &str,
150	part: &ResumePart,
151	chunks: &[ResumeChunk],
152	) -> crate::Result<()>;
153
154	/// Forget `hash` entirely: the upload finished, or its session is gone.
155	fn clear(&self, hash: &str) -> crate::Result<()>;
156	}
157
158	#[cfg(test)]
159	mod tests {
160	use super::*;
161
162	fn rec(parts: &[u32]) -> ResumeRecord {
163	ResumeRecord {
164	session: ResumeSession {
165	upload_id: "u".into(),
166	part_size: 8,
167	part_count: 4,
168	size_bytes: 32,
169	},
170	age_secs: 0,
171	parts: parts
172	.iter()
173	.map(\|n\| ResumePart {
174	part_number: *n,
175	etag: format!("e{n}"),
176	})
177	.collect(),
178	chunks: vec![],
179	}
180	}
181
182	#[test]
183	fn contiguous_parts_resume_after_the_last_one() {
184	assert_eq!(rec(&[1, 2, 3]).first_missing_part(), 4);
185	assert_eq!(rec(&[1, 2, 3]).usable_parts().len(), 3);
186	}
187
188	#[test]
189	fn no_parts_resumes_from_the_first() {
190	assert_eq!(rec(&[]).first_missing_part(), 1);
191	assert!(rec(&[]).usable_parts().is_empty());
192	}
193
194	#[test]
195	fn a_gap_truncates_the_usable_run() {
196	// 3 is present but unreachable: S3 cannot assemble past the hole at 2.
197	let r = rec(&[1, 3]);
198	assert_eq!(r.first_missing_part(), 2);
199	assert_eq!(r.usable_parts().len(), 1);
200	}
201
202	#[test]
203	fn chunk_lookup_finds_by_index_not_position() {
204	let mut r = rec(&[1]);
205	r.chunks = vec![
206	ResumeChunk {
207	index: 4,
208	nonce: [4u8; BLOB_NONCE_LEN],
209	plain_sha: [0; 32],
210	},
211	ResumeChunk {
212	index: 9,
213	nonce: [9u8; BLOB_NONCE_LEN],
214	plain_sha: [0; 32],
215	},
216	];
217	assert_eq!(r.chunk(9).unwrap().nonce[0], 9);
218	assert!(r.chunk(5).is_none());
219	}
220	}
221