Skip to main content

max / synckit

14.2 KB · 413 lines History Blame Raw
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 }
413