Better journal mismatch error (#44)

Author: slinkydeveloper

src/error.rs

@@ -18,7 +18,7 @@ pub(crate) struct CommandMetadata {
 
 impl fmt::Display for CommandMetadata {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "{:?} ", self.ty)?;
+        write!(f, "{} ", self.ty)?;
         if let Some(name) = &self.name {
             write!(f, "[{}]", name)?;
         } else {
@@ -171,3 +171,133 @@ impl From<CommandType> for MessageType {
         }
     }
 }
+
+impl TryFrom<MessageType> for CommandType {
+    type Error = MessageType;
+
+    fn try_from(value: MessageType) -> Result<Self, Self::Error> {
+        match value {
+            MessageType::InputCommand => Ok(CommandType::Input),
+            MessageType::OutputCommand => Ok(CommandType::Output),
+            MessageType::GetLazyStateCommand | MessageType::GetEagerStateCommand => {
+                Ok(CommandType::GetState)
+            }
+            MessageType::GetLazyStateKeysCommand | MessageType::GetEagerStateKeysCommand => {
+                Ok(CommandType::GetStateKeys)
+            }
+            MessageType::SetStateCommand => Ok(CommandType::SetState),
+            MessageType::ClearStateCommand => Ok(CommandType::ClearState),
+            MessageType::ClearAllStateCommand => Ok(CommandType::ClearAllState),
+            MessageType::GetPromiseCommand => Ok(CommandType::GetPromise),
+            MessageType::PeekPromiseCommand => Ok(CommandType::PeekPromise),
+            MessageType::CompletePromiseCommand => Ok(CommandType::CompletePromise),
+            MessageType::SleepCommand => Ok(CommandType::Sleep),
+            MessageType::CallCommand => Ok(CommandType::Call),
+            MessageType::OneWayCallCommand => Ok(CommandType::OneWayCall),
+            MessageType::SendSignalCommand => Ok(CommandType::SendSignal),
+            MessageType::RunCommand => Ok(CommandType::Run),
+            MessageType::AttachInvocationCommand => Ok(CommandType::AttachInvocation),
+            MessageType::GetInvocationOutputCommand => Ok(CommandType::GetInvocationOutput),
+            MessageType::CompleteAwakeableCommand => Ok(CommandType::CompleteAwakeable),
+            _ => Err(value),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_message_type_to_command_type_conversion() {
+        // Test successful conversions
+        assert_eq!(
+            CommandType::try_from(MessageType::InputCommand).unwrap(),
+            CommandType::Input
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::OutputCommand).unwrap(),
+            CommandType::Output
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::GetLazyStateCommand).unwrap(),
+            CommandType::GetState
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::GetLazyStateKeysCommand).unwrap(),
+            CommandType::GetStateKeys
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::SetStateCommand).unwrap(),
+            CommandType::SetState
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::ClearStateCommand).unwrap(),
+            CommandType::ClearState
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::ClearAllStateCommand).unwrap(),
+            CommandType::ClearAllState
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::GetPromiseCommand).unwrap(),
+            CommandType::GetPromise
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::PeekPromiseCommand).unwrap(),
+            CommandType::PeekPromise
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::CompletePromiseCommand).unwrap(),
+            CommandType::CompletePromise
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::SleepCommand).unwrap(),
+            CommandType::Sleep
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::CallCommand).unwrap(),
+            CommandType::Call
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::OneWayCallCommand).unwrap(),
+            CommandType::OneWayCall
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::SendSignalCommand).unwrap(),
+            CommandType::SendSignal
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::RunCommand).unwrap(),
+            CommandType::Run
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::AttachInvocationCommand).unwrap(),
+            CommandType::AttachInvocation
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::GetInvocationOutputCommand).unwrap(),
+            CommandType::GetInvocationOutput
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::CompleteAwakeableCommand).unwrap(),
+            CommandType::CompleteAwakeable
+        );
+
+        // Test failed conversions
+        assert_eq!(
+            CommandType::try_from(MessageType::Start).err().unwrap(),
+            MessageType::Start
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::End).err().unwrap(),
+            MessageType::End
+        );
+        assert_eq!(
+            CommandType::try_from(MessageType::GetLazyStateCompletionNotification)
+                .err()
+                .unwrap(),
+            MessageType::GetLazyStateCompletionNotification
+        );
+    }
+}

src/fmt.rs

