add more logging

Author: MarinPostma

libsql-wal/src/replication/replicator.rs

@@ -40,17 +40,21 @@ impl<IO: Io> Replicator<IO> {
     ///
     /// In a single replication step, the replicator guarantees that a minimal set of frames is
     /// sent to the replica.
+    #[tracing::instrument(skip(self))]
     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
+                tracing::debug!(next_frame_no = self.next_frame_no);
                 let most_recent_frame_no = *self
                     .new_frame_notifier
                     .wait_for(|fno| *fno >= self.next_frame_no)
                     .await
                     .expect("channel cannot be closed because we hold a ref to the sending end");
 
+                tracing::debug!(most_recent_frame_no, "new frame_no available");
+
                 let mut commit_frame_no = 0;
                 // we have stuff to replicate
                 if most_recent_frame_no >= self.next_frame_no {
@@ -66,6 +70,7 @@ impl<IO: Io> Replicator<IO> {
 
                         let mut stream = stream.peekable();
 
+                        tracing::debug!(replicated_until, "replicating from current log");
                         loop {
                             let Some(frame) = stream.next().await else { break };
                             let mut frame = frame.map_err(|e| Error::CurrentSegment(e.into()))?;
@@ -88,6 +93,7 @@ impl<IO: Io> Replicator<IO> {
                                 .stream_pages_from(replicated_until, self.next_frame_no, &mut seen).await;
                             tokio::pin!(stream);
 
+                        tracing::debug!(replicated_until, "replicating from tail");
                             let mut stream = stream.peekable();
 
                             let should_replicate_from_storage = replicated_until != self.next_frame_no;
@@ -110,6 +116,7 @@ impl<IO: Io> Replicator<IO> {
                         // Replicating from sealed segments was not enough, so we replicate from
                         // durable storage
                         if let Some(replicated_until) = replicated_until {
+                            tracing::debug!("replicating from durable storage");
                             let stream = self
                                 .shared
                                 .stored_segments

libsql-wal/src/segment/list.rs

@@ -78,6 +78,7 @@ where
 
     /// Checkpoints as many segments as possible to the main db file, and return the checkpointed
     /// frame_no, if anything was checkpointed
+    #[tracing::instrument(skip_all)]
     pub async fn checkpoint<IO: Io>(
         &self,
         db_file: &IO::File,
@@ -108,6 +109,7 @@ where
         // readers pointing to them
         while let Some(segment) = &*current {
             // skip any segment more recent than until_frame_no
+            tracing::debug!(last_committed = segment.last_committed(), until = until_frame_no);
             if segment.last_committed() <= until_frame_no {
                 if !segment.is_checkpointable() {
                     segs.clear();
@@ -120,6 +122,7 @@ where
 
         // nothing to checkpoint rn
         if segs.is_empty() {
+            tracing::debug!("nothing to checkpoint");
             return Ok(None);
         }
 
@@ -133,6 +136,7 @@ where
         let mut last_replication_index = 0;
         while let Some((k, v)) = union.next() {
             let page_no = u32::from_be_bytes(k.try_into().unwrap());
+            tracing::debug!(page_no);
             let v = v.iter().min_by_key(|i| i.index).unwrap();
             let offset = v.value as u32;
 

libsql-wal/src/segment/sealed.rs

@@ -181,7 +181,9 @@ where
     }
 
     fn is_checkpointable(&self) -> bool {
-        self.read_locks.load(Ordering::Relaxed) == 0
+        let read_locks = self.read_locks.load(Ordering::Relaxed);
+        tracing::debug!(read_locks);
+        read_locks == 0
     }
 
     fn size_after(&self) -> u32 {

libsql-wal/src/shared_wal.rs

@@ -305,6 +305,7 @@ impl<IO: Io> SharedWal<IO> {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub async fn checkpoint(&self) -> Result<Option<u64>> {
         let durable_frame_no = *self.durable_frame_no.lock();
         let checkpointed_frame_no = self

libsql-wal/src/transaction.rs

@@ -318,7 +318,7 @@ impl<F> WriteTransaction<F> {
             }
         }
 
-        tracing::debug!(id = read_tx.id, "lock released");
+        tracing::trace!(id = read_tx.id, "lock released");
 
         read_tx
     }

libsql-wal/src/wal.rs

@@ -131,7 +131,7 @@ impl<FS: Io> Wal for LibsqlWal<FS> {
         self.last_read_frame_no = Some(tx.max_frame_no);
         self.tx = Some(Transaction::Read(tx));
 
-        tracing::debug!(invalidate_cache, "read started");
+        tracing::trace!(invalidate_cache, "read started");
         Ok(invalidate_cache)
     }
 
@@ -182,9 +182,9 @@ impl<FS: Io> Wal for LibsqlWal<FS> {
         match self.tx.as_mut() {
             Some(tx) => {
                 self.shared.upgrade(tx).map_err(Into::into)?;
-                tracing::debug!("write lock acquired");
+                tracing::trace!("write lock acquired");
             }
-            None => todo!("should acquire read txn first"),
+            None => panic!("should acquire read txn first"),
         }
 
         Ok(())