coven_database/store/store_session/
blob_transitions.rs

1use super::*;
2
3use crate::{CloudOutboxRecords, MakeRemoteIntentState};
4use crate::{OutboxEntry, OutboxOperation, OutboxUploadState};
5use coven_protocol::blob::RowBlobRef;
6
7pub enum PostUpload {
8    Waiting,
9    Cancelled,
10    MadeRemote { root_table: String, root_id: String },
11}
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub enum BlobTransitionRoot {
15    Gated,
16    RemoteRoot,
17    NotGated,
18}
19
20#[derive(Clone)]
21pub struct MaterializedLocalBlob {
22    pub remote: RowBlobRef,
23    pub stored: coven_protocol::blob::locator::StoredBlobRef,
24    pub destination: Option<std::path::PathBuf>,
25}
26
27pub struct MakeRemoteAdmission {
28    pub root_id: String,
29    pub root_label: String,
30    pub uploads: Vec<(RowBlobRef, std::path::PathBuf)>,
31}
32
33impl StoreSession<'_> {
34    fn gated_root_gate_column(&self, root_table: &str) -> Result<&str, DbError> {
35        self.synced_tables
36            .iter()
37            .find(|table| table.name() == root_table)
38            .and_then(|table| table.gate_column())
39            .ok_or_else(|| {
40                DbError::Message(format!(
41                    "blob locality transition root {root_table:?} has no boolean gate column"
42                ))
43            })
44    }
45
46    fn gated_root_locality(
47        &self,
48        root_table: &str,
49        root_id: &str,
50    ) -> Result<Option<bool>, DbError> {
51        let gate_column = self.gated_root_gate_column(root_table)?;
52        crate::query_truth(self.conn, root_table, gate_column, root_id).map_err(DbError::from)
53    }
54
55    #[allow(clippy::too_many_arguments)]
56    fn admit_make_remote_on(
57        &self,
58        connection: &rusqlite::Connection,
59        root_table: &str,
60        root_id: &str,
61        root_label: &str,
62        pin: bool,
63        created_at: &str,
64        uploads: &[(RowBlobRef, std::path::PathBuf)],
65    ) -> Result<Option<bool>, DbError> {
66        let gate_column = self.gated_root_gate_column(root_table)?;
67        let locality = crate::query_truth(connection, root_table, gate_column, root_id)
68            .map_err(DbError::from)?;
69        if locality == Some(false) {
70            let current = Database::row_blob_refs_for_root_on(
71                connection,
72                self.gates,
73                self.synced_tables,
74                root_table,
75                root_id,
76            )?;
77            let supplied_are_current = current.len() == uploads.len()
78                && uploads.iter().enumerate().all(|(index, (verified, _))| {
79                    current.contains(verified)
80                        && !uploads[..index]
81                            .iter()
82                            .any(|(earlier, _)| earlier == verified)
83                });
84            if !supplied_are_current {
85                return Err(DbError::Message(format!(
86                    "blob rows below {root_table:?}/{root_id:?} changed while make_remote verified their sources"
87                )));
88            }
89            Database::insert_make_remote_intent_on(
90                connection, root_table, root_id, root_label, pin,
91            )?;
92            let cloud_outbox = CloudOutboxRecords::new(connection);
93            for (reference, source_path) in uploads {
94                cloud_outbox.enqueue_upload(
95                    root_table,
96                    root_id,
97                    root_label,
98                    reference,
99                    source_path,
100                    pin,
101                    created_at,
102                )?;
103            }
104        }
105        Ok(locality)
106    }
107
108    #[allow(clippy::too_many_arguments)]
109    fn begin_make_remote(
110        &self,
111        root_table: &str,
112        root_id: &str,
113        root_label: &str,
114        pin: bool,
115        created_at: &str,
116        uploads: &[(RowBlobRef, std::path::PathBuf)],
117    ) -> Result<Option<bool>, DbError> {
118        let transaction = self.conn.unchecked_transaction()?;
119        let locality = self.admit_make_remote_on(
120            &transaction,
121            root_table,
122            root_id,
123            root_label,
124            pin,
125            created_at,
126            uploads,
127        )?;
128        transaction.commit().map_err(DbError::from)?;
129        Ok(locality)
130    }
131
132    fn begin_make_remote_batch(
133        &self,
134        root_table: &str,
135        pin: bool,
136        created_at: &str,
137        roots: &[MakeRemoteAdmission],
138    ) -> Result<(), DbError> {
139        let transaction = self.conn.unchecked_transaction()?;
140        for root in roots {
141            if Database::make_remote_intent_state(&transaction, root_table, &root.root_id)?
142                .is_some()
143            {
144                return Err(DbError::Message(format!(
145                    "make_remote for {root_table:?}/{:?} is already in progress",
146                    root.root_id
147                )));
148            }
149        }
150        for root in roots {
151            match self.admit_make_remote_on(
152                &transaction,
153                root_table,
154                &root.root_id,
155                &root.root_label,
156                pin,
157                created_at,
158                &root.uploads,
159            )? {
160                Some(false) => {}
161                Some(true) => {
162                    return Err(DbError::Message(format!(
163                        "root {root_table:?}/{:?} is already Remote",
164                        root.root_id
165                    )));
166                }
167                None => {
168                    return Err(DbError::Message(format!(
169                        "root {root_table:?}/{:?} has no resolvable Local/Remote state",
170                        root.root_id
171                    )));
172                }
173            }
174        }
175        transaction.commit().map_err(DbError::from)
176    }
177
178    fn finalize_created_blob_upload(
179        &mut self,
180        entry: OutboxEntry,
181        root_table: String,
182        root_id: String,
183        row: RowBlobRef,
184        stamp: &str,
185        routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
186        write_id: coven_protocol::write::WriteId,
187    ) -> Result<PostUpload, DbError> {
188        let connection = self.conn;
189        let resolved_root = self
190            .gates
191            .resolve_root_of(connection, row.table(), row.row_id())
192            .map_err(DbError::from)?
193            .ok_or_else(|| {
194                DbError::Message(format!(
195                    "upload row {:?}/{:?} has no gated transition root",
196                    row.table(),
197                    row.row_id()
198                ))
199            })?;
200        if resolved_root != (root_table.clone(), root_id.clone()) {
201            return Err(DbError::Message(format!(
202                "upload row {:?}/{:?} moved from make_remote root {:?}/{:?} to {:?}/{:?}",
203                row.table(),
204                row.row_id(),
205                root_table,
206                root_id,
207                resolved_root.0,
208                resolved_root.1
209            )));
210        }
211        match Database::make_remote_intent_state(connection, &root_table, &root_id)? {
212            Some(MakeRemoteIntentState::Uploading) => {}
213            Some(MakeRemoteIntentState::Publishing(_)) => return Ok(PostUpload::Waiting),
214            Some(MakeRemoteIntentState::Cancelling) => return Ok(PostUpload::Cancelled),
215            None => {
216                return Err(DbError::Message(format!(
217                    "upload for {root_table:?}/{root_id:?} has no make_remote intent"
218                )));
219            }
220        }
221
222        let rows = Database::row_blob_refs_for_root_on(
223            connection,
224            self.gates,
225            self.synced_tables,
226            &root_table,
227            &root_id,
228        )?;
229        let entries = CloudOutboxRecords::new(connection).upload_entries_for_root(
230            self.gates,
231            self.synced_tables,
232            &root_table,
233            &root_id,
234        )?;
235        if rows.len() != entries.len() {
236            return Err(DbError::Message(format!(
237                "make_remote root {root_table:?}/{root_id:?} has {} blob rows but {} exact upload journals",
238                rows.len(),
239                entries.len()
240            )));
241        }
242        if !entries.iter().all(|candidate| {
243            matches!(
244                candidate.operation,
245                OutboxOperation::Upload {
246                    state: OutboxUploadState::Created { .. },
247                    ..
248                }
249            )
250        }) {
251            return Ok(PostUpload::Waiting);
252        }
253        if !entries.iter().any(|candidate| candidate == &entry) {
254            return Err(DbError::Message(
255                "Created upload changed before make_remote finalization".to_string(),
256            ));
257        }
258
259        let gate_column = self
260            .synced_tables
261            .iter()
262            .find(|table| table.name() == root_table)
263            .and_then(|table| table.gate_column())
264            .ok_or_else(|| {
265                DbError::Message(format!(
266                    "make_remote root {root_table:?} has no boolean gate column"
267                ))
268            })?;
269        super::host_write_capture::CapturedStoreWriteTransaction::begin_prepared_blob_transition(
270            connection,
271            self.store_dir,
272            self.synced_tables,
273            self.gates,
274            self.blob_decls,
275            routing_encryption,
276            self.verified_store_authority,
277            write_id.clone(),
278        )?
279        .execute_make_remote(
280            root_table.clone(),
281            root_id.clone(),
282            gate_column.to_string(),
283            stamp.to_string(),
284            rows,
285            write_id,
286        )?;
287        Ok(PostUpload::MadeRemote {
288            root_table,
289            root_id,
290        })
291    }
292
293    #[allow(clippy::too_many_arguments)]
294    fn commit_make_local(
295        &mut self,
296        root_table: &str,
297        root_id: &str,
298        stamp: &str,
299        routing_encryption: Option<&coven_keys::encryption::EncryptionService>,
300        materialized: &[MaterializedLocalBlob],
301        write_id: coven_protocol::write::WriteId,
302    ) -> Result<(), DbError> {
303        let gate_column = self.gated_root_gate_column(root_table)?.to_string();
304        super::host_write_capture::CapturedStoreWriteTransaction::begin_prepared_blob_transition(
305            self.conn,
306            self.store_dir,
307            self.synced_tables,
308            self.gates,
309            self.blob_decls,
310            routing_encryption,
311            self.verified_store_authority,
312            write_id,
313        )?
314        .execute_make_local(
315            root_table.to_string(),
316            root_id.to_string(),
317            gate_column,
318            stamp.to_string(),
319            materialized.to_vec(),
320        )
321        .map(|receipt| receipt.value)
322    }
323
324    fn cancel_make_remote(&self, root_table: &str, root_id: &str) -> Result<(), DbError> {
325        self.gated_root_gate_column(root_table)?;
326        let transaction = self.conn.unchecked_transaction()?;
327        match Database::make_remote_intent_state(&transaction, root_table, root_id)? {
328            Some(MakeRemoteIntentState::Uploading) => {
329                let updated = transaction
330                    .execute(
331                        "UPDATE blob_make_remote_intents SET state = 'cancelling'
332                         WHERE root_table = ?1 AND root_id = ?2 AND state = 'uploading'",
333                        (root_table, root_id),
334                    )
335                    .map_err(DbError::from)?;
336                if updated != 1 {
337                    return Err(DbError::Message(format!(
338                        "make_remote intent {root_table:?}/{root_id:?} cannot enter cancellation"
339                    )));
340                }
341                transaction
342                    .execute(
343                        "UPDATE cloud_outbox
344                         SET attempt_count = 0, last_error = NULL, last_attempt_at = NULL
345                         WHERE operation = 'upload' AND root_table = ?1 AND root_id = ?2",
346                        (root_table, root_id),
347                    )
348                    .map_err(DbError::from)?;
349            }
350            Some(MakeRemoteIntentState::Cancelling) => {}
351            Some(MakeRemoteIntentState::Publishing(write_id)) => {
352                return Err(DbError::Message(format!(
353                    "make_remote for {root_table:?}/{root_id:?} is already publishing as {write_id}"
354                )));
355            }
356            // No intent, but the queue may still hold work for the root — one
357            // already Remote when more was queued for it, or one whose
358            // transition ended while its uploads had not. That work is exactly
359            // what a cancel is for, so it is adopted and unwound. A root with
360            // nothing queued either has genuinely nothing to cancel, and saying
361            // so is how a cancel that arrives after completion is told it
362            // changed nothing.
363            None => {
364                if !Database::adopt_cancelling_intent_from_queue_on(
365                    &transaction,
366                    root_table,
367                    root_id,
368                )? {
369                    return Err(DbError::Message(format!(
370                        "make_remote for {root_table:?}/{root_id:?} does not exist"
371                    )));
372                }
373            }
374        }
375        transaction.commit().map_err(DbError::from)
376    }
377}
378
379impl StoreDatabase {
380    pub async fn gated_root_locality(
381        &self,
382        root_table: &str,
383        root_id: &str,
384    ) -> Result<Option<bool>, DbError> {
385        let root_table = root_table.to_string();
386        let root_id = root_id.to_string();
387        self.call_store(move |session| session.gated_root_locality(&root_table, &root_id))
388            .await
389    }
390
391    #[allow(clippy::too_many_arguments)]
392    pub async fn begin_make_remote(
393        &self,
394        root_table: &str,
395        root_id: &str,
396        root_label: &str,
397        pin: bool,
398        created_at: String,
399        uploads: Vec<(RowBlobRef, std::path::PathBuf)>,
400    ) -> Result<Option<bool>, DbError> {
401        let root_table = root_table.to_string();
402        let root_id = root_id.to_string();
403        let root_label = root_label.to_string();
404        self.call_store(move |session| {
405            session.begin_make_remote(
406                &root_table,
407                &root_id,
408                &root_label,
409                pin,
410                &created_at,
411                &uploads,
412            )
413        })
414        .await
415    }
416
417    pub async fn begin_make_remote_batch(
418        &self,
419        root_table: &str,
420        pin: bool,
421        created_at: String,
422        roots: Vec<MakeRemoteAdmission>,
423    ) -> Result<(), DbError> {
424        let root_table = root_table.to_string();
425        self.call_store(move |session| {
426            session.begin_make_remote_batch(&root_table, pin, &created_at, &roots)
427        })
428        .await
429    }
430
431    pub async fn cancel_make_remote(&self, root_table: &str, root_id: &str) -> Result<(), DbError> {
432        let root_table = root_table.to_string();
433        let root_id = root_id.to_string();
434        self.call_store(move |session| session.cancel_make_remote(&root_table, &root_id))
435            .await
436    }
437
438    /// Complete a Created upload journal entry without exposing a Remote row that lacks
439    /// exact object authority. Every blob-bearing row below the same gated root must
440    /// still match its queued row version and have reached Created. The final transaction
441    /// then flips the gate, clears external-file ownership, records the pending Store
442    /// write, and binds the transition intent to that write together. The intent and
443    /// Created handoffs remain until that Store write activates, so a crash cannot make
444    /// the upload drain mistake a published object for an orphan.
445    pub async fn finalize_created_blob_upload(
446        &self,
447        entry: &OutboxEntry,
448        stamp: String,
449        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
450    ) -> Result<PostUpload, DbError> {
451        let OutboxOperation::Upload {
452            root_table,
453            root_id,
454            row,
455            state,
456            ..
457        } = &entry.operation
458        else {
459            return Err(DbError::Message(
460                "make_remote finalizer received a non-upload outbox entry".to_string(),
461            ));
462        };
463        if !matches!(state, OutboxUploadState::Created { .. }) {
464            return Err(DbError::Message(
465                "make_remote finalizer requires a Created exact upload".to_string(),
466            ));
467        }
468
469        let entry = entry.clone();
470        let root_table = root_table.clone();
471        let root_id = root_id.clone();
472        let row = row.clone();
473        let write_id = self.new_store_write_id();
474        self.call_store(move |session| {
475            session.finalize_created_blob_upload(
476                entry,
477                root_table,
478                root_id,
479                row,
480                &stamp,
481                routing_encryption.as_ref(),
482                write_id,
483            )
484        })
485        .await
486    }
487
488    #[allow(clippy::too_many_arguments)]
489    pub async fn commit_make_local(
490        &self,
491        root_table: &str,
492        root_id: &str,
493        stamp: String,
494        routing_encryption: Option<coven_keys::encryption::EncryptionService>,
495        materialized: Vec<MaterializedLocalBlob>,
496    ) -> Result<(), DbError> {
497        let root_table = root_table.to_string();
498        let root_id = root_id.to_string();
499        let write_id = self.new_store_write_id();
500        self.call_store(move |session| {
501            session.commit_make_local(
502                &root_table,
503                &root_id,
504                &stamp,
505                routing_encryption.as_ref(),
506                &materialized,
507                write_id,
508            )
509        })
510        .await
511    }
512}
513
514/// Set a gated root's locality and stamp the row inside the caller's prepared
515/// blob-transition transaction.
516pub(super) fn write_gate(
517    transaction: &rusqlite::Transaction<'_>,
518    root_table: &str,
519    gate_column: &str,
520    remote: bool,
521    stamp: &str,
522    root_id: &str,
523) -> Result<(), rusqlite::Error> {
524    transaction.execute(
525        &format!(
526            "UPDATE {} SET {} = ?1, _updated_at = ?2 WHERE id = ?3",
527            crate::quote_ident(root_table),
528            crate::quote_ident(gate_column),
529        ),
530        (remote as i64, stamp, root_id),
531    )?;
532    Ok(())
533}
534
535pub(super) fn same_row_blob_version(left: &RowBlobRef, right: &RowBlobRef) -> bool {
536    left.table() == right.table()
537        && left.row_id() == right.row_id()
538        && left.row_stamp() == right.row_stamp()
539        && left.column() == right.column()
540        && left.blob() == right.blob()
541        && left.plaintext_size() == right.plaintext_size()
542        && left.plaintext_hash() == right.plaintext_hash()
543}
coven_database/store/store_session/blob_transitions.rs

coven_database/store/store_session/
blob_transitions.rs
