fix: yield back to executor for abortion

Author: levydsa

libsql/src/database/builder.rs

@@ -645,10 +645,14 @@ cfg_sync! {
             let mut bg_abort: Option<std::sync::Arc<crate::sync::DropAbort>> = None;
 
             if let Some(sync_interval) = sync_interval {
+                let sync_span = tracing::debug_span!("sync_interval");
+                let _enter = sync_span.enter();
+
                 let sync_ctx = db.sync_ctx.as_ref().unwrap().clone();
                 {
                     let mut ctx = sync_ctx.lock().await;
                     crate::sync::bootstrap_db(&mut ctx).await?;
+                    tracing::debug!("finished bootstrap with sync interval");
                 }
 
                 // db.connect creates a local db file, so it is important that we always call
@@ -657,8 +661,15 @@ cfg_sync! {
                 let conn = db.connect()?;
                 let jh = tokio::spawn(
                     async move {
+                        let mut interval = tokio::time::interval(sync_interval);
+
                         loop {
-                            tracing::trace!("trying to sync");
+                            tracing::info!("trying to sync");
+
+                            // NOTE(levy): Sleeping after the sync proceature will cause a delay
+                            // when aborting. I have no idea why.
+                            interval.tick().await;
+
                             let mut ctx = sync_ctx.lock().await;
                             if remote_writes {
                                 if let Err(e) = crate::sync::try_pull(&mut ctx, &conn).await {
@@ -669,10 +680,9 @@ cfg_sync! {
                                     tracing::error!("sync error: {}", e);
                                 }
                             }
-                            tokio::time::sleep(sync_interval).await;
                         }
                     }
-                    .instrument(tracing::info_span!("sync_interval")),
+                    .instrument(tracing::debug_span!("sync interval thread")),
                 );
 
                 bg_abort.replace(std::sync::Arc::new(crate::sync::DropAbort(jh.abort_handle())));

libsql/src/sync.rs

@@ -846,6 +846,8 @@ async fn try_push(
 
     let mut frame_no = start_frame_no;
     while frame_no <= end_frame_no {
+        tokio::task::yield_now().await;
+
         let batch_size = sync_ctx.push_batch_size.min(end_frame_no - frame_no + 1);
         let mut frames = conn.wal_get_frame(frame_no, page_size)?;
         if batch_size > 1 {
@@ -893,6 +895,8 @@ pub async fn try_pull(
     let mut err = None;
 
     loop {
+        tokio::task::yield_now().await;
+
         let generation = sync_ctx.durable_generation();
         let frame_no = sync_ctx.durable_frame_num() + 1;
         match sync_ctx.pull_one_frame(generation, frame_no).await {