fix: Report fatal error from the server to client properly.

.config/nextest.toml

@@ -0,0 +1,2 @@
+[profile.default]
+slow-timeout = { period = "15s", terminate-after = 1 }

.github/workflows/pr.yml

@@ -44,8 +44,14 @@ jobs:
       - name: Build
         run: cargo check --all-targets
 
+      - name: Install cargo-nextest
+        uses: baptiste0928/cargo-install@v3
+        with:
+          crate: cargo-nextest
+          locked: true
+
       - name: Run tests
-        run: cargo test --all-targets
+        run: cargo nextest run --hide-progress-bar
 
   linting:
     needs: build
@@ -78,4 +84,4 @@ jobs:
         run: cargo clippy --all-features -- -D warnings
 
       - name: Format
-        run: cargo fmt -- --check
+        run: cargo fmt -- --check

.gitignore

@@ -1,3 +1,4 @@
 target/
 .idea/
 proptest-regressions/
+mutants.out/

geth-client-tests/src/delete_tests.rs

@@ -1,5 +1,3 @@
-use std::u64;
-
 use bytes::Bytes;
 use fake::{faker::name::en::Name, Fake};
 use geth_client::GrpcClient;

geth-client-tests/src/lib.rs

@@ -16,7 +16,7 @@ pub mod tests {
 
     #[ctor::ctor]
     fn test_init() {
-        let _ = tracing_subscriber::fmt::fmt()
+        tracing_subscriber::fmt::fmt()
             .with_env_filter(EnvFilter::new("geth_engine=debug"))
             // .with_max_level(tracing::Level::DEBUG)
             .with_file(true)

geth-client/src/next/driver.rs

@@ -5,7 +5,7 @@ use eyre::bail;
 use uuid::Uuid;
 
 use geth_common::generated::next::protocol;
-use geth_common::{EndPoint, Operation, OperationIn, OperationOut};
+use geth_common::{EndPoint, Operation, OperationIn, OperationOut, Reply};
 
 use crate::next::{connect_to_node, Command, ConnErr, Connection, Mailbox};
 
@@ -104,6 +104,18 @@ impl Driver {
         tracing::warn!("received an event that is not related to any command");
     }
 
+    pub fn handle_disconnect(&mut self) {
+        tracing::warn!("connection was closed by the server");
+        self.connection = None;
+
+        for (correlation, cmd) in self.registry.drain() {
+            let _ = cmd.resp.send(OperationOut {
+                correlation,
+                reply: Reply::ServerDisconnected,
+            });
+        }
+    }
+
     /// We might consider implementing a retry logic here.
     async fn connect(&mut self) -> eyre::Result<()> {
         let uri = format!("http://{}:{}", self.endpoint.host, self.endpoint.port)

geth-client/src/next/grpc.rs

@@ -103,7 +103,14 @@ impl Client for GrpcClient {
         if let Some(out) = task.recv().await? {
             match out {
                 Reply::AppendStreamCompleted(resp) => Ok(resp),
-                _ => eyre::bail!("unexpected reply when appending events to '{}'", stream_id),
+                Reply::Error(e) => {
+                    eyre::bail!("server error when writing to stream '{}': {}", stream_id, e)
+                }
+                x => eyre::bail!(
+                    "unexpected reply when appending events to '{}' = {:?}",
+                    stream_id,
+                    x
+                ),
             }
         } else {
             eyre::bail!(

geth-client/src/next/mod.rs

@@ -16,6 +16,7 @@ pub mod grpc;
 pub enum Msg {
     Command(Command),
     Event(OperationOut),
+    Disconnected,
 }
 
 #[derive(Clone)]
@@ -48,18 +49,21 @@ pub(crate) async fn connect_to_node(uri: &Uri, mailbox: Mailbox) -> Result<Conne
         while let Some(reply) = stream_response.next().await {
             match reply {
                 Err(e) => {
-                    tracing::error!("Error receiving response: {:?}", e);
-                    break;
+                    tracing::error!("error receiving response: {:?}", e);
+                    // TODO - Needs find a way to handle unexpected errors so we avoid
+                    // stalled operations.
                 }
 
                 Ok(out) => {
                     if mailbox.send(Msg::Event(out.into())).is_err() {
                         tracing::warn!("seems main connection is closed");
-                        break;
+                        return;
                     }
                 }
             }
         }
+
+        let _ = mailbox.send(Msg::Disconnected);
     });
 
     Ok(connection)
@@ -75,6 +79,10 @@ pub(crate) async fn multiplex_loop(mut driver: Driver, mut receiver: UnboundedRe
             }
 
             Msg::Event(event) => driver.handle_event(event),
+
+            Msg::Disconnected => {
+                driver.handle_disconnect();
+            }
         }
     }
 }

geth-common/src/client.rs

@@ -11,13 +11,15 @@ use crate::{
     SubscriptionConfirmation,
 };
 
+#[derive(Debug)]
 pub enum SubscriptionEvent {
     EventAppeared(Record),
     Confirmed(SubscriptionConfirmation),
     CaughtUp,
     Unsubscribed(UnsubscribeReason),
 }
 
+#[derive(Debug)]
 pub enum UnsubscribeReason {
     User,
     Server,

geth-common/src/lib.rs

@@ -36,7 +36,7 @@ impl From<next::protocol::Ident> for Uuid {
     }
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct EndPoint {
     pub host: String,
     pub port: u16,
@@ -113,6 +113,7 @@ impl From<next::protocol::OperationIn> for OperationIn {
     }
 }
 
+#[derive(Debug)]
 pub enum Reply {
     AppendStreamCompleted(AppendStreamCompleted),
     StreamRead(StreamRead),
@@ -121,6 +122,8 @@ pub enum Reply {
     ProgramsListed(ProgramListed),
     ProgramKilled(ProgramKilled),
     ProgramObtained(ProgramObtained),
+    ServerDisconnected,
+    Error(String),
 }
 
 pub struct OperationOut {
@@ -157,6 +160,7 @@ impl From<next::protocol::OperationOut> for OperationOut {
             operation_out::Operation::ProgramKilled(resp) => Reply::ProgramKilled(resp.into()),
 
             operation_out::Operation::ProgramGot(resp) => Reply::ProgramObtained(resp.into()),
+            operation_out::Operation::Error(e) => Reply::Error(e),
         };
 
         Self { correlation, reply }
@@ -165,6 +169,7 @@ impl From<next::protocol::OperationOut> for OperationOut {
 
 impl TryFrom<OperationOut> for next::protocol::OperationOut {
     type Error = eyre::Report;
+
     fn try_from(value: OperationOut) -> eyre::Result<Self> {
         let correlation = Some(value.correlation.into());
         let operation = match value.reply {
@@ -183,6 +188,12 @@ impl TryFrom<OperationOut> for next::protocol::OperationOut {
             Reply::ProgramKilled(resp) => operation_out::Operation::ProgramKilled(resp.into()),
 
             Reply::ProgramObtained(resp) => operation_out::Operation::ProgramGot(resp.into()),
+
+            Reply::ServerDisconnected => {
+                eyre::bail!("not supposed to send server disconnected message to the server");
+            }
+
+            Reply::Error(e) => operation_out::Operation::Error(e),
         };
 
         Ok(Self {
@@ -1074,6 +1085,7 @@ impl<A> ReadStreamCompleted<A> {
     }
 }
 
+#[derive(Debug)]
 pub enum StreamRead {
     EndOfStream,
     EventAppeared(Record),
@@ -1116,6 +1128,7 @@ impl TryFrom<StreamRead> for operation_out::StreamRead {
     }
 }
 
+#[derive(Debug)]
 pub enum SubscriptionConfirmation {
     StreamName(String),
     ProcessId(Uuid),
@@ -1205,6 +1218,7 @@ impl From<operation_out::subscription_event::Error> for SubscriptionError {
     }
 }
 
+#[derive(Debug)]
 pub enum DeleteStreamCompleted {
     Success(WriteResult),
     Error(DeleteError),
@@ -1220,6 +1234,7 @@ impl DeleteStreamCompleted {
     }
 }
 
+#[derive(Debug)]
 pub enum DeleteError {
     StreamDeleted,
     WrongExpectedRevision(WrongExpectedRevisionError),

geth-consensus/src/tests/storage/in_mem.rs

@@ -67,8 +67,7 @@ impl PersistentStorage for InMemStorage {
 
         self.inner
             .iter()
-            .find(|e| e.index == entry_id.index && e.term == entry_id.term)
-            .is_some()
+            .any(|e| e.index == entry_id.index && e.term == entry_id.term)
     }
 }
 
@@ -81,7 +80,7 @@ struct InMemIter<'a> {
     init: bool,
 }
 
-impl<'a> IterateEntries for InMemIter<'a> {
+impl IterateEntries for InMemIter<'_> {
     fn next(&mut self) -> std::io::Result<Option<Entry>> {
         if self.count >= self.limit {
             return Ok(None);

geth-domain/src/index/tests/fs/lsm.rs

@@ -75,8 +75,10 @@ fn test_fs_lsm_mem_table_scan() -> io::Result<()> {
 /// When scanning, it will access to sstables.
 #[test]
 fn test_fs_lsm_sync() -> io::Result<()> {
-    let mut setts = LsmSettings::default();
-    setts.mem_table_max_size = MEM_TABLE_ENTRY_SIZE;
+    let setts = LsmSettings {
+        mem_table_max_size: MEM_TABLE_ENTRY_SIZE,
+        ..Default::default()
+    };
 
     let temp = TempDir::default();
     let root = PathBuf::from(temp.as_ref());
@@ -135,8 +137,8 @@ fn test_fs_lsm_serialization() -> io::Result<()> {
 
     assert_eq!(lsm.logical_position, actual.logical_position);
 
-    let actual_table_1 = actual.levels.get(&0).unwrap().front().clone().unwrap();
-    let actual_table_2 = actual.levels.get(&1).unwrap().front().clone().unwrap();
+    let actual_table_1 = actual.levels.get(&0).unwrap().front().unwrap();
+    let actual_table_2 = actual.levels.get(&1).unwrap().front().unwrap();
 
     assert_eq!(table1.id, actual_table_1.id);
     assert_eq!(table1.metas, actual_table_1.metas);

geth-domain/src/index/tests/fs/ss_table.rs

@@ -1,5 +1,5 @@
+use std::io;
 use std::path::PathBuf;
-use std::{io, u64};
 
 use temp_testdir::TempDir;
 

geth-domain/src/index/tests/in_mem/block.rs

@@ -1,4 +1,4 @@
-use std::{io, u64};
+use std::io;
 
 use geth_mikoshi::InMemoryStorage;
 

geth-domain/src/index/tests/in_mem/lsm.rs

@@ -1,4 +1,4 @@
-use std::{io, u64};
+use std::io;
 
 use geth_common::IteratorIO;
 use geth_mikoshi::InMemoryStorage;
@@ -107,8 +107,10 @@ fn test_in_mem_lsm_mem_table_scan_backward() -> io::Result<()> {
 /// When scanning, it will access to in-mem sstables.
 #[test]
 fn test_in_mem_lsm_sync() -> io::Result<()> {
-    let mut setts = LsmSettings::default();
-    setts.mem_table_max_size = MEM_TABLE_ENTRY_SIZE;
+    let setts = LsmSettings {
+        mem_table_max_size: MEM_TABLE_ENTRY_SIZE,
+        ..Default::default()
+    };
 
     let mut lsm = Lsm::new(setts, InMemoryStorage::new());
 

geth-domain/src/index/tests/in_mem/mem_table.rs

@@ -1,4 +1,4 @@
-use std::{io, u64};
+use std::io;
 
 use geth_common::IteratorIO;
 use geth_mikoshi::InMemoryStorage;

geth-domain/src/index/tests/in_mem/ss_table.rs

@@ -1,4 +1,4 @@
-use std::{io, u64};
+use std::io;
 
 use geth_common::IteratorIO;
 use geth_mikoshi::InMemoryStorage;