Fix reverse iterator in RocksDB

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### Fixed
 - [2366](https://github.com/FuelLabs/fuel-core/pull/2366): The `importer_gas_price_for_block` metric is properly collected.
 - [2369](https://github.com/FuelLabs/fuel-core/pull/2369): The `transaction_insertion_time_in_thread_pool_milliseconds` metric is properly collected.
+- [2389](https://github.com/FuelLabs/fuel-core/pull/2389): Fix construction of reverse iterator in RocksDB.
 
 ### Changed
 

crates/fuel-core/src/state/rocks_db.rs

@@ -443,12 +443,9 @@ where
     }
 
     /// RocksDB prefix iteration doesn't support reverse order,
-    /// but seeking the start key and iterating in reverse order works.
-    /// So we can create a workaround. We need to find the next available
-    /// element and use it as an anchor for reverse iteration,
-    /// but skip the first element to jump on the previous prefix.
-    /// If we can't find the next element, we are at the end of the list,
-    /// so we can use `IteratorMode::End` to start reverse iteration.
+    /// so we need to to force the RocksDB iterator to order
+    /// all the prefix in the iterator so that we can take the next prefix
+    /// as start of iterator and iterate in reverse.
     fn reverse_prefix_iter<T>(
         &self,
         prefix: &[u8],
@@ -457,28 +454,19 @@ where
     where
         T: ExtractItem,
     {
-        let maybe_next_item = next_prefix(prefix.to_vec())
-            .and_then(|next_prefix| {
-                self.iter_store(
-                    column,
-                    Some(next_prefix.as_slice()),
-                    None,
-                    IterDirection::Forward,
-                )
-                .next()
-            })
-            .and_then(|res| res.ok());
-
-        if let Some((next_start_key, _)) = maybe_next_item {
-            let iter_mode = IteratorMode::From(
-                next_start_key.as_slice(),
-                rocksdb::Direction::Reverse,
-            );
+        let reverse_iterator = next_prefix(prefix.to_vec()).map(|next_prefix| {
+            let mut opts = self.read_options();
+            opts.set_total_order_seek(true);
+            self.iterator::<T>(
+                column,
+                opts,
+                IteratorMode::From(next_prefix.as_slice(), rocksdb::Direction::Reverse),
+            )
+        });
+
+        if let Some(iterator) = reverse_iterator {
             let prefix = prefix.to_vec();
-            self
-                .iterator::<T>(column, self.read_options(), iter_mode)
-                // Skip the element under the `next_start_key` key.
-                .skip(1)
+            iterator
                 .take_while(move |item| {
                     if let Ok(item) = item {
                         T::starts_with(item, prefix.as_slice())
@@ -1218,4 +1206,33 @@ mod tests {
         let _ = open_with_part_of_columns
             .expect("Should open the database with shorter number of columns");
     }
+
+    #[test]
+    fn iter_store__reverse_iterator() {
+        // Given
+        let (mut db, _tmp) = create_db();
+        let value = Arc::new(Vec::new());
+        let key_1 = [1, 1];
+        let key_2 = [2, 2];
+        let key_3 = [2, 3];
+        let key_4 = [3, 0];
+        db.put(&key_1, Column::Metadata, value.clone()).unwrap();
+        db.put(&key_2, Column::Metadata, value.clone()).unwrap();
+        db.put(&key_3, Column::Metadata, value.clone()).unwrap();
+        db.put(&key_4, Column::Metadata, value.clone()).unwrap();
+
+        // When
+        let db_iter = db
+            .iter_store(
+                Column::Metadata,
+                Some(vec![2].as_slice()),
+                None,
+                IterDirection::Reverse,
+            )
+            .map(|item| item.map(|(key, _)| key))
+            .collect::<Vec<_>>();
+
+        // Then
+        assert_eq!(db_iter, vec![Ok(key_3.to_vec()), Ok(key_2.to_vec())]);
+    }
 }

tests/tests/tx.rs

@@ -751,6 +751,56 @@ async fn get_transactions_by_owner_returns_correct_number_of_results(
     assert_eq!(transactions_forward.len(), 5);
 }
 
+#[test_case::test_case(PageDirection::Forward; "forward")]
+#[test_case::test_case(PageDirection::Backward; "backward")]
+#[tokio::test]
+async fn get_transactions_by_owners_multiple_owners_returns_correct_number_of_results(
+    direction: PageDirection,
+) {
+    let alice = Address::from([1; 32]);
+    let bob = Address::from([2; 32]);
+    let charlie = Address::from([3; 32]);
+
+    // Given
+    let mut context = TestContext::new(100).await;
+    let _ = context.transfer(bob, alice, 1).await.unwrap();
+    let _ = context.transfer(bob, alice, 2).await.unwrap();
+    let _ = context.transfer(bob, charlie, 1).await.unwrap();
+    let _ = context.transfer(alice, bob, 3).await.unwrap();
+    let _ = context.transfer(alice, bob, 4).await.unwrap();
+    let _ = context.transfer(alice, bob, 5).await.unwrap();
+    let _ = context.transfer(charlie, alice, 1).await.unwrap();
+    let _ = context.transfer(charlie, bob, 2).await.unwrap();
+    let _ = context.transfer(charlie, bob, 3).await.unwrap();
+    let _ = context.transfer(charlie, bob, 4).await.unwrap();
+    let _ = context.transfer(charlie, bob, 5).await.unwrap();
+
+    let client = context.client;
+    let all_transactions_with_direction = PaginationRequest {
+        cursor: None,
+        results: 10,
+        direction,
+    };
+
+    // When
+    let response = client
+        .transactions_by_owner(&bob, all_transactions_with_direction)
+        .await
+        .unwrap();
+
+    let transactions = response
+        .results
+        .into_iter()
+        .map(|tx| {
+            assert!(matches!(tx.status, TransactionStatus::Success { .. }));
+            tx.transaction
+        })
+        .collect_vec();
+
+    // Then
+    assert_eq!(transactions.len(), 10);
+}
+
 #[test_case::test_case(PageDirection::Forward; "forward")]
 #[test_case::test_case(PageDirection::Backward; "backward")]
 #[tokio::test]