raw: fix reverse_scan failing across multiple regions

When reverse scanning a range spanning multiple regions, `retryable_scan`
selected the region based on `start_key` (lower bound). However, TiKV
reverse scan starts from the upper bound, causing `key_not_in_region`
errors when bounds are in different regions.

Fix by selecting the region containing `end_key` for reverse scans and
adjusting continuation logic to move backward through regions. Also
handle the edge case when iteration lands exactly on a region boundary.

Fixes #512

Signed-off-by: Hugh <HueCodes@users.noreply.github.com>

Author: HueCodes

src/raw/client.rs

@@ -759,20 +759,42 @@ impl<PdC: PdClient> Client<PdC> {
             });
         }
         let backoff = DEFAULT_STORE_BACKOFF;
-        let mut range = range.into().encode_keyspace(self.keyspace, KeyMode::Raw);
+        let range = range.into().encode_keyspace(self.keyspace, KeyMode::Raw);
         let mut result = Vec::new();
         let mut current_limit = limit;
         let (start_key, end_key) = range.clone().into_keys();
-        let mut current_key: Key = start_key;
+
+        // For forward scan: current_key tracks the lower bound, moving forward
+        // For reverse scan: current_key tracks the upper bound, moving backward
+        let mut current_key: Key = if reverse {
+            // Start from the upper bound for reverse scan
+            end_key.clone().unwrap_or_default()
+        } else {
+            start_key.clone()
+        };
 
         while current_limit > 0 {
-            let scan_args = ScanInnerArgs {
-                start_key: current_key.clone(),
-                end_key: end_key.clone(),
-                limit: current_limit,
-                key_only,
-                reverse,
-                backoff: backoff.clone(),
+            // Build scan arguments based on direction:
+            // - Forward: scan from current_key (moving lower bound) to end_key (fixed upper bound)
+            // - Reverse: scan from start_key (fixed lower bound) to current_key (moving upper bound)
+            let scan_args = if reverse {
+                ScanInnerArgs {
+                    start_key: start_key.clone(),
+                    end_key: Some(current_key.clone()),
+                    limit: current_limit,
+                    key_only,
+                    reverse,
+                    backoff: backoff.clone(),
+                }
+            } else {
+                ScanInnerArgs {
+                    start_key: current_key.clone(),
+                    end_key: end_key.clone(),
+                    limit: current_limit,
+                    key_only,
+                    reverse,
+                    backoff: backoff.clone(),
+                }
             };
             let (res, next_key) = self.retryable_scan(scan_args).await?;
 
@@ -784,11 +806,29 @@ impl<PdC: PdClient> Client<PdC> {
                 current_limit -= kvs.len() as u32;
                 result.append(&mut kvs);
             }
-            if end_key.clone().is_some_and(|ek| ek <= next_key) {
-                break;
+
+            // Determine if we should continue to the next region
+            if reverse {
+                // For reverse scan: next_key is the region's start_key (lower boundary)
+                // Stop if next_key is empty (reached the beginning) or
+                // if we've reached/passed the lower bound of our scan range
+                if next_key.is_empty() || next_key <= start_key {
+                    break;
+                }
+                // Safety: if next_key >= current_key, we've made no progress
+                // (shouldn't happen with boundary fix, but prevents infinite loop)
+                if next_key >= current_key {
+                    break;
+                }
+                current_key = next_key;
             } else {
+                // For forward scan: next_key is the region's end_key (upper boundary)
+                // Stop if next_key is empty (reached the end) or
+                // if we've reached/passed the upper bound of our scan range
+                if next_key.is_empty() || end_key.clone().is_some_and(|ek| ek <= next_key) {
+                    break;
+                }
                 current_key = next_key;
-                range = BoundRange::new(std::ops::Bound::Included(current_key.clone()), range.to);
             }
         }
 
@@ -807,8 +847,42 @@ impl<PdC: PdClient> Client<PdC> {
     ) -> Result<(Option<RawScanResponse>, Key)> {
         let start_key = scan_args.start_key;
         let end_key = scan_args.end_key;
+        let reverse = scan_args.reverse;
         loop {
-            let region = self.rpc.clone().region_for_key(&start_key).await?;
+            // For forward scan: select region containing start_key (lower bound)
+            // For reverse scan: select region containing end_key (upper bound)
+            // because TiKV reverse scan starts from the upper bound and goes backward.
+            //
+            // When reverse=true, new_raw_scan_request swaps the keys so that TiKV receives
+            // end_key as its start_key. We must select the region containing that key.
+            let region_lookup_key: &Key = if reverse {
+                match &end_key {
+                    Some(ek) if !ek.is_empty() => ek,
+                    // If no upper bound, fall back to start_key (though this case
+                    // is documented as unsupported for reverse scan)
+                    _ => &start_key,
+                }
+            } else {
+                &start_key
+            };
+            let mut region = self.rpc.clone().region_for_key(region_lookup_key).await?;
+
+            // For reverse scan: if the lookup key equals the region's start_key exactly,
+            // we're at a boundary and need the previous region. This happens when iterating
+            // backward and current_key lands on a region boundary.
+            if reverse {
+                let region_start = region.start_key();
+                if !region_start.is_empty() && *region_lookup_key == region_start {
+                    // Find the previous region by looking up a key just before this boundary.
+                    // Truncating the last byte gives a lexicographically smaller key.
+                    let mut prev_key: Vec<u8> = region_start.into();
+                    prev_key.pop();
+                    if !prev_key.is_empty() {
+                        region = self.rpc.clone().region_for_key(&prev_key.into()).await?;
+                    }
+                    // If prev_key is empty, we're at the start of the keyspace
+                }
+            }
             let store = self.rpc.clone().store_for_id(region.id()).await?;
             let request = new_raw_scan_request(
                 (start_key.clone(), end_key.clone()).into(),
@@ -833,7 +907,14 @@ impl<PdC: PdClient> Client<PdC> {
                             return Err(RegionError(Box::new(err)));
                         }
                     }
-                    Ok((Some(r), region.end_key()))
+                    // For forward scan: next region starts at this region's end_key
+                    // For reverse scan: next region ends at this region's start_key
+                    let next_key = if reverse {
+                        region.start_key()
+                    } else {
+                        region.end_key()
+                    };
+                    Ok((Some(r), next_key))
                 }
                 Err(err) => Err(err),
             };