From e79bed2dbe4077f02479d073574c9f7a7ab03b85 Mon Sep 17 00:00:00 2001 From: "nadav.govari" Date: Wed, 30 Sep 2026 10:56:38 -0400 Subject: [PATCH 1/3] Return num bytes on truncation --- src/lib.rs | 1 + src/mem/mod.rs | 2 +- src/mem/queue.rs | 21 +++++++++++++++++---- src/mem/queues.rs | 8 ++++++-- src/multi_record_log.rs | 5 +++-- src/tests.rs | 35 +++++++++++++++++++++++++++++++++++ 6 files changed, 63 insertions(+), 9 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 7ecbbdd..5d2bc2e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -60,6 +60,7 @@ pub struct TruncateOutcome { /// 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, } diff --git a/src/mem/mod.rs b/src/mem/mod.rs index 00dd137..63175ce 100644 --- a/src/mem/mod.rs +++ b/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::{EvictedRecords, MemQueue}; pub(crate) use self::queues::MemQueues; pub use self::summary::{QueueSummary, QueuesSummary}; diff --git a/src/mem/queue.rs b/src/mem/queue.rs index 9b9c3a7..fe8ea92 100644 --- a/src/mem/queue.rs +++ b/src/mem/queue.rs @@ -15,6 +15,12 @@ struct RecordMeta { position: u64, } +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct EvictedRecords { + pub num_records: usize, + pub num_bytes: usize, +} + #[derive(Default)] pub(crate) struct MemQueue { // Concatenated records @@ -164,17 +170,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) -> usize { + pub fn truncate_head(&mut self, truncate_range: RangeToInclusive) -> EvictedRecords { let truncate_up_to_pos = truncate_range.end; if self.start_position > truncate_up_to_pos { - return 0; + return EvictedRecords::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(); self.record_metas.clear(); - return record_count; + return EvictedRecords { + num_records: record_count, + num_bytes, + }; } let first_record_to_keep = self .position_to_idx(truncate_up_to_pos + 1) @@ -188,7 +198,10 @@ impl MemQueue { self.concatenated_records .truncate_head(..start_offset_to_keep); self.start_position = truncate_up_to_pos + 1; - first_record_to_keep + EvictedRecords { + num_records: first_record_to_keep, + num_bytes: start_offset_to_keep, + } } pub fn size(&self) -> usize { diff --git a/src/mem/queues.rs b/src/mem/queues.rs index ec775f8..0bcf306 100644 --- a/src/mem/queues.rs +++ b/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::{EvictedRecords, MemQueue, QueuesSummary}; use crate::rolling::FileNumber; use crate::Record; @@ -154,7 +154,11 @@ impl MemQueues { /// /// If there are no records `<= position`, the method will /// not do anything. - pub fn truncate(&mut self, queue: &str, position: RangeToInclusive) -> Option { + pub fn truncate( + &mut self, + queue: &str, + position: RangeToInclusive, + ) -> Option { if let Ok(queue) = self.get_queue_mut(queue) { Some(queue.truncate_head(position)) } else { diff --git a/src/multi_record_log.rs b/src/multi_record_log.rs index 5f67989..aa85f50 100644 --- a/src/multi_record_log.rs +++ b/src/multi_record_log.rs @@ -270,11 +270,12 @@ 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, + evicted_records: evicted_records.num_records, + evicted_bytes: evicted_records.num_bytes, wal_bytes_written: num_bytes_written, }) } diff --git a/src/tests.rs b/src/tests.rs index 7306d17..ae3ef59 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -270,6 +270,41 @@ fn test_multi_insert_truncate() { } } +#[test] +fn test_truncate_reports_evicted_bytes() { + let tempdir = tempfile::tempdir().unwrap(); + { + let mut multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); + multi_record_log.create_queue("queue").unwrap(); + multi_record_log + .append_records( + "queue", + None, + [b"hello".as_slice(), b"happy", b"tax", b"payer"].into_iter(), + ) + .unwrap(); + + let truncate_outcome = multi_record_log.truncate("queue", ..=1).unwrap(); + assert_eq!(truncate_outcome.evicted_records, 2); + assert_eq!(truncate_outcome.evicted_bytes, 10); + + let truncate_outcome = multi_record_log.truncate("queue", ..=1).unwrap(); + assert_eq!(truncate_outcome.evicted_records, 0); + assert_eq!(truncate_outcome.evicted_bytes, 0); + } + { + let mut multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); + + let truncate_outcome = multi_record_log.truncate("queue", ..=2).unwrap(); + assert_eq!(truncate_outcome.evicted_records, 1); + assert_eq!(truncate_outcome.evicted_bytes, 3); + + let truncate_outcome = multi_record_log.truncate("queue", ..=10).unwrap(); + assert_eq!(truncate_outcome.evicted_records, 1); + assert_eq!(truncate_outcome.evicted_bytes, 5); + } +} + #[test] fn test_truncate_range_correct_pos() { let tempdir = tempfile::tempdir().unwrap(); From 853b5365a7f0b72f43e14f55192298714513c8e4 Mon Sep 17 00:00:00 2001 From: "nadav.govari" Date: Wed, 7 Oct 2026 10:01:02 -0400 Subject: [PATCH 2/3] Return queue sizes on operation outcomes and queue summary --- src/lib.rs | 4 ++ src/mem/queue.rs | 5 ++ src/mem/queues.rs | 4 ++ src/mem/summary.rs | 1 + src/multi_record_log.rs | 4 ++ src/proptests.rs | 50 +++++++++++----- src/tests.rs | 129 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 182 insertions(+), 15 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 5d2bc2e..191595c 100644 --- a/src/lib.rs +++ b/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, @@ -57,6 +59,8 @@ 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, diff --git a/src/mem/queue.rs b/src/mem/queue.rs index fe8ea92..4cffb50 100644 --- a/src/mem/queue.rs +++ b/src/mem/queue.rs @@ -40,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(), @@ -204,6 +205,10 @@ impl MemQueue { } } + pub fn num_bytes(&self) -> usize { + self.concatenated_records.len() + } + pub fn size(&self) -> usize { self.concatenated_records.len() + self.record_metas.len() * std::mem::size_of::() diff --git a/src/mem/queues.rs b/src/mem/queues.rs index 0bcf306..f17fd68 100644 --- a/src/mem/queues.rs +++ b/src/mem/queues.rs @@ -149,6 +149,10 @@ impl MemQueues { Ok(self.get_queue(queue)?.next_position()) } + pub fn num_bytes(&self, queue: &str) -> Result { + Ok(self.get_queue(queue)?.num_bytes()) + } + /// Removes records up to the supplied `position`, /// including the position itself. /// diff --git a/src/mem/summary.rs b/src/mem/summary.rs index 71f62b2..65fa5ce 100644 --- a/src/mem/summary.rs +++ b/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, pub file_number: Option, diff --git a/src/multi_record_log.rs b/src/multi_record_log.rs index aa85f50..e9ac51a 100644 --- a/src/multi_record_log.rs +++ b/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, }) @@ -274,6 +277,7 @@ impl MultiRecordLog { num_bytes_written += self.run_gc_if_necessary()?; self.persist_on_policy()?; Ok(TruncateOutcome { + 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, diff --git a/src/proptests.rs b/src/proptests.rs index 9a7b509..c7cf929 100644 --- a/src/proptests.rs +++ b/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); } } } diff --git a/src/tests.rs b/src/tests.rs index ae3ef59..f44e01b 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -2,6 +2,7 @@ use std::borrow::Cow; use bytes::Buf; +use crate::error::{AppendError, TruncateError}; use crate::{MultiRecordLog, Record}; fn read_all_records<'a>(multi_record_log: &'a MultiRecordLog, queue: &str) -> Vec> { @@ -287,10 +288,12 @@ fn test_truncate_reports_evicted_bytes() { let truncate_outcome = multi_record_log.truncate("queue", ..=1).unwrap(); assert_eq!(truncate_outcome.evicted_records, 2); assert_eq!(truncate_outcome.evicted_bytes, 10); + assert_eq!(truncate_outcome.queue_size_bytes, 8); let truncate_outcome = multi_record_log.truncate("queue", ..=1).unwrap(); assert_eq!(truncate_outcome.evicted_records, 0); assert_eq!(truncate_outcome.evicted_bytes, 0); + assert_eq!(truncate_outcome.queue_size_bytes, 8); } { let mut multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); @@ -298,13 +301,139 @@ fn test_truncate_reports_evicted_bytes() { let truncate_outcome = multi_record_log.truncate("queue", ..=2).unwrap(); assert_eq!(truncate_outcome.evicted_records, 1); assert_eq!(truncate_outcome.evicted_bytes, 3); + assert_eq!(truncate_outcome.queue_size_bytes, 5); let truncate_outcome = multi_record_log.truncate("queue", ..=10).unwrap(); assert_eq!(truncate_outcome.evicted_records, 1); assert_eq!(truncate_outcome.evicted_bytes, 5); + assert_eq!(truncate_outcome.queue_size_bytes, 0); } } +#[test] +fn test_queue_bytes_after_writes_truncation_and_recovery() { + let tempdir = tempfile::tempdir().unwrap(); + let mut multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); + multi_record_log.create_queue("queue").unwrap(); + multi_record_log.create_queue("other-queue").unwrap(); + let outcome = multi_record_log + .append_record("other-queue", None, b"untouched".as_slice()) + .unwrap(); + assert_eq!(outcome.queue_size_bytes, 9); + assert_eq!(multi_record_log.summary().queues["queue"].num_bytes, 0); + + let mut retained_bytes = 0; + for iteration in 0..32 { + let outcome = multi_record_log + .append_records( + "queue", + None, + [b"hello".as_slice(), b"happy", b"tax", b"payer"].into_iter(), + ) + .unwrap(); + assert_eq!(outcome.queue_size_bytes, retained_bytes + 18); + assert_eq!(outcome.last_position, Some(iteration * 4 + 3)); + + let outcome = multi_record_log + .append_records("queue", None, std::iter::empty::<&[u8]>()) + .unwrap(); + assert_eq!(outcome.last_position, None); + assert_eq!(outcome.queue_size_bytes, retained_bytes + 18); + assert_eq!(outcome.wal_bytes_written, 0); + + let outcome = multi_record_log + .append_record("queue", Some(iteration * 4 + 3), b"ignored".as_slice()) + .unwrap(); + assert_eq!(outcome.last_position, None); + assert_eq!(outcome.queue_size_bytes, retained_bytes + 18); + assert_eq!(outcome.wal_bytes_written, 0); + + let outcome = multi_record_log + .truncate("queue", ..=iteration * 4 + 1) + .unwrap(); + assert_eq!(outcome.evicted_bytes, retained_bytes + 10); + assert_eq!(outcome.queue_size_bytes, 8); + retained_bytes = 8; + + let outcome = multi_record_log + .truncate("queue", ..=iteration * 4 + 1) + .unwrap(); + assert_eq!(outcome.evicted_bytes, 0); + assert_eq!(outcome.queue_size_bytes, 8); + + drop(multi_record_log); + multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); + let summary = multi_record_log.summary(); + assert_eq!(summary.queues["queue"].num_bytes, 8); + assert_eq!(summary.queues["other-queue"].num_bytes, 9); + assert_eq!( + multi_record_log + .range("queue", ..) + .unwrap() + .collect::>(), + [ + Record::new(iteration * 4 + 2, b"tax"), + Record::new(iteration * 4 + 3, b"payer"), + ] + ); + let actual_bytes: usize = multi_record_log + .range("queue", ..) + .unwrap() + .map(|record| record.payload.len()) + .sum(); + assert_eq!(summary.queues["queue"].num_bytes, actual_bytes); + } + + let outcome = multi_record_log.truncate("queue", ..=127).unwrap(); + assert_eq!(outcome.queue_size_bytes, 0); + drop(multi_record_log); + let multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); + assert_eq!(multi_record_log.summary().queues["queue"].num_bytes, 0); +} + +#[test] +fn test_rejected_operations_preserve_queue_and_wal() { + let tempdir = tempfile::tempdir().unwrap(); + let mut multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); + multi_record_log.create_queue("queue").unwrap(); + multi_record_log + .append_records("queue", None, [b"hello".as_slice(), b"world!"].into_iter()) + .unwrap(); + let usage_before = multi_record_log.resource_usage(); + + assert!(matches!( + multi_record_log.append_record("missing", None, b"hello".as_slice()), + Err(AppendError::MissingQueue(queue)) if queue == "missing" + )); + assert!(matches!( + multi_record_log.append_records("missing", None, std::iter::empty::<&[u8]>()), + Err(AppendError::MissingQueue(queue)) if queue == "missing" + )); + assert!(matches!( + multi_record_log.truncate("missing", ..=10), + Err(TruncateError::MissingQueue(queue)) if queue == "missing" + )); + assert!(matches!( + multi_record_log.append_record("queue", Some(0), b"rejected".as_slice()), + Err(AppendError::Past) + )); + assert_eq!(multi_record_log.resource_usage(), usage_before); + assert_eq!(multi_record_log.summary().queues["queue"].num_bytes, 11); + assert_eq!( + read_all_records(&multi_record_log, "queue"), + [b"hello".as_slice(), b"world!"] + ); + + drop(multi_record_log); + let multi_record_log = MultiRecordLog::open(tempdir.path()).unwrap(); + assert!(!multi_record_log.queue_exists("missing")); + assert_eq!(multi_record_log.summary().queues["queue"].num_bytes, 11); + assert_eq!( + read_all_records(&multi_record_log, "queue"), + [b"hello".as_slice(), b"world!"] + ); +} + #[test] fn test_truncate_range_correct_pos() { let tempdir = tempfile::tempdir().unwrap(); From 582ddb131181a053635e6904d3190491a3ee556c Mon Sep 17 00:00:00 2001 From: "nadav.govari" Date: Wed, 7 Oct 2026 10:04:18 -0400 Subject: [PATCH 3/3] PR comments --- README.md | 8 ++++++++ src/mem/mod.rs | 2 +- src/mem/queue.rs | 14 +++++++------- src/mem/queues.rs | 4 ++-- 4 files changed, 18 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index c4352c6..73c9082 100644 --- a/README.md +++ b/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". diff --git a/src/mem/mod.rs b/src/mem/mod.rs index 63175ce..c1badd0 100644 --- a/src/mem/mod.rs +++ b/src/mem/mod.rs @@ -3,7 +3,7 @@ mod queues; mod rolling_buffer; mod summary; -pub(crate) use self::queue::{EvictedRecords, MemQueue}; +pub(crate) use self::queue::{MemEvictedRecords, MemQueue}; pub(crate) use self::queues::MemQueues; pub use self::summary::{QueueSummary, QueuesSummary}; diff --git a/src/mem/queue.rs b/src/mem/queue.rs index 4cffb50..c482639 100644 --- a/src/mem/queue.rs +++ b/src/mem/queue.rs @@ -16,7 +16,7 @@ struct RecordMeta { } #[derive(Debug, Default, Clone, Copy)] -pub(crate) struct EvictedRecords { +pub(crate) struct MemEvictedRecords { pub num_records: usize, pub num_bytes: usize, } @@ -171,19 +171,19 @@ 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) -> EvictedRecords { + pub fn truncate_head(&mut self, truncate_range: RangeToInclusive) -> MemEvictedRecords { let truncate_up_to_pos = truncate_range.end; if self.start_position > truncate_up_to_pos { - return EvictedRecords::default(); + 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 EvictedRecords { - num_records: record_count, + return MemEvictedRecords { + num_records, num_bytes, }; } @@ -199,7 +199,7 @@ impl MemQueue { self.concatenated_records .truncate_head(..start_offset_to_keep); self.start_position = truncate_up_to_pos + 1; - EvictedRecords { + MemEvictedRecords { num_records: first_record_to_keep, num_bytes: start_offset_to_keep, } diff --git a/src/mem/queues.rs b/src/mem/queues.rs index f17fd68..d985cad 100644 --- a/src/mem/queues.rs +++ b/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::{EvictedRecords, MemQueue, QueuesSummary}; +use crate::mem::{MemEvictedRecords, MemQueue, QueuesSummary}; use crate::rolling::FileNumber; use crate::Record; @@ -162,7 +162,7 @@ impl MemQueues { &mut self, queue: &str, position: RangeToInclusive, - ) -> Option { + ) -> Option { if let Ok(queue) = self.get_queue_mut(queue) { Some(queue.truncate_head(position)) } else {