fix(segments): Calculate absolute_offset and Position properly

rumqttd/src/lib.rs

@@ -144,8 +144,7 @@ impl ConsoleSettings {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
-#[derive(Default)]
+#[derive(Debug, Serialize, Deserialize, Clone, Default)]
 pub enum Transport {
     #[serde(rename = "tcp")]
     #[default]
@@ -157,8 +156,6 @@ pub enum Transport {
     },
 }
 
-
-
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct ClientAuth {
     certs: PathBuf,

rumqttd/src/link/mod.rs

@@ -8,4 +8,3 @@ pub mod remote;
 #[cfg(feature = "websockets")]
 pub mod shadow;
 pub mod timer;
-

rumqttd/src/segments/mod.rs

@@ -160,6 +160,9 @@ where
 
     fn apply_retention(&mut self) {
         if self.active_segment().size() >= self.max_segment_size as u64 {
+            // Read absolute_offset before applying memory retention, incase there is only 1
+            // segment allowed
+            let absolute_offset = self.active_segment().next_offset();
             // If active segment is full and segments are full, apply retention policy
             if self.memory_segments_count() >= self.max_mem_segments {
                 self.segments.pop_front();
@@ -168,7 +171,6 @@ where
 
             // Pushing a new segment into segments and updating tail automatically changes active
             // segment to new empty one.
-            let absolute_offset = self.active_segment().next_offset();
             self.segments
                 .push_back(Segment::with_offset(absolute_offset));
             self.tail += 1;
@@ -230,7 +232,7 @@ where
                         end: (cursor.0, offset),
                     });
                 }
-                // no offset returned -> we reached end / invalid file
+                // no offset returned -> we reached end
                 // if len unfulfilled -> try next segment with remaining length
                 SegmentPosition::Done(next_offset) => {
                     // this condition is needed in case cursor.1 > 0 (when user provies cursor.1
@@ -274,7 +276,7 @@ where
                 // debug!("start: {:?}, end: ({}, {}) done", orig_cursor, cursor.0, absolute_offset);
                 Ok(Position::Done {
                     start,
-                    end: (cursor.0, absolute_offset),
+                    end: (cursor.0 + 1, absolute_offset),
                 })
             }
         }
@@ -387,7 +389,7 @@ mod tests {
                     next,
                     Done {
                         start: (0, 9),
-                        end: (0, 10)
+                        end: (1, 10)
                     }
                 );
                 continue;
@@ -428,7 +430,7 @@ mod tests {
             next,
             Done {
                 start: (0, 0),
-                end: (0, 10)
+                end: (1, 10)
             }
         );
 
@@ -443,7 +445,7 @@ mod tests {
             next,
             Done {
                 start: (0, 0),
-                end: (0, 10)
+                end: (1, 10)
             }
         );
 
@@ -458,13 +460,14 @@ mod tests {
             next,
             Done {
                 start: (0, 5),
-                end: (0, 10)
+                end: (1, 10)
             }
         );
 
         // Read again after after appending again
         let mut out = Vec::new();
         let next = log.readv((0, 10), 20, &mut out).unwrap();
+        // This should have end: (1, 10)
         assert_eq!(
             next,
             Done {
@@ -484,7 +487,7 @@ mod tests {
             next,
             Done {
                 start: (0, 10),
-                end: (0, 20)
+                end: (1, 20)
             }
         );
     }
@@ -549,7 +552,7 @@ mod tests {
             next,
             Done {
                 start: (2, 200),
-                end: (2, 300)
+                end: (3, 300)
             }
         );
     }
@@ -595,6 +598,7 @@ mod tests {
                 end: (3, 35)
             }
         );
+        // This should have end: (4, 40) as well
         let next = log.readv((3, 40), 5, &mut out).unwrap();
         assert_eq!(
             next,
@@ -643,7 +647,7 @@ mod tests {
             next,
             Done {
                 start: (3, 30),
-                end: (3, 40)
+                end: (4, 40)
             }
         );
     }
@@ -688,4 +692,40 @@ mod tests {
             }
         );
     }
+
+    #[test]
+    fn reads_which_end_exactly_at_active_segments_end() {
+        let max_segment_size = 1024;
+        let packet_size: u64 = 1024;
+        let mut log = CommitLog::new(max_segment_size, 1).unwrap();
+
+        log.append(random_payload(0, packet_size));
+        assert_eq!(log.head, 0);
+        assert_eq!(log.tail, 0);
+
+        let mut out = Vec::new();
+        let next = log.readv((0, 0), 1, &mut out).unwrap();
+        assert_eq!(
+            next,
+            Done {
+                start: (0, 0),
+                end: (1, 1)
+            }
+        );
+
+        log.append(random_payload(1, packet_size));
+        assert_eq!(log.head, 1);
+        assert_eq!(log.tail, 1);
+
+        let mut out = Vec::new();
+
+        let next = log.readv((0, 1), 1, &mut out).unwrap();
+        assert_eq!(
+            next,
+            Done {
+                start: (1, 1),
+                end: (2, 2)
+            }
+        );
+    }
 }