coven_database/
blob_declarations.rs

1//! The resolved blob-declaration model: coven's own derivation of which rows
2//! carry blobs and where each blob's columns live.
3//!
4//! The gate-sibling of [`Gates`](crate::Gates). A host declares per
5//! table, via [`SyncedTable::carries_blob`], the columns that locate a blob (its
6//! id, optional readable cloud path, and encryption-scope column) plus the
7//! namespace and retention class. [`BlobDecls::from_tables`] resolves those column
8//! *names* to indices against the live schema for the database handle — the same
9//! `PRAGMA table_info` name→index resolution the gate runs — so coven reads a
10//! row's blob straight off a changeset row or a live `SELECT` with no per-row host
11//! callback.
12//!
13//! From that one model coven derives every blob set it needs:
14//! [`BlobDecls::ref_from_change`] over a changeset row (push upload / pull
15//! download / apply-side local-copy drop),
16//! [`BlobDecls::publication_blobs_in_db`] over the whole database, and
17//! [`BlobDecls::row_for_blob_in_namespace`] to map a blob back to its row by namespace
18//! (the read-path locality dispatch and the make-Remote completion check).
19//!
20//! A declaration's three blob properties — [`Provenance`] (the Local story),
21//! [`CacheFill`] (the Remote story), and [`BlobReplacement`] (whether the row may be
22//! repointed at a different blob) — are described by the blob concept tree in
23//! the replication layer. The last of them is enforced here, because it is a rule about a
24//! row's `(blob id, cloud path)` pair and this is the one place coven reads that pair off
25//! a row: a replaceable blob's readable path must name its blob, and a write-once row may
26//! never be repointed. Together they are what keeps a cloud object from ever being
27//! rewritten with different bytes.
28
29use std::collections::HashMap;
30
31use rusqlite::{Connection, OptionalExtension};
32
33use crate::{quote_ident, table_columns as session_table_columns};
34use coven_foundation::changeset::{ChangeOp, RowChange};
35use coven_protocol::blob::{
36    cloud_path_names_blob, BlobRef, BlobReplacement, BlobScope, CacheFill, Provenance,
37};
38use coven_protocol::synced_schema::SyncedTable;
39
40/// Why building the blob-declaration model failed.
41#[derive(Debug)]
42pub enum BlobDeclError {
43    /// A declared blob column is absent from the table's live schema.
44    MissingColumn { table: String, column: String },
45    /// A schema read (`PRAGMA table_info`) failed.
46    Sqlite(rusqlite::Error),
47    /// A captured host changeset could not be read.
48    Changeset(crate::ChangesetError),
49    /// A row's declared size column is negative.
50    InvalidSize { table: String, value: i64 },
51    /// A row names a blob but has no content hash.
52    MissingHash { table: String, row_id: String },
53    /// New and old changeset walks produced different row counts.
54    ChangesetWalkMismatch { old_count: usize, new_count: usize },
55    /// A blob-bearing INSERT or UPDATE has no primary key.
56    MissingPublicationPrimaryKey { table: String },
57    /// The transaction row named by a blob-bearing change is absent.
58    MissingPublicationRow { table: String, primary_key: String },
59    /// The transaction row named by a blob-bearing change no longer carries a blob.
60    MissingPublicationBlob { table: String, primary_key: String },
61    /// The transaction row carries a different blob than its change introduced.
62    PublicationBlobMismatch {
63        table: String,
64        primary_key: String,
65        changed_blob_id: String,
66        row_blob_id: String,
67    },
68    /// A [`Replaceable`](BlobReplacement::Replaceable) blob's readable cloud path does not
69    /// name the blob it carries, so the row's next blob would be keyed at this blob's
70    /// cloud object and overwrite it. See `cloud_path_names_blob`.
71    CloudPathNotKeyedByBlob {
72        table: String,
73        blob_id: String,
74        cloud_path: String,
75    },
76    /// A [`WriteOnce`](BlobReplacement::WriteOnce) row was repointed at a different blob.
77    /// Its cloud path is a stable readable name — that is what write-once buys — so the
78    /// new blob would be keyed at the old blob's cloud object and overwrite it.
79    WriteOnceBlobRepointed { table: String, blob_id: String },
80}
81
82impl std::fmt::Display for BlobDeclError {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        match self {
85            BlobDeclError::MissingColumn { table, column } => {
86                write!(
87                    f,
88                    "blob declaration names column {column:?} absent from {table:?}"
89                )
90            }
91            BlobDeclError::Sqlite(e) => write!(f, "blob declaration schema read failed: {e}"),
92            BlobDeclError::Changeset(error) => {
93                write!(f, "blob declaration changeset read failed: {error}")
94            }
95            BlobDeclError::InvalidSize { table, value } => {
96                write!(f, "blob declaration found invalid size in {table}: {value}")
97            }
98            BlobDeclError::MissingHash { table, row_id } => write!(
99                f,
100                "blob-bearing row {table:?}/{row_id:?} has no content hash"
101            ),
102            BlobDeclError::ChangesetWalkMismatch {
103                old_count,
104                new_count,
105            } => write!(
106                f,
107                "blob declaration changeset walk mismatch: old={old_count}, new={new_count}"
108            ),
109            BlobDeclError::MissingPublicationPrimaryKey { table } => {
110                write!(f, "blob-bearing Store write row in {table:?} has no primary key")
111            }
112            BlobDeclError::MissingPublicationRow { table, primary_key } => write!(
113                f,
114                "blob-bearing Store write row {table:?}/{primary_key:?} is absent before commit"
115            ),
116            BlobDeclError::MissingPublicationBlob { table, primary_key } => write!(
117                f,
118                "blob-bearing Store write row {table:?}/{primary_key:?} no longer carries a blob"
119            ),
120            BlobDeclError::PublicationBlobMismatch {
121                table,
122                primary_key,
123                changed_blob_id,
124                row_blob_id,
125            } => write!(
126                f,
127                "blob-bearing Store write row {table:?}/{primary_key:?} changed from introduced blob \
128                 {changed_blob_id:?} to {row_blob_id:?} before commit"
129            ),
130            BlobDeclError::CloudPathNotKeyedByBlob {
131                table,
132                blob_id,
133                cloud_path,
134            } => write!(
135                f,
136                "replaceable blob {blob_id} in {table} has cloud path {cloud_path:?}, which does \
137                 not name it: the path's file name must be the blob id, or end with -{blob_id} \
138                 before its extension, so that replacing the blob moves its cloud key rather \
139                 than overwriting its cloud object"
140            ),
141            BlobDeclError::WriteOnceBlobRepointed { table, blob_id } => write!(
142                f,
143                "write-once row in {table} was repointed at blob {blob_id}: a write-once blob's \
144                 cloud path is a stable readable name, so the new blob would overwrite the cloud \
145                 object of the blob it replaced. Declare the table replaceable (and key its path \
146                 by its blob id) if its rows are meant to be repointed"
147            ),
148        }
149    }
150}
151
152impl std::error::Error for BlobDeclError {}
153
154impl From<rusqlite::Error> for BlobDeclError {
155    fn from(e: rusqlite::Error) -> Self {
156        BlobDeclError::Sqlite(e)
157    }
158}
159
160/// Exact row facts captured with a durable Store write for one blob-bearing row.
161#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct PublicationBlob {
163    pub table: String,
164    pub row_id: String,
165    pub row_stamp: String,
166    pub column: String,
167    pub blob: BlobRef,
168    pub plaintext_size: u64,
169    pub plaintext_hash: String,
170}
171
172/// A blob-bearing table's columns resolved to indices in the live schema (the
173/// same order a changeset reports its columns, so an index reads either source).
174struct TableBlob {
175    namespace: String,
176    provenance: Provenance,
177    fill: CacheFill,
178    /// Index of the blob-id column.
179    id_col: usize,
180    /// Index of the plaintext-size column.
181    size_col: usize,
182    /// Index of the content-hash column.
183    hash_col: usize,
184    /// Name of the blob-id column. The index reads a row top-to-bottom; the name
185    /// keys a lookup the other way ([`BlobDecls::row_for_blob_in_namespace`]: which row
186    /// carries a given blob id), so both directions resolve off the same declaration.
187    id_col_name: String,
188    /// Index of the readable cloud-path column, if declared.
189    cloud_path_col: Option<usize>,
190    /// The encryption scope, fixed per table by the declaration.
191    scope: BlobScope,
192    /// Whether this table's row may be repointed at a different blob, and so which rule
193    /// keeps its cloud object from ever being rewritten. See [`TableBlob::blob_ref`] and
194    /// [`TableBlob::ref_from_change`].
195    replacement: BlobReplacement,
196}
197
198impl TableBlob {
199    /// Build the [`BlobRef`] for one of this table's rows from the per-row `id`,
200    /// `scope`, and `cloud_path` plus this table's fixed namespace, provenance, and
201    /// cache fill. Shared by changeset and live-row readers.
202    ///
203    /// The gate a [`Replaceable`](BlobReplacement::Replaceable) blob's readable cloud path
204    /// passes through: it must name the blob ([`cloud_path_names_blob`]), so that a row
205    /// repointed at a new blob keys it at a *new* cloud object rather than over the one it
206    /// replaced. Every blob set coven derives — the push scan, the pull scan, the snapshot
207    /// backfill, the transitions — is built here, so there is no path around it. A
208    /// [`WriteOnce`](BlobReplacement::WriteOnce) blob is exempt: its row is never
209    /// repointed ([`TableBlob::ref_from_change`] refuses that), so its object is written
210    /// once and its path is free to be a stable readable name.
211    fn blob_ref(
212        &self,
213        table: &str,
214        id: String,
215        scope: BlobScope,
216        cloud_path: Option<String>,
217    ) -> Result<BlobRef, BlobDeclError> {
218        if self.replacement == BlobReplacement::Replaceable {
219            if let Some(path) = cloud_path.as_deref() {
220                if !cloud_path_names_blob(path, &id) {
221                    return Err(BlobDeclError::CloudPathNotKeyedByBlob {
222                        table: table.to_string(),
223                        blob_id: id,
224                        cloud_path: path.to_string(),
225                    });
226                }
227            }
228        }
229        Ok(BlobRef {
230            namespace: self.namespace.clone(),
231            id,
232            scope,
233            cloud_path,
234            provenance: self.provenance,
235            fill: self.fill,
236        })
237    }
238
239    /// The blob a changeset row references.
240    ///
241    /// The gate a [`WriteOnce`](BlobReplacement::WriteOnce) row passes through. A
242    /// changeset UPDATE marks only the columns whose values changed. The decoded row
243    /// also carries old values for unchanged columns, so write-once enforcement must
244    /// inspect that marker before treating its blob id as a repointing. A real repoint
245    /// is refused here, where the change is read, rather than discovered as a corrupted
246    /// bucket later.
247    fn ref_from_change(
248        &self,
249        table: &str,
250        change: &RowChange,
251    ) -> Result<Option<BlobRef>, BlobDeclError> {
252        let Some(id) = change.col(self.id_col).map(str::to_string) else {
253            return Ok(None);
254        };
255        if self.replacement == BlobReplacement::WriteOnce
256            && change.op == ChangeOp::Update
257            && change.column_changed(self.id_col)
258        {
259            return Err(BlobDeclError::WriteOnceBlobRepointed {
260                table: table.to_string(),
261                blob_id: id,
262            });
263        }
264        let cloud_path = self
265            .cloud_path_col
266            .and_then(|i| change.col(i))
267            .map(str::to_string);
268        self.blob_ref(table, id, self.scope.clone(), cloud_path)
269            .map(Some)
270    }
271
272    /// Build the [`BlobRef`] for a live `SELECT *` row of this table, or `None` when
273    /// the row's blob id is NULL. The resolved
274    /// indices address a `SELECT *` row in schema order, exactly as they address a
275    /// changeset row.
276    fn ref_from_row(
277        &self,
278        table: &str,
279        row: &rusqlite::Row<'_>,
280    ) -> Result<Option<BlobRef>, BlobDeclError> {
281        let Some(id) = row.get::<_, Option<String>>(self.id_col)? else {
282            return Ok(None);
283        };
284        let cloud_path = match self.cloud_path_col {
285            Some(i) => row.get::<_, Option<String>>(i)?,
286            None => None,
287        };
288        self.blob_ref(table, id, self.scope.clone(), cloud_path)
289            .map(Some)
290    }
291
292    fn size_from_row(&self, table: &str, row: &rusqlite::Row<'_>) -> Result<u64, BlobDeclError> {
293        let value = row.get::<_, i64>(self.size_col)?;
294        u64::try_from(value).map_err(|_| BlobDeclError::InvalidSize {
295            table: table.to_string(),
296            value,
297        })
298    }
299
300    fn hash_from_row(
301        &self,
302        table: &str,
303        row_id: &str,
304        row: &rusqlite::Row<'_>,
305    ) -> Result<String, BlobDeclError> {
306        row.get::<_, Option<String>>(self.hash_col)?
307            .ok_or_else(|| BlobDeclError::MissingHash {
308                table: table.to_string(),
309                row_id: row_id.to_string(),
310            })
311    }
312}
313
314/// The blob declarations for a database handle, resolved from the declared set +
315/// the live schema at open. A synced table absent from this map carries no blob.
316pub struct BlobDecls {
317    tables: HashMap<String, TableBlob>,
318}
319
320#[cfg(any(test, feature = "test-utils"))]
321thread_local! {
322    static FROM_TABLES_CALLS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
323}
324
325#[cfg(any(test, feature = "test-utils"))]
326pub fn reset_from_tables_call_count() {
327    FROM_TABLES_CALLS.with(|calls| calls.set(0));
328}
329
330#[cfg(any(test, feature = "test-utils"))]
331pub fn from_tables_call_count() -> usize {
332    FROM_TABLES_CALLS.with(std::cell::Cell::get)
333}
334
335impl BlobDecls {
336    /// Resolve every [`SyncedTable::carries_blob`] declaration's column names to
337    /// indices against the live schema, mirroring
338    /// [`Gates::from_tables`](crate::Gates::from_tables). A declared
339    /// column absent from the table is a host error surfaced here, never a silent
340    /// drop.
341    pub(crate) fn from_tables(
342        conn: &Connection,
343        tables: &[SyncedTable],
344    ) -> Result<Self, BlobDeclError> {
345        #[cfg(any(test, feature = "test-utils"))]
346        FROM_TABLES_CALLS.with(|calls| calls.set(calls.get() + 1));
347
348        let mut map = HashMap::new();
349        for t in tables {
350            let Some(decl) = t.blob() else {
351                continue;
352            };
353            // Column names in declared (schema) order — the index of a name here
354            // is the index a changeset reports for that column.
355            let cols = session_table_columns(conn, t.name()).map_err(BlobDeclError::from)?;
356            let index_of = |column: &str| -> Result<usize, BlobDeclError> {
357                cols.iter()
358                    .position(|c| c == column)
359                    .ok_or_else(|| BlobDeclError::MissingColumn {
360                        table: t.name().to_string(),
361                        column: column.to_string(),
362                    })
363            };
364
365            let id_col = index_of(&decl.id_column)?;
366            let size_col = index_of(&decl.size_column)?;
367            let hash_col = index_of(&decl.hash_column)?;
368            let cloud_path_col = match &decl.cloud_path_column {
369                Some(c) => Some(index_of(c)?),
370                None => None,
371            };
372
373            map.insert(
374                t.name().to_string(),
375                TableBlob {
376                    namespace: decl.namespace.clone(),
377                    provenance: decl.provenance,
378                    fill: decl.fill,
379                    id_col,
380                    size_col,
381                    hash_col,
382                    id_col_name: decl.id_column.clone(),
383                    cloud_path_col,
384                    scope: decl.scope.clone(),
385                    replacement: decl.replacement,
386                },
387            );
388        }
389        Ok(BlobDecls { tables: map })
390    }
391
392    /// Install connection-local guards that keep a blob cleanup intent exclusive
393    /// until its filesystem deletion finishes. A cleanup intent is committed
394    /// before the database releases the row; while either cleanup queue holds it,
395    /// no INSERT or UPDATE may make the same `(namespace, blob id)` live again.
396    /// TEMP triggers keep this runtime guard out of snapshots and use each
397    /// declaration's resolved blob-id column rather than assuming the row primary
398    /// key carries the blob id.
399    pub(crate) fn install_cleanup_guards(&self, conn: &Connection) -> Result<(), BlobDeclError> {
400        for (table, blob) in &self.tables {
401            let table_ident = quote_ident(table);
402            let id_ident = quote_ident(&blob.id_col_name);
403            let namespace_literal: String =
404                conn.query_row("SELECT quote(?1)", [&blob.namespace], |row| row.get(0))?;
405            for (trigger_kind, event_clause) in [
406                ("insert", "BEFORE INSERT".to_string()),
407                ("update", format!("BEFORE UPDATE OF {id_ident}")),
408            ] {
409                let trigger = quote_ident(&format!(
410                    "{}{trigger_kind}_{table}",
411                    super::COVEN_CLEANUP_GUARD_PREFIX
412                ));
413                conn.execute_batch(&format!(
414                    "CREATE TEMP TRIGGER {trigger} \
415                     {event_clause} ON main.{table_ident} \
416                     WHEN NEW.{id_ident} IS NOT NULL AND (\
417                         EXISTS (\
418                             SELECT 1 FROM local_cleanup_intents \
419                             WHERE namespace = {namespace_literal} \
420                               AND blob_id = NEW.{id_ident}\
421                         ) OR EXISTS (\
422                             SELECT 1 FROM published_blob_drop_intents \
423                             WHERE namespace = {namespace_literal} \
424                               AND blob_id = NEW.{id_ident}\
425                         )\
426                     ) \
427                     BEGIN \
428                         SELECT RAISE(ABORT, 'blob local cleanup in progress'); \
429                     END;"
430                ))?;
431            }
432        }
433        Ok(())
434    }
435
436    /// The blob a single changeset row references, or `None` when the row's table
437    /// carries no blob or the blob id is absent/NULL. Reads the declared columns
438    /// off the changeset row (which reports columns in schema order, the order the
439    /// resolved indices address).
440    pub fn ref_from_change(&self, change: &RowChange) -> Result<Option<BlobRef>, BlobDeclError> {
441        let Some(tb) = self.tables.get(&change.table) else {
442            return Ok(None);
443        };
444        tb.ref_from_change(&change.table, change)
445    }
446
447    /// The exact blob reference and declared size owned by an INSERT or UPDATE,
448    /// completed from that row inside the transaction that produced the change.
449    /// The decoded change supplies the blob identity, including an unchanged id,
450    /// while the live transaction row supplies its full cloud path and size before
451    /// a later write can repoint or delete it.
452    pub(crate) fn publication_blob_from_change(
453        &self,
454        conn: &Connection,
455        change: &RowChange,
456    ) -> Result<Option<PublicationBlob>, BlobDeclError> {
457        if !matches!(change.op, ChangeOp::Insert | ChangeOp::Update) {
458            return Ok(None);
459        }
460        let Some(tb) = self.tables.get(&change.table) else {
461            return Ok(None);
462        };
463        let Some(changed_blob) = tb.ref_from_change(&change.table, change)? else {
464            return Ok(None);
465        };
466        let pk = change
467            .pk()
468            .ok_or_else(|| BlobDeclError::MissingPublicationPrimaryKey {
469                table: change.table.clone(),
470            })?;
471        let sql = format!("SELECT * FROM {} WHERE id = ?1", quote_ident(&change.table));
472        let mut statement = conn.prepare(&sql)?;
473        let mut rows = statement.query([pk])?;
474        let row = rows
475            .next()?
476            .ok_or_else(|| BlobDeclError::MissingPublicationRow {
477                table: change.table.clone(),
478                primary_key: pk.to_string(),
479            })?;
480        let publication = publication_blob_from_row(&change.table, tb, row)?;
481        let blob = &publication.blob;
482        if blob.id != changed_blob.id {
483            return Err(BlobDeclError::PublicationBlobMismatch {
484                table: change.table.clone(),
485                primary_key: pk.to_string(),
486                changed_blob_id: changed_blob.id,
487                row_blob_id: blob.id.clone(),
488            });
489        }
490        Ok(Some(publication))
491    }
492
493    pub(crate) fn publication_blob_for_row(
494        &self,
495        conn: &Connection,
496        table: &str,
497        row_id: &str,
498    ) -> Result<Option<PublicationBlob>, BlobDeclError> {
499        let Some(blob) = self.tables.get(table) else {
500            return Ok(None);
501        };
502        let sql = format!("SELECT * FROM {} WHERE id = ?1", quote_ident(table));
503        let mut statement = conn.prepare(&sql)?;
504        let mut rows = statement.query([row_id])?;
505        rows.next()?
506            .map(|row| publication_blob_from_row(table, blob, row))
507            .transpose()
508    }
509
510    /// Require every inserted or updated blob-bearing row in `changeset` to
511    /// carry complete final content facts before its transaction can commit.
512    pub(crate) fn validate_changed_rows(
513        &self,
514        conn: &Connection,
515        changeset: &[u8],
516    ) -> Result<(), BlobDeclError> {
517        let changes = crate::walk_changeset(changeset).map_err(BlobDeclError::Changeset)?;
518        for change in changes {
519            if !matches!(change.op, ChangeOp::Insert | ChangeOp::Update)
520                || !self.tables.contains_key(&change.table)
521            {
522                continue;
523            }
524            let row_id =
525                change
526                    .pk()
527                    .ok_or_else(|| BlobDeclError::MissingPublicationPrimaryKey {
528                        table: change.table.clone(),
529                    })?;
530            match self.publication_blob_for_row(conn, &change.table, row_id) {
531                Ok(_) | Err(BlobDeclError::MissingPublicationBlob { .. }) => {}
532                Err(error) => return Err(error),
533            }
534        }
535        Ok(())
536    }
537
538    /// Every exact blob-bearing row version currently present in `conn`.
539    pub(crate) fn publication_blobs_in_db(
540        &self,
541        conn: &Connection,
542    ) -> Result<Vec<PublicationBlob>, BlobDeclError> {
543        let mut out = Vec::new();
544        for (table, blob) in &self.tables {
545            let sql = format!("SELECT * FROM {}", quote_ident(table));
546            let mut statement = conn.prepare(&sql)?;
547            let mut rows = statement.query([])?;
548            while let Some(row) = rows.next()? {
549                let Some(reference) = blob.ref_from_row(table, row)? else {
550                    continue;
551                };
552                let row_id = row.get::<_, String>("id")?;
553                out.push(PublicationBlob {
554                    table: table.clone(),
555                    row_id: row_id.clone(),
556                    row_stamp: row.get("_updated_at")?,
557                    column: blob.id_col_name.clone(),
558                    blob: reference,
559                    plaintext_size: blob.size_from_row(table, row)?,
560                    plaintext_hash: blob.hash_from_row(table, &row_id, row)?,
561                });
562            }
563        }
564        out.sort_by(|left, right| {
565            (&left.table, &left.row_id, &left.column, &left.row_stamp).cmp(&(
566                &right.table,
567                &right.row_id,
568                &right.column,
569                &right.row_stamp,
570            ))
571        });
572        Ok(out)
573    }
574
575    /// The `(table, primary key)` of the row carrying `blob_id` in the table declared
576    /// for `namespace` — the carrying table resolved from the blob's own namespace
577    /// (part of its address), not by scanning every blob-bearing table. `None` when no
578    /// declared table owns `namespace`, or that table has no row with the id. Both the
579    /// read path (locality dispatch) and the make-Remote completion check use this, so
580    /// a blob id that collides across namespaces always reads the right table's gate,
581    /// never the first id match.
582    pub(crate) fn row_for_blob_in_namespace(
583        &self,
584        conn: &Connection,
585        namespace: &str,
586        blob_id: &str,
587    ) -> Result<Option<(String, String)>, BlobDeclError> {
588        let Some((table, tb)) = self.table_for_namespace(namespace) else {
589            return Ok(None);
590        };
591        let sql = format!(
592            "SELECT id FROM {} WHERE {} = ?1",
593            quote_ident(table),
594            quote_ident(&tb.id_col_name),
595        );
596        conn.query_row(&sql, [blob_id], |row| row.get::<_, String>(0))
597            .optional()
598            .map(|primary_key| primary_key.map(|primary_key| (table.clone(), primary_key)))
599            .map_err(BlobDeclError::from)
600    }
601
602    /// The one `(table, declaration)` whose blob namespace is `namespace`, or
603    /// `None` when no declared table owns it. The namespace is part of a blob's
604    /// address, so this resolves the carrying table without scanning every
605    /// blob-bearing table's rows.
606    fn table_for_namespace(&self, namespace: &str) -> Option<(&String, &TableBlob)> {
607        self.tables.iter().find(|(_, tb)| tb.namespace == namespace)
608    }
609
610    /// Whether a live row still needs the logical-id-keyed local source for this
611    /// blob. A row needs that source exactly when its current stamp has no installed
612    /// remote locator binding.
613    pub(crate) fn local_copy_is_referenced(
614        &self,
615        conn: &Connection,
616        namespace: &str,
617        blob_id: &str,
618    ) -> Result<bool, BlobDeclError> {
619        let Some((table, blob)) = self.table_for_namespace(namespace) else {
620            return Ok(false);
621        };
622        let sql = format!(
623            "SELECT EXISTS(
624                 SELECT 1 FROM {table} AS live
625                 WHERE CAST(live.{blob_column} AS TEXT) = ?1
626                   AND NOT EXISTS (
627                       SELECT 1 FROM row_blob_locators AS binding
628                       WHERE binding.table_name = ?2
629                         AND binding.row_id = CAST(live.id AS TEXT)
630                         AND binding.column_name = ?3
631                         AND binding.row_stamp = CAST(live._updated_at AS TEXT)
632                   )
633             )",
634            table = quote_ident(table),
635            blob_column = quote_ident(&blob.id_col_name),
636        );
637        conn.query_row(
638            &sql,
639            rusqlite::params![blob_id, table, blob.id_col_name],
640            |row| row.get(0),
641        )
642        .map_err(BlobDeclError::from)
643    }
644
645    /// Whether any live row still carries this logical blob ID, independent of
646    /// whether that row currently resolves to a local source or an exact remote
647    /// locator.
648    pub(crate) fn blob_id_is_referenced(
649        &self,
650        conn: &Connection,
651        namespace: &str,
652        blob_id: &str,
653    ) -> Result<bool, BlobDeclError> {
654        let Some((table, blob)) = self.table_for_namespace(namespace) else {
655            return Ok(false);
656        };
657        let sql = format!(
658            "SELECT EXISTS(
659                 SELECT 1 FROM {table}
660                 WHERE CAST({blob_column} AS TEXT) = ?1
661             )",
662            table = quote_ident(table),
663            blob_column = quote_ident(&blob.id_col_name),
664        );
665        conn.query_row(&sql, [blob_id], |row| row.get(0))
666            .map_err(BlobDeclError::from)
667    }
668
669    /// Whether a live row's current stamp still names one exact locator. A row
670    /// that merely reuses the same logical blob id under another locator does not
671    /// retain this locator's cache or pinned file.
672    pub(crate) fn exact_copy_is_referenced(
673        &self,
674        conn: &Connection,
675        namespace: &str,
676        blob_id: &str,
677        locator_hash: coven_protocol::store_commit::ObjectHash,
678    ) -> Result<bool, BlobDeclError> {
679        let Some((table, blob)) = self.table_for_namespace(namespace) else {
680            return Ok(false);
681        };
682        let sql = format!(
683            "SELECT EXISTS(
684                 SELECT 1
685                 FROM {table} AS live
686                 JOIN row_blob_locators AS binding
687                   ON binding.table_name = ?2
688                  AND binding.row_id = CAST(live.id AS TEXT)
689                  AND binding.column_name = ?3
690                  AND binding.row_stamp = CAST(live._updated_at AS TEXT)
691                 JOIN blob_locators AS locator
692                   ON locator.remote_object_id = binding.remote_object_id
693                 WHERE CAST(live.{blob_column} AS TEXT) = ?1
694                   AND locator.locator_hash = ?4
695             )",
696            table = quote_ident(table),
697            blob_column = quote_ident(&blob.id_col_name),
698        );
699        conn.query_row(
700            &sql,
701            rusqlite::params![blob_id, table, blob.id_col_name, locator_hash.to_string()],
702            |row| row.get(0),
703        )
704        .map_err(BlobDeclError::from)
705    }
706}
707
708fn publication_blob_from_row(
709    table: &str,
710    blob: &TableBlob,
711    row: &rusqlite::Row<'_>,
712) -> Result<PublicationBlob, BlobDeclError> {
713    let row_id = row.get::<_, String>("id")?;
714    let reference =
715        blob.ref_from_row(table, row)?
716            .ok_or_else(|| BlobDeclError::MissingPublicationBlob {
717                table: table.to_string(),
718                primary_key: row_id.clone(),
719            })?;
720    let plaintext_hash = blob.hash_from_row(table, &row_id, row)?;
721    Ok(PublicationBlob {
722        table: table.to_string(),
723        row_id,
724        row_stamp: row.get("_updated_at")?,
725        column: blob.id_col_name.clone(),
726        blob: reference,
727        plaintext_size: blob.size_from_row(table, row)?,
728        plaintext_hash,
729    })
730}
731
732#[cfg(test)]
733mod tests {
734    use super::*;
735    use coven_protocol::blob::{CacheFill, Provenance};
736    use coven_protocol::synced_schema::{BlobDecl, RowIdentity};
737    use rusqlite::session::Session;
738
739    fn capture_update(conn: &Connection, sql: &str) -> RowChange {
740        let mut session = Session::new(conn).expect("create session");
741        session.attach(Some("files")).expect("attach files");
742        conn.execute(sql, []).expect("update file row");
743        let mut changeset = Vec::new();
744        session
745            .changeset_strm(&mut changeset)
746            .expect("extract changeset");
747        crate::walk_changeset(&changeset)
748            .expect("walk changeset")
749            .into_iter()
750            .next()
751            .expect("captured update")
752    }
753
754    fn write_once_decl(id_column: Option<&str>) -> BlobDecl {
755        let decl =
756            BlobDecl::new("files", Provenance::HostProvided, CacheFill::CacheEager).write_once();
757        match id_column {
758            Some(column) => decl.with_id_column(column),
759            None => decl,
760        }
761    }
762
763    #[test]
764    fn unrelated_update_does_not_repoint_a_primary_key_blob() {
765        let conn = Connection::open_in_memory().expect("open connection");
766        conn.execute_batch(
767            "CREATE TABLE files (
768                 id TEXT PRIMARY KEY,
769                 title TEXT NOT NULL,
770                 size INTEGER NOT NULL,
771                 hash TEXT NOT NULL
772             );
773             INSERT INTO files VALUES ('blob-a', 'before', 1, 'hash-a');",
774        )
775        .expect("create file row");
776        let declarations = BlobDecls::from_tables(
777            &conn,
778            &[SyncedTable::new("files", RowIdentity::IndependentUuid)
779                .carries_blob(write_once_decl(None))],
780        )
781        .expect("resolve declarations");
782
783        let change = capture_update(&conn, "UPDATE files SET title = 'after'");
784
785        let blob = declarations
786            .ref_from_change(&change)
787            .expect("read unrelated update")
788            .expect("unchanged blob reference remains available");
789        assert_eq!(blob.id, "blob-a");
790    }
791
792    #[test]
793    fn changing_a_write_once_blob_column_is_rejected() {
794        let conn = Connection::open_in_memory().expect("open connection");
795        conn.execute_batch(
796            "CREATE TABLE files (
797                 id TEXT PRIMARY KEY,
798                 blob_id TEXT NOT NULL,
799                 size INTEGER NOT NULL,
800                 hash TEXT NOT NULL
801             );
802             INSERT INTO files VALUES ('row-a', 'blob-a', 1, 'hash-a');",
803        )
804        .expect("create file row");
805        let declarations = BlobDecls::from_tables(
806            &conn,
807            &[SyncedTable::new("files", RowIdentity::IndependentUuid)
808                .carries_blob(write_once_decl(Some("blob_id")))],
809        )
810        .expect("resolve declarations");
811
812        let change = capture_update(&conn, "UPDATE files SET blob_id = 'blob-b'");
813
814        assert!(matches!(
815            declarations.ref_from_change(&change),
816            Err(BlobDeclError::WriteOnceBlobRepointed { blob_id, .. })
817                if blob_id == "blob-b"
818        ));
819    }
820}
coven_database/blob_declarations.rs

coven_database/
blob_declarations.rs
