find_segment now uses FindSegmentReq

Author: MarinPostma

libsql-wal/src/replication/storage.rs

@@ -9,6 +9,7 @@ use zerocopy::FromZeroes;
 
 use crate::io::buf::ZeroCopyBoxIoBuf;
 use crate::segment::Frame;
+use crate::storage::backend::FindSegmentReq;
 use crate::storage::Storage;
 
 use super::Result;
@@ -43,9 +44,10 @@ where
         mut current: u64,
         until: u64,
     ) -> Pin<Box<dyn Stream<Item = Result<Box<Frame>>> + Send + 'a>> {
+        
         Box::pin(async_stream::try_stream! {
             loop {
-                let key = self.storage.find_segment(&self.namespace, current, None).await?;
+                let key = self.storage.find_segment(&self.namespace, FindSegmentReq::EndFrameNoLessThan(current), None).await?;
                 let index = self.storage.fetch_segment_index(&self.namespace, &key, None).await?;
                 let mut pages = index.into_stream();
                 let mut maybe_seg = None;

libsql-wal/src/storage/async_storage.rs

@@ -13,7 +13,7 @@ use crate::io::{FileExt, Io, StdIO};
 use crate::segment::compacted::CompactedSegment;
 use crate::segment::Segment;
 
-use super::backend::Backend;
+use super::backend::{Backend, FindSegmentReq};
 use super::scheduler::Scheduler;
 use super::{OnStoreCallback, RestoreOptions, Storage, StoreSegmentRequest};
 
@@ -228,7 +228,7 @@ where
     async fn find_segment(
         &self,
         namespace: &NamespaceName,
-        frame_no: u64,
+        req: FindSegmentReq,
         config_override: Option<Self::Config>,
     ) -> super::Result<super::SegmentKey> {
         let config = config_override.unwrap_or_else(|| self.backend.default_config());
@@ -237,7 +237,7 @@ where
             .find_segment(
                 &config,
                 namespace,
-                super::backend::FindSegmentReq::Frame(frame_no),
+                req,
             )
             .await?;
         Ok(key)

libsql-wal/src/storage/mod.rs

@@ -207,7 +207,7 @@ pub trait Storage: Send + Sync + 'static {
     fn find_segment(
         &self,
         namespace: &NamespaceName,
-        frame_no: u64,
+        frame_no: FindSegmentReq,
         config_override: Option<Self::Config>,
     ) -> impl Future<Output = Result<SegmentKey>> + Send;
 
@@ -306,7 +306,7 @@ where
     fn find_segment(
         &self,
         namespace: &NamespaceName,
-        frame_no: u64,
+        frame_no: FindSegmentReq,
         config_override: Option<Self::Config>,
     ) -> impl Future<Output = Result<SegmentKey>> + Send {
         async move {
@@ -415,7 +415,7 @@ impl Storage for NoStorage {
     async fn find_segment(
         &self,
         _namespace: &NamespaceName,
-        _frame_no: u64,
+        _frame_no: FindSegmentReq,
         _config_override: Option<Self::Config>,
     ) -> Result<SegmentKey> {
         unimplemented!()
@@ -564,7 +564,7 @@ impl<IO: Io> Storage for TestStorage<IO> {
     async fn find_segment(
         &self,
         namespace: &NamespaceName,
-        frame_no: u64,
+        req: FindSegmentReq,
         _config_override: Option<Self::Config>,
     ) -> Result<SegmentKey> {
         let inner = self.inner.lock().await;