Skip to main content

coven_core/blob/
mod.rs

1//! The blob engine: coven's single owner of a blob's whole durability lifecycle.
2//!
3//! coven syncs opaque encrypted blobs referenced by DB rows. By default it owns
4//! the cloud layout (the content-addressed `{namespace}/{ab}/{cd}/{id}`) and
5//! encryption; the host decides which rows carry blobs, where their plaintext
6//! lives locally, and how each is scoped for encryption. A home configured for
7//! the unobfuscated blob-path scheme instead stores each blob at the consumer's
8//! readable [`BlobRef::cloud_path`] so the bucket is browsable.
9//!
10//! # The coven concept tree
11//!
12//! A blob has two **declared** properties — [`Provenance`] (its Local story) and
13//! [`CacheFill`] (its Remote story) — and one **state**, locality, flipped by the
14//! transitions. The cache is a *mechanism* that serves Remote blobs; it is not a
15//! kind of blob.
16//!
17//! ```text
18//! A blob the host declares with:
19//!
20//! provenance — its LOCAL story: where the bytes live when Local, and the
21//!              Remote→Local path requirement
22//!    ├─ user-provided   the user's file at a path; coven references it.
23//!    │                  Remote→Local writes the bytes back to a user file → NEEDS A PATH.
24//!    └─ host-provided   bae hands coven the data; coven keeps it in its local store.
25//!                       Remote→Local restores it to the local store → no path.
26//!
27//! cache fill — its REMOTE story: how a device gets the bytes when the release is
28//!              Remote. A cache-mechanism setting; applies to ANY blob, regardless of
29//!              provenance, once it is Remote.
30//!    ├─ CacheEager   fetched into the cache on pull, with the SQL row   (covers)
31//!    └─ CacheLazy    fetched into the cache on first read               (audio — big, fetch what you play)
32//!
33//! and a current state:
34//!
35//! locality
36//!    ├─ Local    bytes on-device — the user's path (user-provided) or coven's local store (host-provided)
37//!    └─ Remote   bytes in the cloud; each device's local copy is a CACHE copy, filled
38//!                per `cache fill`, kept-or-evicted per `pin`
39//!
40//! namespace (bucket)   the blob's category — release_files · covers · artist_images
41//!
42//! transitions
43//!    ├─ Local → Remote   upload the bytes; now cache-distributed to every device per cache fill
44//!    └─ Remote → Local   bring the bytes back to a local file — path required iff user-provided
45//!
46//! cache budget   per-NAMESPACE size limit; each namespace evicts independently, so
47//!                evicting release_files (big) never touches covers (small reserved slice)
48//! pin            keep one specific Remote blob's cache copy from eviction (e.g. a
49//!                release the user pinned for offline)
50//! ```
51//!
52//! ## The cache vs local files
53//!
54//! The cache holds local copies of **Remote** blobs (filled per `cache fill`,
55//! evicted per budget unless pinned). It is **segmented by namespace**: each
56//! namespace has its own configurable cache budget and evicts independently, so
57//! evicting `release_files` (big) never touches `covers` (a small reserved slice). A
58//! `CacheEager` cover that falls out of its namespace budget shows a placeholder
59//! until the next read re-fetches it — covers are not pinned. A **Local** blob is not
60//! in the cache: a user-provided Local blob is the user's file at its path (an
61//! external ref); a host-provided Local blob is in coven's local store (see
62//! [`local_files`]). The cache is the mechanism for *remoteness* — so
63//! `CacheEager`/`CacheLazy`/pin/budget describe a blob only while it is Remote, never
64//! while it is Local.
65//!
66//! # The engine's halves
67//!
68//! This module is the engine; its halves move a blob through its lifecycle:
69//!
70//! - [`cache`] — the device-local cache for **Remote** blobs: bytes on disk keyed
71//!   by exact locator hash, with the folder a file lives in as the only retention truth
72//!   (`storage/pinned/` protected, `storage/cache/` evictable). Read (whole and
73//!   ranged), pin/unpin, clear, and budget eviction.
74//! - [`local_files`] — coven's own copy of a **host-provided Local** blob, in the
75//!   local store (`storage/local/<namespace>/<id>`). Never evicted; the budget
76//!   sweep never walks it. Store, read, drop.
77//! - [`upload`] — the cloud-write half: drain the durable upload queue, sealing
78//!   each blob under its scope and writing it to the cloud with coalesced progress,
79//!   so a local-only blob becomes uploaded. The sync cycle calls the drain
80//!   each round before it pushes.
81//! - [`delete`] — the cloud-delete half: turn a queued deletion into a signed
82//!   cloud tombstone, hold the blob for a convergence grace so a lagging peer
83//!   isn't stranded, then GC the blob once the grace has passed. The sync cycle
84//!   drains tombstones and runs the GC each round after it pulls.
85//!
86//! The types below ([`BlobRef`], [`BlobScope`], [`Provenance`],
87//! [`CacheFill`], [`BlobTransitionObserver`]) are the vocabulary both halves and
88//! the host speak. Which rows carry blobs is not a runtime callback but a per-table
89//! declaration ([`crate::sync::session::BlobDecl`]) coven resolves into a
90//! [`decl::BlobDecls`] each cycle to derive the blob set itself.
91//!
92//! coven also owns the two locality transitions ([`transition`]): `make_remote`
93//! (Local → Remote: upload the bytes, then flip the gate) and `make_local`
94//! (Remote → Local: bring each blob back to a local file, then retract). The
95//! make-Remote *completion* — flipping the gate the instant the last user-provided
96//! upload lands — lives in the [`upload`] drain, the one place that knows an upload
97//! just succeeded.
98
99pub mod cache;
100#[cfg(test)]
101mod cache_tests;
102pub mod decl;
103pub mod delete;
104#[doc(hidden)]
105pub mod local_cleanup;
106pub mod local_files;
107pub mod locator;
108#[cfg(test)]
109mod row_ref_tests;
110pub mod transition;
111pub mod upload;
112
113pub use delete::BLOB_TOMBSTONE_GRACE;
114
115use sha2::{Digest, Sha256};
116
117/// The content hash a blob-bearing row carries: the lowercase-hex SHA-256 of the
118/// blob's plaintext bytes, computed at import and stored in the row's blob columns
119/// alongside the declared size. The row is carried in a signed changeset (and in a
120/// signed snapshot), so this hash is signed by the row's author — that is what
121/// makes it authoritative: on download coven hashes the decrypted plaintext and
122/// requires equality with the row's hash, so the bytes are pinned by the author,
123/// not by the cloud key they happened to arrive under. A host computes this over a
124/// blob's plaintext at import and writes it into the row's declared hash column,
125/// the same way it writes the plaintext length into the size column.
126pub fn content_hash(plaintext: &[u8]) -> String {
127    hex::encode(Sha256::digest(plaintext))
128}
129
130/// An incremental SHA-256 over a blob's plaintext, so the streaming download path
131/// verifies a large blob's content hash without ever holding the whole plaintext
132/// in memory — feed each decrypted chunk to [`update`](Self::update) as it is read,
133/// then [`verify`](Self::verify) against the row's hash before the bytes are
134/// committed to the cache. The hex-encoded digest matches [`content_hash`] over the
135/// same bytes.
136pub struct ContentHasher(Sha256);
137
138impl ContentHasher {
139    pub fn new() -> Self {
140        ContentHasher(Sha256::new())
141    }
142
143    /// Fold the next plaintext chunk into the running digest.
144    pub fn update(&mut self, chunk: &[u8]) {
145        self.0.update(chunk);
146    }
147
148    /// The lowercase-hex digest of everything fed so far.
149    pub fn finish(self) -> String {
150        hex::encode(self.0.finalize())
151    }
152}
153
154impl Default for ContentHasher {
155    fn default() -> Self {
156        Self::new()
157    }
158}
159
160/// How many blob transfers coven runs at once in each of its two transfer loops:
161/// the upload drain ([`upload::drain_uploads`]) and the pin/download loop
162/// ([`cache::pin`]). An open-time blob-engine tunable the host sets on the builder,
163/// carried on [`Database`](crate::database::Database) alongside the other open-time
164/// blob config and read back by each loop, which holds `&Database`.
165///
166/// Each bound is a [`NonZeroUsize`], so a zero — which would leave a loop admitting
167/// nothing and never completing — is unrepresentable rather than clamped or rejected
168/// at open. `one_at_a_time()` (both `1`) is the default: transfers run one at a
169/// time in queue order.
170///
171/// [`NonZeroUsize`]: std::num::NonZeroUsize
172#[derive(Debug, Clone, Copy, PartialEq, Eq)]
173pub struct TransferLimits {
174    /// Maximum concurrent blob uploads in one upload-drain pass.
175    pub uploads: std::num::NonZeroUsize,
176    /// Maximum concurrent blob downloads (fetches) in one pin call.
177    pub downloads: std::num::NonZeroUsize,
178}
179
180impl TransferLimits {
181    /// One at a time in each loop.
182    pub fn one_at_a_time() -> Self {
183        Self {
184            uploads: std::num::NonZeroUsize::MIN,
185            downloads: std::num::NonZeroUsize::MIN,
186        }
187    }
188}
189
190// The cache's own tests: real `Database` + `TestStore` over a temp store
191// dir, asserting hits/misses, the pinned/cache folder split, and pin/unpin/clear.
192// These drive a real temp directory on the filesystem. See [`cache`].
193// The upload drain's tests: real `Database` (the `cloud_outbox` queue) driven
194// against `InMemoryCloudHome`/`FailingCloudHome`, asserting record-and-continue,
195// per-entry backoff, scope-resolved sealing, and the observer callbacks. See
196// [`upload`].
197#[cfg(test)]
198mod upload_tests;
199// The coven-owned make-Remote / make-Local transition tests: multi-device
200// make_remote + make_local through the real cycle, cancel both directions, the
201// drain's completion flip, durable cancellation, crash-idempotency at each commit
202// boundary, and a round-trip. Uses a `watch` cancel signal + `run_single_sync_cycle`,
203// See [`transition`].
204#[cfg(test)]
205mod transition_tests;
206// The local-files store's tests: store/read round-trip, a host-provided Local blob
207// surviving a budget sweep (the sweep never walks `local/`), and drop. These
208// drive a real temp directory. See [`local_files`].
209#[cfg(test)]
210mod local_files_tests;
211// The delete half's tests: tombstone signing, the drain that writes tombstones,
212// the graced GC that reclaims exact immutable objects, and the delete-outbox row
213// shape. Driven against `InMemoryCloudHome` and
214// `TestStore`. See [`delete`].
215#[cfg(test)]
216mod delete_tests;
217
218/// Which key encrypts a blob, as a host names it on a [`BlobRef`].
219///
220/// The host names *what* a blob is scoped to — the whole store or a derived
221/// per-scope key — never the raw key bytes. Storage and encryption consume this
222/// same type; there is no key material in it to leak.
223#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
224pub enum BlobScope {
225    /// The store master key — every member reads it.
226    Master,
227    /// A per-scope key derived from the master key (e.g. one key per item).
228    Derived(String),
229}
230
231impl BlobScope {
232    /// Serialize for the `cloud_outbox.scope` column. The audio outbox persists
233    /// the scope at enqueue and resolves it to a key at drain, so the string
234    /// must round-trip every variant. The variant tag is split from the payload
235    /// at the first `:`; the payload (a derived scope name) is stored verbatim,
236    /// so it may itself contain `:`.
237    pub fn to_outbox_str(&self) -> String {
238        match self {
239            BlobScope::Master => "master".to_string(),
240            BlobScope::Derived(s) => format!("derived:{s}"),
241        }
242    }
243
244    /// Parse a `cloud_outbox.scope` value written by [`Self::to_outbox_str`].
245    /// Returns `None` on an unknown tag (a corrupt row), which the drain surfaces
246    /// rather than silently defaulting to the master key.
247    pub fn from_outbox_str(s: &str) -> Option<Self> {
248        match s.split_once(':') {
249            None if s == "master" => Some(BlobScope::Master),
250            Some(("derived", rest)) => Some(BlobScope::Derived(rest.to_string())),
251            _ => None,
252        }
253    }
254}
255
256/// A blob's **Local story**: where its bytes live while the blob is Local, and
257/// whether bringing it back from Remote needs a destination path. Orthogonal to
258/// [`CacheFill`] (the Remote story) — a blob declares both.
259///
260/// The cache never enters into this: a Local blob is not a cache copy. Provenance
261/// decides which of the two Local homes holds it, and what `make_local` does to
262/// restore it.
263#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
264pub enum Provenance {
265    /// The user's own file at a path; coven references it but does not own it
266    /// (tracked as an external ref — see `local_blob_refs`). `make_local` writes
267    /// the bytes back to a user file, so it **needs a destination path**.
268    UserProvided,
269    /// The host hands coven the data; coven keeps its own copy in the local store
270    /// (`storage/local/<namespace>/<id>`, see [`local_files`]). `make_local`
271    /// restores it to the local store, so it needs **no path**.
272    HostProvided,
273}
274
275/// A blob's **Remote story**: how a device gets the bytes once the blob is Remote.
276/// A cache-mechanism setting — it describes a blob only while Remote — that applies
277/// to ANY blob regardless of [`Provenance`]. Orthogonal to provenance; a blob
278/// declares both.
279///
280/// Both classes are declared per blob and are global (every device reads the same
281/// class from the blob's [`BlobRef`]); the difference is what a device does with
282/// the blob on pull. The distinction has to be a declared property and not a
283/// per-device choice: device B, deciding during its own pull whether to fetch a
284/// blob, can only read the blob's declared class — it cannot see what device A
285/// chose locally.
286#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
287pub enum CacheFill {
288    /// Fetched into the cache on pull, right away, on every device — part of
289    /// "having the store" (e.g. cover art, so the grid renders from local bytes
290    /// without a fetch). The cache copy is evictable + re-fetchable, not pinned.
291    CacheEager,
292    /// Not fetched on pull: a pulling device skips it and fetches it into the cache
293    /// on first read — e.g. audio, which is big and streams on demand.
294    CacheLazy,
295}
296
297/// A blob's **replacement story**: whether the row carrying it may ever be repointed at
298/// a different blob. Orthogonal to [`Provenance`] and [`CacheFill`]; a blob declares all
299/// three.
300///
301/// It exists because a cloud object must never be rewritten with different bytes. The
302/// pull verifies an object against its row's content hash and a position advances only over
303/// a fully-realized changeset, so a key whose content can change leaves a device that
304/// pulls an older changeset unable to satisfy it — wedged there for good, not merely
305/// missing a blob. Two declarations reach that guarantee by different routes, and coven
306/// enforces whichever one the blob declares:
307///
308/// - [`Replaceable`](Self::Replaceable) — the row may be repointed, so the *key* must
309///   move with the blob: a readable `cloud_path` has to name its blob
310///   ([`crate::blob::decl::cloud_path_names_blob`]), and a replacement then writes a new
311///   object beside the one it replaces instead of over it.
312/// - [`WriteOnce`](Self::WriteOnce) — the row is never repointed, so the object at its
313///   key is written once and there is nothing to protect it from. Its path is free to be
314///   a stable, fully readable name. coven refuses the repointing.
315///
316/// `Replaceable` is the default: its guarantee is the airtight one (the key itself
317/// carries the blob id, so no path can ever be reused), while `WriteOnce` is a weaker
318/// contract a consumer opts into knowingly — see its docs.
319#[derive(Debug, Clone, Copy, PartialEq, Eq)]
320pub enum BlobReplacement {
321    /// The row may be repointed at a new blob id — replacing a cover, swapping an
322    /// attachment. Requires a readable `cloud_path` that names its blob, so that the
323    /// replacement's fresh blob id yields a fresh key.
324    Replaceable,
325    /// The row is never repointed: the blob it names when it is inserted is the blob it
326    /// names for life. Repointing one is refused.
327    ///
328    /// This buys a stable, fully readable cloud path — `Live at Leeds/01 Sonata.flac`
329    /// rather than `01 Sonata-0ef7a1c9.flac` — for content that is written once and never
330    /// rewritten: an imported file, whose bytes are what they are.
331    ///
332    /// **What coven enforces, and what it does not.** coven refuses to repoint the row,
333    /// which is the reuse it can see. It cannot see a consumer *deleting* a row and
334    /// inserting a different blob at the same `cloud_path` — the deleted row is gone, and
335    /// coven keeps no history of the paths it has used. Declaring `WriteOnce` is therefore
336    /// also a promise that the path is never reused by a different blob. Derive it from
337    /// data that never repeats and it holds by construction: a path carrying a freshly
338    /// minted id for the thing being imported can never be handed out twice.
339    WriteOnce,
340}
341
342/// A blob a row references: its cloud identity, encryption scope, and the two
343/// declared properties ([`provenance`](BlobRef::provenance) +
344/// [`fill`](BlobRef::fill)). coven derives it from the row's declared columns
345/// ([`crate::sync::session::BlobDecl`]) via [`decl::BlobDecls`]. Where its bytes
346/// live depends on its locality and provenance: a user-provided Local blob is the
347/// user's file at its path; a host-provided Local blob is in coven's local store
348/// (`storage/local/<namespace>/<id>`); a Remote blob's device-local copy is a cache
349/// copy (`storage/pinned/<namespace>/<locator-hash>` /
350/// `storage/cache/<namespace>/<locator-hash>`, built
351/// from the validated namespace + exact locator hash — see [`cache`]).
352#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
353pub struct BlobRef {
354    /// Cloud namespace, e.g. `"images"`. Becomes `{namespace}/{ab}/{cd}/{id}`
355    /// under the hashed scheme, or `{namespace}/{cloud_path}` under the plain one.
356    pub namespace: String,
357    /// Blob id (typically the id of the blob-bearing row).
358    pub id: String,
359    /// Encryption scope for this blob.
360    pub scope: BlobScope,
361    /// The consumer's readable cloud-relative path for this blob, e.g.
362    /// `"Artist - Album/cover.jpg"`. Used as the object key under `namespace` when
363    /// the home's [`crate::sync::cloud_storage::BlobPathScheme`] is `Plain`;
364    /// ignored when `Hashed`. `None` is only valid for a `Hashed` home — a `Plain`
365    /// home with no `cloud_path` is a surfaced error, never a silent fallback.
366    pub cloud_path: Option<String>,
367    /// The blob's **Local story**: where its bytes live while Local, and whether
368    /// `make_local` needs a destination path. See [`Provenance`].
369    pub provenance: Provenance,
370    /// The blob's **Remote story**: whether a pulling device fetches it into the
371    /// cache right away ([`CacheFill::CacheEager`]) or on first read
372    /// ([`CacheFill::CacheLazy`]). See [`CacheFill`].
373    pub fill: CacheFill,
374}
375
376/// One exact blob-bearing row version. A reference becomes stale when the live
377/// row stamp or any declared blob value changes.
378#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
379#[serde(deny_unknown_fields)]
380pub struct RowBlobRef {
381    table: String,
382    row_id: String,
383    row_stamp: String,
384    column: String,
385    blob: BlobRef,
386    plaintext_size: u64,
387    plaintext_hash: crate::sync::store_commit::ObjectHash,
388    authority: RowBlobAuthority,
389    stored: Option<locator::StoredBlobRef>,
390}
391
392/// The authority state that determines where one row version's blob lives.
393/// A remote-audience blob remains `PendingRemote` while its verified plaintext
394/// is local and no cloud object has been created; `Remote` carries the exact
395/// package authority needed to open its committed object.
396#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
397#[serde(rename_all = "snake_case", deny_unknown_fields)]
398pub enum RowBlobAuthority {
399    Local,
400    PendingRemote(locator::RemoteAudience),
401    Remote(crate::sync::audience_package::PackageAudience),
402}
403
404impl<'de> serde::Deserialize<'de> for RowBlobRef {
405    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
406    where
407        D: serde::Deserializer<'de>,
408    {
409        #[derive(serde::Deserialize)]
410        #[serde(deny_unknown_fields)]
411        struct Fields {
412            table: String,
413            row_id: String,
414            row_stamp: String,
415            column: String,
416            blob: BlobRef,
417            plaintext_size: u64,
418            plaintext_hash: crate::sync::store_commit::ObjectHash,
419            authority: RowBlobAuthority,
420            stored: Option<locator::StoredBlobRef>,
421        }
422
423        let fields = Fields::deserialize(deserializer)?;
424        Self::new(
425            fields.table,
426            fields.row_id,
427            fields.row_stamp,
428            fields.column,
429            fields.blob,
430            fields.plaintext_size,
431            fields.plaintext_hash,
432            fields.authority,
433            fields.stored,
434        )
435        .map_err(serde::de::Error::custom)
436    }
437}
438
439impl RowBlobAuthority {
440    pub fn audience(&self) -> crate::sync::circle::Audience {
441        match self {
442            Self::Local => crate::sync::circle::Audience::Local,
443            Self::PendingRemote(locator::RemoteAudience::Store) => {
444                crate::sync::circle::Audience::Store
445            }
446            Self::PendingRemote(locator::RemoteAudience::Circle(circle_id)) => {
447                crate::sync::circle::Audience::Circle(*circle_id)
448            }
449            Self::Remote(crate::sync::audience_package::PackageAudience::Store) => {
450                crate::sync::circle::Audience::Store
451            }
452            Self::Remote(crate::sync::audience_package::PackageAudience::Circle {
453                circle_id,
454                ..
455            }) => crate::sync::circle::Audience::Circle(*circle_id),
456        }
457    }
458}
459
460impl RowBlobRef {
461    #[allow(clippy::too_many_arguments)]
462    pub(crate) fn new(
463        table: String,
464        row_id: String,
465        row_stamp: String,
466        column: String,
467        blob: BlobRef,
468        plaintext_size: u64,
469        plaintext_hash: crate::sync::store_commit::ObjectHash,
470        authority: RowBlobAuthority,
471        stored: Option<locator::StoredBlobRef>,
472    ) -> Result<Self, String> {
473        let remote = match &authority {
474            RowBlobAuthority::Local => None,
475            RowBlobAuthority::PendingRemote(audience) => Some(audience.clone()),
476            RowBlobAuthority::Remote(package) => Some(package.remote_audience()),
477        };
478        match (&authority, remote.as_ref(), stored.as_ref()) {
479            (RowBlobAuthority::Local, None, None)
480            | (RowBlobAuthority::PendingRemote(_), Some(_), None) => {}
481            (RowBlobAuthority::Remote(_), Some(expected), Some(stored))
482                if &stored.locator().audience() == expected => {}
483            (RowBlobAuthority::Local, None, Some(_)) => {
484                return Err("Local row blob carries a remote locator".to_string());
485            }
486            (RowBlobAuthority::PendingRemote(_), Some(_), Some(_)) => {
487                return Err("pending remote row blob carries a cloud locator".to_string());
488            }
489            (RowBlobAuthority::Remote(_), Some(_), None) => {
490                return Err("remote row blob has no exact locator".to_string());
491            }
492            (RowBlobAuthority::Remote(_), Some(expected), Some(stored)) => {
493                return Err(format!(
494                    "row audience {expected:?} differs from locator audience {:?}",
495                    stored.locator().audience()
496                ));
497            }
498            _ => unreachable!("authority determines whether a remote audience exists"),
499        }
500        if let Some(stored) = &stored {
501            let locator = stored.locator();
502            if locator.namespace() != blob.namespace {
503                return Err(format!(
504                    "row blob namespace {:?} differs from locator namespace {:?}",
505                    blob.namespace,
506                    locator.namespace()
507                ));
508            }
509            if locator.blob_id() != blob.id {
510                return Err(format!(
511                    "row blob id {:?} differs from locator id {:?}",
512                    blob.id,
513                    locator.blob_id()
514                ));
515            }
516            if locator.plaintext_size() != plaintext_size
517                || locator.plaintext_hash() != plaintext_hash
518            {
519                return Err(
520                    "row blob plaintext size or hash differs from its exact locator".to_string(),
521                );
522            }
523            match locator {
524                locator::BlobLocator::Opaque {
525                    scope,
526                    key_fingerprint,
527                    ..
528                } => {
529                    if scope != &blob.scope {
530                        return Err(
531                            "row blob encryption scope differs from its exact locator".to_string()
532                        );
533                    }
534                    if let RowBlobAuthority::Remote(
535                        crate::sync::audience_package::PackageAudience::Circle {
536                            key_fingerprint: expected,
537                            ..
538                        },
539                    ) = &authority
540                    {
541                        if key_fingerprint != expected {
542                            return Err(
543                                "row blob Circle key differs from its exact locator".to_string()
544                            );
545                        }
546                    }
547                }
548                locator::BlobLocator::Browsable { cloud_path, .. } => {
549                    if blob.cloud_path.as_deref() != Some(cloud_path) {
550                        return Err(
551                            "row blob cloud path differs from its exact locator".to_string()
552                        );
553                    }
554                }
555            }
556        }
557        Ok(Self {
558            table,
559            row_id,
560            row_stamp,
561            column,
562            blob,
563            plaintext_size,
564            plaintext_hash,
565            authority,
566            stored,
567        })
568    }
569
570    pub fn table(&self) -> &str {
571        &self.table
572    }
573
574    pub fn row_id(&self) -> &str {
575        &self.row_id
576    }
577
578    pub fn row_stamp(&self) -> &str {
579        &self.row_stamp
580    }
581
582    pub fn column(&self) -> &str {
583        &self.column
584    }
585
586    pub fn blob(&self) -> &BlobRef {
587        &self.blob
588    }
589
590    pub fn plaintext_size(&self) -> u64 {
591        self.plaintext_size
592    }
593
594    pub fn plaintext_hash(&self) -> crate::sync::store_commit::ObjectHash {
595        self.plaintext_hash
596    }
597
598    pub fn authority(&self) -> &RowBlobAuthority {
599        &self.authority
600    }
601
602    pub fn audience(&self) -> crate::sync::circle::Audience {
603        self.authority.audience()
604    }
605
606    pub fn stored(&self) -> Option<&locator::StoredBlobRef> {
607        self.stored.as_ref()
608    }
609}
610
611/// Notified about coven's blob transitions, for host-specific bookkeeping and UI:
612/// per-blob upload progress while a make_remote uploads, per-blob materialize
613/// progress while a make_local copies files back, and a completion hook per
614/// direction the host turns into its own UI event.
615///
616/// The host no longer drives the transition — coven owns flipping the gate and
617/// deciding when a cycle publishes — so this observer only *reports*. The upload
618/// callbacks fire as the drain works: `on_blob_upload_started` before each
619/// attempt, `on_blob_upload_progress` zero or more times as encrypted bytes reach
620/// the cloud (backends that can't report sub-file progress call it once at the end
621/// with `bytes_done == bytes_total`), `on_blob_uploaded` on success (notification
622/// only — coven, not the host, flips the gate and breaks the drain to publish),
623/// and `on_blob_upload_failed` when an attempt fails and its entry stays queued.
624///
625/// `on_root_made_remote` / `on_root_made_local` fire whenever coven *completes* a
626/// transition — including one resumed after a restart — so the host's own
627/// row-updated event survives a restart rather than being lost with an in-memory
628/// flag. `on_blob_materialize_progress` moves a make_local's per-file progress bar.
629///
630/// `should_skip_uploads` lets the host pause the upload pipeline without touching
631/// the queue: the sync cycle consults it before draining so a paused queue still
632/// accepts new entries but doesn't drain ([`upload::drain_uploads`] checks once at
633/// the top of each entry; in-flight uploads complete normally).
634///
635#[async_trait::async_trait]
636pub trait BlobTransitionObserver: Send + Sync {
637    /// An upload attempt for this blob is starting now.
638    async fn on_blob_upload_started(&self, blob_id: &str);
639
640    /// `bytes_done` of `bytes_total` encrypted bytes have reached the cloud for
641    /// this in-flight blob. `bytes_done` is cumulative and monotonic within one
642    /// upload attempt. The default is a no-op so observers that don't surface
643    /// sub-file progress don't need a stub.
644    async fn on_blob_upload_progress(&self, blob_id: &str, bytes_done: u64, bytes_total: u64) {
645        let _ = (blob_id, bytes_done, bytes_total);
646    }
647
648    /// The blob was uploaded to the cloud successfully — notification only. coven
649    /// owns flipping the gate and breaking the drain to publish a completed
650    /// make_remote.
651    async fn on_blob_uploaded(&self, blob_id: &str);
652
653    /// An upload attempt failed; the entry remains queued for retry.
654    async fn on_blob_upload_failed(&self, blob_id: &str, error: &str);
655
656    /// If true, the sync cycle skips the upload drain this round and
657    /// [`upload::drain_uploads`] short-circuits before pulling the next queued
658    /// entry. The default is `false` so existing implementations don't need a stub.
659    fn should_skip_uploads(&self) -> bool {
660        false
661    }
662
663    /// coven completed a make_remote of `(root_table, root_id)`: every blob is
664    /// uploaded and the gate is flipped true (the subtree publishes this cycle).
665    /// Fires for a restart-resumed completion too, so the host's row-updated event
666    /// is not lost to a crash. The default is a no-op.
667    async fn on_root_made_remote(&self, root_table: &str, root_id: &str) {
668        let _ = (root_table, root_id);
669    }
670
671    /// coven completed a make_local of `(root_table, root_id)`: every blob is back
672    /// to a local file (a user file for user-provided, the local store for
673    /// host-provided), the gate is flipped false (the subtree retracts from peers),
674    /// and the cloud blobs are queued for tombstoning. The default is a no-op.
675    async fn on_root_made_local(&self, root_table: &str, root_id: &str) {
676        let _ = (root_table, root_id);
677    }
678
679    /// `done` of `total` of a make_local's blobs have been materialized back to a
680    /// local file, so the host can move a per-file progress bar. The default is a
681    /// no-op.
682    async fn on_blob_materialize_progress(
683        &self,
684        root_table: &str,
685        root_id: &str,
686        blob_id: &str,
687        done: u64,
688        total: u64,
689    ) {
690        let _ = (root_table, root_id, blob_id, done, total);
691    }
692}