Merge branch 'main' into libsql-transactions

Author: LucioFranco

.github/release-drafter.yml

@@ -1,5 +1,7 @@
-name-template: 'v$RESOLVED_VERSION'
-tag-template: 'v$RESOLVED_VERSION'
+name-template: 'SQLD v$RESOLVED_VERSION'
+tag-template: 'libsql-server-v$RESOLVED_VERSION'
+tag-prefix: "libsql-server-v"
+include-paths: ["libsql-server"]
 categories:
   - title: '🚀 Features'
     labels:

.github/workflows/nemesis.yml

@@ -1,12 +1,8 @@
 name: Nemesis Tests
 
 on:
-  push:
-    branches: [ "main" ]
-  pull_request_target:
-    branches: [ "main" ]
-  merge_group:
-    branches: [ "main" ]
+  schedule:
+    - cron: '* */4 * * *'
 
 env:
   CARGO_TERM_COLOR: always
@@ -16,17 +12,8 @@ env:
   # RUSTFLAGS: "-D warnings"
 
 jobs:
-  authorize:
-    environment: ${{ github.event_name == 'pull_request_target' &&
-      github.event.pull_request.head.repo.full_name != github.repository &&
-      'external' || 'internal' }}
-    runs-on: ubuntu-latest
-    steps:
-      - run: true
-
   test-nemesis:
-    needs: authorize
-    runs-on: self-hosted
+    runs-on: ubuntu-latest
     name: Run Nemesis Tests
     env:
       RUSTFLAGS: -D warnings
@@ -59,14 +46,14 @@ jobs:
       - name: Cargo build
         run: |
           cargo build
-          mv ./target/debug/sqld /home/ubuntu/.cargo/bin
+          mv ./target/debug/sqld ~/.cargo/bin
           sqld --version
 
       - name: Download MinIO binary
         run: |
           wget -q https://dl.min.io/server/minio/release/linux-amd64/minio -O minio
           chmod +x minio
-          mv minio /home/ubuntu/.cargo/bin
+          mv minio ~/.cargo/bin
           minio --version
 
       - name: Nemesis tests checkout

Cargo.lock

@@ -47,11 +47,11 @@ installers = ["shell", "homebrew"]
 # A GitHub repo to push Homebrew formulas to
 tap = "tursodatabase/homebrew-sqld"
 # Target platforms to build apps for (Rust target-triple syntax)
-targets = ["x86_64-unknown-linux-gnu", "aarch64-apple-darwin", "x86_64-apple-darwin", "x86_64-unknown-linux-musl"]
+targets = ["x86_64-unknown-linux-gnu", "aarch64-apple-darwin", "x86_64-apple-darwin"]
 # Publish jobs to run in CI
 publish-jobs = ["homebrew"]
 # Whether cargo-dist should create a Github Release or use an existing draft
-create-release = false
+create-release = true
 # Publish jobs to run in CI
 pr-run-mode = "plan"
 allow-dirty = ["ci"] # we've edited the release yml to narrow the tag scope

Cargo.toml

@@ -168,6 +168,13 @@ async fn run() -> Result<()> {
     } else {
         options.bucket = std::env::var("LIBSQL_BOTTOMLESS_BUCKET").ok();
     }
