Merge pull request tursodatabase#1615 from avinassh/ss-better-error

Improve error handling in Storage Server RPC calls

Author: avinassh

Cargo.lock

@@ -27,6 +27,7 @@ tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 async-trait = "0.1.80"
 serde = "1.0.203"
 thiserror = "1.0.61"
+prost = "0.12.6"
 
 [features]
 foundation-db = ["foundationdb"]

libsql-storage-server/Cargo.toml

@@ -1,4 +1,6 @@
-use tonic::Status;
+use libsql_storage::rpc;
+use prost::Message;
+use tonic::{Code, Status};
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
@@ -8,11 +10,34 @@ pub enum Error {
     Other(#[from] anyhow::Error),
 }
 
+impl Error {
+    fn code(&self) -> rpc::ErrorCode {
+        match self {
+            Error::WriteConflict => rpc::ErrorCode::WriteConflict,
+            Error::Other(_) => rpc::ErrorCode::InternalError,
+        }
+    }
+}
+
 impl From<Error> for Status {
     fn from(error: Error) -> Self {
-        match error {
-            Error::WriteConflict => Status::aborted("write conflict"),
-            Error::Other(err) => Status::internal(err.to_string()),
+        let status_code = match error.code() {
+            rpc::ErrorCode::InternalError => Code::Internal,
+            rpc::ErrorCode::WriteConflict => Code::Aborted,
+        };
+        let details = rpc::ErrorDetails {
+            message: error.to_string(),
+            code: error.code() as i32,
+        };
+
+        let mut details_buf = Vec::new();
+        if let Err(e) = details.encode(&mut details_buf) {
+            Status::new(
+                Code::Internal,
+                format!("failed to encode error details: {}", e),
+            )
+        } else {
+            Status::with_details(status_code, error.to_string(), details_buf.into())
         }
     }
 }

libsql-storage-server/src/errors.rs

@@ -67,6 +67,16 @@ message DestroyRequest {
 
 message DestroyResponse {}
 
+enum ErrorCode {
+  INTERNAL_ERROR = 0;
+  WRITE_CONFLICT = 1;
+}
+
+message ErrorDetails {
+  ErrorCode code = 1;
+  string message = 2;
+}
+
 service Storage {
   rpc InsertFrames(InsertFramesRequest) returns (InsertFramesResponse) {}
   rpc FindFrame(FindFrameRequest) returns (FindFrameResponse) {}

libsql-storage/proto/storage.proto

@@ -100,6 +100,40 @@ pub struct DestroyRequest {
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct DestroyResponse {}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ErrorDetails {
+    #[prost(enumeration = "ErrorCode", tag = "1")]
+    pub code: i32,
+    #[prost(string, tag = "2")]
+    pub message: ::prost::alloc::string::String,
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum ErrorCode {
+    InternalError = 0,
+    WriteConflict = 1,
+}
+impl ErrorCode {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            ErrorCode::InternalError => "INTERNAL_ERROR",
+            ErrorCode::WriteConflict => "WRITE_CONFLICT",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "INTERNAL_ERROR" => Some(Self::InternalError),
+            "WRITE_CONFLICT" => Some(Self::WriteConflict),
+            _ => None,
+        }
+    }
+}
 /// Generated client implementations.
 pub mod storage_client {
     #![allow(unused_variables, dead_code, missing_docs, clippy::let_unit_value)]

libsql-storage/src/generated/storage.rs

@@ -6,11 +6,12 @@ use std::path::Path;
 use std::sync::{Arc, Mutex};
 
 use crate::local_cache::LocalCache;
-use crate::rpc::Frame;
+use crate::rpc::{ErrorCode, Frame};
 use libsql_sys::ffi::{SQLITE_ABORT, SQLITE_BUSY};
 use libsql_sys::name::{NamespaceName, NamespaceResolver};
 use libsql_sys::rusqlite;
 use libsql_sys::wal::{Result, Vfs, Wal, WalManager};
+use prost::Message;
 use rpc::storage_client::StorageClient;
 use tonic::transport::Channel;
 use tracing::{error, trace, warn};
@@ -365,10 +366,13 @@ impl Wal for DurableWal {
         let resp = binding.insert_frames(req);
         let resp = tokio::task::block_in_place(|| rt.block_on(resp));
         if let Err(e) = resp {
-            if e.code() == tonic::Code::Aborted {
-                return Err(rusqlite::ffi::Error::new(SQLITE_BUSY));
-            }
-            return Err(rusqlite::ffi::Error::new(SQLITE_ABORT));
+            let error_code = rpc::ErrorDetails::decode(e.details())
+                .ok()
+                .and_then(|details| rpc::ErrorCode::try_from(details.code).ok());
+            return match error_code {
+                Some(ErrorCode::WriteConflict) => Err(rusqlite::ffi::Error::new(SQLITE_BUSY)),
+                _ => Err(rusqlite::ffi::Error::new(SQLITE_ABORT)),
+            };
         }
         // TODO: fix parity with storage server frame num with local cache
         self.local_cache