un-erase config type

Author: MarinPostma

libsql-wal/src/storage/async_storage.rs

@@ -1,7 +1,6 @@
 //! `AsyncStorage` is a `Storage` implementation that defer storage to a background thread. The
 //! durable frame_no is notified asynchronously.
 
-use std::any::Any;
 use std::sync::Arc;
 
 use chrono::Utc;
@@ -114,6 +113,7 @@ where
                 }
             }
         }
+
         tracing::info!("Storage shutdown");
         if let Some(notify) = notify_shutdown {
             let _ = notify.send(());
@@ -124,19 +124,12 @@ where
         &self,
         namespace: NamespaceName,
         ret: oneshot::Sender<super::Result<u64>>,
-        config_override: Option<Arc<dyn Any + Send + Sync>>,
+        config_override: Option<B::Config>,
     ) {
         let backend = self.backend.clone();
-        let config = match config_override
-            .map(|c| c.downcast::<B::Config>())
-            .transpose()
-        {
-            Ok(Some(config)) => config,
-            Ok(None) => backend.default_config(),
-            Err(_) => {
-                let _ = ret.send(Err(super::Error::InvalidConfigType));
-                return;
-            }
+        let config = match config_override {
+            Some(config) => config,
+            None => backend.default_config(),
         };
 
         tokio::spawn(async move {
@@ -167,7 +160,7 @@ enum StorageLoopMessage<S, C> {
     Shutdown(oneshot::Sender<()>),
 }
 
-pub struct AsyncStorage<B, S> {
+pub struct AsyncStorage<B: Backend, S> {
     /// send request to the main loop
     job_sender: mpsc::UnboundedSender<StorageLoopMessage<S, B::Config>>,
     force_shutdown: oneshot::Sender<()>,
@@ -195,15 +188,11 @@ where
         config_override: Option<Self::Config>,
         on_store_callback: OnStoreCallback,
     ) {
-        fn into_any<T: Sync + Send + 'static>(t: Arc<T>) -> Arc<dyn Any + Sync + Send> {
-            t
-        }
-
         let req = StoreSegmentRequest {
             namespace: namespace.clone(),
             segment,
             created_at: Utc::now(),
-            storage_config_override: config_override.map(into_any),
+            storage_config_override: config_override,
             on_store_callback,
         };
 

libsql-wal/src/storage/job.rs

@@ -9,13 +9,13 @@ use crate::segment::Segment;
 
 /// A request, with an id
 #[derive(Debug)]
-pub(crate) struct IndexedRequest<T> {
-    pub(crate) request: StoreSegmentRequest<T>,
+pub(crate) struct IndexedRequest<T, C> {
+    pub(crate) request: StoreSegmentRequest<T, C>,
     pub(crate) id: u64,
 }
 
-impl<T> Deref for IndexedRequest<T> {
-    type Target = StoreSegmentRequest<T>;
+impl<T, C> Deref for IndexedRequest<T, C> {
+    type Target = StoreSegmentRequest<T, C>;
 
     fn deref(&self) -> &Self::Target {
         &self.request
@@ -24,10 +24,10 @@ impl<T> Deref for IndexedRequest<T> {
 
 /// A storage Job to be performed
 #[derive(Debug)]
-pub(crate) struct Job<T> {
+pub(crate) struct Job<T, C> {
     /// Segment to store.
     // TODO: implement request batching (merge segment and send).
-    pub(crate) request: IndexedRequest<T>,
+    pub(crate) request: IndexedRequest<T, C>,
 }
 
 // #[repr(transparent)]
@@ -43,13 +43,15 @@ pub(crate) struct Job<T> {
 // }
 //
 impl<Seg> Job<Seg>
+impl<Seg, C> Job<Seg, C>
 where
     Seg: Segment,
+    C: Clone,
 {
     /// Perform the job and return the JobResult. This is not allowed to panic.
-    pub(crate) async fn perform<B, IO>(self, backend: B, io: IO) -> JobResult<Seg>
+    pub(crate) async fn perform<B, IO>(self, backend: B, io: IO) -> JobResult<Seg, C>
     where
-        B: Backend,
+        B: Backend<Config = C>,
         IO: Io,
     {
         let result = self.try_perform(backend, io).await;
@@ -58,7 +60,7 @@ where
 
     async fn try_perform<B, IO>(&self, backend: B, io: IO) -> Result<u64>
     where
-        B: Backend,
+        B: Backend<Config = C>,
         IO: Io,
     {
         let segment = &self.request.segment;
@@ -81,9 +83,6 @@ where
             .request
             .storage_config_override
             .clone()
-            .map(|c| c.downcast::<B::Config>())
-            .transpose()
-            .map_err(|_| super::Error::InvalidConfigType)?
             .unwrap_or_else(|| backend.default_config());
 
         backend.store(&config, meta, tmp, new_index).await?;
@@ -93,9 +92,9 @@ where
 }
 
 #[derive(Debug)]
-pub(crate) struct JobResult<S> {
+pub(crate) struct JobResult<S, C> {
     /// The job that was performed
-    pub(crate) job: Job<S>,
+    pub(crate) job: Job<S, C>,
     /// The outcome of the job: the new durable index, or an error.
     pub(crate) result: Result<u64>,
 }
@@ -457,8 +456,8 @@ mod test {
                 todo!()
             }
 
-            fn default_config(&self) -> Arc<Self::Config> {
-                Arc::new(())
+            fn default_config(&self) -> Self::Config {
+                ()
             }
 
             async fn restore(
@@ -501,7 +500,7 @@ mod test {
 
             async fn fetch_segment_data(
                 self: Arc<Self>,
-                _config: Arc<Self::Config>,
+                _config: Self::Config,
                 _namespace: NamespaceName,
                 _key: SegmentKey,
             ) -> Result<impl FileExt> {

libsql-wal/src/storage/mod.rs

@@ -1,4 +1,3 @@
-use std::any::Any;
 use std::collections::BTreeMap;
 use std::fmt;
 use std::future::Future;
@@ -11,7 +10,6 @@ use chrono::{DateTime, Utc};
 use fst::Map;
 use hashbrown::HashMap;
 use libsql_sys::name::NamespaceName;
-use parking_lot::Mutex;
 use tempfile::{tempdir, TempDir};
 
 use crate::io::{FileExt, Io, StdIO};
@@ -204,7 +202,7 @@ impl Storage for NoStorage {
         &self,
         _namespace: &NamespaceName,
         _seg: Self::Segment,
-        _config: Option<Arc<Self::Config>>,
+        _config: Option<Self::Config>,
         _on_store: OnStoreCallback,
     ) {
     }
@@ -239,7 +237,7 @@ impl Storage for NoStorage {
         &self,
         _namespace: &NamespaceName,
         _frame_no: u64,
-        _config_override: Option<Arc<Self::Config>>,
+        _config_override: Option<Self::Config>,
     ) -> Result<SegmentKey> {
         unimplemented!()
     }
@@ -248,7 +246,7 @@ impl Storage for NoStorage {
         &self,
         _namespace: &NamespaceName,
         _key: &SegmentKey,
-        _config_override: Option<Arc<Self::Config>>,
+        _config_override: Option<Self::Config>,
     ) -> Result<Map<Arc<[u8]>>> {
         unimplemented!()
     }
@@ -442,7 +440,7 @@ impl<IO: Io> Storage for TestStorage<IO> {
     }
 }
 
-pub struct StoreSegmentRequest<S> {
+pub struct StoreSegmentRequest<S, C> {
     namespace: NamespaceName,
     /// Path to the segment. Read-only for bottomless
     segment: S,
@@ -451,12 +449,12 @@ pub struct StoreSegmentRequest<S> {
 
     /// alternative configuration to use with the storage layer.
     /// e.g: S3 overrides
-    storage_config_override: Option<Arc<dyn Any + Send + Sync>>,
+    storage_config_override: Option<C>,
     /// Called after the segment was stored, with the new durable index
     on_store_callback: OnStoreCallback,
 }
 
-impl<S> fmt::Debug for StoreSegmentRequest<S>
+impl<S, C> fmt::Debug for StoreSegmentRequest<S, C>
 where
     S: fmt::Debug,
 {

libsql-wal/src/storage/scheduler.rs

@@ -5,13 +5,13 @@ use super::job::{IndexedRequest, Job, JobResult};
 use super::StoreSegmentRequest;
 use libsql_sys::name::NamespaceName;
 
-struct NamespaceRequests<F> {
-    requests: VecDeque<IndexedRequest<F>>,
+struct NamespaceRequests<F, C> {
+    requests: VecDeque<IndexedRequest<F, C>>,
     /// there's work in flight for this namespace
     in_flight: bool,
 }
 
-impl<F> Default for NamespaceRequests<F> {
+impl<F, C> Default for NamespaceRequests<F, C> {
     fn default() -> Self {
         Self {
             requests: Default::default(),
@@ -28,14 +28,14 @@ impl<F> Default for NamespaceRequests<F> {
 /// processed, because only the most recent segment is checked for durability. This property
 /// ensures that all segments are present up to the max durable index.
 /// It is generic over C: the storage config type (for config overrides), and T, the segment type
-pub(crate) struct Scheduler<T> {
+pub(crate) struct Scheduler<T, C> {
     /// notify new durability index for namespace
-    requests: HashMap<NamespaceName, NamespaceRequests<T>>,
+    requests: HashMap<NamespaceName, NamespaceRequests<T, C>>,
     queue: priority_queue::PriorityQueue<NamespaceName, Reverse<u64>>,
     next_request_id: u64,
 }
 
-impl<T> Scheduler<T> {
+impl<T, C> Scheduler<T, C> {
     pub fn new() -> Self {
         Self {
             requests: Default::default(),
@@ -46,7 +46,7 @@ impl<T> Scheduler<T> {
 
     /// Register a new request with the scheduler
     #[tracing::instrument(skip_all)]
-    pub fn register(&mut self, request: StoreSegmentRequest<T>) {
+    pub fn register(&mut self, request: StoreSegmentRequest<T, C>) {
         // invariant: new segment comes immediately after the latest segment for that namespace. This means:
         // - immediately after the last registered segment, if there is any
         // - immediately after the last durable index
@@ -71,7 +71,7 @@ impl<T> Scheduler<T> {
     /// be scheduled, and returns description of the job to be performed. No other job for this
     /// namespace will be scheduled, until the `JobResult` is reported
     #[tracing::instrument(skip_all)]
-    pub fn schedule(&mut self) -> Option<Job<T>> {
+    pub fn schedule(&mut self) -> Option<Job<T, C>> {
         let (name, _) = self.queue.pop()?;
         let requests = self
             .requests
@@ -90,7 +90,7 @@ impl<T> Scheduler<T> {
     /// Report the job result to the scheduler. If the job result was a success, the request as
     /// removed from the queue, else, the job is rescheduled
     #[tracing::instrument(skip_all, fields(req_id = result.job.request.id))]
-    pub async fn report(&mut self, result: JobResult<T>) {
+    pub async fn report(&mut self, result: JobResult<T, C>) {
         // re-schedule, or report new max durable frame_no for segment
         let name = result.job.request.request.namespace.clone();
         let requests = self
@@ -151,7 +151,7 @@ mod test {
 
     #[tokio::test]
     async fn schedule_simple() {
-        let mut scheduler = Scheduler::<()>::new();
+        let mut scheduler = Scheduler::<(), ()>::new();
 
         let ns1 = NamespaceName::from("test1");
         let ns2 = NamespaceName::from("test2");
@@ -224,7 +224,7 @@ mod test {
 
     #[tokio::test]
     async fn job_error_reschedule() {
-        let mut scheduler = Scheduler::<()>::new();
+        let mut scheduler = Scheduler::<(), ()>::new();
 
         let ns1 = NamespaceName::from("test1");
         let ns2 = NamespaceName::from("test2");
@@ -264,7 +264,7 @@ mod test {
 
     #[tokio::test]
     async fn schedule_while_in_flight() {
-        let mut scheduler = Scheduler::<()>::new();
+        let mut scheduler = Scheduler::<(), ()>::new();
 
         let ns1 = NamespaceName::from("test1");