libsql wal s3 backend (#1554)

* introduce checkpointer

* fmt

* impl IoBuf(Mut) for BytesMut

* add tokio_util dep

* implement FileStreamBody

wrapper stream arounf File, to send as payload to s3

* introduce SegmentKey

* add copy stream to FileExt compat method

* add warning comment

* fix imports

* add read_at and read_at_async methods

* update Backend trait to return segmetn index

* introduce S3Backend

* add s3 keys formatters

* implement Backend for S3Backend

* fix imports

* remove old implementation

* add basic s3 test

* remove unused implementation of FileExt for Vec<u8>

* add net abstraction

* feature gate s3 backend

* s3 backend with custom connector

* fmt

* expose WalRegistry::get_async

* expose seal_current method

* implement get_durable_frame_no

* expose convenient constructor for AsyncStorage

* implement Clone for LibsqlWalManager

* pass config to Storage methods

* add unhandled storage variant

* cargo bucket already exist error

* return optional segment

* remove net abstraction

the connector can be passed to the SdkConfig

* add basic shell to interact with libsql-wal

* fmt

* do not publish libsql wal

Author: MarinPostma

Cargo.lock

@@ -43,6 +43,8 @@ rusqlite = { package = "libsql-rusqlite", path = "vendored/rusqlite", version =
     "limits",
     "hooks",
 ] }
+hyper = { version = "0.14" }
+tower = { version = "0.4.13" }
 
 # Config for 'cargo dist'
 [workspace.metadata.dist]

Cargo.toml

@@ -30,7 +30,7 @@ fallible-iterator = "0.3.0"
 futures = "0.3.25"
 futures-core = "0.3"
 hmac = "0.12"
-hyper = { version = "0.14.23", features = ["http2"] }
+hyper = { workspace = true, features = ["http2"] }
 hyper-rustls = { git = "https://github.com/rustls/hyper-rustls.git", rev = "163b3f5" }
 hyper-tungstenite = "0.11"
 itertools = "0.10.5"
@@ -74,7 +74,7 @@ tokio-tungstenite = "0.20"
 tokio-util = { version = "0.7.8", features = ["io", "io-util"] }
 tonic = { version = "0.11", features = ["tls"] }
 tonic-web = "0.11"
-tower = { version = "0.4.13", features = ["make"] }
+tower = { workspace = true, features = ["make"] }
 tower-http = { version = "0.3.5", features = ["compression-full", "cors", "trace"] }
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
@@ -97,7 +97,7 @@ arbitrary = { version = "1.3.0", features = ["derive_arbitrary"] }
 aws-config = "0.55"
 aws-sdk-s3 = "0.28"
 env_logger = "0.10"
-hyper = { version = "0.14", features = ["client"] }
+hyper = { workspace = true, features = ["client"] }
 insta = { version = "1.26.0", features = ["json"] }
 libsql = { path = "../libsql/"}
 libsql-client = { version = "0.6.5", default-features = false, features = ["reqwest_backend"] }

libsql-server/Cargo.toml

@@ -181,6 +181,8 @@ pub(super) fn get_next_pending_migration_tasks_batch(
     status: MigrationTaskStatus,
     limit: usize,
 ) -> Result<Vec<MigrationTask>, Error> {
+    dbg!(job_id);
+    dbg!(status, status as u64);
     let txn = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
     let tasks = txn
         .prepare(
@@ -190,6 +192,7 @@ pub(super) fn get_next_pending_migration_tasks_batch(
             LIMIT ?",
         )?
         .query_map((job_id, status as u64, limit), |row| {
+            dbg!(row);
             let task_id = row.get::<_, i64>(0)?;
             let namespace = NamespaceName::from_string(row.get::<_, String>(1)?).unwrap();
             let status = MigrationTaskStatus::from_int(row.get::<_, u64>(2)?);
@@ -205,6 +208,7 @@ pub(super) fn get_next_pending_migration_tasks_batch(
         .collect::<Result<Vec<_>, Error>>()?;
 
     for task in tasks.iter() {
+        dbg!(task);
         txn.execute("INSERT INTO enqueued_tasks VALUES (?)", [task.task_id])?;
     }
 

libsql-server/src/schema/db.rs

@@ -299,6 +299,7 @@ impl Scheduler {
         let Some(ref job) = self.current_job else {
             return Ok(());
         };
+        dbg!(job);
         if self.current_batch.is_empty()
             && matches!(
                 *job.status(),
@@ -310,7 +311,7 @@ impl Scheduler {
             const MAX_BATCH_SIZE: usize = 50;
             // get a batch of enqueued tasks
             let job_id = job.job_id();
-            self.current_batch = match *job.status() {
+            self.current_batch = match dbg!(*job.status()) {
                 MigrationJobStatus::WaitingDryRun => {
                     with_conn_async(self.migration_db.clone(), move |conn| {
                         get_next_pending_migration_tasks_batch(

libsql-server/src/schema/scheduler.rs

@@ -2,6 +2,7 @@
 name = "libsql-wal"
 version = "0.1.0"
 edition = "2021"
+publish = false
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -30,8 +31,16 @@ uuid = { version = "1.8.0", features = ["v4"] }
 walkdir = "2.5.0"
 zerocopy = { version = "0.7.32", features = ["derive", "alloc"] }
 
-aws-config = { version = "1", features = ["behavior-version-latest"] }
-aws-sdk-s3 = "1"
+aws-config = { version = "1", optional = true, features = ["behavior-version-latest"] }
+aws-sdk-s3 = { version = "1", optional = true }
+http-body = "1.0.0"
+tokio-util = "0.7.11"
+hyper = { workspace = true, optional = true, features = ["client", "http2"] }
+aws-smithy-runtime = { version = "1.6.2", optional = true, features = ["connector-hyper-0-14-x", "client"]}
+clap = { version = "4.5.9", optional = true, features = ["derive"] }
+inquire = { version = "0.7.5", optional = true }
+tracing-subscriber = { version = "0.3.18", optional = true }
+aws-credential-types = { version = "1.2.0", optional = true }
 
 [dev-dependencies]
 criterion = "0.5.1"
@@ -51,3 +60,24 @@ tokio = { version = "*", features = ["test-util"] }
 [[bench]]
 name = "benchmarks"
 harness = false
+
+[features]
+default = ["s3", "shell-bin"]
+s3 = [
+    "dep:hyper",
+    "dep:aws-smithy-runtime",
+    "dep:aws-sdk-s3",
+    "dep:aws-config",
+    "dep:aws-credential-types",
+]
+shell-bin = [
+    "dep:clap",
+    "dep:inquire",
+    "s3",
+    "dep:tracing-subscriber",
+]
+
+[[bin]]
+name = "shell"
+path = "src/bins/shell/main.rs"
+required-features = ["shell-bin"]