FlowDB

A high-performance time-series storage engine written in Rust, powered by an LSM-tree architecture with WAL, SSTables, and Bloom filters.

Benchmark Results

FlowDB vs RocksDB comparison (100K records, 128B values, batch=100, release build, Apple M-series):

cargo run --release --example flowdb-vs-rocksdb

Features

• LSM-tree storage with WAL (write-ahead log) for crash recovery
• Per-record WAL checksums — corruption detected on replay, bad records rejected
• Config validation — invalid configs (e.g. time_bucket_secs=0) rejected at startup instead of crashing
• Frozen memtable backpressure — writes stall when flush can't keep up, preventing unbounded memory growth
• HTTP batch & size limits — max 10 000 records/batch, 4 KB keys, 64 KB values (matches UDP protocol)
• Graceful shutdown — SIGTERM/SIGINT triggers flush + clean exit
• Bounded UDP rate limiter — cap at 100 K entries with periodic idle cleanup
• SST reader eviction — stale readers removed after GC/compaction
• Lazy scan iterator (RocksDB-style ScanIterator) for bounded-memory range scans
• get_latest(key) for retrieving the most recent record by key
• Bloom filters for fast point query negative checks
• Dual compression: lz4 for flush (speed), zstd for compaction (ratio)
• Buffered WAL writes (256KB buffer) for reduced syscall overhead
• WAL pre-encoding outside the write lock for better concurrency
• Time-bucketed block index with binary search
• LRU block cache (64 shards, powered by lru crate) with true LRU eviction
• Log-based active memtable (Vec push) with on-freeze BTreeMap conversion
• Hash-based memtable index (avoids key clone on insert)
• Zero-copy owned write path (write_batch_owned)
• Synchronous write path (write_batch_sync) for non-async callers
• Size-tiered compaction with streaming heap merge (low memory footprint)
• Range tombstones (delete_range) for efficient bulk key-range deletion
• Garbage collection (TTL expiry), and point deletes
• HTTP + UDP ingest APIs (feature-gated under server)
• Admin dashboard, metrics, and stats endpoints
• API key authentication
• Config file support (TOML)

Cargo Features

# Full server (default)
flowdb = "0.2"

# Embedded engine only (no HTTP/UDP/auth, smaller binary)
flowdb = { version = "0.2", default-features = false }

Quick Start

Build

cargo build --release

Run Server

# With defaults
./target/release/flowdb-server

# With CLI flags
./target/release/flowdb-server --data-dir ./data --http-addr 0.0.0.0:8080 --api-key mysecret

# With a config file
./target/release/flowdb-server --config flowdb.toml

Config File (TOML)

http_addr = "0.0.0.0:8080"
udp_addr = "0.0.0.0:9090"
api_keys = ["my-api-key"]
max_udp_packet_size = 1400

[engine]
data_dir = "./data"
memtable_size_mb = 64
block_size = 8192
zstd_level = 3
bloom_bits_per_key = 10
wal_segment_size_mb = 64
compaction_threshold = 2
max_frozen_memtables = 2
flush_interval_ms = 1000
gc_interval_secs = 3600
default_ttl_secs = 86400
time_bucket_secs = 3600
index_memory_budget_mb = 256
block_cache_capacity_mb = 128
create_if_missing = true

Configuration Reference

Server Config

Engine Config

HTTP API

Write Records

# JSON write
curl -X POST http://localhost:8080/write \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer my-api-key" \
  -d '{"records": [{"key": "sensor.temp", "ts": 1700000000, "value": "22.5"}]}'

# Binary write (UDP frame format)
curl -X POST http://localhost:8080/write/binary \
  -H "Authorization: Bearer my-api-key" \
  --data-binary @frame.bin

Query Records

# Prefix query
curl "http://localhost:8080/query?prefix=sensor.temp"

# Key range query
curl "http://localhost:8080/query?key_start=sensor.temp&key_end=sensor.tempz"

# Time range query
curl "http://localhost:8080/query?ts_start=1700000000&ts_end=1700003600"

# Prefix + time range
curl "http://localhost:8080/query?prefix=sensor&ts_start=1700000000&ts_end=1700003600"

