1use super::*;
2use crate::{
3 resolve::{Merged, Version},
4 working::{Request, Sealed},
5};
6use onestore::{CommitError, CommitState, RevisionIndex, Stamp, Store, Transaction};
7use rusqlite::OptionalExtension;
8use std::{
9 collections::BTreeMap,
10 sync::{MutexGuard, TryLockError},
11};
12
13/// A single remote file with fresh reads and native-compatible guarded publication.
14/// Errors retain publication state; confirmation checks the stamp, flushes, and notifies
15/// cached readers.
16pub trait Remote {
17 /// Completes a non-blocking operation that returned `WouldBlock`; retrying that
18 /// operation then takes its result. Blocking providers leave this absent.
19 fn pending(&mut self) -> Option<std::pin::Pin<Box<dyn Future<Output = ()> + '_>>> {
20 None
21 }
22 fn read(&mut self) -> io::Result<Vec<u8>>;
23 /// The file's stamp without reading its body or coordinating with writers. While it is
24 /// the last observed image's, synchronization neither reads nor revalidates the file.
25 fn stamp(&mut self) -> io::Result<Stamp>;
26 fn publish(&mut self, transaction: &Transaction) -> std::result::Result<(), CommitError>;
27 fn accepts_edits(&self) -> bool {
28 false
29 }
30 fn publish_edits(
31 &mut self,
32 transaction: &Transaction,
33 edits: &[PendingEdit],
34 revisions: &BTreeMap<ExGuid, ExGuid>,
35 ) -> std::result::Result<(), CommitError> {
36 let _ = (edits, revisions);
37 self.publish(transaction)
38 }
39 /// Confirms that the file still has `base`'s stamp and is durable (`onestore::confirm`).
40 fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), CommitError>;
41 /// The versions a file provider keeps beside the file, as iCloud Drive keeps the commits
42 /// that lost to another device's (unresolved conflict versions); none by default.
43 /// Synchronization merges each into the file, then retires it.
44 fn versions(&mut self) -> io::Result<Vec<Version>> {
45 Ok(Vec::new())
46 }
47 /// A version's image.
48 fn version(&mut self, id: &str) -> io::Result<Vec<u8>> {
49 Err(io::Error::new(
50 io::ErrorKind::NotFound,
51 format!("No version {id}"),
52 ))
53 }
54 /// Retires a version the file now holds everything of; with `keep`, one that is another
55 /// section or cannot be read, kept first as a file of its own beside this one.
56 fn retire(&mut self, id: &str, keep: bool) -> io::Result<()> {
57 let _ = (id, keep);
58 Ok(())
59 }
60}
61
62pub(crate) trait Waiting {
63 fn waiting(&self) -> bool;
64}
65
66impl Waiting for io::Error {
67 fn waiting(&self) -> bool {
68 self.kind() == io::ErrorKind::WouldBlock
69 }
70}
71
72impl Waiting for CommitError {
73 fn waiting(&self) -> bool {
74 self.error.kind() == io::ErrorKind::WouldBlock
75 }
76}
77
78pub(crate) async fn awaited<R: Remote, T, E: Waiting>(
79 remote: &mut R,
80 mut operation: impl FnMut(&mut R) -> std::result::Result<T, E>,
81) -> std::result::Result<T, E> {
82 loop {
83 let result = operation(remote);
84 if result.as_ref().is_err_and(Waiting::waiting)
85 && let Some(pending) = remote.pending()
86 {
87 pending.await;
88 continue;
89 }
90 return result;
91 }
92}
93
94#[derive(Debug, Clone, PartialEq, Eq)]
95pub enum EditStatus {
96 Pending,
97 /// Retained publication attempt; this revision alone may be insufficient to confirm it.
98 AwaitingConfirmation {
99 revision: ExGuid,
100 },
101 /// Historical confirmation; later remote edits or restores may remove the effect.
102 Published {
103 revision: ExGuid,
104 },
105 /// Retired unpublished by a reviewed release; the archive at this path holds the
106 /// edit, its attempt evidence and both images.
107 Archived {
108 archive: String,
109 },
110}
111
112/// What one synchronization step did: the state its batch reached, named by the batch's
113/// newest edit, and the pages a remote change replaced.
114#[derive(Debug, Clone, Default, PartialEq, Eq)]
115pub struct Synced {
116 pub edit: Option<(u64, EditStatus)>,
117 pub changed: Vec<ExGuid>,
118}
119
120/// The queue as a synchronization step starts it.
121struct State {
122 base: Stamp,
123 /// The observed remote image's stamp while it is not the base.
124 remote: Option<Stamp>,
125 /// The first batch waiting on a remote answer to an attempt.
126 blocked: Option<Blocked>,
127 queued: bool,
128}
129
130struct Blocked {
131 batch: i64,
132 revisions: Option<BTreeMap<ExGuid, ExGuid>>,
133}
134
135fn state(connection: &Connection) -> Result<State> {
136 let base = base::base_stamp(connection)?;
137 let blocked = connection
138 .query_row(
139 "SELECT id, revisions FROM batches WHERE attempted=1 ORDER BY id LIMIT 1",
140 [],
141 |row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option<String>>(1)?)),
142 )
143 .optional()?
144 .map(|(batch, revisions)| {
145 Ok::<_, Error>(Blocked {
146 batch,
147 revisions: revisions
148 .map(|revisions| decode_revisions(&revisions))
149 .transpose()?,
150 })
151 })
152 .transpose()?;
153 Ok(State {
154 base,
155 remote: base::stamp(connection, base::Image::Remote)?,
156 blocked,
157 queued: connection
158 .query_row("SELECT EXISTS(SELECT 1 FROM batches)", [], |row| row.get(0))?,
159 })
160}
161
162fn decode_revisions(encoded: &str) -> Result<BTreeMap<ExGuid, ExGuid>> {
163 let revisions: BTreeMap<ExGuid, ExGuid> = serde_json::from_str(encoded)
164 .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
165 if revisions
166 .iter()
167 .any(|(space, revision)| space.guid == [0; 16] || revision.guid == [0; 16])
168 {
169 return Err(io::Error::new(
170 io::ErrorKind::InvalidData,
171 "Cached publication evidence is incomplete",
172 )
173 .into());
174 }
175 Ok(revisions)
176}
177
178/// The revision a batch's seal stored for the first page an edit changes.
179fn revision_of(edit: &onestore::op::Edit, revisions: &BTreeMap<ExGuid, ExGuid>) -> Result<ExGuid> {
180 edit.ops
181 .iter()
182 .find_map(|op| match op {
183 onestore::op::Op::Page { space, .. } => revisions.get(space),
184 onestore::op::Op::Section(_) => None,
185 })
186 .or_else(|| revisions.values().next())
187 .copied()
188 .ok_or_else(|| {
189 io::Error::new(
190 io::ErrorKind::InvalidData,
191 "Cached publication evidence is incomplete",
192 )
193 .into()
194 })
195}
196
197/// The newest edit of a batch.
198fn newest(connection: &Connection, batch: i64) -> Result<u64> {
199 unsigned(
200 connection.query_row("SELECT max(id) FROM edits WHERE batch=?1", [batch], |row| {
201 row.get(0)
202 })?,
203 )
204}
205
206impl Replica {
207 /// Returns a durable receipt or the persisted state of a locally acknowledged edit.
208 pub fn status(&self, id: u64) -> Result<Option<EditStatus>> {
209 status(&*self.lock()?, id, self.section.key.as_ref())
210 }
211
212 /// The last observed remote image. Observation alone does not acknowledge any pending
213 /// edit's remote durability.
214 pub(crate) fn remote_snapshot(&self) -> Result<Vec<u8>> {
215 let connection = self.lock()?;
216 match base::read(&connection, base::Image::Remote)? {
217 Some(image) => Ok(image),
218 None => base::base(&connection),
219 }
220 }
221
222 /// Publishes the oldest unpublished batch, rebasing the queue first when the remote
223 /// changed. Reads the remote image only when its stamp moved; network I/O holds
224 /// synchronization ownership without holding the cache mutex. Uncertain attempts are
225 /// never replayed.
226 /// Versions the remote keeps beside the file merge into it first, each published as one
227 /// more revision and then retired (`resolve.rs`).
228 pub fn sync_once(&self, remote: &mut impl Remote) -> Result<Synced> {
229 crate::task::ready(self.sync_once_async(remote))?
230 }
231
232 /// The same guarded synchronization step, awaiting non-blocking remote operations.
233 pub async fn sync_once_async(&self, remote: &mut impl Remote) -> Result<Synced> {
234 let result = self.sync_once_inner(remote).await;
235 crate::fs::durable().await?;
236 result
237 }
238
239 /// Ownership uses `try_lock`; the cache mutex is released before every await.
240 #[allow(clippy::await_holding_lock)]
241 async fn sync_once_inner(&self, remote: &mut impl Remote) -> Result<Synced> {
242 let _owner = self.sync_owner()?;
243 for version in awaited(remote, Remote::versions)
244 .await
245 .map_err(Error::RemoteIo)?
246 {
247 let image = awaited(remote, |remote| remote.version(&version.id))
248 .await
249 .map_err(Error::RemoteIo)?;
250 let current = awaited(remote, Remote::read)
251 .await
252 .map_err(Error::RemoteIo)?;
253 let device = version.device.as_deref().unwrap_or("Another device");
254 let keep = match crate::resolve::merge(&current, &image, device) {
255 Ok(Merged::Held) => false,
256 Ok(Merged::Publish(transaction)) => {
257 awaited(remote, |remote| remote.publish(&transaction)).await?;
258 false
259 }
260 // A version this cannot merge is kept whole rather than lost.
261 Ok(Merged::Foreign) | Err(Error::Document(_) | Error::Rejected(_)) => true,
262 Err(error) => return Err(error),
263 };
264 awaited(remote, |remote| remote.retire(&version.id, keep))
265 .await
266 .map_err(Error::RemoteIo)?;
267 }
268 let state = state(&*self.lock()?)?;
269 let batched = self.section.key.is_none()
270 && state.queued
271 && state.blocked.is_none()
272 && remote.accepts_edits();
273 let observed = awaited(remote, Remote::stamp)
274 .await
275 .map_err(Error::RemoteIo)?;
276 if let Some(blocked) = &state.blocked
277 && observed == *state.remote.as_ref().unwrap_or(&state.base)
278 {
279 // Nothing new to decide: the remote is as it was when the batch blocked.
280 let id = newest(&*self.lock()?, blocked.batch)?;
281 return Ok(Synced {
282 edit: Some((
283 id,
284 status(&*self.lock()?, id, self.section.key.as_ref())?
285 .unwrap_or(EditStatus::Pending),
286 )),
287 changed: Vec::new(),
288 });
289 }
290 let image = match observed {
291 observed if observed == state.base || batched => None,
292 _ => {
293 let image = awaited(remote, Remote::read)
294 .await
295 .map_err(Error::RemoteIo)?;
296 // Protected elsewhere: a section written anew, which only its key reads.
297 if self.section.key.is_none() && crate::discover::locked(&Store::parse(&image)?) {
298 return Err(Error::RemoteIo(io::Error::new(
299 io::ErrorKind::Unsupported,
300 "Password protected",
301 )));
302 }
303 // A read image is compared whole: a stamp stands for it only when read alone.
304 (base::base(&*self.lock()?)? != image).then_some(image)
305 }
306 };
307 let mut changed = Vec::new();
308 if let Some(image) = image {
309 if let Some(Blocked { batch, revisions }) = &state.blocked {
310 let id = newest(&*self.lock()?, *batch)?;
311 let revisions = revisions.clone().unwrap_or_default();
312 let store = Store::parse(&image)?;
313 if !store.checksum_mismatches.is_empty() {
314 return Err(io::Error::new(
315 io::ErrorKind::InvalidData,
316 "Notebook transaction checksum damage",
317 )
318 .into());
319 }
320 let index = RevisionIndex::parse(&store)?;
321 index.validate_current()?;
322 if index.root != self.root {
323 return Err(io::Error::new(
324 io::ErrorKind::InvalidInput,
325 "Remote snapshot belongs to another document",
326 )
327 .into());
328 }
329 let observed = revisions.iter().all(|(space, revision)| {
330 index
331 .spaces
332 .get(space)
333 .is_some_and(|space| space.revisions.contains_key(revision))
334 });
335 let sealed = working::sealed(&*self.lock()?)?
336 .filter(|sealed| sealed.batch == *batch)
337 .and_then(|sealed| sealed.transaction);
338 // Without its revisions the attempt still counts once the remote holds
339 // every page it changed as it changed them; its receipts then name the
340 // remote's revisions.
341 let receipts = if observed {
342 None
343 } else {
344 let equal = self.holds(&image, sealed.as_ref(), &revisions)?;
345 if !equal {
346 base::write(&*self.lock()?, base::Image::Remote, &image)?;
347 return Ok(Synced {
348 edit: Some((
349 id,
350 status(&*self.lock()?, id, self.section.key.as_ref())?
351 .unwrap_or(EditStatus::Pending),
352 )),
353 changed,
354 });
355 }
356 Some(
357 revisions
358 .keys()
359 .filter_map(|space| Some((*space, index.active(*space).ok()?)))
360 .collect(),
361 )
362 };
363 let stamp = Stamp::of(&image)?;
364 if let Err(error) = awaited(remote, |remote| remote.confirm(&stamp)).await {
365 if error.state == CommitState::Committed {
366 self.acknowledge(*batch, sealed.as_ref(), receipts.as_ref())?;
367 }
368 return Err(error.into());
369 }
370 self.acknowledge(*batch, sealed.as_ref(), receipts.as_ref())?;
371 let revision = self.receipt(id)?;
372 changed = self.rebase(Some(image))?;
373 return Ok(Synced {
374 edit: Some((id, EditStatus::Published { revision })),
375 changed,
376 });
377 }
378 changed = self.rebase(Some(image))?;
379 } else if let Some(blocked) = &state.blocked {
380 if state.remote.is_some() {
381 // The remote is back at the base: what blocked the queue is gone.
382 base::clear(&*self.lock()?, base::Image::Remote)?;
383 }
384 let id = newest(&*self.lock()?, blocked.batch)?;
385 return Ok(Synced {
386 edit: Some((
387 id,
388 status(&*self.lock()?, id, self.section.key.as_ref())?
389 .unwrap_or(EditStatus::Pending),
390 )),
391 changed,
392 });
393 } else if !state.queued {
394 return Ok(Synced::default());
395 }
396 // The cache lock is released before the section thread, which takes it, seals.
397 let waiting = working::sealed(&*self.lock()?)?;
398 let sealed = match waiting {
399 Some(sealed) => {
400 if sealed.transaction.is_some() {
401 self.lock()?
402 .execute("UPDATE batches SET attempted=1 WHERE id=?1", [sealed.batch])?;
403 }
404 Some(sealed)
405 }
406 None => self.ask(|reply| Request::Seal { reply })?,
407 };
408 let Some(Sealed { batch, transaction }) = sealed else {
409 return Ok(Synced {
410 edit: None,
411 changed,
412 });
413 };
414 let id = newest(&*self.lock()?, batch)?;
415 let Some(transaction) = transaction else {
416 // Edits that changed nothing are published once the remote's image is durable.
417 let base = base::base_stamp(&*self.lock()?)?;
418 if let Err(error) = awaited(remote, |remote| remote.confirm(&base)).await {
419 if error.state == CommitState::Committed {
420 self.acknowledge(batch, None, None)?;
421 }
422 return Err(error.into());
423 }
424 self.acknowledge(batch, None, None)?;
425 let revision = self.receipt(id)?;
426 return Ok(Synced {
427 edit: Some((id, EditStatus::Published { revision })),
428 changed,
429 });
430 };
431 if let Err(error) = crate::fs::durable().await {
432 self.lock()?
433 .execute("UPDATE batches SET attempted=0 WHERE id=?1", [batch])?;
434 return Err(error.into());
435 }
436 let published = if self.section.key.is_none() && remote.accepts_edits() {
437 let (edits, revisions) = {
438 let connection = self.lock()?;
439 let edits = queue::load(&connection, None, Some(batch))?;
440 let revisions = decode_revisions(&connection.query_row(
441 "SELECT revisions FROM batches WHERE id=?1",
442 [batch],
443 |row| row.get::<_, String>(0),
444 )?)?;
445 (edits, revisions)
446 };
447 awaited(remote, |remote| {
448 remote.publish_edits(&transaction, &edits, &revisions)
449 })
450 .await
451 } else {
452 awaited(remote, |remote| remote.publish(&transaction)).await
453 };
454 match published {
455 Ok(()) => {}
456 Err(error) if error.state == CommitState::NotCommitted => {
457 self.lock()?
458 .execute("UPDATE batches SET attempted=0 WHERE id=?1", [batch])?;
459 return Err(error.into());
460 }
461 Err(error) if error.state == CommitState::Committed => {
462 self.acknowledge(batch, Some(&transaction), None)?;
463 return Err(error.into());
464 }
465 Err(error) => return Err(error.into()),
466 }
467 self.acknowledge(batch, Some(&transaction), None)?;
468 let revision = self.receipt(id)?;
469 if batched {
470 changed.extend(
471 self.rebase(Some(
472 awaited(remote, Remote::read)
473 .await
474 .map_err(Error::RemoteIo)?,
475 ))?,
476 );
477 changed.sort();
478 changed.dedup();
479 }
480 Ok(Synced {
481 edit: Some((id, EditStatus::Published { revision })),
482 changed,
483 })
484 }
485
486 /// Whether `image` holds every page in `revisions` as the base with `sealed` has it.
487 fn holds(
488 &self,
489 image: &[u8],
490 sealed: Option<&Transaction>,
491 revisions: &BTreeMap<ExGuid, ExGuid>,
492 ) -> Result<bool> {
493 let base = base::base(&*self.lock()?)?;
494 let (local, remote) = (onestore::Arena::default(), onestore::Arena::default());
495 let mut local = working::open(&local, base, self.section.key.as_ref())?;
496 if let Some(sealed) = sealed {
497 local.replay(sealed)?;
498 }
499 let mut remote = working::open(&remote, image.to_vec(), self.section.key.as_ref())?;
500 // A page's versions change without its page changing.
501 let versions = |section: &mut onestore::Section<'_>| {
502 section.versions().map(|versions| {
503 versions
504 .into_iter()
505 .filter(|(page, _)| revisions.contains_key(page))
506 .collect::<Vec<_>>()
507 })
508 };
509 Ok(revisions.keys().all(
510 |space| matches!((local.page(*space), remote.page(*space)), (Ok(a), Ok(b)) if a == b),
511 ) && versions(&mut local)? == versions(&mut remote)?)
512 }
513
514 /// Asks the section thread to replay the queue on `image`, returning the pages the
515 /// remote changed.
516 fn rebase(&self, image: Option<Vec<u8>>) -> Result<Vec<ExGuid>> {
517 self.ask(|reply| Request::Rebase { image, reply })
518 }
519
520 fn receipt(&self, id: u64) -> Result<ExGuid> {
521 match status(&*self.lock()?, id, self.section.key.as_ref())? {
522 Some(EditStatus::Published { revision }) => Ok(revision),
523 _ => Err(io::Error::other("A published edit has no receipt").into()),
524 }
525 }
526
527 /// Records a published batch: the base becomes the image it left, each of its edits
528 /// gets a receipt naming the revisions its seal stored, or `revisions`, and it leaves
529 /// the queue.
530 fn acknowledge(
531 &self,
532 batch: i64,
533 transaction: Option<&Transaction>,
534 revisions: Option<&BTreeMap<ExGuid, ExGuid>>,
535 ) -> Result<()> {
536 let mut connection = self.lock()?;
537 let database = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
538 if let Some(transaction) = transaction {
539 base::publish(&database, transaction)?;
540 }
541 let revisions = match revisions {
542 Some(revisions) => revisions.clone(),
543 None => decode_revisions(&database.query_row(
544 "SELECT revisions FROM batches WHERE id=?1",
545 [batch],
546 |row| row.get::<_, String>(0),
547 )?)?,
548 };
549 for queued in queue::load(&database, self.section.key.as_ref(), Some(batch))? {
550 database.execute(
551 "INSERT INTO receipts(edit_id, revision) VALUES (?1, ?2)",
552 params![
553 signed(queued.id)?,
554 revision_of(&queued.edit, &revisions)?.to_string()
555 ],
556 )?;
557 }
558 database.execute("DELETE FROM batches WHERE id=?1", [batch])?;
559 queue::collect(&database)?;
560 database.commit()?;
561 Ok(())
562 }
563
564 /// Retires the uncertain attempt of the batch holding edit `id` after review. The
565 /// queue is first exported to `archive` (a new file), which is the record: no receipt
566 /// is written. `Mine` publishes the batch again against the current remote; `Theirs`
567 /// abandons every unpublished edit, which become `Archived`, and the local pages return
568 /// to the remote's.
569 pub fn release(&self, id: u64, archive: &Path, resolution: Resolution) -> Result<()> {
570 let owner = self.sync_owner()?;
571 let batch: Option<i64> = self
572 .lock()?
573 .query_row(
574 "SELECT id FROM batches WHERE attempted=1 AND id=(SELECT batch FROM edits WHERE id=?1)",
575 [signed(id)?],
576 |row| row.get(0),
577 )
578 .optional()?;
579 let Some(batch) = batch else {
580 return Err(io::Error::new(
581 io::ErrorKind::InvalidInput,
582 "The edit has no uncertain attempt",
583 )
584 .into());
585 };
586 self.export_recovery(archive)?;
587 {
588 let mut connection = self.lock()?;
589 let transaction =
590 connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
591 match resolution {
592 Resolution::Mine => {
593 transaction.execute("UPDATE batches SET attempted=0 WHERE id=?1", [batch])?;
594 }
595 Resolution::Theirs => {
596 transaction.execute(
597 "INSERT INTO archived(edit_id, archive) SELECT id, ?1 FROM edits",
598 [archive.to_string_lossy().into_owned()],
599 )?;
600 transaction.execute("DELETE FROM batches", [])?;
601 if let Some(remote) = base::read(&transaction, base::Image::Remote)? {
602 base::write(&transaction, base::Image::Base, &remote)?;
603 base::clear(&transaction, base::Image::Remote)?;
604 }
605 queue::collect(&transaction)?;
606 }
607 }
608 transaction.commit()?;
609 }
610 if resolution == Resolution::Theirs {
611 self.ask(|reply| Request::Reopen { reply })?;
612 }
613 drop(owner);
614 self.wake_sync();
615 Ok(())
616 }
617
618 fn sync_owner(&self) -> Result<MutexGuard<'_, ()>> {
619 Ok(self
620 .synchronization
621 .try_lock()
622 .map_err(|error| match error {
623 TryLockError::WouldBlock => io::Error::from(io::ErrorKind::WouldBlock),
624 TryLockError::Poisoned(_) => {
625 io::Error::other("Synchronization owner panicked; reopen the cache")
626 }
627 })?)
628 }
629
630 /// Whether nothing is queued and the remote still has the base's stamp, or the queue
631 /// is blocked on a remote that has not changed since; reads neither image and does not
632 /// lock the cache during remote I/O.
633 pub(crate) async fn settled(&self, remote: &mut impl Remote) -> Result<bool> {
634 let state = state(&*self.lock()?)?;
635 let expected = match &state.blocked {
636 Some(_) => state.remote.unwrap_or(state.base),
637 None if !state.queued => state.base,
638 None => return Ok(false),
639 };
640 Ok(awaited(remote, Remote::stamp)
641 .await
642 .map_err(Error::RemoteIo)?
643 == expected
644 && awaited(remote, Remote::versions)
645 .await
646 .map_err(Error::RemoteIo)?
647 .is_empty())
648 }
649}
650
651pub(crate) fn status(
652 connection: &Connection,
653 id: u64,
654 key: Option<&onestore::protected::Key>,
655) -> Result<Option<EditStatus>> {
656 let id = signed(id)?;
657 if let Some(revision) = connection
658 .query_row(
659 "SELECT revision FROM receipts WHERE edit_id=?1",
660 [id],
661 |row| row.get::<_, String>(0),
662 )
663 .optional()?
664 {
665 return Ok(Some(EditStatus::Published {
666 revision: revision.parse()?,
667 }));
668 }
669 if let Some(archive) = connection
670 .query_row(
671 "SELECT archive FROM archived WHERE edit_id=?1",
672 [id],
673 |row| row.get::<_, String>(0),
674 )
675 .optional()?
676 {
677 return Ok(Some(EditStatus::Archived { archive }));
678 }
679 let record: Option<(bool, Option<String>, String)> = connection
680 .query_row(
681 "SELECT batches.attempted, batches.revisions, edits.edit
682 FROM edits JOIN batches ON batches.id=edits.batch WHERE edits.id=?1",
683 [id],
684 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
685 )
686 .optional()?;
687 Ok(match record {
688 None => None,
689 Some((true, revisions, edit)) => {
690 let revisions = decode_revisions(revisions.as_deref().unwrap_or("{}"))?;
691 let edit = queue::parse(key, &edit)?;
692 let revision = revision_of(&edit, &revisions)?;
693 Some(EditStatus::AwaitingConfirmation { revision })
694 }
695 Some((false, ..)) => Some(EditStatus::Pending),
696 })
697}