Add retries to TableIterator on out-of-retention errors (#42503)

GitOrigin-RevId: 439a9f008ea71d8e2978afdef3eb06221ac6268e

Author: goffrie

crates/common/src/knobs.rs

@@ -183,6 +183,10 @@ pub static DEFAULT_DOCUMENTS_PAGE_SIZE: LazyLock<u32> =
 pub static DOCUMENTS_IN_MEMORY: LazyLock<usize> =
     LazyLock::new(|| env_config("DOCUMENTS_IN_MEMORY", 512));
 
+/// Number of times to retry on out-of-retention errors in TableIterator
+pub static TABLE_ITERATOR_MAX_RETRIES: LazyLock<u32> =
+    LazyLock::new(|| env_config("TABLE_ITERATOR_MAX_RETRIES", 1));
+
 /// Length of the HTTP server TCP backlog.
 pub static HTTP_SERVER_TCP_BACKLOG: LazyLock<u32> =
     LazyLock::new(|| env_config("HTTP_SERVER_TCP_BACKLOG", 256));

crates/database/src/table_iteration.rs

@@ -17,7 +17,10 @@ use common::{
         IndexKeyBytes,
     },
     interval::Interval,
-    knobs::DOCUMENTS_IN_MEMORY,
+    knobs::{
+        DOCUMENTS_IN_MEMORY,
+        TABLE_ITERATOR_MAX_RETRIES,
+    },
     persistence::{
         new_static_repeatable_recent,
         DocumentLogEntry,
@@ -41,6 +44,7 @@ use common::{
     },
     value::ResolvedDocumentId,
 };
+use errors::ErrorMetadataAnyhowExt;
 use futures::{
     pin_mut,
     stream,
@@ -563,27 +567,37 @@ impl<RT: Runtime> TableIteratorInner<RT> {
         tablet_id: TabletId,
         cursor: &mut TableScanCursor,
     ) -> anyhow::Result<(Vec<(IndexKeyBytes, LatestDocument)>, RepeatableTimestamp)> {
-        let ts = self.new_ts().await?;
-        let repeatable_persistence = RepeatablePersistence::new(
-            self.persistence.clone(),
-            ts,
-            self.retention_validator.clone(),
-        );
-        let reader = repeatable_persistence.read_snapshot(ts)?;
-        let stream = reader.index_scan(
-            index_id,
-            tablet_id,
-            &cursor.interval(),
-            Order::Asc,
-            self.page_size,
-        );
-        let documents_in_page: Vec<_> = stream.take(self.page_size).try_collect().await?;
-        if documents_in_page.len() < self.page_size {
-            cursor.advance(CursorPosition::End)?;
-        } else if let Some((index_key, ..)) = documents_in_page.last() {
-            cursor.advance(CursorPosition::After(index_key.clone()))?;
+        for attempt in 0.. {
+            let ts = self.new_ts().await?;
+            let repeatable_persistence = RepeatablePersistence::new(
+                self.persistence.clone(),
+                ts,
+                self.retention_validator.clone(),
+            );
+            let reader = repeatable_persistence.read_snapshot(ts)?;
+            let stream = reader.index_scan(
+                index_id,
+                tablet_id,
+                &cursor.interval(),
+                Order::Asc,
+                self.page_size,
+            );
+            let documents_in_page: Vec<_> = match stream.take(self.page_size).try_collect().await {
+                Ok(docs) => docs,
+                Err(e) if attempt < *TABLE_ITERATOR_MAX_RETRIES && e.is_out_of_retention() => {
+                    tracing::warn!("TableIterator hit out-of-retention error {e}, retrying...");
+                    continue;
+                },
+                Err(e) => return Err(e),
+            };
+            if documents_in_page.len() < self.page_size {
+                cursor.advance(CursorPosition::End)?;
+            } else if let Some((index_key, ..)) = documents_in_page.last() {
+                cursor.advance(CursorPosition::After(index_key.clone()))?;
+            }
+            return Ok((documents_in_page, ts));
         }
-        Ok((documents_in_page, ts))
+        unreachable!()
     }
 
     /// Load the revisions of documents visible at `self.snapshot_ts`.