coven_database/store/store_session/
host_write_blob_transaction.rs

1use super::*;
2
3pub struct HostWriteBlobTransaction<'transaction, 'connection> {
4    store: StoreTransaction<'transaction, 'connection>,
5    verified_authority: &'transaction mut VerifiedStoreAuthority,
6    created_payload_files: &'transaction mut Vec<PathBuf>,
7}
8
9impl<'transaction, 'connection> HostWriteBlobTransaction<'transaction, 'connection> {
10    pub(super) fn new(
11        store: StoreTransaction<'transaction, 'connection>,
12        verified_authority: &'transaction mut VerifiedStoreAuthority,
13        created_payload_files: &'transaction mut Vec<PathBuf>,
14    ) -> Self {
15        Self {
16            store,
17            verified_authority,
18            created_payload_files,
19        }
20    }
21
22    /// Retain the verified plaintext source in the same transaction as its
23    /// captured write. Files created before SQL commit join capture rollback.
24    pub fn retain_source_plaintext(
25        &mut self,
26        fact: &StoreWriteBlobFact,
27        source: &std::path::Path,
28    ) -> Result<(), DbError> {
29        let (hash, size) = PayloadStore::new(self.store.transaction, self.store.store_dir)
30            .file_writer(source)?
31            .commit_for_capture(self.created_payload_files)?;
32        if hash != fact.plaintext_hash || size != fact.plaintext_size {
33            return Err(DbError::Message(format!(
34                "captured blob {}/{}/{} source differs from its declared plaintext",
35                fact.table, fact.row_id, fact.column,
36            )));
37        }
38        Ok(())
39    }
40
41    pub fn circle_blob_opening_protection(
42        &mut self,
43        root: &coven_protocol::store_commit::StoreRootRef,
44        circle_id: coven_protocol::circle::CircleId,
45        expected_control: &coven_protocol::circle::CircleControlCoord,
46        expected_key_fingerprint: coven_keys::encryption::KeyFingerprint,
47    ) -> Result<coven_protocol::objects::BlobSpoolProtection, DbError> {
48        self.store.circle_blob_opening_protection(
49            self.verified_authority,
50            root,
51            circle_id,
52            expected_control,
53            expected_key_fingerprint,
54        )
55    }
56
57    pub fn external_local_path(
58        &self,
59        fact: &StoreWriteBlobFact,
60    ) -> Result<Option<PathBuf>, DbError> {
61        let stored = self
62            .store
63            .transaction
64            .query_row(
65                "SELECT path, plaintext_size, plaintext_hash
66                 FROM local_blob_refs
67                 WHERE table_name = ?1 AND row_id = ?2 AND column_name = ?3
68                   AND namespace = ?4 AND blob_id = ?5
69                 ORDER BY row_stamp DESC LIMIT 1",
70                rusqlite::params![
71                    fact.table,
72                    fact.row_id,
73                    fact.column,
74                    fact.blob.namespace,
75                    fact.blob.id,
76                ],
77                |row| {
78                    Ok((
79                        row.get::<_, String>(0)?,
80                        row.get::<_, i64>(1)?,
81                        row.get::<_, String>(2)?,
82                    ))
83                },
84            )
85            .optional()
86            .map_err(DbError::from)?;
87        let Some((path, size, hash)) = stored else {
88            return Ok(None);
89        };
90        let size = u64::try_from(size).map_err(|_| {
91            DbError::Message("registered external blob has a negative size".to_string())
92        })?;
93        if size != fact.plaintext_size || hash != fact.plaintext_hash.to_string() {
94            return Err(DbError::Message(
95                "registered external blob identity differs from the moved row".to_string(),
96            ));
97        }
98        Ok(Some(PathBuf::from(path)))
99    }
100}
101
102pub(super) fn rollback_captured_payload_files(
103    directory: &coven_foundation::store_dir::StoreDir,
104    files: Vec<PathBuf>,
105    operation: DbError,
106) -> DbError {
107    let mut failures = Vec::new();
108    for path in files.into_iter().rev() {
109        let cleanup = std::fs::remove_file(&path)
110            .map_err(|source| {
111                coven_foundation::atomic_file::FileError::at(
112                    "remove captured payload",
113                    &path,
114                    source,
115                )
116            })
117            .and_then(|()| directory.sync_parent_dir_blocking(&path));
118        if let Err(error) = cleanup {
119            failures.push(crate::StagedBlobRollbackFailure {
120                path,
121                reason: error.into(),
122            });
123        }
124    }
125    if failures.is_empty() {
126        operation
127    } else {
128        DbError::AudienceBlobRollbackFailed {
129            operation: Box::new(operation),
130            rollback: crate::StagedBlobRollbackFailures(failures),
131        }
132    }
133}
134
135impl StoreSession<'_> {
136    fn stage_captured_blob_source(
137        &self,
138        fact: StoreWriteBlobFact,
139        staged: coven_foundation::local_file::AtomicStagedFile,
140    ) -> Result<coven_foundation::local_file::AtomicStagedFile, DbError> {
141        let materialize = (|| {
142            if fact.audience_move != Some(StoreWriteBlobMoveMaterialization::Payload) {
143                return Err(DbError::Message(
144                    "blob fact does not retain a captured payload".into(),
145                ));
146            }
147            let mut output = std::fs::OpenOptions::new()
148                .write(true)
149                .truncate(true)
150                .open(staged.path())
151                .map_err(|error| {
152                    coven_foundation::atomic_file::FileError::at(
153                        "open captured blob stage",
154                        staged.path(),
155                        error,
156                    )
157                })?;
158            let size = PayloadStore::new(self.conn, self.store_dir)
159                .copy_verified(fact.plaintext_hash, &mut output)?;
160            if size != fact.plaintext_size {
161                return Err(DbError::Message(
162                    "captured blob payload size differs from its fact".into(),
163                ));
164            }
165            Ok(())
166        })();
167        match materialize {
168            Ok(()) => Ok(staged),
169            Err(operation) => {
170                let path = staged.path().to_path_buf();
171                match staged.discard_blocking() {
172                    Ok(()) => Err(operation),
173                    Err(cleanup) => Err(DbError::AudienceBlobRollbackFailed {
174                        operation: Box::new(operation),
175                        rollback: crate::StagedBlobRollbackFailures(vec![
176                            crate::StagedBlobRollbackFailure {
177                                path,
178                                reason: cleanup.into(),
179                            },
180                        ]),
181                    }),
182                }
183            }
184        }
185    }
186}
187
188impl StoreDatabase {
189    /// Fill an unpublished file from an audience move's exact captured source.
190    pub async fn stage_captured_blob_source(
191        &self,
192        fact: StoreWriteBlobFact,
193        staged: coven_foundation::local_file::AtomicStagedFile,
194    ) -> Result<coven_foundation::local_file::AtomicStagedFile, DbError> {
195        self.call_store(move |session| session.stage_captured_blob_source(fact, staged))
196            .await
197    }
198}
coven_database/store/store_session/host_write_blob_transaction.rs

coven_database/store/store_session/
host_write_blob_transaction.rs
