[mqtt] Ring buffer write succeeds when no space to fit a new insert (A…

…zure#4825) There was an issue when ring buffer calculated free space available to insert a new publication.
ggjjj · Jul 22, 2021 · 87b4225 · 87b4225
1 parent 25ff198
commit 87b4225
Show file tree

Hide file tree

Showing 3 changed files with 165 additions and 178 deletions.
diff --git a/mqtt/mqtt-bridge/src/messages.rs b/mqtt/mqtt-bridge/src/messages.rs
@@ -165,9 +165,17 @@ where
                 if let Some(publication) = forward_publication {
                     debug!("saving message to store");
                     return match self.store.push(&publication) {
-                        Ok(_) |
-                        // If we are full we are dropping the message on ground.
-                        Err(PersistError::RingBuffer(RingBufferError::Full)) => Ok(Handled::Fully),
+                        Ok(_) => Ok(Handled::Fully),
+                        Err(
+                            err
+                            @
+                            PersistError::RingBuffer(RingBufferError::InsufficientSpace {
+                                ..
+                            }),
+                        ) => {
+                            error!(error = %err, "dropping incoming publication");
+                            Ok(Handled::Fully)
+                        }
                         Err(err) => Err(BridgeError::Store(err)),
                     };
                 }

diff --git a/mqtt/mqtt-bridge/src/persist/waking_state/ring_buffer/error.rs b/mqtt/mqtt-bridge/src/persist/waking_state/ring_buffer/error.rs
@@ -26,8 +26,8 @@ pub enum RingBufferError {
     #[error("Flushing failed. Caused by {0}")]
     Flush(std::io::Error),
 
-    #[error("Buffer is full and messages must be drained to continue")]
-    Full,
+    #[error("Storage has insufficient space to insert data: required: {required}b, but only {free}b available")]
+    InsufficientSpace { free: u64, required: u64 },
 
     #[error("Unable to create file. Caused by {0}")]
     FileCreate(std::io::Error),