refactor: preparations for returning errors from governor

src/governor/mod.rs

@@ -28,6 +28,7 @@ use crate::storage::Clusterwide;
 use crate::storage::ToEntryIter as _;
 use crate::tlog;
 use crate::traft::error::Error;
+use crate::traft::error::ErrorInfo;
 use crate::traft::network::ConnectionPool;
 use crate::traft::node::global;
 use crate::traft::node::Status;
@@ -178,13 +179,13 @@ impl Loop {
         #[derive(Debug)]
         enum OnError {
             Retry(Error),
-            Abort,
+            Abort(ErrorInfo),
         }
         impl From<OnError> for Error {
             fn from(e: OnError) -> Error {
                 match e {
                     OnError::Retry(e) => e,
-                    OnError::Abort => Error::other("schema change was aborted"),
+                    OnError::Abort(_) => unreachable!("we never convert Abort to Error"),
                 }
             }
         }
@@ -457,11 +458,11 @@ impl Loop {
                                         );
                                         Ok(())
                                     }
-                                    Ok(rpc::ddl_apply::Response::Abort { reason }) => {
-                                        tlog!(Error, "failed to apply schema change on instance: {reason}";
+                                    Ok(rpc::ddl_apply::Response::Abort { cause }) => {
+                                        tlog!(Error, "failed to apply schema change on instance: {cause}";
                                             "instance_id" => %instance_id,
                                         );
-                                        Err(OnError::Abort)
+                                        Err(OnError::Abort(cause))
                                     }
                                     Err(e) => {
                                         tlog!(Warning, "failed calling proc_apply_schema_change: {e}";
@@ -474,7 +475,7 @@ impl Loop {
                         }
                         // TODO: don't hard code timeout
                         let res = try_join_all(fs).timeout(Duration::from_secs(3)).await;
-                        if let Err(TimeoutError::Failed(OnError::Abort)) = res {
+                        if let Err(TimeoutError::Failed(OnError::Abort(_cause))) = res {
                             next_op = Op::DdlAbort;
                             return Ok(());
                         }
@@ -525,14 +526,14 @@ impl Loop {
                                         tlog!(Error, "failed to call proc_load_plugin_dry_run: {e}";
                                             "instance_id" => %instance_id
                                         );
-                                        Err(OnError::Abort)
+                                        Err(e)
                                     }
                                 }
                             });
                         }
 
                         if let Err(e) = try_join_all(fs).timeout(Duration::from_secs(5)).await {
-                            tlog!(Error, "Plugin installation aborted: {e:?}");
+                            tlog!(Error, "Plugin installation aborted: {e}");
                             return Ok(());
                         }
 
@@ -569,22 +570,20 @@ impl Loop {
                             fs.push(async move {
                                 match resp.await {
                                     Ok(rpc::enable_plugin::Response::Ok) => {
-                                        tlog!(Info, "load plugin on instance";
-                                            "instance_id" => %instance_id,
-                                        );
+                                        tlog!(Info, "enable plugin on instance"; "instance_id" => %instance_id);
                                         Ok(())
                                     }
-                                    Ok(rpc::enable_plugin::Response::Abort { reason }) => {
-                                        tlog!(Error, "failed to load plugin at instance: {reason}";
+                                    Ok(rpc::enable_plugin::Response::Abort { cause }) => {
+                                        tlog!(Error, "failed to enable plugin at instance: {cause}";
                                             "instance_id" => %instance_id,
                                         );
-                                        Err(OnError::Abort)
+                                        Err(OnError::Abort(cause))
                                     }
                                     Err(Error::Timeout) => {
-                                        tlog!(Error, "failed to load plugin at instance: timeout";
+                                        tlog!(Error, "failed to enable plugin at instance: timeout";
                                             "instance_id" => %instance_id,
                                         );
-                                        Err(OnError::Abort)
+                                        Err(OnError::Abort(ErrorInfo::timeout(instance_id.clone(), "failed to enable plugin")))
                                     }
                                     Err(e) => {
                                         tlog!(Warning, "failed calling proc_load_plugin: {e}";
@@ -597,7 +596,7 @@ impl Loop {
                         }
 
                         let enable_result = try_join_all(fs).timeout(on_start_timeout.add(Duration::from_secs(1))).await;
-                        if let Err(TimeoutError::Failed(OnError::Abort)) = enable_result {
+                        if let Err(TimeoutError::Failed(OnError::Abort(_cause))) = enable_result {
                             next_op = Some(rollback_op);
                             return Ok(());
                         }

src/plugin/mod.rs

@@ -6,6 +6,7 @@ pub mod topology;
 
 use once_cell::unsync;
 use picoplugin::background::ServiceId;
+use picoplugin::error_code::ErrorCode;
 use picoplugin::plugin::interface::ServiceBox;
 use serde::{Deserialize, Serialize};
 use std::fmt::{Display, Formatter};
@@ -14,7 +15,7 @@ use std::io;
 use std::path::{Path, PathBuf};
 use std::rc::Rc;
 use std::time::Duration;
-use tarantool::error::BoxError;
+use tarantool::error::{BoxError, IntoBoxError};
 use tarantool::fiber;
 use tarantool::time::Instant;
 
@@ -61,8 +62,6 @@ pub enum PluginError {
     ManifestNotFound(String, io::Error),
     #[error("Error while parsing manifest `{0}`, reason: {1}")]
     InvalidManifest(String, Box<dyn std::error::Error>),
-    #[error("`{0}` service defenition not found")]
-    ServiceDefenitionNotFound(String),
     #[error("Read plugin_dir: {0}")]
     ReadPluginDir(#[from] io::Error),
     #[error("Invalid shared object file: {0}")]
@@ -99,6 +98,16 @@ pub enum PluginError {
     AmbiguousEnableCandidate,
 }
 
+impl IntoBoxError for PluginError {
+    #[inline(always)]
+    fn error_code(&self) -> u32 {
+        match self {
+            Self::ServiceNotFound { .. } => ErrorCode::NoSuchService as _,
+            _ => ErrorCode::Other as _,
+        }
+    }
+}
+
 #[derive(thiserror::Error, Debug)]
 pub enum PluginCallbackError {
     #[error("on_start: {0}")]

src/rpc/ddl_apply.rs

@@ -7,9 +7,11 @@ use crate::storage::{ddl_rename_function_on_master, Clusterwide};
 use crate::storage::{local_schema_version, set_local_schema_version};
 use crate::tlog;
 use crate::traft::error::Error as TraftError;
+use crate::traft::error::ErrorInfo;
 use crate::traft::node;
 use crate::traft::{RaftIndex, RaftTerm};
 use std::time::Duration;
+use tarantool::error::IntoBoxError;
 use tarantool::error::{BoxError, TarantoolErrorCode};
 use tarantool::transaction::{transaction, TransactionError};
 
@@ -60,7 +62,14 @@ crate::define_rpc_request! {
             Ok(()) => Ok(Response::Ok),
             Err(TransactionError::RolledBack(Error::Aborted(err))) => {
                 tlog!(Warning, "schema change aborted: {err}");
-                Ok(Response::Abort { reason: err})
+
+                let instance_id = node.raft_storage.instance_id()?.expect("should be persisted before procs are defined");
+                let cause = ErrorInfo {
+                    error_code: err.error_code(),
+                    message: err.to_string(),
+                    instance_id,
+                };
+                Ok(Response::Abort { cause })
             }
             Err(err) => {
                 tlog!(Warning, "applying schema change failed: {err}");
@@ -80,7 +89,7 @@ crate::define_rpc_request! {
         Ok,
         /// Schema change failed on this instance and should be aborted on the
         /// whole cluster.
-        Abort { reason: String },
+        Abort { cause: ErrorInfo },
     }
 }
 
@@ -89,7 +98,7 @@ pub enum Error {
     /// Schema change failed on this instance and should be aborted on the
     /// whole cluster.
     #[error("{0}")]
-    Aborted(String),
+    Aborted(TraftError),
 
     #[error("{0}")]
     Other(TraftError),
@@ -127,7 +136,7 @@ pub fn apply_schema_change(
                         tlog!(Error, "{}:{}: {e}", file, line);
                     }
                 }
-                return Err(Error::Aborted(e.to_string()));
+                return Err(Error::Aborted(e.into()));
             }
         }
 
@@ -139,13 +148,13 @@ pub fn apply_schema_change(
 
             let abort_reason = ddl_drop_space_on_master(id).map_err(Error::Other)?;
             if let Some(e) = abort_reason {
-                return Err(Error::Aborted(e.to_string()));
+                return Err(Error::Aborted(e.into()));
             }
         }
 
         Ddl::CreateProcedure { id, .. } => {
             if let Err(e) = ddl_create_function_on_master(storage, id) {
-                return Err(Error::Aborted(e.to_string()));
+                return Err(Error::Aborted(e));
             }
         }
 
@@ -156,7 +165,7 @@ pub fn apply_schema_change(
 
             let abort_reason = ddl_drop_function_on_master(id).map_err(Error::Other)?;
             if let Some(e) = abort_reason {
-                return Err(Error::Aborted(e.to_string()));
+                return Err(Error::Aborted(e.into()));
             }
         }
 
@@ -166,7 +175,7 @@ pub fn apply_schema_change(
             ..
         } => {
             if let Err(e) = ddl_rename_function_on_master(storage, routine_id, new_name) {
-                return Err(Error::Aborted(e.to_string()));
+                return Err(Error::Aborted(e));
             }
         }
 
@@ -174,7 +183,7 @@ pub fn apply_schema_change(
             space_id, index_id, ..
         } => {
             if let Err(e) = ddl_create_index_on_master(storage, space_id, index_id) {
-                return Err(Error::Aborted(e.to_string()));
+                return Err(Error::Aborted(e));
             }
         }
 
@@ -186,13 +195,13 @@ pub fn apply_schema_change(
             }
 
             if let Err(e) = ddl_drop_index_on_master(space_id, index_id) {
-                return Err(Error::Aborted(e.to_string()));
+                return Err(Error::Aborted(e));
             }
         }
     }
 
     if let Err(e) = set_local_schema_version(version) {
-        return Err(Error::Aborted(e.to_string()));
+        return Err(Error::Aborted(e.into()));
     }
 
     Ok(())

src/rpc/enable_plugin.rs

 use crate::plugin::PluginOp;
 use crate::tlog;
 use crate::traft::error::Error;
+use crate::traft::error::ErrorInfo;
 use crate::traft::node;
 use crate::traft::{RaftIndex, RaftTerm};
 use std::time::Duration;
+use tarantool::error::IntoBoxError;
 
 crate::define_rpc_request! {
     /// Forces the target instance to actually enable the plugin locally.
@@ -37,7 +39,14 @@ crate::define_rpc_request! {
              Ok(()) => Ok(Response::Ok),
              Err(err) => {
                 tlog!(Warning, "plugin enabling aborted: {err}");
-                Ok(Response::Abort { reason: err.to_string() })
+
+                let instance_id = node.raft_storage.instance_id()?.expect("should be persisted before procs are defined");
+                let cause = ErrorInfo {
+                    error_code: err.error_code(),
+                    message: err.to_string(),
+                    instance_id,
+                };
+                Ok(Response::Abort { cause })
             }
         }
     }
@@ -53,6 +62,6 @@ crate::define_rpc_request! {
         Ok,
         /// Plugin loaded failed on this instance and should be aborted on the
         /// whole cluster.
-        Abort { reason: String },
+        Abort { cause: ErrorInfo },
     }
 }

src/traft/error.rs

@@ -4,8 +4,8 @@ use crate::error_code::ErrorCode;
 use crate::instance::InstanceId;
 use crate::plugin::PluginError;
 use crate::traft::{RaftId, RaftTerm};
-use tarantool::error::BoxError;
 use tarantool::error::IntoBoxError;
+use tarantool::error::{BoxError, TarantoolErrorCode};
 use tarantool::fiber::r#async::timeout;
 use tarantool::tlua::LuaError;
 use thiserror::Error;
@@ -217,3 +217,49 @@ impl IntoBoxError for Error {
         }
     }
 }
+
+////////////////////////////////////////////////////////////////////////////////
+// ErrorInfo
+////////////////////////////////////////////////////////////////////////////////
+
+/// This is a serializable version of [`BoxError`].
+///
+/// TODO<https://git.picodata.io/picodata/picodata/tarantool-module/-/issues/221> just make BoxError serializable.
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
+pub struct ErrorInfo {
+    pub error_code: u32,
+    pub message: String,
+    pub instance_id: InstanceId,
+}
+
+impl ErrorInfo {
+    #[inline(always)]
+    pub fn timeout(instance_id: InstanceId, message: impl Into<String>) -> Self {
+        Self {
+            error_code: TarantoolErrorCode::Timeout as _,
+            message: message.into(),
+            instance_id,
+        }
+    }
+}
+
+impl std::fmt::Display for ErrorInfo {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "[instance_id:{}] ", self.instance_id)?;
+        if let Some(c) = ErrorCode::from_i64(self.error_code as _) {
+            write!(f, "{c:?}")?;
+        } else if let Some(c) = TarantoolErrorCode::from_i64(self.error_code as _) {
+            write!(f, "{c:?}")?;
+        } else {
+            write!(f, "#{}", self.error_code)?;
+        }
+        write!(f, ": {}", self.message)
+    }
+}
+
+impl IntoBoxError for ErrorInfo {
+    #[inline]
+    fn error_code(&self) -> u32 {
+        self.error_code
+    }
+}