fix: shutdown and entry payload (#1)

Signed-off-by: tison <[email protected]>

Cargo.toml

@@ -45,13 +45,16 @@ version = "0.1.0"
 
 [workspace.dependencies]
 backon = { version = "1.0", features = ["tokio-sleep"] }
+base64 = { version = "0.22" }
+better-panic = { version = "0.3" }
 build-data = { version = "0.2" }
 byteorder = { version = "1.5" }
 clap = { version = "4.5", features = ["derive"] }
 const_format = { version = "0.2" }
 ctrlc = "3.4"
 error-stack = { version = "0.5" }
 fastrace = { version = "0.7", features = ["enable"] }
+flexbuffers = { version = "2.0" }
 futures = "0.3"
 gix-discover = "0.35"
 insta = { version = "1.38", features = ["json"] }

api/protos/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+extern crate core;
+
 pub mod config;
 pub mod property;
 pub mod rpc;

api/protos/src/rpc.rs

@@ -12,11 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::ops::Range;
+
 use serde::Deserialize;
 use serde::Serialize;
 
 use crate::property::TopicProps;
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Entry {
+    /// Index of this log entry, assigned by the broker when the message is published. Guaranteed
+    /// to be unique within the topic. It must not be populated by the publisher in a write
+    /// call.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub index: Option<i64>,
+    /// A padded, base64-encoded string of bytes, encoded with a URL and filename safe alphabet
+    /// (sometimes referred to as "web-safe" or "base64url"). Defined by [RFC4648].
+    ///
+    /// [RFC4648]: https://datatracker.ietf.org/doc/html/rfc4648
+    pub data: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct CreateLogRequest {
     pub name: String,
@@ -31,14 +47,13 @@ pub struct CreateLogResponse {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AppendLogRequest {
     pub name: String,
-    pub data: Vec<u8>,
-    pub entry_cnt: i32,
+    pub entries: Vec<Entry>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AppendLogResponse {
-    pub start_offset: i64,
-    pub end_offset: i64,
+    /// The half-open offset range of the appended entry.
+    pub offsets: Range<i64>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -47,18 +62,9 @@ pub struct ReadLogRequest {
     pub offset: i64,
 }
 
-#[derive(Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ReadLogResponse {
-    pub data: Vec<u8>,
-}
-
-// TODO(tisonkun): adopts bytes debug fmt or find other robust ways
-impl std::fmt::Debug for ReadLogResponse {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("ReadLogResponse")
-            .field("data", &String::from_utf8_lossy(&self.data))
-            .finish()
-    }
+    pub entries: Vec<Entry>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]

cmd/morax/src/command.rs

@@ -69,19 +69,25 @@ impl CommandStart {
         };
 
         morax_runtime::init(&config.runtime);
-        ctrlc::set_handler(move || {
-            morax_runtime::shutdown();
-        })
-        .change_context_lazy(|| Error("failed to setup ctrl-c signal handle".to_string()))?;
+        morax_telemetry::init(&config.telemetry);
+
+        error_stack::Report::set_color_mode(error_stack::fmt::ColorMode::None);
+        error_stack::Report::set_charset(error_stack::fmt::Charset::Ascii);
 
         let rt = morax_runtime::make_runtime("morax-main", "morax-main", 1);
         rt.block_on(async move {
-            morax_telemetry::init(&config.telemetry);
             let state = morax_server::start(config.server)
                 .await
                 .change_context_lazy(|| {
                     Error("A fatal error has occurred in Morax server process.".to_string())
                 })?;
+
+            let shutdown_handle = state.shutdown_handle();
+            ctrlc::set_handler(move || {
+                shutdown_handle();
+            })
+            .change_context_lazy(|| Error("failed to setup ctrl-c signal handle".to_string()))?;
+
             state.await_shutdown().await;
             Ok(())
         })

crates/kafka-broker/src/network.rs

@@ -69,10 +69,6 @@ pub async fn start_broker(
                 log::info!("Morax Server is closing");
                 return Ok(());
             }
-            _ = morax_runtime::wait_shutdown() => {
-                log::info!("Morax Server is shutting down");
-                return Ok(());
-            }
             socket = broker_listener.accept() => socket,
         };
 
@@ -100,10 +96,6 @@ async fn process_packet(
                 log::info!("Morax Server is closing");
                 return Ok(());
             }
-            _ = morax_runtime::wait_shutdown() => {
-                log::info!("Morax Server is shutting down");
-                return Ok(());
-            }
             closed = process_packet_one(&mut socket, &remote_addr, &broker) => {
                 if closed? {
                     return Ok(());

crates/runtime/Cargo.toml

@@ -26,6 +26,7 @@ version.workspace = true
 test = []
 
 [dependencies]
+better-panic = { workspace = true }
 error-stack = { workspace = true }
 fastrace = { workspace = true }
 futures = { workspace = true }
@@ -38,7 +39,7 @@ tokio = { workspace = true }
 tokio-util = { workspace = true }
 
 [dev-dependencies]
-pollster = { version = "0.3", features = ['macro'] }
+pollster = { version = "0.3" }
 
 [lints]
 workspace = true

crates/runtime/src/global.rs

@@ -12,10 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
 use std::sync::OnceLock;
 
-use latches::task::Latch;
 use serde::Deserialize;
 use serde::Serialize;
 
@@ -45,7 +43,6 @@ struct GlobalRuntimes {
     meta_runtime: Runtime,
     data_runtime: Runtime,
     bg_runtime: Runtime,
-    shutdown: Arc<Latch>,
 }
 
 static GLOBAL_RUNTIMES: OnceLock<GlobalRuntimes> = OnceLock::new();
@@ -84,23 +81,31 @@ pub fn init(opts: &RuntimeOptions) {
     GLOBAL_RUNTIMES.get_or_init(|| do_initialize_runtimes(opts));
 }
 
+fn setup_panic_hook() {
+    std::panic::set_hook(Box::new(move |info| {
+        let backtrace = std::backtrace::Backtrace::capture();
+        log::error!("panic occurred: {info}\nbacktrace:\n{backtrace}");
+        better_panic::Settings::auto().create_panic_handler()(info);
+        shutdown();
+    }));
+}
+
 fn do_initialize_runtimes(opts: &RuntimeOptions) -> GlobalRuntimes {
     log::info!("initializing global runtimes: {opts:?}");
 
+    setup_panic_hook();
+
     let api_runtime = make_runtime("api_runtime", "api_thread", opts.api_runtime_threads);
     let exec_runtime = make_runtime("exec_runtime", "exec_thread", opts.exec_runtime_threads);
     let meta_runtime = make_runtime("meta_runtime", "meta_thread", opts.meta_runtime_threads);
     let data_runtime = make_runtime("data_runtime", "data_thread", opts.data_runtime_threads);
     let bg_runtime = make_runtime("bg_runtime", "bg_thread", opts.bg_runtime_threads);
-    let shutdown = Arc::new(Latch::new(1));
-
     GlobalRuntimes {
         api_runtime,
         exec_runtime,
         meta_runtime,
         data_runtime,
         bg_runtime,
-        shutdown,
     }
 }
 
@@ -129,20 +134,12 @@ pub fn bg_runtime() -> &'static Runtime {
 }
 
 pub fn shutdown() {
-    let runtimes = fetch_runtimes_or_default();
-    runtimes.shutdown.count_down();
-}
-
-pub async fn wait_shutdown() {
-    fetch_runtimes_or_default().shutdown.wait().await;
+    log::info!("The Global Runtimes are shutting down.");
+    std::process::exit(1);
 }
 
 #[cfg(test)]
 mod tests {
-    use core::panic;
-
-    use pollster::FutureExt;
-
     use super::*;
 
     #[test]
@@ -180,10 +177,4 @@ mod tests {
             assert_eq!(out, "hello")
         }
     }
-
-    #[test]
-    fn test_task_panic() {
-        let _fut = exec_runtime().spawn(async { panic!("test panic") });
-        wait_shutdown().block_on();
-    }
 }