fix replicator bug

Author: MarinPostma

libsql-wal/src/replication/replicator.rs

@@ -40,20 +40,20 @@ impl<IO: Io> Replicator<IO> {
     ///
     /// In a single replication step, the replicator guarantees that a minimal set of frames is
     /// sent to the replica.
-    pub fn frame_stream(&mut self) -> impl Stream<Item = Result<Box<Frame>>> + '_ {
+    pub fn into_frame_stream(mut self) -> impl Stream<Item = Result<Box<Frame>>> + Send {
         async_stream::try_stream! {
             loop {
                 // First we decide up to what frame_no we want to replicate in this step. If we are
                 // already up to date, wait for something to happen
                 let most_recent_frame_no = *self
                     .new_frame_notifier
-                    .wait_for(|fno| *fno > self.next_frame_no)
+                    .wait_for(|fno| *fno >= self.next_frame_no)
                     .await
                     .expect("channel cannot be closed because we hold a ref to the sending end");
 
                 let mut commit_frame_no = 0;
                 // we have stuff to replicate
-                if most_recent_frame_no > self.next_frame_no {
+                if most_recent_frame_no >= self.next_frame_no {
                     // first replicate the most recent version of each page from the current
                     // segment. We also return how far we have replicated from the current log
                     let current = self.shared.current.load();