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/lib.rs b/src/lib.rs index 7ecbbdd..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,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, } diff --git a/src/mem/mod.rs b/src/mem/mod.rs index 00dd137..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::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 9b9c3a7..c482639 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 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) -> usize { + 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 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 { diff --git a/src/mem/queues.rs b/src/mem/queues.rs index ec775f8..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::{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 { + 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) -> 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/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 5f67989..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, }) @@ -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, }) } 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 7306d17..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> { @@ -270,6 +271,169 @@ 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); + 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(); + + 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();