Delete / Patch

# Delete a single record
curl -X DELETE "http://localhost:8080/record?key=sensor.temp&ts=1700000000"

# Delete a key range (range tombstone, all keys in [key_start, key_end))
curl -X DELETE "http://localhost:8080/range?key_start=sensor.a&key_end=sensor.z"

curl -X PATCH http://localhost:8080/record \
  -H "Content-Type: application/json" \
  -d '{"key": "sensor.temp", "ts": 1700000000, "value": "23.1"}'

Admin & Monitoring

curl http://localhost:8080/health          # Health check
curl http://localhost:8080/stats           # Engine stats (JSON)
curl http://localhost:8080/metrics         # Prometheus-style metrics
curl http://localhost:8080/admin           # Web dashboard

curl -X POST http://localhost:8080/admin/flush     # Force memtable flush
curl -X POST http://localhost:8080/admin/compact   # Trigger compaction
curl -X POST http://localhost:8080/admin/gc        # Run garbage collection

Rust Library Usage

use flowdb::{
    Engine, Config, Record, Query, ScanRange, ScanIterator, ReadOptions,
};

let config = Config::default();
let engine = Engine::open(config).await?;

// Write
let records = vec![Record {
    key: "sensor.temp".into(),
    ts: 1700000000,
    expire_at: i64::MAX,
    value: b"22.5".to_vec(),
}];
engine.write_batch(&records).await?;

// Zero-copy write (moves key/value, no clones)
engine.write_batch_owned(records).await?;

// Delete a range of keys (range tombstone)
engine.delete_range("sensor.a", "sensor.z").await?;

// ── Eager query (returns Vec<Record>) ──
let results = engine.query_by_prefix("sensor.").await?;
let results = engine.query_prefix_time_range("sensor.", 1700000000, 1700003600).await?;

// ── Lazy scan iterator (RocksDB-style, recommended for large ranges) ──

// Prefix scan — yields records one-at-a-time, bounded memory
let iter: ScanIterator = engine.scan_prefix("sensor.")?;
for result in iter {
    let record = result?;
    println!("{} @ {} = {:?}", record.key, record.ts, record.value);
}

// Prefix + time range scan
let iter = engine.scan_prefix_time_range("sensor.", 1700000000, 1700003600)?;

// Full key+time range scan with ReadOptions
let iter = engine.scan_opt(
    ScanRange::prefix_time_range("sensor.", 1700000000, 1700003600),
    &ReadOptions { fill_cache: true, verify_checksums: true },
)?;

// Key range scan
let iter = engine.scan(ScanRange::key_range("sensor.a", "sensor.z"))?;

// Full scan
let iter = engine.scan(ScanRange::all())?;

// Take only first N records (lazy — doesn't read the rest)
let first_10: Vec<Record> = engine
    .scan_prefix("sensor.")?
    .take(10)
    .map(|r| r.unwrap())
    .collect();

// ── get_latest: retrieve the most recent record for a key ──
let latest = engine.get_latest("sensor.temp").await?; // Option<Record>

// Shutdown
engine.shutdown().await?;

ScanRange Builders

Engine API Reference

Architecture

Write Path:
  Client → encode_batch() (outside lock) → WriteWorker mutex → WAL (buffered) + MemTable
                                                                          ↓ (when full)
                                                                      Flush → SSTable

Read Path:
  Query → Active MemTable → Frozen MemTables → Block Index → Bloom Filter → SSTable (LRU cached)
  Scan   → ScanIterator (lazy merge heap over memtable + SST block sources)

Background:
  Flush:    MemTable → SSTable (sorted, lz4-compressed, bloom-filtered)
  Compact:  Size-tiered merge (streaming heap merge, zstd-compressed)
  GC:       Remove fully-expired SSTables
  Delete:   Point deletes (tombstones) + Range deletes (range tombstones)

Benchmarks

# Stress test
cargo run --release --bin flowdb-stress

# FlowDB vs RocksDB comparison
cargo run --release --example flowdb-vs-rocksdb

License

MIT.