Add document retention check to previous_revisions (#24014)

Enforce document retention in `previous_revisions`. This is a follow-up on [this](get-convex/convex#23587) PR.

GitOrigin-RevId: f2219dd6f48c6f6ab2c46cedbccec691f4069612

Author: jordanhunt22

crates/common/src/persistence.rs

@@ -353,6 +353,7 @@ pub trait PersistenceReader: Send + Sync + 'static {
     async fn previous_revisions(
         &self,
         ids: BTreeSet<(InternalDocumentId, Timestamp)>,
+        retention_validator: Arc<dyn RetentionValidator>,
     ) -> anyhow::Result<
         BTreeMap<(InternalDocumentId, Timestamp), (Timestamp, Option<ResolvedDocument>)>,
     >;
@@ -553,7 +554,9 @@ impl RepeatablePersistence {
             // Reading documents <ts, so ts-1 needs to be repeatable.
             anyhow::ensure!(*ts <= self.upper_bound.succ()?);
         }
-        self.reader.previous_revisions(ids).await
+        self.reader
+            .previous_revisions(ids, self.retention_validator.clone())
+            .await
     }
 
     pub fn read_snapshot(&self, at: RepeatableTimestamp) -> anyhow::Result<PersistenceSnapshot> {

crates/common/src/testing/persistence_test_suite.rs

@@ -1643,7 +1643,9 @@ pub async fn persistence_previous_revisions<P: Persistence>(p: Arc<P>) -> anyhow
         (id8, 3, 1, true),
     ];
     assert_eq!(
-        reader.previous_revisions(queries).await?,
+        reader
+            .previous_revisions(queries, Arc::new(NoopRetentionValidator))
+            .await?,
         expected
             .into_iter()
             .map(|(id, ts, prev_ts, exists)| (

crates/common/src/testing/test_persistence.rs

@@ -278,6 +278,7 @@ impl PersistenceReader for TestPersistence {
     async fn previous_revisions(
         &self,
         ids: BTreeSet<(InternalDocumentId, Timestamp)>,
+        _retention_validator: Arc<dyn RetentionValidator>,
     ) -> anyhow::Result<
         BTreeMap<(InternalDocumentId, Timestamp), (Timestamp, Option<ResolvedDocument>)>,
     > {

crates/sqlite/src/lib.rs

@@ -4,6 +4,7 @@
 #![feature(coroutines)]
 
 use std::{
+    cmp,
     collections::{
         BTreeMap,
         BTreeSet,
@@ -533,36 +534,44 @@ impl PersistenceReader for SqlitePersistence {
     async fn previous_revisions(
         &self,
         ids: BTreeSet<(InternalDocumentId, Timestamp)>,
+        retention_validator: Arc<dyn RetentionValidator>,
     ) -> anyhow::Result<
         BTreeMap<(InternalDocumentId, Timestamp), (Timestamp, Option<ResolvedDocument>)>,
     > {
-        let inner = self.inner.lock();
         let mut out = BTreeMap::new();
-        for (id, ts) in ids {
-            let mut stmt = inner.connection.prepare(PREV_REV_QUERY)?;
-            let internal_id = id.internal_id();
-            let params = params![&id.table().0[..], &internal_id[..], &u64::from(ts)];
-            let mut row_iter = stmt.query_map(params, load_document_row)?;
-            if let Some(row) = row_iter.next() {
-                let (id, prev_ts, table, json_value, deleted) = row?;
-                let id = InternalId::try_from(id)?;
-                let table = TableId(table.try_into()?);
-                let prev_ts = Timestamp::try_from(prev_ts)?;
-                let document_id = table.id(id);
-                let document = if !deleted {
-                    let json_value = json_value.ok_or_else(|| {
-                        anyhow::anyhow!("Unexpected NULL json_value at {} {}", id, prev_ts)
-                    })?;
-                    let json_value: serde_json::Value = serde_json::from_str(&json_value)?;
-                    let value: ConvexValue = json_value.try_into()?;
-                    let document = ResolvedDocument::from_database(table, value)?;
-                    Some(document)
-                } else {
-                    None
-                };
-                out.insert((document_id, ts), (prev_ts, document));
+        let mut min_ts = Timestamp::MAX;
+        {
+            let inner = self.inner.lock();
+            for (id, ts) in ids {
+                let mut stmt = inner.connection.prepare(PREV_REV_QUERY)?;
+                let internal_id = id.internal_id();
+                let params = params![&id.table().0[..], &internal_id[..], &u64::from(ts)];
+                let mut row_iter = stmt.query_map(params, load_document_row)?;
+                if let Some(row) = row_iter.next() {
+                    let (id, prev_ts, table, json_value, deleted) = row?;
+                    let id = InternalId::try_from(id)?;
+                    let table = TableId(table.try_into()?);
+                    let prev_ts = Timestamp::try_from(prev_ts)?;
+                    let document_id = table.id(id);
+                    let document = if !deleted {
+                        let json_value = json_value.ok_or_else(|| {
+                            anyhow::anyhow!("Unexpected NULL json_value at {} {}", id, prev_ts)
+                        })?;
+                        let json_value: serde_json::Value = serde_json::from_str(&json_value)?;
+                        let value: ConvexValue = json_value.try_into()?;
+                        let document = ResolvedDocument::from_database(table, value)?;
+                        Some(document)
+                    } else {
+                        None
+                    };
+                    min_ts = cmp::min(ts, min_ts);
+                    out.insert((document_id, ts), (prev_ts, document));
+                }
             }
         }
+        retention_validator
+            .validate_document_snapshot(min_ts)
+            .await?;
         Ok(out)
     }