Skip to main content

max / synckit

19.7 KB · 556 lines History Blame Raw
1 //! Content-addressed blob synchronization.
2 //!
3 //! Blob sync is a separate pass the scheduler interleaves *after* the metadata
4 //! pull, and it is non-fatal: a failed blob never aborts a sync cycle. An app
5 //! declares its blob policy via [`BlobPolicy`], which rows own a blob, where it
6 //! lives on disk, and how to reflect local-vs-cloud presence, and the engine
7 //! owns the invariant machinery: dedup-aware upload, integrity-verified download,
8 //! and atomic writes.
9 //!
10 //! Like [`super::sync`], the client is abstracted behind a trait
11 //! ([`BlobTransport`]) so the engine tests end-to-end against an in-memory store,
12 //! and `SyncKitClient` satisfies it in production.
13
14 use std::future::Future;
15 use std::path::{Path, PathBuf};
16 use std::sync::Arc;
17
18 use rusqlite::Connection;
19 use sha2::{Digest, Sha256};
20
21 use super::db::DbSource;
22 use crate::client::{BlobUploadOutcome, SyncKitClient};
23 use crate::error::{Result, SyncKitError};
24
25 /// A content-addressed blob: its SHA-256 hash (the address), file extension, and
26 /// plaintext size in bytes.
27 #[derive(Debug, Clone, PartialEq, Eq)]
28 pub struct BlobRef {
29 pub hash: String,
30 pub ext: String,
31 pub size: u64,
32 }
33
34 /// The blob operations the engine needs from a server. A seam for testing and
35 /// the reference a future non-Rust SDK mirrors; [`SyncKitClient`] forwards to its
36 /// inherent `blob_*` methods.
37 pub trait BlobTransport: Sync {
38 /// Upload the file at `path` under the content address `hash`, or report
39 /// that the server already holds it. Takes a path rather than bytes so the
40 /// implementation can stream: the engine must never be the reason a blob is
41 /// read whole into memory.
42 fn upload_file(
43 &self,
44 hash: &str,
45 path: &Path,
46 ) -> impl Future<Output = Result<BlobUploadOutcome>> + Send;
47 fn confirm(&self, hash: &str, size: u64) -> impl Future<Output = Result<()>> + Send;
48 fn download_url(&self, hash: &str) -> impl Future<Output = Result<String>> + Send;
49 fn download(&self, hash: &str, url: &str) -> impl Future<Output = Result<Vec<u8>>> + Send;
50 }
51
52 // The trait declares `-> impl Future + Send`; implementing with `async fn` is
53 // equivalent (the Send bound is enforced by the trait) and reads cleaner.
54 impl BlobTransport for SyncKitClient {
55 async fn upload_file(&self, hash: &str, path: &Path) -> Result<BlobUploadOutcome> {
56 SyncKitClient::blob_upload_streaming(self, hash, path).await
57 }
58 async fn confirm(&self, hash: &str, size: u64) -> Result<()> {
59 SyncKitClient::blob_confirm(self, hash, size as i64).await
60 }
61 async fn download_url(&self, hash: &str) -> Result<String> {
62 SyncKitClient::blob_download_url(self, hash).await
63 }
64 async fn download(&self, hash: &str, url: &str) -> Result<Vec<u8>> {
65 SyncKitClient::blob_download(self, hash, url).await
66 }
67 }
68
69 /// Which rows own blobs, where they live, and how to reflect their presence.
70 ///
71 /// The three query/path methods are required; the presence hooks default to
72 /// no-ops for an app that tracks presence purely by disk existence (GoingsOn). An
73 /// app with an explicit presence flag (audiofiles' `cloud_only`) overrides them.
74 pub trait BlobPolicy: Send + Sync {
75 /// Rows whose bytes should be uploaded. May return the full candidate set,
76 /// the engine skips any whose [`local_path`](Self::local_path) is absent.
77 fn pending_uploads(&self, conn: &Connection) -> Result<Vec<BlobRef>>;
78 /// Rows whose bytes are wanted. The engine skips any already present on disk.
79 fn pending_downloads(&self, conn: &Connection) -> Result<Vec<BlobRef>>;
80 /// The on-disk location for a blob.
81 fn local_path(&self, blob: &BlobRef) -> PathBuf;
82 /// Reconcile presence flags before the passes (e.g. mark missing files cloud-only).
83 fn reconcile(&self, _conn: &Connection) -> Result<()> {
84 Ok(())
85 }
86 /// Reflect that a blob's bytes are now on the server.
87 fn on_uploaded(&self, _conn: &Connection, _blob: &BlobRef) -> Result<()> {
88 Ok(())
89 }
90 /// Reflect that a blob's bytes are now on local disk.
91 fn on_downloaded(&self, _conn: &Connection, _blob: &BlobRef) -> Result<()> {
92 Ok(())
93 }
94 }
95
96 /// Result of a blob sync pass.
97 #[derive(Debug, Default, PartialEq, Eq)]
98 pub struct BlobOutcome {
99 pub uploaded: u64,
100 pub downloaded: u64,
101 }
102
103 type Policy = Arc<dyn BlobPolicy>;
104
105 fn join_err(e: &tokio::task::JoinError) -> SyncKitError {
106 SyncKitError::Internal(format!("blocking task failed: {e}"))
107 }
108
109 fn io_err(context: &str, e: &std::io::Error) -> SyncKitError {
110 SyncKitError::Internal(format!("blob {context}: {e}"))
111 }
112
113 /// First 8 chars of a hash for logs, panic-safe on short/multibyte input.
114 fn short(hash: &str) -> &str {
115 hash.get(..8).unwrap_or(hash)
116 }
117
118 /// Run a policy database method on a blocking thread with a fresh connection.
119 async fn on_conn<F, R>(db: &DbSource, policy: &Policy, f: F) -> Result<R>
120 where
121 F: FnOnce(&dyn BlobPolicy, &Connection) -> Result<R> + Send + 'static,
122 R: Send + 'static,
123 {
124 let db = db.clone();
125 let policy = policy.clone();
126 tokio::task::spawn_blocking(move || f(policy.as_ref(), &db.open()?))
127 .await
128 .map_err(|e| join_err(&e))?
129 }
130
131 /// The full blob pass: reconcile, then upload local, then download missing.
132 /// Non-fatal by construction, per-blob failures are logged and skipped.
133 pub async fn sync_blobs<T: BlobTransport>(
134 db: &DbSource,
135 client: &T,
136 policy: &Policy,
137 ) -> Result<BlobOutcome> {
138 on_conn(db, policy, |p, c| p.reconcile(c)).await?;
139 let uploaded = upload_blobs(db, client, policy).await?;
140 let downloaded = download_blobs(db, client, policy).await?;
141 Ok(BlobOutcome {
142 uploaded,
143 downloaded,
144 })
145 }
146
147 /// Upload every pending local blob the server doesn't already have.
148 pub async fn upload_blobs<T: BlobTransport>(
149 db: &DbSource,
150 client: &T,
151 policy: &Policy,
152 ) -> Result<u64> {
153 let pending = on_conn(db, policy, |p, c| p.pending_uploads(c)).await?;
154 let mut uploaded = 0u64;
155 for blob in pending {
156 match upload_one(db, client, policy, &blob).await {
157 Ok(true) => uploaded += 1,
158 Ok(false) => {}
159 Err(e) => tracing::warn!(
160 hash = short(&blob.hash),
161 "blob upload failed (non-fatal): {e}"
162 ),
163 }
164 }
165 Ok(uploaded)
166 }
167
168 async fn upload_one<T: BlobTransport>(
169 db: &DbSource,
170 client: &T,
171 policy: &Policy,
172 blob: &BlobRef,
173 ) -> Result<bool> {
174 let path = policy.local_path(blob);
175 if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
176 return Ok(false); // no local file to upload
177 }
178 // The transport streams from the path, so a blob is never read whole into
179 // memory here regardless of size, and it runs its own dedup check before
180 // touching the file.
181 if client.upload_file(&blob.hash, &path).await? == BlobUploadOutcome::AlreadyPresent {
182 // Content-addressed dedup: the server says it holds this hash, so the
183 // bytes stay home. Two consequences worth knowing, both bounded:
184 // - The claim is unverified. A server that answers `true` wrongly
185 // withholds the blob from other devices; it cannot substitute one,
186 // because `download_one` re-hashes before committing the file.
187 // - `on_uploaded` is deliberately not fired: nothing was uploaded,
188 // and the confirm call that pairs with it never happened. A policy
189 // using that hook to clear a presence flag will keep re-offering
190 // the blob on every pass, costing one round trip each time.
191 // Harmless for the no-op default (the only implementors today), but
192 // a policy with an explicit flag needs its own way to settle the
193 // dedup case.
194 return Ok(true);
195 }
196 client.confirm(&blob.hash, blob.size).await?;
197
198 let blob = blob.clone();
199 on_conn(db, policy, move |p, c| p.on_uploaded(c, &blob)).await?;
200 Ok(true)
201 }
202
203 /// Download every wanted blob not already on local disk.
204 pub async fn download_blobs<T: BlobTransport>(
205 db: &DbSource,
206 client: &T,
207 policy: &Policy,
208 ) -> Result<u64> {
209 let pending = on_conn(db, policy, |p, c| p.pending_downloads(c)).await?;
210 let mut downloaded = 0u64;
211 for blob in pending {
212 let path = policy.local_path(&blob);
213 if tokio::fs::try_exists(&path).await.unwrap_or(false) {
214 continue; // already have it locally
215 }
216 match download_one(db, client, policy, &blob).await {
217 Ok(true) => downloaded += 1,
218 Ok(false) => {}
219 Err(e) => tracing::warn!(
220 hash = short(&blob.hash),
221 "blob download failed (non-fatal): {e}"
222 ),
223 }
224 }
225 Ok(downloaded)
226 }
227
228 /// Download a single blob by reference, unconditionally (the GUI "download now"
229 /// path). Verifies the content hash before committing the file.
230 pub async fn download_one<T: BlobTransport>(
231 db: &DbSource,
232 client: &T,
233 policy: &Policy,
234 blob: &BlobRef,
235 ) -> Result<bool> {
236 let path = policy.local_path(blob);
237 let url = client.download_url(&blob.hash).await?;
238 let data = client.download(&blob.hash, &url).await?;
239
240 // Verify the plaintext hashes to its address before it lands under that name.
241 // The SDK's blob_download also re-verifies; this guards the store regardless
242 // of transport (and is what the in-memory tests exercise).
243 let actual = hex::encode(Sha256::digest(&data));
244 if actual != blob.hash {
245 return Err(SyncKitError::IntegrityFailed {
246 expected: blob.hash.clone(),
247 actual,
248 });
249 }
250
251 if let Some(parent) = path.parent() {
252 tokio::fs::create_dir_all(parent)
253 .await
254 .map_err(|e| io_err("mkdir", &e))?;
255 }
256 // Atomic: write to a sibling temp then rename, so a crash never leaves a
257 // half-written file under the content-addressed name.
258 let tmp = {
259 let mut s = path.clone().into_os_string();
260 s.push(".downloading");
261 PathBuf::from(s)
262 };
263 tokio::fs::write(&tmp, &data)
264 .await
265 .map_err(|e| io_err("write", &e))?;
266 tokio::fs::rename(&tmp, &path)
267 .await
268 .map_err(|e| io_err("rename", &e))?;
269
270 let blob = blob.clone();
271 on_conn(db, policy, move |p, c| p.on_downloaded(c, &blob)).await?;
272 Ok(true)
273 }
274
275 #[cfg(test)]
276 mod tests {
277 use super::*;
278 use std::collections::HashMap;
279 use std::sync::Mutex;
280
281 /// In-memory content-addressed blob store standing in for the server.
282 #[derive(Clone, Default)]
283 struct FakeBlobs {
284 store: Arc<Mutex<HashMap<String, Vec<u8>>>>,
285 }
286 impl FakeBlobs {
287 fn put_raw(&self, hash: &str, bytes: Vec<u8>) {
288 self.store.lock().unwrap().insert(hash.into(), bytes);
289 }
290 fn has(&self, hash: &str) -> bool {
291 self.store.lock().unwrap().contains_key(hash)
292 }
293 }
294 impl BlobTransport for FakeBlobs {
295 async fn upload_file(&self, hash: &str, path: &Path) -> Result<BlobUploadOutcome> {
296 if self.has(hash) {
297 return Ok(BlobUploadOutcome::AlreadyPresent);
298 }
299 let data = std::fs::read(path).map_err(|e| io_err("read", &e))?;
300 self.store.lock().unwrap().insert(hash.to_string(), data);
301 Ok(BlobUploadOutcome::Uploaded)
302 }
303 async fn confirm(&self, _hash: &str, _size: u64) -> Result<()> {
304 Ok(())
305 }
306 async fn download_url(&self, hash: &str) -> Result<String> {
307 Ok(format!("fake://get/{hash}"))
308 }
309 async fn download(&self, hash: &str, _url: &str) -> Result<Vec<u8>> {
310 self.store
311 .lock()
312 .unwrap()
313 .get(hash)
314 .cloned()
315 .ok_or_else(|| SyncKitError::Internal("blob not on server".into()))
316 }
317 }
318
319 /// A policy backed by a `blobmeta(hash, ext, size, cloud_only)` table.
320 struct TestPolicy {
321 dir: PathBuf,
322 }
323 impl BlobPolicy for TestPolicy {
324 fn pending_uploads(&self, conn: &Connection) -> Result<Vec<BlobRef>> {
325 rows(
326 conn,
327 "SELECT hash, ext, size FROM blobmeta WHERE cloud_only = 0",
328 )
329 }
330 fn pending_downloads(&self, conn: &Connection) -> Result<Vec<BlobRef>> {
331 rows(
332 conn,
333 "SELECT hash, ext, size FROM blobmeta WHERE cloud_only = 1",
334 )
335 }
336 fn local_path(&self, blob: &BlobRef) -> PathBuf {
337 self.dir.join(format!("{}.{}", blob.hash, blob.ext))
338 }
339 fn reconcile(&self, conn: &Connection) -> Result<()> {
340 // Flip cloud_only=0 rows whose file is missing to cloud_only=1.
341 let missing = rows(
342 conn,
343 "SELECT hash, ext, size FROM blobmeta WHERE cloud_only = 0",
344 )?
345 .into_iter()
346 .filter(|b| !self.local_path(b).exists())
347 .collect::<Vec<_>>();
348 for b in missing {
349 conn.execute(
350 "UPDATE blobmeta SET cloud_only = 1 WHERE hash = ?1",
351 [&b.hash],
352 )?;
353 }
354 Ok(())
355 }
356 fn on_downloaded(&self, conn: &Connection, blob: &BlobRef) -> Result<()> {
357 conn.execute(
358 "UPDATE blobmeta SET cloud_only = 0 WHERE hash = ?1",
359 [&blob.hash],
360 )?;
361 Ok(())
362 }
363 }
364
365 fn rows(conn: &Connection, sql: &str) -> Result<Vec<BlobRef>> {
366 let mut stmt = conn.prepare(sql)?;
367 let out = stmt
368 .query_map([], |r| {
369 Ok(BlobRef {
370 hash: r.get(0)?,
371 ext: r.get(1)?,
372 size: r.get::<_, i64>(2)? as u64,
373 })
374 })?
375 .collect::<rusqlite::Result<Vec<_>>>()?;
376 Ok(out)
377 }
378
379 fn hash_of(content: &[u8]) -> String {
380 hex::encode(Sha256::digest(content))
381 }
382
383 fn tempdir() -> PathBuf {
384 use std::sync::atomic::{AtomicU64, Ordering};
385 static N: AtomicU64 = AtomicU64::new(0);
386 let mut p = std::env::temp_dir();
387 p.push(format!(
388 "synckit_b6_{}_{}",
389 std::process::id(),
390 N.fetch_add(1, Ordering::Relaxed)
391 ));
392 std::fs::create_dir_all(&p).unwrap();
393 p
394 }
395
396 fn setup(dir: &std::path::Path) -> (DbSource, Policy) {
397 let db_path = dir.join("meta.db");
398 let db = DbSource::path(&db_path);
399 db.open()
400 .unwrap()
401 .execute_batch("CREATE TABLE blobmeta (hash TEXT PRIMARY KEY, ext TEXT, size INT, cloud_only INT);")
402 .unwrap();
403 let content = dir.join("content");
404 std::fs::create_dir_all(&content).unwrap();
405 let policy: Policy = Arc::new(TestPolicy { dir: content });
406 (db, policy)
407 }
408
409 fn seed_meta(db: &DbSource, hash: &str, ext: &str, size: u64, cloud_only: i64) {
410 db.open()
411 .unwrap()
412 .execute(
413 "INSERT INTO blobmeta (hash, ext, size, cloud_only) VALUES (?1, ?2, ?3, ?4)",
414 (hash, ext, size as i64, cloud_only),
415 )
416 .unwrap();
417 }
418
419 #[tokio::test]
420 async fn upload_sends_local_blob_and_dedups() {
421 let dir = tempdir();
422 let (db, policy) = setup(&dir);
423 let server = FakeBlobs::default();
424 let content = b"blob-one".to_vec();
425 let h = hash_of(&content);
426 seed_meta(&db, &h, "wav", content.len() as u64, 0);
427 std::fs::write(
428 policy.local_path(&BlobRef {
429 hash: h.clone(),
430 ext: "wav".into(),
431 size: 0,
432 }),
433 &content,
434 )
435 .unwrap();
436
437 assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 1);
438 assert!(server.has(&h));
439 // Second run: server already_exists → deduped, no error, still counted handled.
440 assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 1);
441 }
442
443 #[tokio::test]
444 async fn upload_skips_when_local_file_absent() {
445 let dir = tempdir();
446 let (db, policy) = setup(&dir);
447 let server = FakeBlobs::default();
448 seed_meta(&db, &hash_of(b"nope"), "wav", 4, 0); // no file on disk
449 assert_eq!(upload_blobs(&db, &server, &policy).await.unwrap(), 0);
450 }
451
452 #[tokio::test]
453 async fn download_verifies_writes_and_clears_cloud_only() {
454 let dir = tempdir();
455 let (db, policy) = setup(&dir);
456 let server = FakeBlobs::default();
457 let content = b"downloaded-bytes".to_vec();
458 let h = hash_of(&content);
459 server.put_raw(&h, content.clone());
460 seed_meta(&db, &h, "flac", content.len() as u64, 1);
461
462 assert_eq!(download_blobs(&db, &server, &policy).await.unwrap(), 1);
463 let path = policy.local_path(&BlobRef {
464 hash: h.clone(),
465 ext: "flac".into(),
466 size: 0,
467 });
468 assert_eq!(std::fs::read(&path).unwrap(), content);
469 // on_downloaded cleared cloud_only.
470 let co: i64 = db
471 .open()
472 .unwrap()
473 .query_row("SELECT cloud_only FROM blobmeta WHERE hash=?1", [&h], |r| {
474 r.get(0)
475 })
476 .unwrap();
477 assert_eq!(co, 0);
478 // No temp file left behind.
479 assert!(
480 !path
481 .with_file_name(format!("{h}.flac.downloading"))
482 .exists()
483 );
484 }
485
486 #[tokio::test]
487 async fn download_rejects_corrupt_bytes_and_writes_nothing() {
488 let dir = tempdir();
489 let (db, policy) = setup(&dir);
490 let server = FakeBlobs::default();
491 let h = hash_of(b"the-real-content");
492 server.put_raw(&h, b"WRONG BYTES".to_vec()); // hash(bytes) != h
493 seed_meta(&db, &h, "wav", 16, 1);
494
495 // Non-fatal: download_blobs logs and counts 0; file not written.
496 assert_eq!(download_blobs(&db, &server, &policy).await.unwrap(), 0);
497 let path = policy.local_path(&BlobRef {
498 hash: h.clone(),
499 ext: "wav".into(),
500 size: 0,
501 });
502 assert!(!path.exists());
503 // And download_one surfaces the integrity error directly.
504 let err = download_one(
505 &db,
506 &server,
507 &policy,
508 &BlobRef {
509 hash: h,
510 ext: "wav".into(),
511 size: 16,
512 },
513 )
514 .await;
515 assert!(matches!(err, Err(SyncKitError::IntegrityFailed { .. })));
516 }
517
518 #[tokio::test]
519 async fn download_skips_when_already_present() {
520 let dir = tempdir();
521 let (db, policy) = setup(&dir);
522 let server = FakeBlobs::default();
523 let content = b"already-here".to_vec();
524 let h = hash_of(&content);
525 seed_meta(&db, &h, "wav", content.len() as u64, 1);
526 let path = policy.local_path(&BlobRef {
527 hash: h.clone(),
528 ext: "wav".into(),
529 size: 0,
530 });
531 std::fs::write(&path, &content).unwrap();
532 // Server does NOT have it; if the engine tried to download it would fail,
533 // but it must skip because the file is present.
534 assert_eq!(download_blobs(&db, &server, &policy).await.unwrap(), 0);
535 }
536
537 #[tokio::test]
538 async fn reconcile_flags_missing_file_cloud_only() {
539 let dir = tempdir();
540 let (db, policy) = setup(&dir);
541 let server = FakeBlobs::default();
542 // A row claims local presence (cloud_only=0) but no file exists.
543 let h = hash_of(b"ghost");
544 seed_meta(&db, &h, "wav", 5, 0);
545 sync_blobs(&db, &server, &policy).await.unwrap();
546 let co: i64 = db
547 .open()
548 .unwrap()
549 .query_row("SELECT cloud_only FROM blobmeta WHERE hash=?1", [&h], |r| {
550 r.get(0)
551 })
552 .unwrap();
553 assert_eq!(co, 1, "reconcile flips a missing-file row to cloud_only");
554 }
555 }
556