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            crate::blob_records::record_stored_locator_on(conn, stored)?;
154        }
155        Ok(())
156    }
157
158    pub(crate) fn row_blob_refs_for_root_on(
159        conn: &Connection,
160        gates: &Gates,
161        tables: &[SyncedTable],
162        root_table: &str,
163        root_id: &str,
164    ) -> Result<Vec<RowBlobRef>, DbError> {
165        let mut rows = gates
166            .subtree_rows(conn, root_table, root_id)
167            .map_err(DbError::from)?
168            .into_iter()
169            .collect::<Vec<_>>();
170        rows.sort();
171        let tables = tables
172            .iter()
173            .map(|table| (table.name(), table))
174            .collect::<BTreeMap<_, _>>();
175        rows.into_iter()
176            .filter_map(|(table_name, row_id)| {
177                tables
178                    .get(table_name.as_str())
179                    .filter(|table| table.blob().is_some())
180                    .map(|table| Self::row_blob_ref_on(conn, gates, table, &row_id))
181            })
182            .collect()
183    }
184
185    pub(crate) fn stored_blob_reference_state_on(
186        conn: &Connection,
187        gates: &Gates,
188        tables: &[SyncedTable],
189        stored: &StoredBlobRef,
190    ) -> Result<StoredBlobReferenceState, DbError> {
191        let exact_object_id = remote_object_id(stored.object()).to_string();
192        let mut statement = conn
193            .prepare(
194                "SELECT table_name, row_id, row_stamp FROM row_blob_locators
195                 WHERE remote_object_id = ?1",
196            )
197            .map_err(DbError::from)?;
198        let bindings = statement
199            .query_map([exact_object_id], |row| {
200                Ok((
201                    row.get::<_, String>(0)?,
202                    row.get::<_, String>(1)?,
203                    row.get::<_, String>(2)?,
204                ))
205            })
206            .map_err(DbError::from)?
207            .collect::<Result<Vec<_>, _>>()
208            .map_err(DbError::from)?;
209        drop(statement);
210        let mut unresolved = false;
211        for (table_name, row_id, row_stamp) in bindings {
212            let table = tables
213                .iter()
214                .find(|candidate| candidate.name() == table_name)
215                .ok_or_else(|| {
216                    DbError::Message(format!(
217                        "stored blob binding names undeclared table {table_name:?}"
218                    ))
219                })?;
220            let declaration = table.blob().ok_or_else(|| {
221                DbError::Message(format!(
222                    "stored blob binding names table {table_name:?} without a blob declaration"
223                ))
224            })?;
225            let Some(live) = live_blob_row(conn, &table_name, &row_id, declaration)? else {
226                continue;
227            };
228            if live.stamp != row_stamp {
229                continue;
230            }
231            // A row reaches the cloud in two different ways: a gated root is
232            // kept, while an audience-scoped root is addressed to a non-Local
233            // audience. An absent row or unreachable audience parent leaves the
234            // locality unresolved; it cannot prove that the blob is unreferenced.
235            let remote = if !gates.table_is_scoped(&table_name) {
236                gates
237                    .root_kept_of(conn, &table_name, &row_id)
238                    .map_err(DbError::from)?
239            } else {
240                match crate::live_row_audience(conn, gates, &table_name, &row_id) {
241                    Ok(audience) => Some(audience != coven_protocol::circle::Audience::Local),
242                    Err(
243                        crate::GateError::MissingAudienceRow { .. }
244                        | crate::GateError::MissingAudienceParent { .. },
245                    ) => None,
246                    Err(error) => return Err(DbError::from(error)),
247                }
248            };
249            match remote {
250                Some(false) => continue,
251                None => {
252                    unresolved = true;
253                    continue;
254                }
255                Some(true) => {}
256            }
257            let reference = Self::row_blob_ref_on(conn, gates, table, &row_id)?;
258            if matches!(reference.authority(), RowBlobAuthority::Remote(_))
259                && reference.stored() == Some(stored)
260            {
261                return Ok(StoredBlobReferenceState::LiveRemote);
262            }
263        }
264        Ok(if unresolved {
265            StoredBlobReferenceState::Unresolved
266        } else {
267            StoredBlobReferenceState::NotLiveRemote
268        })
269    }
270
271    /// The exact current blob-bearing row version for `row_id`. A row that is
272    /// not there is an error: a caller naming one row is asking about a row it
273    /// believes exists.
274    pub(crate) fn row_blob_ref_on(
275        conn: &Connection,
276        gates: &Gates,
277        table: &SyncedTable,
278        row_id: &str,
279    ) -> Result<RowBlobRef, DbError> {
280        Self::live_row_blob_ref_on(conn, gates, table, row_id)?.ok_or_else(|| {
281            DbError::Message(format!(
282                "blob-bearing row {:?}/{row_id:?} does not exist",
283                table.name()
284            ))
285        })
286    }
287
288    pub(crate) fn validate_row_blob_ref_on(
289        conn: &Connection,
290        gates: &Gates,
291        table: &SyncedTable,
292        reference: &RowBlobRef,
293    ) -> Result<(), DbError> {
294        let current = Self::row_blob_ref_on(conn, gates, table, reference.row_id())?;
295        if &current != reference {
296            return Err(DbError::Message(format!(
297                "row blob reference {:?}/{:?}/{:?} at {:?} is stale",
298                reference.table(),
299                reference.row_id(),
300                reference.column(),
301                reference.row_stamp()
302            )));
303        }
304        Ok(())
305    }
306
307    /// The same reference for a row that may not be there, `None` when it is
308    /// not. This is the shape a list-shaped read needs: a caller asking about
309    /// many ids at once holds ids it has not checked, and one naming no live
310    /// blob-bearing row is an answer about that id, not a failed read.
311    pub(crate) fn live_row_blob_ref_on(
312        conn: &Connection,
313        gates: &Gates,
314        table: &SyncedTable,
315        row_id: &str,
316    ) -> Result<Option<RowBlobRef>, DbError> {
317        let declaration = table.blob().ok_or_else(|| {
318            DbError::Message(format!(
319                "synced table {:?} has no blob declaration",
320                table.name()
321            ))
322        })?;
323        let Some(row) = live_blob_row(conn, table.name(), row_id, declaration)? else {
324            return Ok(None);
325        };
326        let audience =
327            gate::live_row_audience(conn, gates, table.name(), row_id).map_err(|error| {
328                DbError::context(
329                    format!(
330                        "resolve blob row audience for {:?}/{row_id:?}",
331                        table.name()
332                    ),
333                    error,
334                )
335            })?;
336        let (authority, stored) = match RemoteAudience::try_from(audience.clone()) {
337            Err(_) if audience == Audience::Local => (RowBlobAuthority::Local, None),
338            Err(error) => {
339                return Err(DbError::context(
340                    format!(
341                        "blob row {:?}/{row_id:?} has invalid audience",
342                        table.name()
343                    ),
344                    error,
345                ));
346            }
347            Ok(remote_audience) => {
348                let installed: Option<(String, String)> = conn
349                    .query_row(
350                        "SELECT binding.audience_authority, locator.remote_object_id
351                         FROM row_blob_locators AS binding
352                         JOIN blob_locators AS locator
353                           ON locator.remote_object_id = binding.remote_object_id
354                         WHERE binding.table_name = ?1
355                           AND binding.row_id = ?2
356                           AND binding.column_name = ?3
357                           AND binding.row_stamp = ?4",
358                        rusqlite::params![table.name(), row_id, declaration.id_column, row.stamp,],
359                        |row| Ok((row.get(0)?, row.get(1)?)),
360                    )
361                    .optional()
362                    .map_err(DbError::from)?;
363                let exact = if let Some((authority_json, remote_object_id)) = installed {
364                    let package_authority: coven_protocol::audience_package::PackageAudience =
365                        serde_json::from_str(&authority_json).map_err(|error| {
366                            DbError::context(format!("remote blob row {:?}/{row_id:?} has invalid audience authority", table.name()), error)
367                        })?;
368                    let remote_object_id = remote_object_id.parse().map_err(|error| {
369                        DbError::context(
370                            format!(
371                                "remote blob row {:?}/{row_id:?} has invalid prepared object id",
372                                table.name()
373                            ),
374                            error,
375                        )
376                    })?;
377                    let remote = load_remote_object_on(conn, remote_object_id)?;
378                    if !remote.is_activated_stored_blob() {
379                        return Err(DbError::Message(format!(
380                            "remote blob row {:?}/{row_id:?} references a blob without activated ownership",
381                            table.name()
382                        )));
383                    }
384                    let locator = crate::blob_records::carried_blob_locator(
385                        &remote,
386                        &format!(
387                            "remote blob row {:?}/{row_id:?} has invalid locator",
388                            table.name()
389                        ),
390                    )?;
391                    let stored = StoredBlobRef::new(locator, remote.object().clone()).map_err(
392                        |error| {
393                            DbError::context(format!("remote blob row {:?}/{row_id:?} has invalid stored blob reference", table.name()), error)
394                        },
395                    )?;
396                    Some((package_authority, stored))
397                } else {
398                    CloudOutboxRecords::new(conn)
399                        .created_upload_handoff(
400                            table.name(),
401                            row_id,
402                            &declaration.id_column,
403                            &row.stamp,
404                        )?
405                        .map(|handoff| (handoff.authority, handoff.stored))
406                };
407                let Some((package_authority, stored)) = exact else {
408                    return RowBlobRef::new(
409                        table.name().to_string(),
410                        row_id.to_string(),
411                        row.stamp,
412                        declaration.id_column.clone(),
413                        BlobRef {
414                            namespace: declaration.namespace.clone(),
415                            id: row.blob_id,
416                            scope: declaration.scope.clone(),
417                            cloud_path: row.cloud_path,
418                            provenance: declaration.provenance,
419                            fill: declaration.fill,
420                        },
421                        row.plaintext_size,
422                        row.plaintext_hash,
423                        RowBlobAuthority::PendingRemote(remote_audience),
424                        None,
425                    )
426                    .map(Some)
427                    .map_err(DbError::from);
428                };
429                if package_authority.remote_audience() != remote_audience {
430                    return Err(DbError::Message(format!(
431                        "remote blob row {:?}/{row_id:?} has audience authority {:?}, expected {remote_audience:?}",
432                        table.name(),
433                        package_authority
434                    )));
435                }
436                validate_live_blob_locator(
437                    table.name(),
438                    row_id,
439                    &declaration.id_column,
440                    &row.stamp,
441                    &stored,
442                    declaration,
443                    &row,
444                    &remote_audience,
445                )?;
446                (RowBlobAuthority::Remote(package_authority), Some(stored))
447            }
448        };
449        let blob = BlobRef {
450            namespace: declaration.namespace.clone(),
451            id: row.blob_id.clone(),
452            scope: declaration.scope.clone(),
453            cloud_path: row.cloud_path.clone(),
454            provenance: declaration.provenance,
455            fill: declaration.fill,
456        };
457        RowBlobRef::new(
458            table.name().to_string(),
459            row_id.to_string(),
460            row.stamp,
461            declaration.id_column.clone(),
462            blob,
463            row.plaintext_size,
464            row.plaintext_hash,
465            authority,
466            stored,
467        )
468        .map(Some)
469        .map_err(DbError::from)
470    }
471
472    #[cfg(any(test, feature = "test-utils"))]
473    pub async fn row_blob_ref(&self, table: &str, row_id: &str) -> Result<RowBlobRef, DbError> {
474        crate::StoreDatabase::new(self)
475            .row_blob_ref(table, row_id)
476            .await
477    }
478}
coven_database/blob_bindings.rs

coven_database/
blob_bindings.rs
