segment-buffer

Durable bounded queue with zstd+CBOR segment files, ack-based deletion, and filename-based crash recovery.

Extracted from monitor365, proven on 597M+ events.

Why?

There are many disk-backed queues in the Rust ecosystem, but none offer this combination:

• In-memory bounded buffer that spills to disk on a batch/interval trigger (not always write-through)
• zstd + CBOR compression for efficient storage
• Ack-based deletion — segments are removed only after the consumer confirms receipt
• Filename-based crash recovery — ls the directory and you see the state; no WAL, no metadata DB
• Optional AES-256-GCM encryption at rest via a pluggable SegmentCipher trait
• MPMC — multiple writers and readers via parking_lot::Mutex

Install

cargo add segment-buffer
# optional, for the built-in AES-256-GCM cipher:
cargo add segment-buffer --features encryption

Quickstart

use segment_buffer::{SegmentBuffer, SegmentConfig};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Clone)]
struct MyItem { id: u64 }

let buffer = SegmentBuffer::<MyItem>::open("/tmp/my-buffer", SegmentConfig::default())?;

// Append items (auto-flushes at the batch threshold or flush interval)
let seq = buffer.append(MyItem { id: 1 })?;

// Read items back (from on-disk segments + in-memory pending)
let items = buffer.read_from(0, 1000)?;

// Delete acknowledged items — a segment is removed when its end_seq <= acked_seq
let deleted = buffer.delete_acked(last_acked_seq)?;
# Ok::<(), Box<dyn std::error::Error>>(())

Encryption at rest

Enable the encryption feature and supply any SegmentCipher. The built-in AesGcmCipher writes [12-byte nonce][ciphertext + GCM tag] per segment, byte-compatible with monitor365 so existing encrypted segments read without migration.

use segment_buffer::{AesGcmCipher, SegmentBuffer, SegmentConfig};

let cipher = AesGcmCipher::new(&key); // 32-byte AES-256 key
let buffer = SegmentBuffer::<MyItem>::open(
    "/tmp/my-buffer",
    SegmentConfig { cipher: Some(Box::new(cipher)), ..Default::default() },
)?;

See examples/encrypted.rs for a runnable end-to-end example.

How it works

append(item) ─► unflushed: Vec<T>   (in-memory, inside the Mutex)
                    │
                    ▼   batch full  OR  flush_interval elapsed  OR  flush()
              take() the batch, compute start_seq/end_seq INSIDE the lock
                    │
                    ▼   (lock released — mutex is never held across file I/O)
              CBOR ─► zstd ─► [optional cipher.encrypt]
                    │
                    ▼
              write seg_*.zst.tmp ─► fsync ─► atomic rename to seg_*.zst
                    │
                    ▼   (lock re-acquired)
              approx_disk_bytes += len

read_from(start, limit) scans on-disk segments (sorted by start) then drains the in-memory tail. delete_acked(seq) removes every segment whose end <= seq and advances head_seq. Crash recovery is just: delete .tmp debris, parse the remaining filenames. No WAL, no metadata database.

Backpressure

The crate ships metrics, not policy. store_pressure() returns approx_disk_bytes / max_size_bytes ∈ [0.0, 1.0]; is_overloaded() is > 0.9. You define priority thresholds — see examples/backpressure.rs.

Comparison

Status

v0.1.0 — extracted from monitor365, fully decoupled, zero monitor365 dependencies. See FEATURES.md for the honest capability inventory and ROADMAP.md for direction. Changes live in CHANGELOG.md.

License

Licensed under the Apache License, Version 2.0.