1use super::{
2 publication_state::PreparedStoreWriteState, MergeMaterializationTransaction, StoreDatabase,
3 StoreSession, StoreTransactionOutcome, VerifiedStoreTransaction,
4};
5use crate::{
6 candidate_graph_exact_objects, load_prepared_audience_objects_on, load_remote_object_on,
7 CloudOutboxRecords, Database, DbError, OwnedVerifiedMergeMaterialization, PreparedAudienceBlob,
8 RetainedPackageApplication, LOCAL_DEVICE_ID_STATE_KEY,
9};
10use coven_protocol::remote_object::remote_object_id;
11use coven_protocol::store_commit::{StoreBatchCommit, VerifiedStoreBatchCommit};
12use coven_protocol::write::{PublishedPosition, PublishedWrite, WriteId, WriteStatus};
13
14impl VerifiedStoreTransaction<'_, '_, '_, '_> {
15 fn complete_prepared_store_write(
16 &mut self,
17 accepted_publication: crate::StoreCommitPublicationOutcome,
18 routing_key: Option<coven_protocol::circle::RowRoutingKey>,
19 ) -> Result<
20 (
21 Option<OwnedVerifiedMergeMaterialization>,
22 (WriteId, WriteStatus),
23 ),
24 DbError,
25 > {
26 let state = &mut *self.authority;
27 let gates = self.gates;
28 let synced_tables = self.synced_tables;
29 let store_transaction = self.store;
30 let tx = store_transaction.transaction;
31 let local_device_id = crate::required_protocol_state_on(tx, LOCAL_DEVICE_ID_STATE_KEY)?;
32 let prepared_count: i64 = tx
33 .query_row(
34 "SELECT COUNT(*) FROM store_writes WHERE prepared IS NOT NULL",
35 [],
36 |row| row.get(0),
37 )
38 .map_err(DbError::from)?;
39 if prepared_count != 1 {
40 return Err(DbError::Message(format!(
41 "Store publication expected one prepared write, found {prepared_count}"
42 )));
43 }
44 let (stored_write_id, raw_status, raw_prepared): (String, String, String) = tx
45 .query_row(
46 "SELECT write_id, status, prepared FROM store_writes
47 WHERE prepared IS NOT NULL ORDER BY ordinal LIMIT 1",
48 [],
49 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
50 )
51 .map_err(DbError::from)?;
52 let current_status: WriteStatus = serde_json::from_str(&raw_status)
53 .map_err(|error| DbError::context("prepared Store write status", error))?;
54 if current_status != WriteStatus::Publishing {
55 return Err(DbError::Message(format!(
56 "prepared Store write has non-publishing status {current_status:?}"
57 )));
58 }
59 let prepared: PreparedStoreWriteState = serde_json::from_str(&raw_prepared)
60 .map_err(|error| DbError::context("prepared Store write", error))?;
61 let PreparedStoreWriteState {
62 commit,
63 history_evidence,
64 local_cleanup,
65 ..
66 } = prepared;
67 let active = super::active_store_publication::load_active_store_publication_on(tx)?
68 .ok_or_else(|| {
69 DbError::Message(format!(
70 "publishing write {stored_write_id} has no active Store publication"
71 ))
72 })?;
73 let publication = active.attempt()?.clone();
74 let accepted = match &publication.entry.payload {
75 coven_protocol::store_commit::StorePublicationPayload::Commit(reference) => {
76 reference.clone()
77 }
78 coven_protocol::store_commit::StorePublicationPayload::Snapshot(_) => {
79 return Err(DbError::Message(
80 "prepared Store write contains a snapshot publication".to_string(),
81 ));
82 }
83 };
84 let root = state.root().clone();
85 let unverified: StoreBatchCommit = serde_json::from_slice(commit.semantic_bytes())
86 .map_err(|error| DbError::context("prepared Store commit", error))?;
87 let registration =
88 super::verified_store_authority::VerifiedRegistrationLookup::activated_registration_on(
89 state,
90 crate::store::store_session::StoreRecords::new(
91 self.store.transaction,
92 self.store.store_dir,
93 ),
94 &root,
95 &unverified.author_registration,
96 )?;
97 let expected_stream =
98 coven_protocol::store_commit::StreamActivation::device_authorized_stream_id(
99 root.store_root_hash,
100 &unverified.author_registration,
101 coven_protocol::store_commit::StreamAnchorDomain::StoreAnnouncements,
102 );
103 if accepted.coord.stream_id != expected_stream
104 || accepted.object != *commit.prepared().reference()
105 {
106 return Err(DbError::Message(
107 "accepted Store publication differs from the exact prepared commit".to_string(),
108 ));
109 }
110 let commit_value = VerifiedStoreBatchCommit::parse(
111 commit.semantic_bytes(),
112 root.store_root_hash,
113 &accepted,
114 ®istration,
115 )
116 .map_err(|error| DbError::context("outbound commit", error))?;
117 if active.owner()
118 != &crate::ActiveStorePublicationOwner::StoreWrite(commit_value.write_id.clone())
119 || active.commit_reservation()
120 != Some((
121 &commit_value.write_id,
122 &commit_value.author_registration,
123 &commit_value.reference().coord,
124 ))
125 {
126 return Err(DbError::Message(format!(
127 "prepared write {stored_write_id} differs from active Store publication {:?}",
128 active.owner()
129 )));
130 }
131 let accepted_publication =
132 accepted_publication.resolve_installed_on(store_transaction, &commit_value)?;
133 let materialize = accepted_publication.requires_materialization();
134 let publication: crate::AcceptedStoreCommitEvidence = match &accepted_publication {
135 crate::StoreCommitPublicationOutcome::Accepted { interval, .. } => {
136 install_accepted_commit_publication_on(tx, &publication, &commit_value, interval)?
137 .into()
138 }
139 crate::StoreCommitPublicationOutcome::Installed(_) => {
140 accepted_publication.install_on(store_transaction, &commit_value)?
141 }
142 };
143 if commit_value.write_id.as_str() != stored_write_id {
144 return Err(DbError::Message(
145 "prepared write id differs from signed commit".to_string(),
146 ));
147 }
148 let write_id = commit_value.write_id.clone();
149 let commit = commit_value.value();
150 let commit_ref = commit_value.reference();
151 let remaining_spools: i64 = tx
152 .query_row(
153 "SELECT COUNT(*) FROM store_write_blobs
154 WHERE write_id = ?1 AND spool_path IS NOT NULL",
155 [write_id.as_str()],
156 |row| row.get(0),
157 )
158 .map_err(DbError::from)?;
159 if remaining_spools != 0 {
160 return Err(DbError::Message(format!(
161 "prepared write {write_id} retains {remaining_spools} uploaded blob spool(s)"
162 )));
163 }
164 let audiences = load_prepared_audience_objects_on(tx, self.store.store_dir, &write_id)?;
165 let retained_packages = audiences
166 .packages
167 .iter()
168 .map(|package| package.package().clone())
169 .collect::<Vec<_>>();
170 for package in &audiences.packages {
171 package
172 .package()
173 .validate_blob_uploader(&commit.author_registration)
174 .map_err(DbError::from)?;
175 }
176 let mut object_ids = std::collections::BTreeSet::new();
177 object_ids.insert(remote_object_id(&commit_ref.object));
178 object_ids.extend(
179 candidate_graph_exact_objects(commit)?
180 .iter()
181 .map(remote_object_id),
182 );
183 object_ids.extend(
184 audiences
185 .blobs
186 .iter()
187 .map(PreparedAudienceBlob::remote_object_id),
188 );
189 for object_id in object_ids {
190 let remote = load_remote_object_on(tx, object_id)?
191 .into_activated(commit_ref)
192 .map_err(|error| {
193 DbError::context(format!("activate remote object {object_id}"), error)
194 })?;
195 let state = serde_json::to_string(&remote)
196 .map_err(|error| DbError::context("serialize activated remote object", error))?;
197 let updated = tx
198 .execute(
199 "UPDATE remote_objects SET state = ?2 WHERE object_id = ?1",
200 (object_id.to_string(), state),
201 )
202 .map_err(DbError::from)?;
203 if updated != 1 {
204 return Err(DbError::Message(format!(
205 "remote object {object_id} disappeared during activation"
206 )));
207 }
208 }
209 for package in &retained_packages {
210 for binding in package.blob_bindings() {
211 crate::blob_records::record_stored_locator_on(tx, binding.blob())?;
212 }
213 }
214 let retained = if materialize {
215 let merge_transaction = MergeMaterializationTransaction::from_store(self.store);
216 let retained = merge_transaction.record_materialized_merge_commit(
217 state,
218 &root,
219 &commit_value,
220 &[],
221 &publication,
222 &history_evidence,
223 &retained_packages,
224 (!retained_packages.is_empty())
225 .then_some(RetainedPackageApplication::LocallyAuthored),
226 )?;
227 state.insert_verified(retained.clone())?;
228 let replayed = state.replay_projection_watching_on(
229 store_transaction,
230 self.blob_decls,
231 gates,
232 synced_tables,
233 routing_key.as_ref(),
234 &std::collections::BTreeSet::new(),
235 crate::ReplayJournal::Owed,
236 coven_protocol::membership::LocalStoreMembership::Current,
237 commit_ref,
238 )?;
239 match replayed.watched_outcome() {
240 Some(super::WatchedReplayOutcome::Applied) => {}
241 Some(super::WatchedReplayOutcome::Held(reason)) => {
242 return Err(DbError::Message(format!(
243 "accepted local Store publication held during replay: {reason:?}"
244 )));
245 }
246 None => {
247 return Err(DbError::Message(
248 "accepted local Store publication was absent from replay".to_string(),
249 ));
250 }
251 }
252 replayed.install_on(self)?;
253 Some(retained)
254 } else {
255 None
256 };
257 let status = finish_store_write_publication_on(
258 store_transaction,
259 &write_id,
260 &audiences,
261 local_cleanup,
262 PublishedWrite::Commit(PublishedPosition {
263 device_id: local_device_id,
264 commit: accepted,
265 }),
266 &active,
267 )?;
268 Ok((retained, (write_id, status)))
269 }
270}
271
272pub(super) fn finish_store_write_publication_on(
273 store: super::StoreTransaction<'_, '_>,
274 write_id: &WriteId,
275 audiences: &crate::PreparedAudienceObjects,
276 local_cleanup: crate::StoreBatchLocalCleanup,
277 published: PublishedWrite,
278 active: &crate::ActiveStorePublication,
279) -> Result<WriteStatus, DbError> {
280 let tx = store.transaction;
281 let coord = published.coord();
282 let cloud_outbox = CloudOutboxRecords::new(tx);
283 let mut consumed_uploads = 0;
284 for package in &audiences.packages {
285 for binding in package.package().blob_bindings() {
286 if cloud_outbox.consume_created_upload_handoff(package.package(), binding)? {
287 consumed_uploads += 1;
288 }
289 }
290 }
291 match Database::make_remote_publication_root_on(tx, write_id)? {
292 Some((root_table, root_id)) => {
293 if consumed_uploads == 0 {
294 return Err(DbError::Message(format!(
295 "make_remote publication {write_id} for {root_table:?}/{root_id:?} contains no Created upload handoff"
296 )));
297 }
298 let remaining: i64 = tx
299 .query_row(
300 "SELECT COUNT(*) FROM cloud_outbox
301 WHERE operation = 'upload' AND root_table = ?1 AND root_id = ?2",
302 (&root_table, &root_id),
303 |row| row.get(0),
304 )
305 .map_err(DbError::from)?;
306 if remaining != 0 {
307 return Err(DbError::Message(format!(
308 "make_remote publication {write_id} left {remaining} upload handoff(s) for {root_table:?}/{root_id:?}"
309 )));
310 }
311 Database::complete_make_remote_publication_on(tx, write_id)?;
312 }
313 None if consumed_uploads != 0 => {
314 return Err(DbError::Message(format!(
315 "Store write {write_id} consumed Created upload handoffs without a make_remote publication intent"
316 )));
317 }
318 None => {}
319 }
320 for drop in local_cleanup.drops {
321 tx.execute(
322 "INSERT INTO published_blob_drop_intents
323 (seq, namespace, blob_id, size, plaintext_hash, locator_hash, disposition)
324 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
325 ON CONFLICT(seq, namespace, blob_id, locator_hash) DO NOTHING",
326 rusqlite::params![
327 Database::sequence_to_sqlite(&coord.stream_id.to_string(), coord.sequence(),)?,
328 drop.namespace,
329 drop.id,
330 i64::try_from(drop.size).map_err(|_| DbError::Message(
331 "outbound local cleanup size exceeds SQLite integer".to_string()
332 ))?,
333 drop.plaintext_hash.to_string(),
334 drop.locator_hash.to_string(),
335 drop.disposition.as_db(),
336 ],
337 )
338 .map_err(DbError::from)?;
339 }
340 tx.execute(
341 "DELETE FROM store_write_packages WHERE write_id = ?1",
342 [write_id.as_str()],
343 )
344 .map_err(DbError::from)?;
345 tx.execute(
346 "DELETE FROM store_write_blobs WHERE write_id = ?1",
347 [write_id.as_str()],
348 )
349 .map_err(DbError::from)?;
350 retain_local_replay_blob_leases(tx, store.store_dir, write_id)?;
351 let cleared = tx
352 .execute(
353 "UPDATE store_writes SET prepared = NULL
354 WHERE write_id = ?1 AND prepared IS NOT NULL",
355 [write_id.as_str()],
356 )
357 .map_err(DbError::from)?;
358 if cleared != 1 {
359 return Err(DbError::Message(
360 "prepared Store write disappeared".to_string(),
361 ));
362 }
363 super::active_store_publication::clear_active_store_publication_on(tx, active)?;
364 let status = WriteStatus::Published(Box::new(published));
365 Database::set_write_status_on(tx, write_id, &status)?;
366 Ok(status)
367}
368
369fn install_accepted_commit_publication_on(
370 transaction: &rusqlite::Transaction<'_>,
371 attempt: &coven_protocol::prepared_commit::PreparedStorePublication,
372 commit: &VerifiedStoreBatchCommit,
373 accepted: &crate::AcceptedStorePublicationInterval,
374) -> Result<crate::AcceptedStoreCommitPublication, DbError> {
375 let previous =
376 super::observed_store_publication::load_store_current_publication_on(transaction)?;
377 if previous.record() != &attempt.previous
378 || previous.observed_version() != Some(&attempt.previous_version)
379 {
380 return Err(DbError::Message(
381 "accepted Store commit extends a stale local publication boundary".to_string(),
382 ));
383 }
384 if accepted.interval().previous() != &*attempt.previous
385 || accepted.interval().current() != &attempt.replacement
386 {
387 return Err(DbError::Message(
388 "accepted Store publication interval differs from its prepared boundaries".to_string(),
389 ));
390 }
391 let unverified = attempt.entry.clone();
392 let reference = coven_protocol::store_commit::StorePublicationRef::from_entry(
393 &unverified,
394 attempt.entry_object.clone(),
395 )
396 .map_err(|error| DbError::context("accepted Store publication reference", error))?;
397 let publication = accepted
398 .accepted_commit(commit)
399 .map_err(|error| DbError::context("accepted Store commit publication", error))?;
400 if publication.entry() != &attempt.entry || publication.reference() != &reference {
401 return Err(DbError::Message(
402 "accepted Store publication differs from the prepared entry".to_string(),
403 ));
404 }
405 super::observed_store_publication::install_store_publication_interval_on(
406 transaction,
407 &previous,
408 accepted,
409 )?;
410 Ok(publication)
411}
412
413fn retain_local_replay_blob_leases(
414 tx: &rusqlite::Transaction<'_>,
415 store_dir: &coven_foundation::store_dir::StoreDir,
416 write_id: &WriteId,
417) -> Result<(), DbError> {
418 let records = super::StoreRecords::new(tx, store_dir);
419 let partitions = records.store_write_partitions(write_id.as_str())?;
420 let local_rows = partitions
421 .local
422 .iter()
423 .map(|partition| crate::walk_changeset(&partition.changeset))
424 .collect::<Result<Vec<_>, _>>()?
425 .into_iter()
426 .flatten()
427 .filter(|change| {
428 !crate::is_routing_table(&change.table)
429 && !matches!(change.op, coven_foundation::changeset::ChangeOp::Delete)
430 })
431 .filter_map(|change| {
432 let row_id = change.pk()?.to_string();
433 Some((change.table, row_id))
434 })
435 .collect::<std::collections::BTreeSet<_>>();
436 let raw_facts: String = tx
437 .query_row(
438 "SELECT blob_facts FROM store_writes WHERE write_id = ?1",
439 [write_id.as_str()],
440 |row| row.get(0),
441 )
442 .map_err(DbError::from)?;
443 let facts: crate::StoreWriteBlobFacts = serde_json::from_str(&raw_facts)
444 .map_err(|error| DbError::context("published Store write blob facts", error))?;
445 let retained = facts
446 .blobs
447 .into_iter()
448 .filter(|fact| {
449 fact.blob.provenance == coven_protocol::blob::Provenance::HostProvided
450 && local_rows.contains(&(fact.table.clone(), fact.row_id.clone()))
451 })
452 .map(|fact| (fact.blob.namespace, fact.blob.id))
453 .collect::<std::collections::BTreeSet<_>>();
454 let leases = crate::query_mapped_rows(
455 tx,
456 "SELECT namespace, blob_id FROM store_write_blob_leases
457 WHERE write_id = ?1 ORDER BY namespace, blob_id",
458 [write_id.as_str()],
459 |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
460 )?;
461 for (namespace, blob_id) in leases {
462 if retained.contains(&(namespace.clone(), blob_id.clone())) {
463 continue;
464 }
465 tx.execute(
466 "DELETE FROM store_write_blob_leases
467 WHERE write_id = ?1 AND namespace = ?2 AND blob_id = ?3",
468 (write_id.as_str(), namespace, blob_id),
469 )
470 .map_err(DbError::from)?;
471 }
472 Ok(())
473}
474
475impl StoreSession<'_> {
476 fn complete_prepared_store_write(
477 &mut self,
478 accepted_publication: crate::StoreCommitPublicationOutcome,
479 routing_key: Option<coven_protocol::circle::RowRoutingKey>,
480 ) -> Result<
481 (
482 Option<OwnedVerifiedMergeMaterialization>,
483 (WriteId, WriteStatus),
484 ),
485 DbError,
486 > {
487 self.verified_store_transaction(move |transaction| {
488 let result =
489 transaction.complete_prepared_store_write(accepted_publication, routing_key)?;
490 Ok(StoreTransactionOutcome::Commit(result))
491 })
492 }
493}
494
495impl StoreDatabase {
496 pub async fn complete_prepared_store_write(
497 &self,
498 accepted_publication: crate::StoreCommitPublicationOutcome,
499 routing_key: Option<coven_protocol::circle::RowRoutingKey>,
500 ) -> Result<Option<OwnedVerifiedMergeMaterialization>, DbError> {
501 let (materialization, (write_id, status)) = self
502 .call_store(move |session| {
503 session.complete_prepared_store_write(accepted_publication, routing_key)
504 })
505 .await?;
506 self.notify_write_status(write_id, status);
507 Ok(materialization)
508 }
509}