Change the meaning of persistence_reader_stream_revision_pairs prev_rev.document being None (#41170)

Currently prev_rev.document is None if the previous revision is a tombstone (`deleted=1`),
and the stream returns an error if the revision does not exist. However, a prev_ts should
never be a tombstone, so turn that into an error. On the other hand, it is possible for the previous
revision to not exist at all - if it has been garbage collected. Normally this should be forbidden by
the retention validator, but the document retention worker intentionally looks at rows out of retention,
and it shouldn't encounter an error.

GitOrigin-RevId: f08ea7ca3b9a8ed9c0fd367dc40ff74bcb58ff29

Author: goffrie

crates/common/src/persistence_helpers.rs

@@ -99,12 +99,17 @@ pub(crate) async fn persistence_reader_stream_revision_pairs<'a, P: PersistenceR
             let rev = DocumentRevision { ts, document };
             let prev_rev = prev_ts
                 .map(|prev_ts| {
-                    let entry = prev_revs
+                    let document = prev_revs
                         .remove(&DocumentPrevTsQuery { id, ts, prev_ts })
-                        .with_context(|| format!("prev_ts is missing for {id}@{ts}: {prev_ts}"))?;
+                        .map(|entry| {
+                            entry.value.with_context(|| {
+                                format!("prev_ts {prev_ts} of {id}@{ts} points to a deleted value?")
+                            })
+                        })
+                        .transpose()?;
                     anyhow::Ok(DocumentRevision {
-                        ts: entry.ts,
-                        document: entry.value,
+                        ts: prev_ts,
+                        document,
                     })
                 })
                 .transpose()?;

crates/common/src/testing/persistence_test_suite.rs

@@ -63,6 +63,10 @@ use crate::{
         PersistenceIndexEntry,
         TimestampRange,
     },
+    persistence_helpers::{
+        DocumentRevision,
+        RevisionPair,
+    },
     query::Order,
     testing::{
         self,
@@ -247,6 +251,13 @@ macro_rules! run_persistence_test_suite {
             persistence_test_suite::persistence_previous_revisions(::std::sync::Arc::new(p)).await
         }
 
+        #[tokio::test]
+        async fn test_persistence_load_revision_pairs() -> anyhow::Result<()> {
+            let $db = $create_db;
+            let p = $create_persistence;
+            persistence_test_suite::test_load_revision_pairs(::std::sync::Arc::new(p)).await
+        }
+
         #[tokio::test]
         async fn test_persistence_table_stats() -> anyhow::Result<()> {
             let $db = $create_db;
@@ -2121,6 +2132,71 @@ pub async fn persistence_previous_revisions<P: Persistence>(p: Arc<P>) -> anyhow
     Ok(())
 }
 
+pub async fn test_load_revision_pairs<P: Persistence>(p: Arc<P>) -> anyhow::Result<()> {
+    let mut id_generator = TestIdGenerator::new();
+    let table: TableName = str::parse("table")?;
+
+    let id = id_generator.user_generate(&table);
+    let documents = vec![
+        doc(id, 2, Some(2), Some(1))?,
+        doc(id, 3, Some(3), Some(2))?,
+        doc(id, 4, None, Some(3))?,
+    ];
+    p.write(&documents, &[], ConflictStrategy::Error).await?;
+    let revision_pairs: Vec<_> = p
+        .reader()
+        .load_revision_pairs(
+            None,
+            TimestampRange::new(Timestamp::must(2)..),
+            Order::Asc,
+            1,
+            Arc::new(NoopRetentionValidator),
+        )
+        .try_collect()
+        .await?;
+    assert_eq!(
+        revision_pairs,
+        vec![
+            RevisionPair {
+                id: id.into(),
+                rev: DocumentRevision {
+                    ts: Timestamp::must(2),
+                    document: documents[0].value.clone(),
+                },
+                prev_rev: Some(DocumentRevision {
+                    ts: Timestamp::must(1),
+                    // Odd semantics here: None means that the revision is
+                    // garbage collected, *not* that it is a tombstone
+                    document: None
+                }),
+            },
+            RevisionPair {
+                id: id.into(),
+                rev: DocumentRevision {
+                    ts: Timestamp::must(3),
+                    document: documents[1].value.clone(),
+                },
+                prev_rev: Some(DocumentRevision {
+                    ts: Timestamp::must(2),
+                    document: documents[0].value.clone(),
+                }),
+            },
+            RevisionPair {
+                id: id.into(),
+                rev: DocumentRevision {
+                    ts: Timestamp::must(4),
+                    document: None,
+                },
+                prev_rev: Some(DocumentRevision {
+                    ts: Timestamp::must(3),
+                    document: documents[1].value.clone(),
+                }),
+            },
+        ]
+    );
+    Ok(())
+}
+
 pub async fn persistence_table_stats<P: Persistence>(p: Arc<P>) -> anyhow::Result<()> {
     let reader = p.reader();