coven_database/
blob_bindings.rs

1use crate::blob_records::live_blob_row;
2use crate::blob_records::validate_live_blob_locator;
3use crate::cloud_outbox_records::CloudOutboxRecords;
4use crate::remote_object_records::load_remote_object_on;
5use crate::remote_object_records::persist_exact_remote_object_on;
6use crate::remote_object_records::update_remote_object_on;
7
8use super::*;
9
10pub(crate) fn install_pulled_package_activation_on(
11    conn: &Connection,
12    store_dir: &coven_foundation::store_dir::StoreDir,
13    commit_ref: &StoreBatchCommitRef,
14    domain: SharedLiveSetObjectDomain,
15    object: &ExactObjectRef,
16    package: &AudiencePackage,
17) -> Result<(), DbError> {
18    let object_id = remote_object_id(object);
19    let exists: bool = conn
20        .query_row(
21            "SELECT EXISTS(SELECT 1 FROM remote_objects WHERE object_id = ?1)",
22            [object_id.to_string()],
23            |row| row.get(0),
24        )
25        .map_err(DbError::from)?;
26    if exists {
27        let remote = load_remote_object_on(conn, object_id)?;
28        let mut remote = if matches!(remote, RemoteObjectRecord::CandidateExclusive(_)) {
29            remote.into_activated(commit_ref).map_err(|error| {
30                DbError::context(
31                    format!("activate locally prepared pulled package {object_id}"),
32                    error,
33                )
34            })?
35        } else {
36            remote
37        };
38        remote
39            .merge_package_activation(&domain, package, commit_ref)
40            .map_err(|error| {
41                DbError::context(
42                    format!("merge pulled package activation {object_id}"),
43                    error,
44                )
45            })?;
46        update_remote_object_on(conn, object_id, &remote)
47    } else {
48        let remote =
49            RemoteObjectRecord::activated_external_package(domain, package, commit_ref.clone())
50                .map_err(|error| {
51                    DbError::context(
52                        format!("construct pulled package activation {object_id}"),
53                        error,
54                    )
55                })?;
56        persist_exact_remote_object_on(conn, store_dir, &remote, "pulled audience package")
57    }
58}
59
60pub(crate) fn install_pulled_merge_membership_activations_on(
61    conn: &Connection,
62    store_dir: &coven_foundation::store_dir::StoreDir,
63    commit_ref: &StoreBatchCommitRef,
64    remotes: &[coven_protocol::remote_object::ClosedRemoteObject],
65) -> Result<(), DbError> {
66    let mut object_ids = BTreeSet::new();
67    for expected in remotes {
68        let object_id = expected.object_id();
69        if !object_ids.insert(object_id) {
70            return Err(DbError::Message(
71                "pulled Merge membership closure repeats an exact object".to_string(),
72            ));
73        }
74        let existing = conn
75            .query_row(
76                "SELECT EXISTS(SELECT 1 FROM remote_objects WHERE object_id = ?1)",
77                [object_id.to_string()],
78                |row| row.get::<_, bool>(0),
79            )
80            .map_err(DbError::from)?;
81        if existing {
82            let mut remote = load_remote_object_on(conn, object_id)?;
83            remote
84                .merge_retained_authority_activation(expected, commit_ref)
85                .map_err(|error| {
86                    DbError::context(
87                        format!("merge pulled Merge membership authority {object_id}"),
88                        error,
89                    )
90                })?;
91            update_remote_object_on(conn, object_id, &remote)?;
92        } else {
93            persist_exact_remote_object_on(
94                conn,
95                store_dir,
96                expected,
97                "pulled Merge membership authority",
98            )?;
99        }
100    }
101    Ok(())
102}
103
104impl Database {
105    // ---- Materialized Store commit ledger ----
106
107    pub(crate) fn install_pulled_blob_activations_on(
108        conn: &Connection,
109        package: &AudiencePackage,
110        owner: &StoreBatchCommitRef,
111    ) -> Result<(), DbError> {
112        if package.commit_coord() != &owner.coord {
113            return Err(DbError::Message(
114                "pulled blob package coordinate differs from its activating commit".to_string(),
115            ));
116        }
117        for binding in package.blob_bindings() {
118            let stored = binding.blob();
119            let object_id = remote_object_id(stored.object());
120            let exists: bool = conn
121                .query_row(
122                    "SELECT EXISTS(SELECT 1 FROM remote_objects WHERE object_id = ?1)",
123                    [object_id.to_string()],
124                    |row| row.get(0),
125                )
126                .map_err(DbError::from)?;
127            let remote = if exists {
128                let mut remote = load_remote_object_on(conn, object_id)?;
129                remote
130                    .merge_blob_activation(stored, owner)
131                    .map_err(|error| {
132                        DbError::context(format!("merge pulled blob activation {object_id}"), error)
133                    })?;
134                remote
135            } else {
136                RemoteObjectRecord::activated_blob(stored, owner.clone())
137                    .map_err(|error| {
138                        DbError::context(
139                            format!("construct pulled blob activation {object_id}"),
140                            error,
141                        )
142                    })?
143                    .into_record()
144            };
145            let state = serde_json::to_string(&remote)
146                .map_err(|error| DbError::context("serialize pulled blob activation", error))?;
147            conn.execute(
148                "INSERT INTO remote_objects (object_id, state) VALUES (?1, ?2) \
149                 ON CONFLICT(object_id) DO UPDATE SET state = excluded.state",
150                rusqlite::params![object_id.to_string(), state],
151            )
152            .map_err(DbError::from)?;
153        }
154        Ok(())
155    }
156
157    pub(crate) fn row_blob_refs_for_root_on(
158        conn: &Connection,
159        gates: &Gates,
160        tables: &[SyncedTable],
161        root_table: &str,
162        root_id: &str,
163    ) -> Result<Vec<RowBlobRef>, DbError> {
164        let mut rows = gates
165            .subtree_rows(conn, root_table, root_id)
166            .map_err(DbError::from)?
167            .into_iter()
168            .collect::<Vec<_>>();
169        rows.sort();
170        let tables = tables
171            .iter()
172            .map(|table| (table.name(), table))
173            .collect::<BTreeMap<_, _>>();
174        rows.into_iter()
175            .filter_map(|(table_name, row_id)| {
176                tables
177                    .get(table_name.as_str())
178                    .filter(|table| table.blob().is_some())
179                    .map(|table| Self::row_blob_ref_on(conn, gates, table, &row_id))
180            })
181            .collect()
182    }
183
184    pub(crate) fn stored_blob_reference_state_on(
185        conn: &Connection,
186        gates: &Gates,
187        tables: &[SyncedTable],
188        stored: &StoredBlobRef,
189    ) -> Result<StoredBlobReferenceState, DbError> {
190        let exact_object_id = remote_object_id(stored.object()).to_string();
191        let mut statement = conn
192            .prepare(
193                "SELECT table_name, row_id, row_stamp FROM row_blob_locators
194                 WHERE remote_object_id = ?1",
195            )
196            .map_err(DbError::from)?;
197        let bindings = statement
198            .query_map([exact_object_id], |row| {
199                Ok((
200                    row.get::<_, String>(0)?,
201                    row.get::<_, String>(1)?,
202                    row.get::<_, String>(2)?,
203                ))
204            })
205            .map_err(DbError::from)?
206            .collect::<Result<Vec<_>, _>>()
207            .map_err(DbError::from)?;
208        drop(statement);
209        let mut unresolved = false;
210        for (table_name, row_id, row_stamp) in bindings {
211            let table = tables
212                .iter()
213                .find(|candidate| candidate.name() == table_name)
214                .ok_or_else(|| {
215                    DbError::Message(format!(
216                        "stored blob binding names undeclared table {table_name:?}"
217                    ))
218                })?;
219            let declaration = table.blob().ok_or_else(|| {
220                DbError::Message(format!(
221                    "stored blob binding names table {table_name:?} without a blob declaration"
222                ))
223            })?;
224            let Some(live) = live_blob_row(conn, &table_name, &row_id, declaration)? else {
225                continue;
226            };
227            if live.stamp != row_stamp {
228                continue;
229            }
230            // A row reaches the cloud in two different ways: a gated root is
231            // kept, while an audience-scoped root is addressed to a non-Local
232            // audience. An absent row or unreachable audience parent leaves the
233            // locality unresolved; it cannot prove that the blob is unreferenced.
234            let remote = if !gates.table_is_scoped(&table_name) {
235                gates
236                    .root_kept_of(conn, &table_name, &row_id)
237                    .map_err(DbError::from)?
238            } else {
239                match crate::live_row_audience(conn, gates, &table_name, &row_id) {
240                    Ok(audience) => Some(audience != coven_protocol::circle::Audience::Local),
241                    Err(
242                        crate::GateError::MissingAudienceRow { .. }
243                        | crate::GateError::MissingAudienceParent { .. },
244                    ) => None,
245                    Err(error) => return Err(DbError::from(error)),
246                }
247            };
248            match remote {
249                Some(false) => continue,
250                None => {
251                    unresolved = true;
252                    continue;
253                }
254                Some(true) => {}
255            }
256            let reference = Self::row_blob_ref_on(conn, gates, table, &row_id)?;
257            if matches!(reference.authority(), RowBlobAuthority::Remote(_))
258                && reference.stored() == Some(stored)
259            {
260                return Ok(StoredBlobReferenceState::LiveRemote);
261            }
262        }
263        Ok(if unresolved {
264            StoredBlobReferenceState::Unresolved
265        } else {
266            StoredBlobReferenceState::NotLiveRemote
267        })
268    }
269
270    /// The exact current blob-bearing row version for `row_id`. A row that is
271    /// not there is an error: a caller naming one row is asking about a row it
272    /// believes exists.
273    pub(crate) fn row_blob_ref_on(
274        conn: &Connection,
275        gates: &Gates,
276        table: &SyncedTable,
277        row_id: &str,
278    ) -> Result<RowBlobRef, DbError> {
279        Self::live_row_blob_ref_on(conn, gates, table, row_id)?.ok_or_else(|| {
280            DbError::Message(format!(
281                "blob-bearing row {:?}/{row_id:?} does not exist",
282                table.name()
283            ))
284        })
285    }
286
287    pub(crate) fn validate_row_blob_ref_on(
288        conn: &Connection,
289        gates: &Gates,
290        table: &SyncedTable,
291        reference: &RowBlobRef,
292    ) -> Result<(), DbError> {
293        let current = Self::row_blob_ref_on(conn, gates, table, reference.row_id())?;
294        if &current != reference {
295            return Err(DbError::Message(format!(
296                "row blob reference {:?}/{:?}/{:?} at {:?} is stale",
297                reference.table(),
298                reference.row_id(),
299                reference.column(),
300                reference.row_stamp()
301            )));
302        }
303        Ok(())
304    }
305
306    /// The same reference for a row that may not be there, `None` when it is
307    /// not. This is the shape a list-shaped read needs: a caller asking about
308    /// many ids at once holds ids it has not checked, and one naming no live
309    /// blob-bearing row is an answer about that id, not a failed read.
310    pub(crate) fn live_row_blob_ref_on(
311        conn: &Connection,
312        gates: &Gates,
313        table: &SyncedTable,
314        row_id: &str,
315    ) -> Result<Option<RowBlobRef>, DbError> {
316        let declaration = table.blob().ok_or_else(|| {
317            DbError::Message(format!(
318                "synced table {:?} has no blob declaration",
319                table.name()
320            ))
321        })?;
322        let Some(row) = live_blob_row(conn, table.name(), row_id, declaration)? else {
323            return Ok(None);
324        };
325        let audience =
326            gate::live_row_audience(conn, gates, table.name(), row_id).map_err(|error| {
327                DbError::context(
328                    format!(
329                        "resolve blob row audience for {:?}/{row_id:?}",
330                        table.name()
331                    ),
332                    error,
333                )
334            })?;
335        let (authority, stored) = match RemoteAudience::try_from(audience.clone()) {
336            Err(_) if audience == Audience::Local => (RowBlobAuthority::Local, None),
337            Err(error) => {
338                return Err(DbError::context(
339                    format!(
340                        "blob row {:?}/{row_id:?} has invalid audience",
341                        table.name()
342                    ),
343                    error,
344                ));
345            }
346            Ok(remote_audience) => {
347                let installed: Option<(String, String)> = conn
348                    .query_row(
349                        "SELECT binding.audience_authority, locator.remote_object_id
350                         FROM row_blob_locators AS binding
351                         JOIN blob_locators AS locator
352                           ON locator.remote_object_id = binding.remote_object_id
353                         WHERE binding.table_name = ?1
354                           AND binding.row_id = ?2
355                           AND binding.column_name = ?3
356                           AND binding.row_stamp = ?4",
357                        rusqlite::params![table.name(), row_id, declaration.id_column, row.stamp,],
358                        |row| Ok((row.get(0)?, row.get(1)?)),
359                    )
360                    .optional()
361                    .map_err(DbError::from)?;
362                let exact = if let Some((authority_json, remote_object_id)) = installed {
363                    let package_authority: coven_protocol::audience_package::PackageAudience =
364                        serde_json::from_str(&authority_json).map_err(|error| {
365                            DbError::context(format!("remote blob row {:?}/{row_id:?} has invalid audience authority", table.name()), error)
366                        })?;
367                    let remote_object_id = remote_object_id.parse().map_err(|error| {
368                        DbError::context(
369                            format!(
370                                "remote blob row {:?}/{row_id:?} has invalid prepared object id",
371                                table.name()
372                            ),
373                            error,
374                        )
375                    })?;
376                    let remote = load_remote_object_on(conn, remote_object_id)?;
377                    if !remote.is_activated_stored_blob() {
378                        return Err(DbError::Message(format!(
379                            "remote blob row {:?}/{row_id:?} references a blob without activated ownership",
380                            table.name()
381                        )));
382                    }
383                    let locator = crate::blob_records::carried_blob_locator(
384                        &remote,
385                        &format!(
386                            "remote blob row {:?}/{row_id:?} has invalid locator",
387                            table.name()
388                        ),
389                    )?;
390                    let stored = StoredBlobRef::new(locator, remote.object().clone()).map_err(
391                        |error| {
392                            DbError::context(format!("remote blob row {:?}/{row_id:?} has invalid stored blob reference", table.name()), error)
393                        },
394                    )?;
395                    Some((package_authority, stored))
396                } else {
397                    CloudOutboxRecords::new(conn)
398                        .created_upload_handoff(
399                            table.name(),
400                            row_id,
401                            &declaration.id_column,
402                            &row.stamp,
403                        )?
404                        .map(|handoff| (handoff.authority, handoff.stored))
405                };
406                let Some((package_authority, stored)) = exact else {
407                    return RowBlobRef::new(
408                        table.name().to_string(),
409                        row_id.to_string(),
410                        row.stamp,
411                        declaration.id_column.clone(),
412                        BlobRef {
413                            namespace: declaration.namespace.clone(),
414                            id: row.blob_id,
415                            scope: declaration.scope.clone(),
416                            cloud_path: row.cloud_path,
417                            provenance: declaration.provenance,
418                            fill: declaration.fill,
419                        },
420                        row.plaintext_size,
421                        row.plaintext_hash,
422                        RowBlobAuthority::PendingRemote(remote_audience),
423                        None,
424                    )
425                    .map(Some)
426                    .map_err(DbError::from);
427                };
428                if package_authority.remote_audience() != remote_audience {
429                    return Err(DbError::Message(format!(
430                        "remote blob row {:?}/{row_id:?} has audience authority {:?}, expected {remote_audience:?}",
431                        table.name(), package_authority
432                    )));
433                }
434                validate_live_blob_locator(
435                    table.name(),
436                    row_id,
437                    &declaration.id_column,
438                    &row.stamp,
439                    &stored,
440                    declaration,
441                    &row,
442                    &remote_audience,
443                )?;
444                (RowBlobAuthority::Remote(package_authority), Some(stored))
445            }
446        };
447        let blob = BlobRef {
448            namespace: declaration.namespace.clone(),
449            id: row.blob_id.clone(),
450            scope: declaration.scope.clone(),
451            cloud_path: row.cloud_path.clone(),
452            provenance: declaration.provenance,
453            fill: declaration.fill,
454        };
455        RowBlobRef::new(
456            table.name().to_string(),
457            row_id.to_string(),
458            row.stamp,
459            declaration.id_column.clone(),
460            blob,
461            row.plaintext_size,
462            row.plaintext_hash,
463            authority,
464            stored,
465        )
466        .map(Some)
467        .map_err(DbError::from)
468    }
469
470    #[cfg(any(test, feature = "test-utils"))]
471    pub async fn row_blob_ref(&self, table: &str, row_id: &str) -> Result<RowBlobRef, DbError> {
472        crate::StoreDatabase::new(self)
473            .row_blob_ref(table, row_id)
474            .await
475    }
476}
coven_database/blob_bindings.rs

coven_database/
blob_bindings.rs
