Return num bytes on truncation

README.md

@@ -34,6 +34,14 @@ pub struct MultiRecordLog {
 }
 ```
 
+`QueueSummary::num_bytes` reports the total payload bytes retained in a queue.
+`AppendOutcome::queue_size_bytes` and `TruncateOutcome::queue_size_bytes` report this
+same total after the operation, including operations that do not change the queue.
+These values exclude WAL framing and in-memory record metadata. They use the
+existing payload buffer length and are available after recovery without scanning
+records. `TruncateOutcome::evicted_bytes` remains the number of payload bytes
+removed by that operation.
+
 # Non-goals
 
 This is not Kafka. This recordlog is designed for a "small amount of data".

src/lib.rs

@@ -45,6 +45,8 @@ pub struct ResourceUsage {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct AppendOutcome {
+    /// Total size of the queue after this append operation.
+    pub queue_size_bytes: usize,
     /// Position of the last record appended, or `None` for an idempotent no-op
     /// (empty payloads, or `position_opt` already past the queue's head).
     pub last_position: Option<u64>,
@@ -57,9 +59,12 @@ pub struct AppendOutcome {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct TruncateOutcome {
+    /// Total size of the queue after this truncate operation.
+    pub queue_size_bytes: usize,
     /// Number of records evicted from the in-memory queue by this call. This is a delta:
     /// a truncate at a position already covered by a previous truncate reports `0`.
     pub evicted_records: usize,
+    pub evicted_bytes: usize,
     pub wal_bytes_written: u64,
 }
 

src/mem/mod.rs

@@ -3,7 +3,7 @@ mod queues;
 mod rolling_buffer;
 mod summary;
 
-pub(crate) use self::queue::MemQueue;
+pub(crate) use self::queue::{MemEvictedRecords, MemQueue};
 pub(crate) use self::queues::MemQueues;
 pub use self::summary::{QueueSummary, QueuesSummary};
 

src/mem/queue.rs

@@ -15,6 +15,12 @@ struct RecordMeta {
     position: u64,
 }
 
+#[derive(Debug, Default, Clone, Copy)]
+pub(crate) struct MemEvictedRecords {
+    pub num_records: usize,
+    pub num_bytes: usize,
+}
+
 #[derive(Default)]
 pub(crate) struct MemQueue {
     // Concatenated records
@@ -34,6 +40,7 @@ impl MemQueue {
 
     pub fn summary(&self) -> QueueSummary {
         QueueSummary {
+            num_bytes: self.num_bytes(),
             start: self.start_position(),
             end: self.last_position(),
             file_number: self.first_file_number(),
@@ -164,17 +171,21 @@ impl MemQueue {
     ///
     /// If truncating to a future position, make the queue go forward to that position.
     /// Return the number of record removed.
-    pub fn truncate_head(&mut self, truncate_range: RangeToInclusive<u64>) -> usize {
+    pub fn truncate_head(&mut self, truncate_range: RangeToInclusive<u64>) -> MemEvictedRecords {
         let truncate_up_to_pos = truncate_range.end;
         if self.start_position > truncate_up_to_pos {
-            return 0;
+            return MemEvictedRecords::default();
         }
         if truncate_up_to_pos + 1 >= self.next_position() {
             self.start_position = truncate_up_to_pos + 1;
+            let num_bytes = self.concatenated_records.len();
             self.concatenated_records.clear();
-            let record_count = self.record_metas.len();
+            let num_records = self.record_metas.len();
             self.record_metas.clear();
-            return record_count;
+            return MemEvictedRecords {
+                num_records,
+                num_bytes,
+            };
         }
         let first_record_to_keep = self
             .position_to_idx(truncate_up_to_pos + 1)
@@ -188,7 +199,14 @@ impl MemQueue {
         self.concatenated_records
             .truncate_head(..start_offset_to_keep);
         self.start_position = truncate_up_to_pos + 1;
-        first_record_to_keep
+        MemEvictedRecords {
+            num_records: first_record_to_keep,
+            num_bytes: start_offset_to_keep,
+        }
+    }
+
+    pub fn num_bytes(&self) -> usize {
+        self.concatenated_records.len()
     }
 
     pub fn size(&self) -> usize {

src/mem/queues.rs

@@ -4,7 +4,7 @@ use std::ops::{RangeBounds, RangeToInclusive};
 use tracing::{info, warn};
 
 use crate::error::{AlreadyExists, AppendError, MissingQueue};
-use crate::mem::{MemQueue, QueuesSummary};
+use crate::mem::{MemEvictedRecords, MemQueue, QueuesSummary};
 use crate::rolling::FileNumber;
 use crate::Record;
 
@@ -149,12 +149,20 @@ impl MemQueues {
         Ok(self.get_queue(queue)?.next_position())
     }
 
+    pub fn num_bytes(&self, queue: &str) -> Result<usize, MissingQueue> {
+        Ok(self.get_queue(queue)?.num_bytes())
+    }
+
     /// Removes records up to the supplied `position`,
     /// including the position itself.
     ///
     /// If there are no records `<= position`, the method will
     /// not do anything.
-    pub fn truncate(&mut self, queue: &str, position: RangeToInclusive<u64>) -> Option<usize> {
+    pub fn truncate(
+        &mut self,
+        queue: &str,
+        position: RangeToInclusive<u64>,
+    ) -> Option<MemEvictedRecords> {
         if let Ok(queue) = self.get_queue_mut(queue) {
             Some(queue.truncate_head(position))
         } else {

src/mem/summary.rs

@@ -4,6 +4,7 @@ use serde::Serialize;
 
 #[derive(Default, Serialize, Debug)]
 pub struct QueueSummary {
+    pub num_bytes: usize,
     pub start: u64,
     pub end: Option<u64>,
     pub file_number: Option<u64>,

src/multi_record_log.rs

@@ -182,6 +182,7 @@ impl MultiRecordLog {
             // we accept position in the future, and move forward as required.
             if position + 1 == next_position {
                 return Ok(AppendOutcome {
+                    queue_size_bytes: self.in_mem_queues.num_bytes(queue)?,
                     last_position: None,
                     wal_bytes_written: 0,
                 });
@@ -198,6 +199,7 @@ impl MultiRecordLog {
             self.multi_record_spare_buffer = multi_record_spare_buffer;
             // empty transaction: don't persist it
             return Ok(AppendOutcome {
+                queue_size_bytes: self.in_mem_queues.num_bytes(queue)?,
                 last_position: None,
                 wal_bytes_written: 0,
             });
@@ -223,6 +225,7 @@ impl MultiRecordLog {
 
         self.multi_record_spare_buffer = multi_record_spare_buffer;
         Ok(AppendOutcome {
+            queue_size_bytes: self.in_mem_queues.num_bytes(queue)?,
             last_position: Some(max_position),
             wal_bytes_written: num_bytes_written,
         })
@@ -270,11 +273,13 @@ impl MultiRecordLog {
         let evicted_records = self
             .in_mem_queues
             .truncate(queue, truncate_range)
-            .unwrap_or(0);
+            .unwrap_or_default();
         num_bytes_written += self.run_gc_if_necessary()?;
         self.persist_on_policy()?;
         Ok(TruncateOutcome {
-            evicted_records,
+            queue_size_bytes: self.in_mem_queues.num_bytes(queue)?,
+            evicted_records: evicted_records.num_records,
+            evicted_bytes: evicted_records.num_bytes,
             wal_bytes_written: num_bytes_written,
         })
     }

src/proptests.rs

@@ -10,6 +10,14 @@ use tempfile::TempDir;
 use crate::record::{MultiPlexedRecord, MultiRecord};
 use crate::{MultiRecordLog, Record, Serializable};
 
+fn retained_payload_bytes(record_log: &MultiRecordLog, queue: &str) -> usize {
+    record_log
+        .range(queue, ..)
+        .unwrap()
+        .map(|record| record.payload.len())
+        .sum()
+}
+
 struct PropTestEnv {
     tempdir: TempDir,
     record_log: MultiRecordLog,
@@ -56,6 +64,12 @@ impl PropTestEnv {
                 self.truncate(queue, pos);
             }
         }
+        for (queue, summary) in self.record_log.summary().queues {
+            assert_eq!(
+                summary.num_bytes,
+                retained_payload_bytes(&self.record_log, &queue)
+            );
+        }
     }
 
     pub fn reload(&mut self) {
@@ -72,21 +86,23 @@ impl PropTestEnv {
         let state = self.state.get_mut(queue).unwrap();
 
         let new_pos = state.0.end + skip_one_pos as u64;
-        let res = self
+        let outcome = self
             .record_log
             .append_records(queue, Some(new_pos), std::iter::once(&b"BB"[..]))
-            .unwrap()
-            .last_position
             .unwrap();
+        assert_eq!(outcome.last_position, Some(new_pos));
+        assert_eq!(
+            outcome.queue_size_bytes,
+            retained_payload_bytes(&self.record_log, queue)
+        );
 
-        assert!(self
+        let repeated_outcome = self
             .record_log
             .append_records(queue, Some(new_pos), std::iter::once(&b"BB"[..]))
-            .unwrap()
-            .last_position
-            .is_none());
+            .unwrap();
+        assert!(repeated_outcome.last_position.is_none());
+        assert_eq!(repeated_outcome.queue_size_bytes, outcome.queue_size_bytes);
 
-        assert_eq!(new_pos, res);
         state.0.end = new_pos + 1;
         state.1 += 1;
     }
@@ -103,6 +119,10 @@ impl PropTestEnv {
                 std::iter::repeat(&self.block_to_write[..]).take(count as usize),
             )
             .unwrap();
+        assert_eq!(
+            outcome.queue_size_bytes,
+            retained_payload_bytes(&self.record_log, queue)
+        );
 
         if count != 0 {
             let res = outcome.last_position.unwrap();
@@ -114,21 +134,21 @@ impl PropTestEnv {
 
     pub fn truncate(&mut self, queue: &str, pos: u64) {
         let state = self.state.get_mut(queue).unwrap();
+        let previous_bytes = retained_payload_bytes(&self.record_log, queue);
+        let outcome = self.record_log.truncate(queue, ..=pos).unwrap();
+        let remaining_bytes = retained_payload_bytes(&self.record_log, queue);
+        assert_eq!(outcome.queue_size_bytes, remaining_bytes);
+        assert_eq!(outcome.evicted_bytes, previous_bytes - remaining_bytes);
         if state.0.contains(&pos) {
             state.0.start = pos + 1;
-            state.1 -= self
-                .record_log
-                .truncate(queue, ..=pos)
-                .unwrap()
-                .evicted_records as u64;
+            state.1 -= outcome.evicted_records as u64;
         } else if pos >= state.0.end {
             // advance the queue to the position.
             state.0 = (pos + 1)..(pos + 1);
             state.1 = 0;
-            self.record_log.truncate(queue, ..=pos).unwrap();
         } else {
             // should be a no-op
-            self.record_log.truncate(queue, ..=pos).unwrap();
+            assert_eq!(outcome.evicted_records, 0);
         }
     }
 }