sqld: Add extended_code support

sqld-libsql-bindings/src/lib.rs

@@ -111,7 +111,7 @@ impl<W: WalHook> Connection<W> {
     ///
     /// # Safety
     /// The caller is responsible for the returned pointer.
-    pub unsafe fn handle(&mut self) -> *mut sqlite3 {
+    pub unsafe fn handle(&self) -> *mut sqlite3 {
         self.conn.handle()
     }
 }

sqld/Cargo.toml

@@ -69,7 +69,7 @@ hyper-rustls = { git = "https://github.com/rustls/hyper-rustls.git", rev = "163b
 rustls-pemfile = "1.0.3"
 rustls = "0.21.7"
 async-stream = "0.3.5"
-libsql = { git = "https://github.com/tursodatabase/libsql.git", rev = "bea8863", optional = true }
+libsql = { git = "https://github.com/tursodatabase/libsql", rev = "8847ca05c", optional = true }
 metrics = "0.21.1"
 metrics-exporter-prometheus = "0.12.1"
 

sqld/proto/proxy.proto

@@ -42,6 +42,7 @@ message Error {
 
     ErrorCode code = 1;
     string message = 2;
+    int32 extended_code = 3;
 }
 
 message ResultRows {

sqld/src/connection/libsql.rs

@@ -374,7 +374,7 @@ impl<W: WalHook> Connection<W> {
         current_frame_no_receiver: watch::Receiver<Option<FrameNo>>,
         state: Arc<TxnState<W>>,
     ) -> Result<Self> {
-        let mut conn = open_conn(
+        let conn = open_conn(
             path,
             wal_methods,
             hook_ctx,
@@ -519,7 +519,14 @@ impl<W: WalHook> Connection<W> {
             match self.execute_query(&step.query, builder) {
                 // builder error interupt the execution of query. we should exit immediately.
                 Err(e @ Error::BuilderError(_)) => return Err(e),
-                Err(e) => {
+                Err(mut e) => {
+                    if let Error::RusqliteError(err) = e {
+                        let extended_code =
+                            unsafe { rusqlite::ffi::sqlite3_extended_errcode(self.conn.handle()) };
+
+                        e = Error::RusqliteErrorExtended(err, extended_code as i32);
+                    };
+
                     builder.step_error(e)?;
                     enabled = false;
                     (0, None)
@@ -570,6 +577,7 @@ impl<W: WalHook> Connection<W> {
             .map_err(Error::LibSqlInvalidQueryParams)?;
 
         let mut qresult = stmt.raw_query();
+
         builder.begin_rows()?;
         while let Some(row) = qresult.next()? {
             builder.begin_row()?;

sqld/src/error.rs

@@ -20,6 +20,8 @@ pub enum Error {
     IOError(#[from] std::io::Error),
     #[error(transparent)]
     RusqliteError(#[from] rusqlite::Error),
+    #[error("{0}")]
+    RusqliteErrorExtended(rusqlite::Error, i32),
     #[error("Failed to execute query via RPC. Error code: {}, message: {}", .0.code, .0.message)]
     RpcQueryError(crate::rpc::proxy::rpc::Error),
     #[error("Failed to execute queries via RPC protocol: `{0}`")]
@@ -106,6 +108,7 @@ impl IntoResponse for Error {
             LibSqlTxBusy => self.format_err(StatusCode::TOO_MANY_REQUESTS),
             IOError(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR),
             RusqliteError(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR),
+            RusqliteErrorExtended(_, _) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR),
             RpcQueryError(_) => self.format_err(StatusCode::BAD_REQUEST),
             RpcQueryExecutionError(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR),
             DbValueError(_) => self.format_err(StatusCode::BAD_REQUEST),

sqld/src/rpc/proxy.rs

@@ -37,20 +37,23 @@ pub mod rpc {
 
     impl From<SqldError> for Error {
         fn from(other: SqldError) -> Self {
-            Error {
-                message: other.to_string(),
-                code: ErrorCode::from(other).into(),
-            }
-        }
-    }
-
-    impl From<SqldError> for ErrorCode {
-        fn from(other: SqldError) -> Self {
-            match other {
+            let code = match other {
                 SqldError::LibSqlInvalidQueryParams(_) => ErrorCode::SqlError,
                 SqldError::LibSqlTxTimeout => ErrorCode::TxTimeout,
                 SqldError::LibSqlTxBusy => ErrorCode::TxBusy,
                 _ => ErrorCode::Internal,
+            };
+
+            let extended_code = if let SqldError::RusqliteErrorExtended(_, code) = &other {
+                *code
+            } else {
+                0
+            };
+
+            Error {
+                message: other.to_string(),
+                code: code as i32,
+                extended_code,
             }
         }
     }

sqld/tests/embedded_replica/mod.rs

@@ -0,0 +1,107 @@
+use std::path::PathBuf;
+
+use crate::common::http::Client;
+use crate::common::net::{init_tracing, TestServer, TurmoilAcceptor, TurmoilConnector};
+use libsql::Database;
+use serde_json::json;
+use sqld::config::{AdminApiConfig, RpcServerConfig, UserApiConfig};
+use turmoil::{Builder, Sim};
+use uuid::Uuid;
+
+fn make_primary(sim: &mut Sim, path: PathBuf) {
+    init_tracing();
+    sim.host("primary", move || {
+        let path = path.clone();
+        async move {
+            let server = TestServer {
+                path: path.into(),
+                user_api_config: UserApiConfig {
+                    http_acceptor: Some(TurmoilAcceptor::bind(([0, 0, 0, 0], 8080)).await?),
+                    ..Default::default()
+                },
+                admin_api_config: Some(AdminApiConfig {
+                    acceptor: TurmoilAcceptor::bind(([0, 0, 0, 0], 9090)).await?,
+                    connector: TurmoilConnector,
+                    disable_metrics: false,
+                }),
+                rpc_server_config: Some(RpcServerConfig {
+                    acceptor: TurmoilAcceptor::bind(([0, 0, 0, 0], 4567)).await?,
+                    tls_config: None,
+                }),
+                disable_namespaces: false,
+                disable_default_namespace: true,
+                ..Default::default()
+            };
+
+            server.start().await?;
+
+            Ok(())
+        }
+    });
+}
+
+#[test]
+fn embedded_replica() {
+    let mut sim = Builder::new().build();
+
+    let tmp_dir_name = Uuid::new_v4().simple().to_string();
+
+    let tmp = std::env::temp_dir().join(tmp_dir_name);
+
+    tracing::debug!("tmp dir: {:?}", tmp);
+
+    // We need to ensure that libsql's init code runs before we do anything
+    // with rusqlite in sqld. This is because libsql has saftey checks and
+    // needs to configure the sqlite api. Thus if we init sqld first
+    // it will fail. To work around this we open a temp db in memory db
+    // to ensure we run libsql's init code first. This DB is not actually
+    // used in the test only for its run once init code.
+    //
+    // This does change the serialization mode for sqld but because the mode
+    // that we use in libsql is safer than the sqld one it is still safe.
+    let db = Database::open_in_memory().unwrap();
+    db.connect().unwrap();
+
+    make_primary(&mut sim, tmp.to_path_buf());
+
+    sim.client("client", async move {
+        let client = Client::new();
+        client
+            .post("http://primary:9090/v1/namespaces/foo/create", json!({}))
+            .await?;
+
+        let db = Database::open_with_remote_sync_connector(
+            tmp.join("embedded").to_str().unwrap(),
+            "http://foo.primary:8080",
+            "",
+            TurmoilConnector,
+        )?;
+
+        let n = db.sync().await?;
+        assert_eq!(n, 0);
+
+        let conn = db.connect()?;
+
+        conn.execute("CREATE TABLE user (id INTEGER NOT NULL PRIMARY KEY)", ())
+            .await?;
+
+        let n = db.sync().await?;
+        assert_eq!(n, 2);
+
+        let err = conn
+            .execute("INSERT INTO user(id) VALUES (1), (1)", ())
+            .await
+            .unwrap_err();
+
+        let libsql::Error::RemoteSqliteFailure(code, extended_code, _) = err else {
+            panic!()
+        };
+
+        assert_eq!(code, 3);
+        assert_eq!(extended_code, 1555);
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}

sqld/tests/tests.rs

@@ -3,5 +3,6 @@
 
 mod cluster;
 mod common;
+mod embedded_replica;
 mod namespaces;
 mod standalone;