@@ -0,0 +1,100 @@
+use crate::CommandType;
+use std::fmt;
+use std::sync::OnceLock;
+
+pub trait CommandFormatter: Send + Sync + fmt::Debug + 'static {
+    fn ty_display(&self, ty: CommandType) -> &'static str;
+}
+
+static GLOBAL_COMMAND_FORMATTER: OnceLock<Box<dyn CommandFormatter>> = OnceLock::new();
+
+// Public function for the crate user to set the formatter
+pub fn set_error_formatter(formatter: impl CommandFormatter + 'static) {
+    GLOBAL_COMMAND_FORMATTER
+        .set(Box::new(formatter))
+        .expect("Error formatter already set! It can only be set once.");
+}
+
+#[derive(Debug)]
+struct DefaultCommandFormatter;
+
+impl CommandFormatter for DefaultCommandFormatter {
+    fn ty_display(&self, ty: CommandType) -> &'static str {
+        match ty {
+            CommandType::Input => "handler input",
+            CommandType::Output => "handler return",
+            CommandType::GetState => "get state",
+            CommandType::GetStateKeys => "get state keys",
+            CommandType::SetState => "set state",
+            CommandType::ClearState => "clear state",
+            CommandType::ClearAllState => "clear all state",
+            CommandType::GetPromise => "get promise",
+            CommandType::PeekPromise => "peek promise",
+            CommandType::CompletePromise => "complete promise",
+            CommandType::Sleep => "sleep",
+            CommandType::Call => "call",
+            CommandType::OneWayCall => "one way call/send",
+            CommandType::SendSignal => "send signal",
+            CommandType::Run => "run",
+            CommandType::AttachInvocation => "attach invocation",
+            CommandType::GetInvocationOutput => "get invocation output",
+            CommandType::CompleteAwakeable => "complete awakeable",
+            CommandType::CancelInvocation => "cancel invocation",
+        }
+    }
+}
+
+pub(crate) fn display_command_ty(command_type: CommandType) -> &'static str {
+    if let Some(custom_formatter) = GLOBAL_COMMAND_FORMATTER.get() {
+        custom_formatter.ty_display(command_type)
+    } else {
+        DefaultCommandFormatter.ty_display(command_type)
+    }
+}
+
+pub(crate) struct DiffFormatter<'a, 'b> {
+    fmt: &'a mut fmt::Formatter<'b>,
+    indentation: &'static str,
+}
+
+impl<'a, 'b: 'a> DiffFormatter<'a, 'b> {
+    pub(crate) fn new(fmt: &'a mut fmt::Formatter<'b>, indentation: &'static str) -> Self {
+        Self { fmt, indentation }
+    }
+
+    pub(crate) fn write_diff(
+        &mut self,
+        field_name: &'static str,
+        actual: impl fmt::Display,
+        expected: impl fmt::Display,
+    ) -> fmt::Result {
+        write!(
+            self.fmt,
+            "\n{}{field_name}: {actual} != {expected}",
+            self.indentation
+        )
+    }
+
+    pub(crate) fn write_bytes_diff(
+        &mut self,
+        field_name: &'static str,
+        actual: &[u8],
+        expected: &[u8],
+    ) -> fmt::Result {
+        write!(self.fmt, "\n{}{field_name}: ", self.indentation)?;
+        match (std::str::from_utf8(actual), std::str::from_utf8(expected)) {
+            (Ok(actual), Ok(expected)) => {
+                write!(self.fmt, "'{actual}' != '{expected}'",)
+            }
+            (Ok(actual), Err(_)) => {
+                write!(self.fmt, "'{actual}' != {expected:?}")
+            }
+            (Err(_), Ok(expected)) => {
+                write!(self.fmt, "{actual:?} != '{expected}'")
+            }
+            (Err(_), Err(_)) => {
+                write!(self.fmt, "{actual:?} != {expected:?}")
+            }
+        }
+    }
+}

src/lib.rs

@@ -1,4 +1,5 @@
 pub mod error;
+pub mod fmt;
 mod headers;
 #[cfg(feature = "request_identity")]
 mod request_identity;

src/service_protocol/encoding.rs

@@ -14,6 +14,7 @@ use super::*;
 
 use std::mem;
 
+use crate::vm::errors::CommandTypeMismatchError;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use bytes_utils::SegmentedBuf;
 use prost::Message;
@@ -22,11 +23,8 @@ use prost::Message;
 pub enum DecodingError {
     #[error("cannot decode protocol message type {0:?}. Reason: {1:?}")]
     DecodeMessage(MessageType, #[source] prost::DecodeError),
-    #[error("Replayed journal doesn't match the handler code.\nThe handler code generated: {expected:?}\nwhile the replayed entry is: {actual:?}")]
-    UnexpectedMessageType {
-        expected: MessageType,
-        actual: MessageType,
-    },
+    #[error(transparent)]
+    UnexpectedMessageType(CommandTypeMismatchError),
     #[error("expected message type {expected:?} to have field {field}")]
     MissingField {
         expected: MessageType,
@@ -94,10 +92,9 @@ impl RawMessage {
 
     pub fn decode_to<M: RestateMessage>(self) -> Result<M, DecodingError> {
         if self.0.message_type() != M::ty() {
-            return Err(DecodingError::UnexpectedMessageType {
-                expected: M::ty(),
-                actual: self.0.message_type(),
-            });
+            return Err(DecodingError::UnexpectedMessageType(
+                CommandTypeMismatchError::new(self.0.message_type(), M::ty()),
+            ));
         }
         M::decode(self.1).map_err(|e| DecodingError::DecodeMessage(self.0.message_type(), e))
     }

src/service_protocol/header.rs

@@ -8,6 +8,9 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use crate::CommandType;
+use std::fmt;
+
 const COMMAND_ENTRY_MASK: u16 = 0x0400;
 const NOTIFICATION_ENTRY_MASK: u16 = 0x8000;
 const CUSTOM_ENTRY_MASK: u16 = 0xFC00;
@@ -118,6 +121,15 @@ impl MessageType {
     }
 }
 
+impl fmt::Display for MessageType {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match CommandType::try_from(*self) {
+            Ok(ct) => write!(f, "{}", crate::fmt::display_command_ty(ct)),
+            Err(mt) => write!(f, "{:?}", mt),
+        }
+    }
+}
+
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 pub struct MessageHeader {
     ty: MessageType,