+
+    if let Some(ns) = options.namespace.as_deref() {
+        if !ns.starts_with("ns-") {
+            println!("Namespace should start with 'ns-'");
+            std::process::exit(1)
+        }
+    }
     let namespace = options.namespace.as_deref().unwrap_or("ns-default");
     std::env::set_var("LIBSQL_BOTTOMLESS_DATABASE_ID", namespace);
     let database = match options.database.clone() {

bottomless-cli/src/main.rs

@@ -49,7 +49,7 @@ pub struct Replicator {
     /// Last frame which has been confirmed as stored locally outside of WAL file.
     /// Always: [last_committed_frame_no] <= [last_sent_frame_no].
     last_committed_frame_no: Receiver<Result<u32>>,
-    flush_trigger: Sender<()>,
+    flush_trigger: Option<Sender<()>>,
     snapshot_waiter: Receiver<Result<Option<Uuid>>>,
     snapshot_notifier: Arc<Sender<Result<Option<Uuid>>>>,
 
@@ -65,7 +65,7 @@ pub struct Replicator {
     use_compression: CompressionKind,
     max_frames_per_batch: usize,
     s3_upload_max_parallelism: usize,
-    _join_set: JoinSet<()>,
+    join_set: JoinSet<()>,
 }
 
 #[derive(Debug)]
@@ -262,7 +262,7 @@ impl Replicator {
         let next_frame_no = Arc::new(AtomicU32::new(1));
         let last_sent_frame_no = Arc::new(AtomicU32::new(0));
 
-        let mut _join_set = JoinSet::new();
+        let mut join_set = JoinSet::new();
 
         let (frames_outbox, mut frames_inbox) = tokio::sync::mpsc::channel(64);
         let _local_backup = {
@@ -278,7 +278,7 @@ impl Replicator {
             let next_frame_no = next_frame_no.clone();
             let last_sent_frame_no = last_sent_frame_no.clone();
             let batch_interval = options.max_batch_interval;
-            _join_set.spawn(async move {
+            join_set.spawn(async move {
                 loop {
                     let timeout = Instant::now() + batch_interval;
                     let trigger = match timeout_at(timeout, flush_trigger_rx.changed()).await {
@@ -313,7 +313,7 @@ impl Replicator {
             let client = client.clone();
             let bucket = options.bucket_name.clone();
             let max_parallelism = options.s3_upload_max_parallelism;
-            _join_set.spawn(async move {
+            join_set.spawn(async move {
                 let sem = Arc::new(tokio::sync::Semaphore::new(max_parallelism));
                 let mut join_set = JoinSet::new();
                 while let Some(fdesc) = frames_inbox.recv().await {
@@ -353,7 +353,7 @@ impl Replicator {
             generation,
             next_frame_no,
             last_sent_frame_no,
-            flush_trigger,
+            flush_trigger: Some(flush_trigger),
             last_committed_frame_no,
             verify_crc: options.verify_crc,
             db_path,
@@ -365,10 +365,22 @@ impl Replicator {
             use_compression: options.use_compression,
             max_frames_per_batch: options.max_frames_per_batch,
             s3_upload_max_parallelism: options.s3_upload_max_parallelism,
-            _join_set,
+            join_set,
         })
     }
 
+    pub async fn shutdown_gracefully(&mut self) -> Result<()> {
+        let last_frame_no = self.last_known_frame();
+        // drop flush trigger, which will cause background task for local WAL copier to complete
+        self.flush_trigger.take();
+        self.wait_until_committed(last_frame_no).await?;
+        self.wait_until_snapshotted().await?;
+        while let Some(t) = self.join_set.join_next().await {
+            t?;
+        }
+        Ok(())
+    }
+
     pub fn next_frame_no(&self) -> u32 {
         self.next_frame_no.load(Ordering::Acquire)
     }
@@ -624,8 +636,12 @@ impl Replicator {
     }
 
     pub fn request_flush(&self) {
-        tracing::trace!("Requesting flush");
-        let _ = self.flush_trigger.send(());
+        if let Some(tx) = self.flush_trigger.as_ref() {
+            tracing::trace!("Requesting flush");
+            let _ = tx.send(());
+        } else {
+            tracing::warn!("Cannot request flush - replicator is closing");
+        }
     }
 
     // Drops uncommitted frames newer than given last valid frame

bottomless/src/replicator.rs

@@ -14419,16 +14419,6 @@ SQLITE_API void libsql_wasm_engine_free(libsql_wasm_engine_t *);
 #  endif
 #endif
 
-/*
-** Enable SQLITE_USE_SEH by default on MSVC builds.  Only omit
-** SEH support if the -DSQLITE_OMIT_SEH option is given.
-*/
-#if defined(_MSC_VER) && !defined(SQLITE_OMIT_SEH)
-# define SQLITE_USE_SEH 1
-#else
-# undef SQLITE_USE_SEH
-#endif
-
 /*
 ** The SQLITE_THREADSAFE macro must be defined as 0, 1, or 2.
 ** 0 means mutexes are permanently disable and the library is never

libsql-ffi/bundled/src/sqlite3.c

@@ -64,7 +64,7 @@ impl Database for PrimaryDatabase {
         if let Some(replicator) = &self.logger.bottomless_replicator {
             let replicator = replicator.lock().unwrap().take();
             if let Some(mut replicator) = replicator {
-                replicator.wait_until_snapshotted().await?;
+                replicator.shutdown_gracefully().await?;
             }
         }
         Ok(())

libsql-server/src/database.rs

@@ -2,6 +2,7 @@ pub mod db_factory;
 mod dump;
 mod hrana_over_http_1;
 mod result_builder;
+mod trace;
 mod types;
 
 use std::path::Path;
@@ -24,17 +25,16 @@ use serde_json::Number;
 use tokio::sync::{mpsc, oneshot, Notify};
 use tokio::task::JoinSet;
 use tonic::transport::Server;
-use tower_http::trace::DefaultOnResponse;
+
 use tower_http::{compression::CompressionLayer, cors};
-use tracing::{Level, Span};
 
 use crate::auth::{Auth, Authenticated};
 use crate::connection::Connection;
 use crate::database::Database;
 use crate::error::Error;
 use crate::hrana;
 use crate::http::user::types::HttpQuery;
-use crate::metrics::{CLIENT_VERSION, LEGACY_HTTP_CALL};
+use crate::metrics::LEGACY_HTTP_CALL;
 use crate::namespace::{MakeNamespace, NamespaceStore};
 use crate::net::Accept;
 use crate::query::{self, Query};
@@ -308,22 +308,6 @@ where
                 path: self.path,
             };
 
-            fn trace_request<B>(req: &Request<B>, span: &Span) {
-                let _s = span.enter();
-
-                tracing::debug!(
-                    "got request: {} {} {:?}",
-                    req.method(),
-                    req.uri(),
-                    req.headers()
-                );
-                if let Some(v) = req.headers().get("x-libsql-client-version") {
-                    if let Ok(s) = v.to_str() {
-                        metrics::increment_counter!(CLIENT_VERSION, "version" => s.to_string());
-                    }
-                }
-            }
-
             macro_rules! handle_hrana {
                 ($endpoint:expr, $version:expr, $encoding:expr,) => {{
                     async fn handle_hrana<F: MakeNamespace>(
@@ -420,13 +404,11 @@ where
             let router = router
                 .layer(option_layer(self.idle_shutdown_kicker.clone()))
                 .layer(
-                    tower_http::trace::TraceLayer::new_for_http()
-                        .on_request(trace_request)
-                        .on_response(
-                            DefaultOnResponse::new()
-                                .level(Level::DEBUG)
-                                .latency_unit(tower_http::LatencyUnit::Micros),
-                        ),
+                    tower_http::trace::TraceLayer::new_for_grpc()
+                        .on_eos(trace::eos)
+                        .on_request(trace::request)
+                        .on_response(trace::response)
+                        .on_failure(trace::failure),
                 )
                 .layer(CompressionLayer::new())
                 .layer(

libsql-server/src/http/user/mod.rs

@@ -0,0 +1,103 @@
+use std::time::Duration;
+
+use hyper::{http, HeaderMap, Request, Response};
+use tonic::Status;
+use tower_http::{
+    classify::{
+        ClassifiedResponse, ClassifyResponse, GrpcCode, GrpcErrorsAsFailures, GrpcFailureClass,
+    },
+    trace::{DefaultOnResponse, OnResponse},
+};
+use tracing::{Level, Span};
+
+use crate::metrics::CLIENT_VERSION;
+
+pub(crate) fn request<B>(req: &Request<B>, span: &Span) {
+    let _s = span.enter();
+
+    tracing::debug!(
+        "got request: {} {} {:?}",
+        req.method(),
+        req.uri(),
+        req.headers()
+    );
+    if let Some(v) = req.headers().get("x-libsql-client-version") {
+        if let Ok(s) = v.to_str() {
+            metrics::increment_counter!(CLIENT_VERSION, "version" => s.to_string());
+        }
+    }
+}
+
+pub(crate) fn response<B>(res: &Response<B>, latency: Duration, span: &Span) {
+    let on_response = DefaultOnResponse::new()
+        .level(Level::DEBUG)
+        .latency_unit(tower_http::LatencyUnit::Micros);
+
+    let _s = span.enter();
+
+    on_response.on_response(res, latency, span);
+
+    let is_grpc = res
+        .headers()
+        .get(http::header::CONTENT_TYPE)
+        .map_or(false, |value| {
+            value.as_bytes().starts_with("application/grpc".as_bytes())
+        });
+
+    if !is_grpc {
+        let status = res.status();
+        metrics::increment_counter!(
+            "libsql_server_user_http_response",
+            "protocol" => "http",
+            "status" => status.as_str().to_string()
+        );
+
+        if status.is_server_error() {
+            metrics::increment_counter!("libsql_server_user_http_fault", "protcol" => "http");
+        }
+    } else {
+        let grpc = GrpcErrorsAsFailures::new().with_success(GrpcCode::FailedPrecondition);
+
+        let code = match dbg!(grpc.classify_response(res)) {
+            ClassifiedResponse::Ready(Ok(())) => "0".to_string(),
+            ClassifiedResponse::Ready(Err(GrpcFailureClass::Code(code))) => {
+                metrics::increment_counter!("libsql_server_user_http_fault", "protcol" => "grpc");
+                code.to_string()
+            }
+            ClassifiedResponse::Ready(Err(GrpcFailureClass::Error(_))) => return,
+            // TODO(lucio): We need to fix this as this is not correct, right now we
+            // assume that if the grpc-status is not in the init header frame then it will
+            // be sent in the trailers, in our use case most of our gRPC calls will Error
+            // initially on response rather than in the end of the stream.
+            //
+            // The problem here is for some reason on_eos is not getting called allowing us
+            // to inspect the trailer headers. So for now we will short cut and treat any
+            // trailing grpc-status code to be treated as a success.
+            ClassifiedResponse::RequiresEos(_) => "0".to_string(),
+        };
+
+        metrics::increment_counter!(
+            "libsql_server_user_http_response",
+            "protocol" => "grpc",
+            "status" => code,
+        );
+    }
+}
+
+pub(crate) fn eos(trailers: Option<&HeaderMap>, _duration: Duration, _span: &Span) {
+    if let Some(t) = trailers {
+        if let Some(s) = Status::from_header_map(t) {
+            let code = s.code().to_string();
+
+            metrics::increment_counter!(
+                "libsql_server_user_http_response",
+                "protocol" => "grpc",
+                "status" => code,
+            );
+        }
+    }
+}
+
+pub(crate) fn failure<F>(_: F, _: Duration, _: &Span) {
+    metrics::increment_counter!("libsql_server_user_http_fault", "protcol" => "http");
+}