Struct SegmentBuffer 

pub struct SegmentBuffer<T> { /* private fields */ }

Durable bounded queue of T backed by compressed segment files.

Thread-safe via parking_lot::Mutex. All file I/O is synchronous. The mutex is never held across an async boundary because there are no await points.

Create with SegmentBuffer::open, supplying the directory and config.

Implementations

impl<T> SegmentBuffer<T>

where T: Serialize + DeserializeOwned + Clone + Send + 'static,

pub fn open(dir: impl Into<PathBuf>, config: SegmentConfig) -> Result<Self>

Open (or create) a buffer at dir, recovering from any existing segment files.

Recovery is filename-based: it scans the directory to rebuild head_seq / next_seq and deletes leftover .tmp debris. Segment contents are not read until read_from, so a corrupted segment does not fail here — it fails when read.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;

Errors

Returns SegmentError::Io if the directory cannot be created or read.

pub fn append(&self, event: T) -> Result<u64>

Append an item to the buffer. Assigns the next sequence number and auto-flushes if the batch threshold or interval is reached.

Returns the assigned sequence number. The first append returns 0, and the number increments by 1 for each subsequent append.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;

assert_eq!(buf.append(1)?, 0);
assert_eq!(buf.append(2)?, 1);
assert_eq!(buf.append(3)?, 2);

pub fn flush(&self) -> Result<()>

Flush buffered items to a segment file. No-op if nothing is buffered.

Flushing is also triggered automatically by append when the batch threshold (max_batch_events) or interval (flush_interval_secs) is reached. Call this explicitly when you need durability before a known threshold.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;
buf.append(1)?;
buf.append(2)?;

buf.flush()?; // items now durable on disk
assert_eq!(buf.pending_count(), 2);

pub fn read_from(&self, start_seq: u64, limit: usize) -> Result<Vec<T>>

Read up to limit items starting from start_seq (inclusive).

Reads from both on-disk segment files and in-memory pending items. Items are returned in ascending sequence order.

Passing limit = 0 returns an empty Vec without scanning.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;
buf.append(10)?;
buf.append(20)?;
buf.append(30)?;
buf.flush()?;

let items = buf.read_from(0, 100)?;
assert_eq!(items, vec![10, 20, 30]);

// start_seq skips already-read items:
let tail = buf.read_from(2, 100)?;
assert_eq!(tail, vec![30]);

pub fn delete_acked(&self, acked_seq: u64) -> Result<usize>

Delete all on-disk segment files whose items are fully covered by acked_seq.

A segment is deleted when its end_seq <= acked_seq. Returns the number of segment files removed.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;
for i in 0..5u64 {
    buf.append(i)?;
}
buf.flush()?;

// Consumer has processed sequence 0..=4; acknowledge them:
let removed = buf.delete_acked(4)?;
assert_eq!(removed, 1); // one segment file deleted
assert_eq!(buf.pending_count(), 0);

Limitation

Acknowledgement only removes flushed segment files. Items still held in the in-memory pending batch have no segment file to delete, so they remain readable (and counted by SegmentBuffer::pending_count) until they are flushed and acknowledged in a later call. head_seq is clamped so it never advances past the pending window, keeping the backlog count honest.

pub fn latest_sequence(&self) -> u64

The highest sequence number assigned (or 0 if buffer is empty).

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;

assert_eq!(buf.latest_sequence(), 0);
buf.append(7)?;
assert_eq!(buf.latest_sequence(), 0);
buf.append(8)?;
assert_eq!(buf.latest_sequence(), 1);

pub fn pending_count(&self) -> u64

Total items waiting in the buffer (on-disk + in-memory pending).

Equivalent to latest_sequence() - head_seq + 1 when non-empty, 0 when empty. Decreases as delete_acked removes files.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;

assert_eq!(buf.pending_count(), 0);
buf.append(1)?;
buf.append(2)?;
assert_eq!(buf.pending_count(), 2);
buf.flush()?;
assert_eq!(buf.pending_count(), 2); // still pending until acked
buf.delete_acked(1)?;
assert_eq!(buf.pending_count(), 0);

pub fn len(&self) -> u64

Standard len alias for pending_count.

Provided so SegmentBuffer reads like a normal collection at the call site (buf.len(), buf.is_empty()). Same value as pending_count(), kept as u64 because the buffer is proven beyond usize::MAX on 32-bit targets (597M+ events in monitor365).

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;
assert!(buf.is_empty());
buf.append(7)?;
assert_eq!(buf.len(), 1);
assert!(!buf.is_empty());

pub fn is_empty(&self) -> bool

true when there are no items waiting in the buffer (on-disk or in-memory). Equivalent to pending_count() == 0.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;
assert!(buf.is_empty());

pub fn store_pressure(&self) -> f32

Disk usage pressure as a value between 0.0 and 1.0.

Use this to implement your own admission/backpressure policy (e.g. reject low-priority items above 0.90, reject standard items above 0.95). Returns 0.0 when max_size_bytes == 0 (limit disabled).

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let mut cfg = SegmentConfig::default();
cfg.max_size_bytes = 1000; // tiny limit so pressure is observable
let buf: SegmentBuffer<u64> = SegmentBuffer::open(dir.path(), cfg)?;

assert!(buf.store_pressure() < 0.1);

pub fn is_overloaded(&self) -> bool

True when disk usage exceeds 90% of the configured limit.

Convenience wrapper around store_pressure() > 0.9.

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;

assert!(!buf.is_overloaded());

pub fn stats(&self) -> BufferStats

Capture a consistent snapshot of buffer state under a single lock.

Cheaper and more consistent than calling pending_count, latest_sequence, store_pressure etc. individually (which each take the mutex and could observe a flush/delete between calls).

Example

use segment_buffer::{SegmentBuffer, SegmentConfig};
use tempfile::tempdir;

let dir = tempdir()?;
let buf: SegmentBuffer<u64> =
    SegmentBuffer::open(dir.path(), SegmentConfig::default())?;
buf.append(1)?;
buf.append(2)?;

let snapshot = buf.stats();
assert_eq!(snapshot.pending_count, 2);
assert_eq!(snapshot.next_sequence, 2);
assert!(snapshot.store_pressure < 0.01);