fix(server): release lock before async save to prevent deadlock (#1567)

- In `consumer_offsets.rs`, clone the consumer offset's path and
  drop the mutable guard before awaiting the async save.
- Refactor `save_consumer_offset` to accept a new offset value
  and path instead of a ConsumerOffset reference.
- Update error messages and trace logs accordingly.
- Bump version in Cargo.toml from 0.4.210 to 0.4.211.

Author: hubcio

Cargo.lock

@@ -42,7 +42,10 @@ async fn assert_persisted_offset(
     consumer_offset: &ConsumerOffset,
     expected_offsets_count: u32,
 ) {
-    storage.save_consumer_offset(consumer_offset).await.unwrap();
+    storage
+        .save_consumer_offset(consumer_offset.offset, &consumer_offset.path)
+        .await
+        .unwrap();
     let consumer_offsets = storage
         .load_consumer_offsets(consumer_offset.kind, path)
         .await
@@ -51,11 +54,7 @@ async fn assert_persisted_offset(
     assert_eq!(consumer_offsets.len(), expected_offsets_count);
     let loaded_consumer_offset = consumer_offsets.get(expected_offsets_count - 1).unwrap();
 
-    // TODO(hubcio): This is a workaround: sometimes offset is 4, sometimes 5
-    let offset_ok = loaded_consumer_offset.offset == consumer_offset.offset
-        || loaded_consumer_offset.offset == consumer_offset.offset + 1
-        || loaded_consumer_offset.offset == consumer_offset.offset - 1;
-    assert!(offset_ok);
+    assert!(loaded_consumer_offset.offset == consumer_offset.offset);
 
     assert_eq!(loaded_consumer_offset.kind, consumer_offset.kind);
     assert_eq!(

integration/tests/streaming/consumer_offset.rs

@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.4.210"
+version = "0.4.211"
 edition = "2021"
 build = "src/build.rs"
 license = "Apache-2.0"

server/Cargo.toml

@@ -78,14 +78,15 @@ impl Partition {
         let consumer_offsets = self.get_consumer_offsets(kind);
         if let Some(mut consumer_offset) = consumer_offsets.get_mut(&consumer_id) {
             consumer_offset.offset = offset;
+            let path = consumer_offset.path.clone();
+            drop(consumer_offset);
             self.storage
                 .partition
-                .save_consumer_offset(&consumer_offset)
+                .save_consumer_offset(offset, &path)
                 .await
                 .with_error_context(|error| {
                     format!(
-                        "{COMPONENT} (error: {error}) - failed to save consumer offset, consumer ID: {}, offset: {}",
-                        consumer_id, offset
+                        "{COMPONENT} (error: {error}) - failed to save consumer offset, consumer ID: {consumer_id}, offset: {offset}, path: {path}",
                     )
                 })?;
             return Ok(());
@@ -98,7 +99,7 @@ impl Partition {
         let consumer_offset = ConsumerOffset::new(kind, consumer_id, offset, path);
         self.storage
             .partition
-            .save_consumer_offset(&consumer_offset)
+            .save_consumer_offset(offset, &consumer_offset.path)
             .await
             .with_error_context(|error| {
                 format!(

server/src/streaming/partitions/consumer_offsets.rs

@@ -54,7 +54,7 @@ pub struct ConsumerOffset {
     pub kind: ConsumerKind,
     pub consumer_id: u32,
     pub offset: u64,
-    pub path: String,
+    pub path: Arc<String>,
 }
 
 impl ConsumerOffset {
@@ -63,7 +63,7 @@ impl ConsumerOffset {
             kind,
             consumer_id,
             offset,
-            path: format!("{path}/{consumer_id}"),
+            path: Arc::new(format!("{path}/{consumer_id}")),
         }
     }
 }

server/src/streaming/partitions/partition.rs

@@ -335,21 +335,15 @@ impl PartitionStorage for FilePartitionStorage {
         Ok(())
     }
 
-    async fn save_consumer_offset(&self, offset: &ConsumerOffset) -> Result<(), IggyError> {
+    async fn save_consumer_offset(&self, offset: u64, path: &str) -> Result<(), IggyError> {
         self.persister
-            .overwrite(&offset.path, &offset.offset.to_le_bytes())
+            .overwrite(path, &offset.to_le_bytes())
             .await
             .with_error_context(|error| format!(
-                "{COMPONENT} (error: {error}) - failed to overwrite consumer offset with value: {}, kind: {}, consumer ID: {}, path: {}",
-                offset.offset, offset.kind, offset.consumer_id, offset.path,
+                "{COMPONENT} (error: {error}) - failed to overwrite consumer offset with value: {}, path: {}",
+                offset, path,
             ))?;
-        trace!(
-            "Stored consumer offset value: {} for {} with ID: {}, path: {}",
-            offset.offset,
-            offset.kind,
-            offset.consumer_id,
-            offset.path
-        );
+        trace!("Stored consumer offset value: {}, path: {}", offset, path);
         Ok(())
     }
 
@@ -390,7 +384,7 @@ impl PartitionStorage for FilePartitionStorage {
                 continue;
             }
 
-            let path = path.unwrap().to_string();
+            let path = Arc::new(path.unwrap().to_string());
             let consumer_id = consumer_id.unwrap();
             let mut file = file::open(&path)
                 .await

server/src/streaming/partitions/storage.rs

@@ -105,7 +105,8 @@ pub trait PartitionStorage: Send {
     fn delete(&self, partition: &Partition) -> impl Future<Output = Result<(), IggyError>> + Send;
     fn save_consumer_offset(
         &self,
-        offset: &ConsumerOffset,
+        offset: u64,
+        path: &str,
     ) -> impl Future<Output = Result<(), IggyError>> + Send;
     fn load_consumer_offsets(
         &self,
@@ -177,7 +178,7 @@ impl PartitionStorageKind {
             -> Result<(), IggyError>;
         async fn save(&self, partition: &mut Partition) -> Result<(), IggyError>;
         async fn delete(&self, partition: &Partition) -> Result<(), IggyError>;
-        async fn save_consumer_offset(&self, offset: &ConsumerOffset) -> Result<(), IggyError>;
+        async fn save_consumer_offset(&self, offset: u64, path: &str) -> Result<(), IggyError>;
         async fn load_consumer_offsets(
             &self,
             kind: ConsumerKind,