fix: remove potential panics and improve robustness

liboxia-native/src/client.rs

@@ -20,6 +20,7 @@ use crate::shard_manager::{Node, ShardManager, ShardManagerOptions};
 use crate::write_stream_manager::WriteStreamManager;
 use dashmap::DashMap;
 use std::cmp::Ordering;
+use std::fmt;
 use std::sync::Arc;
 use tokio::sync::mpsc::Receiver;
 use tokio::sync::{mpsc, oneshot, Mutex};
@@ -143,6 +144,30 @@ pub enum Notification {
     Unknown(),
 }
 
+impl fmt::Display for Notification {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Notification::KeyCreated(c) => write!(
+                f,
+                "KeyCreated(key={}, version_id={:?})",
+                c.key, c.version_id
+            ),
+            Notification::KeyDeleted(d) => write!(f, "KeyDeleted(key={})", d.key),
+            Notification::KeyModified(m) => write!(
+                f,
+                "KeyModified(key={}, version_id={:?})",
+                m.key, m.version_id
+            ),
+            Notification::KeyRangeDeleted(r) => write!(
+                f,
+                "KeyRangeDeleted(key={}, last={:?})",
+                r.key, r.key_range_last
+            ),
+            Notification::Unknown() => write!(f, "Unknown"),
+        }
+    }
+}
+
 impl From<(String, crate::oxia::Notification)> for Notification {
     fn from(tuple: (String, crate::oxia::Notification)) -> Self {
         let (key, notification) = tuple;

liboxia-native/src/shard_manager.rs

@@ -157,18 +157,19 @@ impl ShardManager {
     pub fn get_shard(&self, key: &str) -> Option<i64> {
         let code = xxhash_rust::xxh32::xxh32(key.as_bytes(), 0);
         for entry in self.inner.current_assignments.iter() {
-            match entry.shard_boundaries.unwrap() {
-                ShardBoundaries::Int32HashRange(range) => {
-                    if range.min_hash_inclusive <= code && code <= range.max_hash_inclusive {
-                        return Some(entry.shard);
+            if let Some(boundaries) = entry.shard_boundaries.as_ref() {
+                match boundaries {
+                    ShardBoundaries::Int32HashRange(range) => {
+                        if range.min_hash_inclusive <= code && code <= range.max_hash_inclusive {
+                            return Some(entry.shard);
+                        }
                     }
                 }
             }
         }
         None
     }
 
-    // todo: consider iterator to avoid clone
     pub fn get_shards_leader(&self) -> HashMap<i64, Node> {
         let mut map = HashMap::with_capacity(self.inner.current_assignments.len());
         for item in self.inner.current_assignments.iter() {
@@ -185,12 +186,14 @@ impl ShardManager {
     pub fn get_shard_leader(&self, key: &str) -> Option<Node> {
         let code = xxhash_rust::xxh32::xxh32(key.as_bytes(), 0);
         for entry in self.inner.current_assignments.iter() {
-            match entry.shard_boundaries.unwrap() {
-                ShardBoundaries::Int32HashRange(range) => {
-                    if range.min_hash_inclusive <= code && code <= range.max_hash_inclusive {
-                        return Some(Node {
-                            service_address: ensure_protocol(entry.leader.clone()),
-                        });
+            if let Some(boundaries) = entry.shard_boundaries.as_ref() {
+                match boundaries {
+                    ShardBoundaries::Int32HashRange(range) => {
+                        if range.min_hash_inclusive <= code && code <= range.max_hash_inclusive {
+                            return Some(Node {
+                                service_address: ensure_protocol(entry.leader.clone()),
+                            });
+                        }
                     }
                 }
             }