validate previous_revisions timestamp is within document retention (#34238)

currently we're validating that the timestamp is within document retention, but only for timestamps that actually have a previous revision. Consider if you query at a too-old-timestamp, where it's very likely the previous revision has been deleted by document retention. Then we don't assert anything, but this is exactly the case where we want to throw an error.

fix it by calculating the min_ts of all the input timestamps, not of all the ones that actually had results.

GitOrigin-RevId: 562ee684de507dc13bcec95e20670cb116ff8d65

Author: ldanilek

crates/mysql/src/lib.rs

@@ -991,13 +991,15 @@ impl<RT: Runtime> PersistenceReader for MySqlReader<RT> {
 
         let mut results = vec![];
 
+        let mut min_ts = Timestamp::MAX;
         for chunk in smart_chunks(&ids) {
             let mut params = vec![];
             for (id, ts) in chunk {
                 params.push(i64::from(*ts).into());
                 params.push(internal_id_param(id.table().0).into());
                 params.push(internal_doc_id_param(*id).into());
                 params.push(i64::from(*ts).into());
+                min_ts = cmp::min(*ts, min_ts);
             }
             let result_stream = client
                 .query_stream(prev_rev_chunk(chunk.len()), params, chunk.len())
@@ -1007,12 +1009,10 @@ impl<RT: Runtime> PersistenceReader for MySqlReader<RT> {
                 results.push(result);
             }
         }
-        let mut min_ts = Timestamp::MAX;
         for row in results.into_iter() {
             let ts: i64 = row.get(6).unwrap();
             let ts = Timestamp::try_from(ts)?;
             let (prev_ts, id, maybe_doc, prev_prev_ts) = self.row_to_document(row)?;
-            min_ts = cmp::min(ts, min_ts);
             anyhow::ensure!(result
                 .insert(
                     (id, ts),

crates/postgres/src/lib.rs

@@ -928,13 +928,15 @@ impl PersistenceReader for PostgresReader {
         // the pipeline at once.
         let mut result_futures = vec![];
 
+        let mut min_ts = Timestamp::MAX;
         for chunk in chunks.by_ref() {
             assert_eq!(chunk.len(), 8);
             let mut params = Vec::with_capacity(16);
             for (id, ts) in chunk {
                 params.push(Param::TableId(id.table()));
                 params.push(internal_doc_id_param(*id));
                 params.push(Param::Ts(i64::from(*ts)));
+                min_ts = cmp::min(*ts, min_ts);
             }
             result_futures.push(client.query_raw(&prev_rev_chunk, params));
         }
@@ -944,10 +946,10 @@ impl PersistenceReader for PostgresReader {
                 internal_doc_id_param(*id),
                 Param::Ts(i64::from(*ts)),
             ];
+            min_ts = cmp::min(*ts, min_ts);
             result_futures.push(client.query_raw(&prev_rev, params));
         }
         let mut result_stream = stream::iter(result_futures).buffered(*PIPELINE_QUERIES);
-        let mut min_ts = Timestamp::MAX;
         while let Some(row_stream) = result_stream.try_next().await? {
             futures::pin_mut!(row_stream);
             while let Some(row) = row_stream.try_next().await? {

crates/sqlite/src/lib.rs

@@ -493,6 +493,7 @@ impl PersistenceReader for SqlitePersistence {
         {
             let inner = self.inner.lock();
             for (id, ts) in ids {
+                min_ts = cmp::min(ts, min_ts);
                 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)];
@@ -515,7 +516,6 @@ impl PersistenceReader for SqlitePersistence {
                         None
                     };
                     let prev_prev_ts = prev_prev_ts.map(Timestamp::try_from).transpose()?;
-                    min_ts = cmp::min(ts, min_ts);
                     out.insert(
                         (document_id, ts),
                         DocumentLogEntry {