remove durable_notifier

use callbacks instead

Author: MarinPostma

libsql-wal/src/storage/async_storage.rs

@@ -144,8 +144,6 @@ enum StorageLoopMessage<C, S> {
 pub struct AsyncStorage<B: Backend, S> {
     /// send request to the main loop
     job_sender: mpsc::UnboundedSender<StorageLoopMessage<B::Config, S>>,
-    /// receiver for the current max durable index
-    durable_notifier: mpsc::Receiver<(NamespaceName, u64)>,
     force_shutdown: oneshot::Sender<()>,
     backend: Arc<B>,
 }
@@ -273,9 +271,8 @@ impl<B: Backend, S> AsyncStorage<B, S> {
         S: Segment,
     {
         let (job_snd, job_rcv) = tokio::sync::mpsc::unbounded_channel();
-        let (durable_notifier_snd, durable_notifier_rcv) = tokio::sync::mpsc::channel(16);
         let (shutdown_snd, shutdown_rcv) = tokio::sync::oneshot::channel();
-        let scheduler = Scheduler::new(durable_notifier_snd);
+        let scheduler = Scheduler::new();
         let storage_loop = AsyncStorageLoop {
             receiver: job_rcv,
             scheduler,
@@ -287,7 +284,6 @@ impl<B: Backend, S> AsyncStorage<B, S> {
 
         let this = Self {
             job_sender: job_snd,
-            durable_notifier: durable_notifier_rcv,
             force_shutdown: shutdown_snd,
             backend: config.backend,
         };

libsql-wal/src/storage/scheduler.rs

@@ -1,8 +1,6 @@
 use std::cmp::Reverse;
 use std::collections::{HashMap, VecDeque};
 
-use tokio::sync::mpsc;
-
 use super::job::{IndexedRequest, Job, JobResult};
 use super::StoreSegmentRequest;
 use libsql_sys::name::NamespaceName;
@@ -32,16 +30,14 @@ impl<C, F> Default for NamespaceRequests<C, F> {
 /// It is generic over C: the storage config type (for config overrides), and T, the segment type
 pub(crate) struct Scheduler<C, T> {
     /// notify new durability index for namespace
-    durable_notifier: mpsc::Sender<(NamespaceName, u64)>,
     requests: HashMap<NamespaceName, NamespaceRequests<C, T>>,
     queue: priority_queue::PriorityQueue<NamespaceName, Reverse<u64>>,
     next_request_id: u64,
 }
 
 impl<C, T> Scheduler<C, T> {
-    pub fn new(durable_notifier: mpsc::Sender<(NamespaceName, u64)>) -> Self {
+    pub fn new() -> Self {
         Self {
-            durable_notifier,
             requests: Default::default(),
             queue: Default::default(),
             next_request_id: Default::default(),
@@ -108,14 +104,6 @@ impl<C, T> Scheduler<C, T> {
             Ok(durable_index) => {
                 tracing::debug!("job success registered");
                 (result.job.request.request.on_store_callback)(durable_index).await;
-                if self
-                    .durable_notifier
-                    .send((name.clone(), durable_index))
-                    .await
-                    .is_err()
-                {
-                    tracing::warn!("durability notifier was closed, proceeding anyway");
-                }
             }
             Err(e) => {
                 tracing::error!("error processing request, re-enqueuing: {e}");
@@ -154,6 +142,7 @@ mod test {
     use std::future::ready;
 
     use chrono::Utc;
+    use tokio::sync::oneshot;
 
     use crate::storage::Error;
     use libsql_sys::name::NamespaceName;
@@ -162,29 +151,34 @@ mod test {
 
     #[tokio::test]
     async fn schedule_simple() {
-        let (sender, mut receiver) = tokio::sync::mpsc::channel(10);
-        let mut scheduler = Scheduler::<(), ()>::new(sender);
+        let mut scheduler = Scheduler::<(), ()>::new();
 
         let ns1 = NamespaceName::from("test1");
         let ns2 = NamespaceName::from("test2");
 
+        let (job_1_snd, job_1_rcv) = oneshot::channel();
         scheduler.register(
             StoreSegmentRequest {
                 namespace: ns1.clone(),
                 segment: (),
                 created_at: Utc::now(),
                 storage_config_override: None,
-                on_store_callback: Box::new(|_| Box::pin(ready(()))),
+                on_store_callback: Box::new(move |n| Box::pin( async move {
+                    let _ = job_1_snd.send(n);
+                })),
             },
         );
 
+        let (job_2_snd, job_2_rcv) = oneshot::channel();
         scheduler.register(
             StoreSegmentRequest {
                 namespace: ns2.clone(),
                 segment: (),
                 created_at: Utc::now(),
                 storage_config_override: None,
-                on_store_callback: Box::new(|_| Box::pin(ready(()))),
+                on_store_callback: Box::new(move |n| Box::pin( async move {
+                    let _ = job_2_snd.send(n);
+                })),
             },
         );
 
@@ -194,7 +188,7 @@ mod test {
                 segment: (),
                 created_at: Utc::now(),
                 storage_config_override: None,
-                on_store_callback: Box::new(|_| Box::pin(ready(()))),
+                on_store_callback: Box::new(move |_| Box::pin(ready(()))),
             },
         );
 
@@ -215,15 +209,15 @@ mod test {
             .await;
 
         assert!(scheduler.schedule().is_none());
-        assert_eq!(receiver.recv().await.unwrap(), (ns2.clone(), 42));
+        assert_eq!(job_2_rcv.await.unwrap(), 42);
 
         scheduler
             .report(JobResult {
                 job: job1,
                 result: Ok(10),
             })
             .await;
-        assert_eq!(receiver.recv().await.unwrap(), (ns1.clone(), 10));
+        assert_eq!(job_1_rcv.await.unwrap(), 10);
 
         let job1 = scheduler.schedule().unwrap();
         assert_eq!(job1.request.request.namespace, ns1);
@@ -232,8 +226,7 @@ mod test {
 
     #[tokio::test]
     async fn job_error_reschedule() {
-        let (sender, _) = tokio::sync::mpsc::channel(10);
-        let mut scheduler = Scheduler::<(), ()>::new(sender);
+        let mut scheduler = Scheduler::<(), ()>::new();
 
         let ns1 = NamespaceName::from("test1");
         let ns2 = NamespaceName::from("test2");
@@ -275,8 +268,7 @@ mod test {
 
     #[tokio::test]
     async fn schedule_while_in_flight() {
-        let (sender, _) = tokio::sync::mpsc::channel(10);
-        let mut scheduler = Scheduler::<(), ()>::new(sender);
+        let mut scheduler = Scheduler::<(), ()>::new();
 
         let ns1 = NamespaceName::from("test1");