Skip to main content

max / makenotwork

Report the upload outcome and stream from the SyncStore engine blob_upload_streaming now returns BlobUploadOutcome rather than (), because every caller has to tell the two cases apart: only an upload that sent bytes has anything to confirm. The SyncStore BlobTransport seam follows. upload_url + upload collapse into one upload_file(hash, path), so the engine hands over a path and can no longer be the reason a blob is read whole into memory; dedup moves inside the transport, where it runs before any read. BlobUploadUrl goes away with them. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 15:52 UTC
Commit: 06e61c055bec6631ee30594f7b20872e3a4093fa
Parent: 436a4eb
8 files changed, +95 insertions, -60 deletions
M shared/synckit-client/docs/architecture.md +11 -4
@@ -233,12 +233,19 @@ Pulling with cursor 0 fetches from the beginning.
233 233 Binary blobs (files, images, audio samples) are encrypted client-side before
234 234 upload:
235 235
236 - 1. `blob_upload_url(hash, size)` requests a presigned S3 PUT URL from the server.
237 - If the blob already exists (content-addressed by hash), no upload is needed.
236 + 1. `blob_upload_url(hash, plaintext_size)` requests a presigned S3 PUT URL from
237 + the server. If the blob already exists (content-addressed by hash), no upload
238 + is needed. The declared size is converted to the ciphertext length before it
239 + is sent, because the server signs it as `Content-Length` and the PUT carries
240 + sealed bytes.
238 241 2. `blob_upload(hash, presigned_url, data)` encrypts the plaintext bytes with the
239 242 master key, **binding the content hash as AEAD associated data**, and PUTs the
240 243 ciphertext to S3.
241 244 3. `blob_confirm(hash, size)` tells the server the upload completed.
245 +
246 + Steps 1 and 2 are the in-memory path, for bytes with no file behind them. Every
247 + consumer app instead calls `blob_upload_streaming(hash, path)` (below), which
248 + covers the same ground with bounded memory and no size ceiling, then step 3.
242 249 4. `blob_download_url(hash)` requests a presigned GET URL.
243 250 5. `blob_download(expected_hash, presigned_url)` fetches the ciphertext, decrypts
244 251 with the hash as AAD, then **re-verifies** that the plaintext hashes to
@@ -621,8 +628,8 @@ pub trait BlobPolicy: Send + Sync {
621 628 pub struct BlobRef { pub hash: String, pub ext: String, pub size: u64 }
622 629 ```
623 630
624 - The engine owns the invariant parts: the `blob_upload_url` → dedup-skip-if-exists
625 - → skip-if-`local_path`-absent → `blob_upload` → `blob_confirm` dance, the
631 + The engine owns the invariant parts: the skip-if-`local_path`-absent →
632 + `upload_file` (which streams and dedups server-side) → `blob_confirm` dance, the
626 633 `blob_download_url` → skip-if-`local_path`-present → `blob_download` → **SHA-256
627 634 re-verify** → atomic temp+rename write, and the single-item `download_one(hash)`
628 635 path for a GUI "download now" action. The policy trait only answers "which blobs,"
M shared/synckit-client/docs/integration_patterns.md +10 -4
@@ -162,16 +162,22 @@ let (changes, new_cursor, has_more) = client.pull(device_id, cursor).await?;
162 162 ### Blob Operations (audiofiles only)
163 163
164 164 ```rust
165 - // Upload
166 - let resp = client.blob_upload_url(hash, size).await?;
167 - client.blob_upload(resp.url, data).await?;
168 - client.blob_confirm(hash, size).await?;
165 + // Upload a file-backed blob. Streams it a part at a time, so peak memory is
166 + // one part rather than the whole file, and dedups server-side before reading.
167 + if client.blob_upload_streaming(hash, &path).await? == BlobUploadOutcome::Uploaded {
168 + client.blob_confirm(hash, size).await?;
169 + }
169 170
170 171 // Download
171 172 let url = client.blob_download_url(hash).await?;
172 173 let data = client.blob_download(url).await?;
173 174 ```
174 175
176 + For bytes already in memory (no file to stream), the one-shot pair still exists:
177 + `blob_upload_url(hash, plaintext_size)` then `blob_upload(hash, url, data)`, then
178 + `blob_confirm`. It buffers the whole ciphertext and the server refuses it above
179 + its one-shot ceiling, so prefer the streaming call whenever there is a path.
180 +
175 181 ---
176 182
177 183 ## First-Run Sequence
M shared/synckit-client/src/client/blob.rs +28 -5
@@ -28,6 +28,21 @@ use super::helpers::{check_response, Idempotency};
28 28 /// Generous (4 GiB) so legitimate large media still flow.
29 29 const MAX_BLOB_BYTES: usize = 4 * 1024 * 1024 * 1024;
30 30
31 + /// What [`SyncKitClient::blob_upload_streaming`] did.
32 + ///
33 + /// The caller needs to tell the two apart: only an upload that sent bytes has
34 + /// anything to confirm.
35 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
36 + pub enum BlobUploadOutcome {
37 + /// The parts were uploaded and assembled. Call [`SyncKitClient::blob_confirm`]
38 + /// next to record the blob server-side.
39 + Uploaded,
40 + /// The server already held this content address, so no bytes were sent and
41 + /// there is nothing to confirm. Server-asserted, not verified — see
42 + /// [`BlobUploadUrlResponse::already_exists`] for what a wrong `true` costs.
43 + AlreadyPresent,
44 + }
45 +
31 46 impl SyncKitClient {
32 47 /// Request a presigned upload URL for a blob.
33 48 /// Returns (upload_url, already_exists). If `already_exists` is true,
@@ -136,10 +151,18 @@ impl SyncKitClient {
136 151 /// caller hashed the file in an earlier pass, so a file edited in between
137 152 /// would otherwise be stored under an address that does not describe it.
138 153 ///
139 - /// The caller still calls [`Self::blob_confirm`] afterwards, exactly as with
140 - /// the one-shot path — this method replaces the transport only.
154 + /// Returns [`BlobUploadOutcome`]: on `Uploaded` the caller calls
155 + /// [`Self::blob_confirm`] next, exactly as with the one-shot path (this
156 + /// method replaces the transport only); on `AlreadyPresent` the server
157 + /// already held the content, nothing was read or sent, and there is nothing
158 + /// to confirm. The dedup check happens before any file read or sealing, so
159 + /// a re-sync of content the server already has costs one round trip.
141 160 #[instrument(skip(self, path), fields(path = %path.display()))]
142 - pub async fn blob_upload_streaming(&self, hash: &str, path: &Path) -> Result<()> {
161 + pub async fn blob_upload_streaming(
162 + &self,
163 + hash: &str,
164 + path: &Path,
165 + ) -> Result<BlobUploadOutcome> {
143 166 let master_key = self.require_master_key()?;
144 167
145 168 let meta = tokio::fs::metadata(path).await.map_err(|e| {
@@ -175,7 +198,7 @@ impl SyncKitClient {
175 198 .await?;
176 199
177 200 if start.already_exists {
178 - return Ok(());
201 + return Ok(BlobUploadOutcome::AlreadyPresent);
179 202 }
180 203
181 204 // Every failure past this point leaves uploaded parts that S3 bills for
@@ -217,7 +240,7 @@ impl SyncKitClient {
217 240 )
218 241 .await?;
219 242
220 - Ok(())
243 + Ok(BlobUploadOutcome::Uploaded)
221 244 }
222 245
223 246 /// Seal the file chunk by chunk, cutting the ciphertext at the server's
M shared/synckit-client/src/client/mod.rs +6 -2
@@ -18,8 +18,11 @@
18 18 //! [`list_devices`](SyncKitClient::list_devices).
19 19 //! - **Push/Pull sync**: [`push`](SyncKitClient::push), [`pull`](SyncKitClient::pull),
20 20 //! [`status`](SyncKitClient::status).
21 - //! - **Blob storage**: [`blob_upload_url`](SyncKitClient::blob_upload_url),
22 - //! [`blob_upload`](SyncKitClient::blob_upload), [`blob_confirm`](SyncKitClient::blob_confirm),
21 + //! - **Blob storage**: [`blob_upload_streaming`](SyncKitClient::blob_upload_streaming)
22 + //! (file-backed, bounded memory, the path for large blobs),
23 + //! [`blob_upload_url`](SyncKitClient::blob_upload_url) +
24 + //! [`blob_upload`](SyncKitClient::blob_upload) (in-memory),
25 + //! [`blob_confirm`](SyncKitClient::blob_confirm),
23 26 //! [`blob_download_url`](SyncKitClient::blob_download_url),
24 27 //! [`blob_download`](SyncKitClient::blob_download).
25 28 //!
@@ -51,6 +54,7 @@
51 54
52 55 mod auth;
53 56 mod blob;
57 + pub use blob::BlobUploadOutcome;
54 58 mod encryption;
55 59 pub(crate) mod helpers;
56 60 mod ota;
M shared/synckit-client/src/lib.rs +1 -1
@@ -80,7 +80,7 @@ pub mod types;
80 80
81 81 // Re-exports for convenience
82 82 pub use client::{validate_api_key, SecretToken, SessionInfo, SyncKitClient, SyncKitConfig, SyncNotifyStream};
83 - pub use client::{OtaArtifactUpload, OtaManifest, OtaRelease};
83 + pub use client::{BlobUploadOutcome, OtaArtifactUpload, OtaManifest, OtaRelease};
84 84 pub use oauth::{generate_oauth_state, generate_pkce, states_match, Pkce};
85 85 pub use client::subscription::{
86 86 AccountInfo, AppPricing, BillingInterval, Cents, CheckoutResponse, PriceQuote,
M shared/synckit-client/src/store/blob.rs +25 -36
@@ -12,14 +12,14 @@
12 12 //! and `SyncKitClient` satisfies it in production.
13 13
14 14 use std::future::Future;
15 - use std::path::PathBuf;
15 + use std::path::{Path, PathBuf};
16 16 use std::sync::Arc;
17 17
18 18 use rusqlite::Connection;
19 19 use sha2::{Digest, Sha256};
20 20
21 21 use super::db::DbSource;
22 - use crate::client::SyncKitClient;
22 + use crate::client::{BlobUploadOutcome, SyncKitClient};
23 23 use crate::error::{Result, SyncKitError};
24 24
25 25 /// A content-addressed blob: its SHA-256 hash (the address), file extension, and
@@ -31,27 +31,19 @@ pub struct BlobRef {
31 31 pub size: u64,
32 32 }
33 33
34 - /// A presigned upload target plus whether the server already has the blob.
35 - pub struct BlobUploadUrl {
36 - pub upload_url: String,
37 - pub already_exists: bool,
38 - }
39 -
40 34 /// The blob operations the engine needs from a server. A seam for testing and
41 35 /// the reference a future non-Rust SDK mirrors; [`SyncKitClient`] forwards to its
42 36 /// inherent `blob_*` methods.
43 37 pub trait BlobTransport: Sync {
44 - fn upload_url(
45 - &self,
46 - hash: &str,
47 - size: u64,
48 - ) -> impl Future<Output = Result<BlobUploadUrl>> + Send;
49 - fn upload(
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(
50 43 &self,
51 44 hash: &str,
52 - url: &str,
53 - data: Vec<u8>,
54 - ) -> impl Future<Output = Result<()>> + Send;
45 + path: &Path,
46 + ) -> impl Future<Output = Result<BlobUploadOutcome>> + Send;
55 47 fn confirm(&self, hash: &str, size: u64) -> impl Future<Output = Result<()>> + Send;
56 48 fn download_url(&self, hash: &str) -> impl Future<Output = Result<String>> + Send;
57 49 fn download(&self, hash: &str, url: &str) -> impl Future<Output = Result<Vec<u8>>> + Send;
@@ -60,12 +52,8 @@ pub trait BlobTransport: Sync {
60 52 // The trait declares `-> impl Future + Send`; implementing with `async fn` is
61 53 // equivalent (the Send bound is enforced by the trait) and reads cleaner.
62 54 impl BlobTransport for SyncKitClient {
63 - async fn upload_url(&self, hash: &str, size: u64) -> Result<BlobUploadUrl> {
64 - let r = SyncKitClient::blob_upload_url(self, hash, size as i64).await?;
65 - Ok(BlobUploadUrl { upload_url: r.upload_url, already_exists: r.already_exists })
66 - }
67 - async fn upload(&self, hash: &str, url: &str, data: Vec<u8>) -> Result<()> {
68 - SyncKitClient::blob_upload(self, hash, url, data).await
55 + async fn upload_file(&self, hash: &str, path: &Path) -> Result<BlobUploadOutcome> {
56 + SyncKitClient::blob_upload_streaming(self, hash, path).await
69 57 }
70 58 async fn confirm(&self, hash: &str, size: u64) -> Result<()> {
71 59 SyncKitClient::blob_confirm(self, hash, size as i64).await
@@ -181,8 +169,10 @@ async fn upload_one<T: BlobTransport>(
181 169 if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
182 170 return Ok(false); // no local file to upload
183 171 }
184 - let target = client.upload_url(&blob.hash, blob.size).await?;
185 - if target.already_exists {
172 + // The transport streams from the path, so a blob is never read whole into
173 + // memory here regardless of size, and it runs its own dedup check before
174 + // touching the file.
175 + if client.upload_file(&blob.hash, &path).await? == BlobUploadOutcome::AlreadyPresent {
186 176 // Content-addressed dedup: the server says it holds this hash, so the
187 177 // bytes stay home. Two consequences worth knowing, both bounded:
188 178 // - The claim is unverified. A server that answers `true` wrongly
@@ -191,14 +181,12 @@ async fn upload_one<T: BlobTransport>(
191 181 // - `on_uploaded` is deliberately not fired: nothing was uploaded,
192 182 // and the confirm call that pairs with it never happened. A policy
193 183 // using that hook to clear a presence flag will keep re-offering
194 - // the blob on every pass, costing one `upload_url` round trip each
195 - // time. Harmless for the no-op default (the only implementors
196 - // today), but a policy with an explicit flag needs its own way to
197 - // settle the dedup case.
184 + // the blob on every pass, costing one round trip each time.
185 + // Harmless for the no-op default (the only implementors today), but
186 + // a policy with an explicit flag needs its own way to settle the
187 + // dedup case.
198 188 return Ok(true);
199 189 }
200 - let data = tokio::fs::read(&path).await.map_err(|e| io_err("read", e))?;
201 - client.upload(&blob.hash, &target.upload_url, data).await?;
202 190 client.confirm(&blob.hash, blob.size).await?;
203 191
204 192 let blob = blob.clone();
@@ -286,12 +274,13 @@ mod tests {
286 274 }
287 275 }
288 276 impl BlobTransport for FakeBlobs {
289 - async fn upload_url(&self, hash: &str, _size: u64) -> Result<BlobUploadUrl> {
290 - Ok(BlobUploadUrl { upload_url: "fake://put".into(), already_exists: self.has(hash) })
291 - }
292 - async fn upload(&self, hash: &str, _url: &str, data: Vec<u8>) -> Result<()> {
277 + async fn upload_file(&self, hash: &str, path: &Path) -> Result<BlobUploadOutcome> {
278 + if self.has(hash) {
279 + return Ok(BlobUploadOutcome::AlreadyPresent);
280 + }
281 + let data = std::fs::read(path).map_err(|e| io_err("read", e))?;
293 282 self.store.lock().unwrap().insert(hash.to_string(), data);
294 - Ok(())
283 + Ok(BlobUploadOutcome::Uploaded)
295 284 }
296 285 async fn confirm(&self, _hash: &str, _size: u64) -> Result<()> {
297 286 Ok(())
M shared/synckit-client/src/store/facade.rs +13 -7
@@ -340,7 +340,8 @@ impl SyncStore<SyncKitClient> {
340 340
341 341 #[cfg(test)]
342 342 mod tests {
343 - use super::super::blob::{BlobTransport, BlobUploadUrl};
343 + use super::super::blob::BlobTransport;
344 + use crate::client::BlobUploadOutcome;
344 345 use super::super::schema::SyncTable;
345 346 use super::*;
346 347 use crate::ids::{AppId, DeviceId, UserId};
@@ -421,13 +422,18 @@ mod tests {
421 422 }
422 423
423 424 impl BlobTransport for Fake {
424 - async fn upload_url(&self, hash: &str, _size: u64) -> Result<BlobUploadUrl> {
425 - let exists = self.blobs.lock().unwrap().contains_key(hash);
426 - Ok(BlobUploadUrl { upload_url: "fake".into(), already_exists: exists })
427 - }
428 - async fn upload(&self, hash: &str, _url: &str, data: Vec<u8>) -> Result<()> {
425 + async fn upload_file(
426 + &self,
427 + hash: &str,
428 + path: &std::path::Path,
429 + ) -> Result<BlobUploadOutcome> {
430 + if self.blobs.lock().unwrap().contains_key(hash) {
431 + return Ok(BlobUploadOutcome::AlreadyPresent);
432 + }
433 + let data = std::fs::read(path)
434 + .map_err(|e| crate::error::SyncKitError::Internal(e.to_string()))?;
429 435 self.blobs.lock().unwrap().insert(hash.to_string(), data);
430 - Ok(())
436 + Ok(BlobUploadOutcome::Uploaded)
431 437 }
432 438 async fn confirm(&self, _hash: &str, _size: u64) -> Result<()> {
433 439 Ok(())
M shared/synckit-client/src/store/mod.rs +1 -1
@@ -29,7 +29,7 @@ pub mod sync;
29 29 pub use apply::{apply_remote_changes, ApplyOutcome};
30 30 pub use blob::{
31 31 download_blobs, download_one, sync_blobs, upload_blobs, BlobOutcome, BlobPolicy, BlobRef,
32 - BlobTransport, BlobUploadUrl,
32 + BlobTransport,
33 33 };
34 34 pub use facade::{SchedulerHandle, SyncConfig, SyncOutcome, SyncStore, SyncStoreBuilder};
35 35 pub use scheduler::{NoopObserver, SyncObserver, SyncState};