1use super::*;
2use crate::store::StoreSession;
3
4impl StoreSession<'_> {
5 fn merge_candidate_terminal_verifications(
6 &mut self,
7 root: &coven_protocol::store_commit::StoreRootRef,
8 write_id: &WriteId,
9 ) -> Result<Vec<TerminalCandidateCleanupVerification>, DbError> {
10 let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
11 let verified_authority = &mut *self.verified_store_authority;
12 let conn = self.conn;
13 let (raw_status, raw_prepared): (String, Option<String>) = conn
14 .query_row(
15 "SELECT status, prepared FROM store_writes WHERE write_id = ?1",
16 [write_id.as_str()],
17 |row| Ok((row.get(0)?, row.get(1)?)),
18 )
19 .map_err(DbError::from)?;
20 let status: WriteStatus = serde_json::from_str(&raw_status)
21 .map_err(|error| DbError::context("Merge cleanup status", error))?;
22 let mut candidates = Vec::new();
23 if let WriteStatus::Resolved(WriteResolution::Retracted { witness }) = status {
24 witness.validate().map_err(DbError::from)?;
25 let candidate = crate::StoreDatabase::load_merge_retraction_cleanup_on(
26 records,
27 verified_authority,
28 witness.original_position().commit(),
29 )?;
30 if candidate.commit.write_id != *write_id {
31 return Err(DbError::Message(
32 "Merge retraction cleanup names another write".to_string(),
33 ));
34 }
35 candidates.push(candidate);
36 } else {
37 let raw_prepared = raw_prepared.ok_or_else(|| {
38 DbError::Message("Merge cleanup has no prepared candidate".to_string())
39 })?;
40 let prepared: PreparedStoreWriteState = serde_json::from_str(&raw_prepared)
41 .map_err(|error| DbError::context("prepared Merge cleanup", error))?;
42 match &prepared {
43 PreparedStoreWriteState::Publication { .. } => {
44 candidates.push(parse_prepared_merge_candidate_on(
45 records,
46 verified_authority,
47 &prepared,
48 )?);
49 }
50 PreparedStoreWriteState::MergeAbandonment {
51 candidate_commit,
52 candidate_head,
53 authority_commit,
54 authority_head,
55 ..
56 } => {
57 candidates.push(parse_prepared_merge_candidate_parts_on(
58 records,
59 verified_authority,
60 candidate_commit.semantic_bytes(),
61 candidate_commit.prepared().reference(),
62 candidate_head.semantic_bytes(),
63 candidate_head.prepared().reference(),
64 )?);
65 candidates.push(parse_prepared_merge_candidate_parts_on(
66 records,
67 verified_authority,
68 authority_commit.semantic_bytes(),
69 authority_commit.prepared().reference(),
70 authority_head.semantic_bytes(),
71 authority_head.prepared().reference(),
72 )?);
73 }
74 }
75 }
76 let mut verifications = Vec::new();
77 for candidate in candidates {
78 if let Some(verification) =
79 terminal_candidate_verification_on(records, verified_authority, root, candidate)?
80 {
81 verifications.push(verification);
82 }
83 }
84 Ok(verifications)
85 }
86
87 fn reconcile_merge_candidate_terminal_head(
88 &mut self,
89 root: &coven_protocol::store_commit::StoreRootRef,
90 write_id: &WriteId,
91 durable: coven_protocol::remote_object::CandidateNonactivation,
92 head_nonactivation: coven_protocol::remote_object::VerifiedCandidateHeadNonactivation,
93 ) -> Result<(), DbError> {
94 let verified_authority = &mut *self.verified_store_authority;
95 let conn = self.conn;
96 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
97 let (raw_status, raw_prepared): (String, Option<String>) = tx
98 .query_row(
99 "SELECT status, prepared FROM store_writes WHERE write_id = ?1",
100 [write_id.as_str()],
101 |row| Ok((row.get(0)?, row.get(1)?)),
102 )
103 .map_err(DbError::from)?;
104 let reference = durable.reference().map_err(DbError::from)?;
105 let mut candidates = Vec::new();
106 let status: WriteStatus = serde_json::from_str(&raw_status)
107 .map_err(|error| DbError::context("Merge cleanup status", error))?;
108 let store_transaction =
109 crate::store::store_session::StoreTransaction::new(&tx, self.store_dir);
110 if let WriteStatus::Resolved(WriteResolution::Retracted { witness }) = status {
111 witness.validate().map_err(DbError::from)?;
112 if witness.original_position().commit() != &reference {
113 return Err(DbError::Message(
114 "fresh excluded-author head evidence differs from the retraction witness"
115 .to_string(),
116 ));
117 }
118 candidates.push(
119 store_transaction.load_merge_retraction_cleanup(verified_authority, &reference)?,
120 );
121 } else {
122 let raw_prepared = raw_prepared.ok_or_else(|| {
123 DbError::Message("Merge cleanup has no prepared candidate".to_string())
124 })?;
125 let prepared: PreparedStoreWriteState = serde_json::from_str(&raw_prepared)
126 .map_err(|error| DbError::context("prepared Merge cleanup", error))?;
127 match &prepared {
128 PreparedStoreWriteState::Publication { commit, head, .. } => {
129 candidates.push(store_transaction.prepared_merge_candidate_parts(
130 verified_authority,
131 commit.semantic_bytes(),
132 commit.prepared().reference(),
133 head.semantic_bytes(),
134 head.prepared().reference(),
135 )?)
136 }
137 PreparedStoreWriteState::MergeAbandonment {
138 candidate_commit,
139 candidate_head,
140 authority_commit,
141 authority_head,
142 ..
143 } => {
144 candidates.push(store_transaction.prepared_merge_candidate_parts(
145 verified_authority,
146 candidate_commit.semantic_bytes(),
147 candidate_commit.prepared().reference(),
148 candidate_head.semantic_bytes(),
149 candidate_head.prepared().reference(),
150 )?);
151 candidates.push(store_transaction.prepared_merge_candidate_parts(
152 verified_authority,
153 authority_commit.semantic_bytes(),
154 authority_commit.prepared().reference(),
155 authority_head.semantic_bytes(),
156 authority_head.prepared().reference(),
157 )?);
158 }
159 }
160 }
161 let candidate = candidates
162 .into_iter()
163 .find(|candidate| candidate.reference == reference)
164 .ok_or_else(|| {
165 DbError::Message(
166 "fresh excluded-author head evidence names another write".to_string(),
167 )
168 })?;
169 store_transaction.validate_terminal_candidate_authority(
170 verified_authority,
171 root,
172 &candidate,
173 &durable,
174 )?;
175 let object_id = remote_object_id(&candidate.head_object);
176 let remote_exists: bool = tx
177 .query_row(
178 "SELECT EXISTS(SELECT 1 FROM remote_objects WHERE object_id = ?1)",
179 [object_id.to_string()],
180 |row| row.get(0),
181 )
182 .map_err(DbError::from)?;
183 if !remote_exists {
184 let inert = load_protocol_inert_object_on(&tx, object_id)?;
185 if inert
186 .candidate_nonactivation_proof(&candidate.reference)
187 .map_err(DbError::from)?
188 != Some(durable.proof())
189 {
190 return Err(DbError::Message(
191 "protocol-inert candidate head carries another proof".to_string(),
192 ));
193 }
194 return tx.commit().map_err(DbError::from);
195 }
196 let mut remote = load_remote_object_on(&tx, object_id)?;
197 let inert = remote
198 .begin_candidate_nonactivation_with_verified_head_nonactivation(
199 durable,
200 &head_nonactivation,
201 )
202 .map_err(|error| {
203 DbError::context(format!("reconcile excluded-author head {object_id}"), error)
204 })?;
205 finish_remote_candidate_nonactivation_on(&tx, object_id, remote, inert)?;
206 tx.commit().map_err(DbError::from)
207 }
208
209 fn adopt_alternate_merge_head(
210 &mut self,
211 write_id: &WriteId,
212 winner: StoreDeviceHead,
213 winner_prepared: PreparedExactObject,
214 ) -> Result<(), DbError> {
215 let verified_authority = &mut *self.verified_store_authority;
216 let conn = self.conn;
217 let tx = conn.unchecked_transaction().map_err(DbError::from)?;
218 let (raw_status, raw_prepared): (String, String) = tx
219 .query_row(
220 "SELECT status, prepared FROM store_writes WHERE write_id = ?1",
221 [write_id.as_str()],
222 |row| Ok((row.get(0)?, row.get(1)?)),
223 )
224 .map_err(DbError::from)?;
225 let status: WriteStatus = serde_json::from_str(&raw_status)
226 .map_err(|error| DbError::context("alternate Merge head status", error))?;
227 if !matches!(status, WriteStatus::Publishing) {
228 return Err(DbError::Message(format!(
229 "Merge candidate {write_id} is not publishing"
230 )));
231 }
232 let prepared: PreparedStoreWriteState = serde_json::from_str(&raw_prepared)
233 .map_err(|error| DbError::context("prepared Merge candidate", error))?;
234 let store_transaction =
235 crate::store::store_session::StoreTransaction::new(&tx, self.store_dir);
236 let publication =
237 store_transaction.prepared_merge_publication(verified_authority, &prepared)?;
238 let root = store_transaction.required_root_authority(verified_authority)?;
239 let registration = store_transaction.activated_registration(
240 verified_authority,
241 &root,
242 &publication.commit.author_registration,
243 )?;
244 let candidate = publication.reference;
245 let verified_winner = StoreDeviceHead::parse_at(
246 &winner.to_bytes(),
247 root.store_root_hash,
248 ®istration,
249 &candidate,
250 )
251 .map_err(|error| DbError::context("verify alternate Merge head", error))?;
252 if verified_winner != winner || winner.commit != candidate {
253 return Err(DbError::Message(
254 "alternate Merge head does not activate the prepared commit".to_string(),
255 ));
256 }
257 replace_prepared_merge_head_remote_on(
258 &tx,
259 self.store_dir,
260 &publication.head_object,
261 &winner,
262 winner_prepared.reference(),
263 &candidate,
264 )?;
265 let replacement_head =
266 DurablePreparedProtocolObject::new(winner.to_bytes(), winner_prepared);
267 let replacement = match prepared {
268 PreparedStoreWriteState::Publication {
269 commit,
270 history_evidence,
271 local_cleanup,
272 completion,
273 ..
274 } => PreparedStoreWriteState::Publication {
275 commit,
276 head: replacement_head,
277 history_evidence,
278 local_cleanup,
279 completion,
280 },
281 PreparedStoreWriteState::MergeAbandonment {
282 candidate_commit,
283 candidate_head,
284 candidate_history_evidence,
285 authority_commit,
286 authority_history_evidence,
287 outcome,
288 local_cleanup,
289 completion,
290 ..
291 } => PreparedStoreWriteState::MergeAbandonment {
292 candidate_commit,
293 candidate_head,
294 candidate_history_evidence,
295 authority_commit,
296 authority_head: replacement_head,
297 authority_history_evidence,
298 outcome,
299 local_cleanup,
300 completion,
301 },
302 };
303 let replacement = serde_json::to_string(&replacement)
304 .map_err(|error| DbError::context("serialize alternate Merge preparation", error))?;
305 let updated = tx
306 .execute(
307 "UPDATE store_writes SET prepared = ?2
308 WHERE write_id = ?1 AND status = '\"publishing\"' AND prepared = ?3",
309 rusqlite::params![write_id.as_str(), replacement, raw_prepared],
310 )
311 .map_err(DbError::from)?;
312 if updated != 1 {
313 return Err(DbError::Message(
314 "prepared Merge write changed during head replacement".to_string(),
315 ));
316 }
317 tx.commit().map_err(DbError::from)
318 }
319}
320
321impl StoreDatabase {
322 pub async fn merge_candidate_terminal_verifications(
323 &self,
324 root: coven_protocol::store_commit::StoreRootRef,
325 write_id: WriteId,
326 ) -> Result<Vec<TerminalCandidateCleanupVerification>, DbError> {
327 self.call_store(move |session| {
328 session.merge_candidate_terminal_verifications(&root, &write_id)
329 })
330 .await
331 }
332
333 pub async fn reconcile_merge_candidate_terminal_head(
334 &self,
335 root: coven_protocol::store_commit::StoreRootRef,
336 write_id: WriteId,
337 verified: coven_protocol::remote_object::VerifiedCandidateNonactivation,
338 ) -> Result<(), DbError> {
339 if !matches!(
340 verified.proof(),
341 coven_protocol::remote_object::CandidateNonactivationProof::AuthorExclusion { .. }
342 | coven_protocol::remote_object::CandidateNonactivationProof::MergeMembershipGrantRevocation { .. }
343 ) {
344 return Err(DbError::Message(
345 "terminal head reconciliation received another proof family".to_string(),
346 ));
347 }
348 let (durable, head_nonactivation) = verified
349 .into_terminal_head_nonactivation()
350 .map_err(DbError::from)?;
351 self.call_store(move |session| {
352 session.reconcile_merge_candidate_terminal_head(
353 &root,
354 &write_id,
355 durable,
356 head_nonactivation,
357 )
358 })
359 .await
360 }
361
362 pub async fn adopt_alternate_merge_head(
363 &self,
364 write_id: WriteId,
365 winner: StoreDeviceHead,
366 winner_prepared: PreparedExactObject,
367 ) -> Result<(), DbError> {
368 self.call_store(move |session| {
369 session.adopt_alternate_merge_head(&write_id, winner, winner_prepared)
370 })
371 .await
372 }
373}