ensure Interval advances (#23440)

return errors if we are fetching a page from an index range and the remaining Interval to query does not shrink. Do this at the lowest levels where we compute the remaining_intervals -- once where we merge it with pending writes, and once where we truncate results based on length and bytes size. and also do this at the highest level where we absolutely require the interval to be advancing -- in the Query loop.

GitOrigin-RevId: 8e74577b5a53314c066a3ebaae5f2e9e54a49bdc

Author: ldanilek

crates/database/src/query/index_range.rs

@@ -213,6 +213,7 @@ impl<T: QueryType> IndexRange<T> {
                 self.version.clone(),
             )
             .await?;
+            anyhow::ensure!(self.unfetched_interval != new_unfetched_interval);
             self.unfetched_interval = new_unfetched_interval;
             self.page_count += 1;
             self.rows_read += page.len();

crates/database/src/transaction_index.rs

@@ -237,6 +237,11 @@ impl TransactionIndex {
                         (None, None) => break,
                     }
                 }
+                if remaining_interval == range_request.interval {
+                    Err(anyhow::anyhow!(
+                        "query for {remaining_interval:?} did not shrink"
+                    ))?;
+                }
                 (range_results, remaining_interval)
             };
             assert!(results.insert(batch_key, item_result).is_none());
@@ -382,6 +387,11 @@ impl TransactionIndex {
                     indexed_fields.clone(),
                     interval_read,
                 )?;
+                if interval_unread == interval {
+                    Err(anyhow::anyhow!(
+                        "query for {interval_unread:?} did not shrink"
+                    ))?;
+                }
                 (out, interval_unread)
             };
             assert!(results.insert(batch_key, result).is_none());