Push events to the parent (Click); beava routes matching records through the derivation chain and updates downstream tables for free. You never push to a derived event directly.
Start with the feature your application actually needs. Add complexity only when the use case requires it.
Test your pipeline
-
Before you push real traffic at it, drive the pipeline with a fixture and assert what comes out. bv.test.fixture spins up an in-process server, lets you push synthetic events, and exposes the same get / batch_get API you use in production:
+
Before you push real traffic at it, drive the pipeline with a fixture and assert what comes out. bv.test.fixture is a pytest-shaped generator: yield from it inside your own @pytest.fixture and it spins up an in-process server, exposing the same get / batch_get API you use in production:
The fixture controls the clock with f.advance_time(...), so you can write deterministic tests for windowed aggregations without sleeping. Run with pytest; no extra config required.
+
The fixture yields a regular bv.App, so the test reads exactly like production code — register, push, get. Run with pytest; no extra config required.
Go deeper
Pipelines are written in Python. The engineer who understands the feature can also ship it; the runtime is a single binary that you talk to over HTTP.
# 1. Register the pipeline{'\n'}
-curl -X POST localhost:8080/pipelines -d @pipeline.py{'\n'}{'\n'}
+curl -X POST localhost:8080/register -d @pipeline.json{'\n'}{'\n'}
# 2. Send a watch{'\n'}
-curl -X POST localhost:8080/events/Watch \
+curl -X POST localhost:8080/push \
{' '}-H 'content-type: application/json' \
-{' '}-d {`'{"user_id":"u1","video_id":"v_abc","watch_seconds":42}'`}{'\n'}{'\n'}
+{' '}-d {`'{"event":"Watch","data":{"user_id":"u1","video_id":"v_abc","watch_seconds":42}}'`}{'\n'}{'\n'}
# 3. Ask what this user cares about in the last hour{'\n'}
-curl localhost:8080/features/UserTagAffinity?user_id=u1{'\n'}
+curl localhost:8080/get/UserTagAffinity/u1{'\n'}
{`# => [{"tag":"woodworking","weight_1h":42.0,...}, ...]`}
# 1. Register the pipeline{'\n'}
-curl -X POST localhost:8080/pipelines -d @pipeline.py{'\n'}{'\n'}
+curl -X POST localhost:8080/register -d @pipeline.json{'\n'}{'\n'}
# 2. Send a watch{'\n'}
-curl -X POST localhost:8080/events/Watch \
+curl -X POST localhost:8080/push \
{' '}-H 'content-type: application/json' \
-{' '}-d {`'{"user_id":"u1","video_id":"v_abc","watch_seconds":42}'`}{'\n'}{'\n'}
+{' '}-d {`'{"event":"Watch","data":{"user_id":"u1","video_id":"v_abc","watch_seconds":42}}'`}{'\n'}{'\n'}
# 3. Ask what this user cares about in the last hour{'\n'}
-curl localhost:8080/features/UserTagAffinity?user_id=u1{'\n'}
+curl localhost:8080/get/UserTagAffinity/u1{'\n'}
{`# => [{"tag":"woodworking","weight_1h":42.0,...}, ...]`}
-{'# curl (recommended) — fetches the platform wheel from latest GH Release\n'}
+{'# pip (recommended) — fetches the platform wheel + bundled server binary\n'}
{'$ pip install beava\n'}
{'\n'}
+{'# brew (macOS / Linux) — server binary on PATH the Homebrew way\n'}
+{'$ brew install beava-dev/beava/beava\n'}
+{'\n'}
{'# docker (zero deps on host — :edge tag rebuilt from main on every push)\n'}
{'$ docker run -p 8080:8080 -p 8081:8081 beavadev/beava:edge\n'}
{'\n'}
@@ -690,14 +693,15 @@
- // with the user_ids visible on screen, gets back the metrics, renders.
+ // actual rows in state. Teaches: your app calls POST /batch_get with
+ // {requests:[{table,key},...]}, gets back {results:[...]}, renders.
const QueryPanel = ({ table, userIds, rowsList, metricKeys, empty, onFirstRun, autoRefreshMs }) => {
// Show all visible users (up to 6 to keep the payload readable).
const ids = userIds.slice(0, 6);
- const curl = `curl -X POST https://your-beava/batch/${table} \\
+ const reqBody = JSON.stringify({ requests: ids.map(k => ({ table, key: k })) });
+ const curl = `curl -X POST https://your-beava/batch_get \\
-H 'Content-Type: application/json' \\
- -d '${JSON.stringify(ids)}'`;
+ -d '${reqBody}'`;
// User-triggered query: the response only appears after "Run query"
// is clicked. Once clicked, auto-refresh kicks in so the snapshot
@@ -720,20 +724,23 @@
{
const rowByUid = new Map(rowsRef.current);
- const obj = {};
+ const results = [];
for (const uid of idsRef.current) {
const r = rowByUid.get(uid);
- if (!r) { obj[uid] = null; continue; }
+ // Server returns a flat feature map per result, in request order
+ // (no table/entity_id/key wrapping).
const o = {};
- for (const k of metricKeys) {
- if (k === 'views_total') o.views_total = r.views_total;
- else if (k === 'views_24h') o.views_24h = r.views_24h;
- else if (k === 'distinct_cats_7d') o.distinct_cats_7d = r.cats.size;
- else if (k === 'last_seen') o.last_seen = new Date(r.lastTsMs).toISOString().replace(/\.\d{3}Z$/, 'Z');
+ if (r) {
+ for (const k of metricKeys) {
+ if (k === 'views_total') o.views_total = r.views_total;
+ else if (k === 'views_24h') o.views_24h = r.views_24h;
+ else if (k === 'distinct_cats_7d') o.distinct_cats_7d = r.cats.size;
+ else if (k === 'last_seen') o.last_seen = new Date(r.lastTsMs).toISOString().replace(/\.\d{3}Z$/, 'Z');
+ }
}
- obj[uid] = o;
+ results.push(o);
}
- return JSON.stringify(obj, null, 2);
+ return JSON.stringify({ results }, null, 2);
}, [metricKeys]);
const doFetch = React.useCallback((opts = {}) => {
@@ -777,7 +784,7 @@
- ↓ how your app reads it: POST /batch/{table}
+ ↓ how your app reads it: POST /batch_get
pip install beava drops a platform wheel from PyPI. The wheel ships the SDK and the Rust server binary together (~4 MB, polars / ruff / uv pattern). If you'd rather run the server in a container, Docker is the alternative.
+
Three ways in, pick whichever fits your stack. pip install beava drops a platform wheel from PyPI that ships the SDK and the Rust server binary together (~4 MB, polars / ruff / uv pattern). brew install puts the server binary on your PATH the Homebrew way. docker run hands you a container on :8080 with a persistent volume.
+
@@ -164,6 +165,22 @@
01 Install beava
The wheel comes from PyPI. The bundled beava binary lands in your Python user-scripts dir (~/.local/bin/beava on Linux, ~/Library/Python/<ver>/bin/beava on macOS), so embed mode (bv.App() with no URL) finds it without extra setup. Pin a specific version with pip install beava==0.0.0. Add --memory-only to skip the WAL entirely.
+
+
+
+ terminal
+
+
+
brew install beava-dev/beava/beava
+
+# server binary on PATH; `beava` is ready to run
+beava --data-dir ./.beava/
+# → HTTP listen : 127.0.0.1:8080
+# → TCP listen : 127.0.0.1:8081 (enabled=true)
+
+
The formula installs the beava server binary into your Homebrew prefix (/opt/homebrew/bin/beava on Apple Silicon, /usr/local/bin/beava on Intel). Roll forward with brew upgrade beava. The Python SDK still comes from pip install beava; install both if you want embed mode and the client in the same env.
+
+
diff --git a/crates/beava-core/src/agg_compile.rs b/crates/beava-core/src/agg_compile.rs
index d551678e..3ea0a5a2 100644
--- a/crates/beava-core/src/agg_compile.rs
+++ b/crates/beava-core/src/agg_compile.rs
@@ -553,13 +553,28 @@ pub fn compile_aggregations_from_nodes(
// Validate group keys.
for (key_idx, key) in keys.iter().enumerate() {
- if !upstream_schema.fields.contains_key(key.as_str()) {
- errors.push(ValidationError {
- code: ErrorCode::AggregationUnknownField,
- path: format!("nodes[{node_idx}].ops[{op_idx}].group_by[{key_idx}]"),
- reason: format!("group_by key '{key}' does not exist in upstream schema"),
- });
- deriv_errors = true;
+ match upstream_schema.fields.get(key.as_str()) {
+ None => {
+ errors.push(ValidationError {
+ code: ErrorCode::AggregationUnknownField,
+ path: format!("nodes[{node_idx}].ops[{op_idx}].group_by[{key_idx}]"),
+ reason: format!(
+ "group_by key '{key}' does not exist in upstream schema"
+ ),
+ });
+ deriv_errors = true;
+ }
+ Some(crate::schema::FieldType::F64) => {
+ errors.push(ValidationError {
+ code: ErrorCode::AggregationUnknownField,
+ path: format!("nodes[{node_idx}].ops[{op_idx}].group_by[{key_idx}]"),
+ reason: format!(
+ "group_by key '{key}' cannot be float-typed; cast to str/int/bool first"
+ ),
+ });
+ deriv_errors = true;
+ }
+ Some(_) => {}
}
}
@@ -1513,6 +1528,33 @@ mod tests {
);
}
+ #[test]
+ fn rule11_rejects_float_group_key() {
+ let nodes = vec![
+ event_node_with_fields(
+ "Txn",
+ &[("user_id", FieldType::Str), ("amount", FieldType::F64)],
+ ),
+ group_by_derivation(
+ "AmountStats",
+ "Txn",
+ vec!["amount"],
+ serde_json::json!({
+ "cnt": {"op": "count", "params": {}}
+ }),
+ ),
+ ];
+ let (_, errors) = compile_aggregations_from_nodes(&nodes, &empty_registry(), &[]);
+ assert!(
+ errors.iter().any(|e| {
+ e.code == ErrorCode::AggregationUnknownField
+ && e.path.contains("group_by")
+ && e.reason.contains("cannot be float-typed")
+ }),
+ "expected float group_by rejection, got: {errors:#?}"
+ );
+ }
+
#[test]
fn rule11_rejects_unknown_op_field() {
let nodes = vec![
diff --git a/crates/beava-core/src/agg_state_table.rs b/crates/beava-core/src/agg_state_table.rs
index a75673c0..2af0f139 100644
--- a/crates/beava-core/src/agg_state_table.rs
+++ b/crates/beava-core/src/agg_state_table.rs
@@ -51,6 +51,7 @@ use std::hash::{Hash, Hasher};
use crate::agg_descriptor::AggregationDescriptor;
use crate::agg_op::AggOp;
use crate::row::{Row, Value};
+use crate::schema::FieldType;
use compact_str::CompactString;
use fxhash::FxBuildHasher;
use hashbrown::HashMap;
@@ -429,7 +430,8 @@ pub fn has_entries_for_name(
///
/// **Snapshot serialization (back-compat):** `iter_sorted` reconstructs
/// canonical `EntityKey` from the three maps, preserving the serialized
-/// snapshot format. `insert_from_entity_key` is the recovery-path inverse.
+/// snapshot format. `insert_from_entity_key` is the recovery-path inverse
+/// and restores each key into the same runtime sub-map the apply path uses.
///
/// **Query path:** `query_feature` takes an `EntityKey` (built by the query
/// parser from URL/JSON parameters) and routes through the appropriate map.
@@ -539,9 +541,70 @@ impl AggStateTable {
}
/// Recovery path: insert a `(EntityKey, Vec)` pair loaded from a
- /// snapshot. Routes via the `multi` sub-map (EntityKey is always canonical
- /// Value::Str pairs in the snapshot format).
+ /// snapshot. Routes through the same sub-map selection as the apply hot
+ /// path, so post-snapshot WAL replay updates the restored row instead of
+ /// creating a duplicate in a different storage shape.
pub fn insert_from_entity_key(&mut self, key: EntityKey, ops: Vec) {
+ self.insert_from_entity_key_with_types(key, ops, None);
+ }
+
+ /// Recovery path with descriptor field types available. Older snapshots
+ /// serialized single numeric/bool/datetime keys as `Value::Str`; use the
+ /// source schema to coerce those legacy keys back into the hot-path shape.
+ pub fn insert_from_entity_key_with_types(
+ &mut self,
+ key: EntityKey,
+ ops: Vec,
+ group_key_types: Option<&[FieldType]>,
+ ) {
+ if self.group_keys.is_empty() {
+ self.group_keys = key.0.iter().map(|(name, _)| name.to_string()).collect();
+ }
+
+ if key.0.len() == 1 {
+ let (_, value) = &key.0[0];
+ let value = match (value, group_key_types.and_then(|types| types.first())) {
+ (Value::Str(s), Some(FieldType::I64)) => s.parse::().ok().map(Value::I64),
+ (Value::Str(s), Some(FieldType::F64)) => s.parse::().ok().map(Value::F64),
+ (Value::Str(s), Some(FieldType::Bool)) => s.parse::().ok().map(Value::Bool),
+ (Value::Str(s), Some(FieldType::Datetime)) => {
+ s.parse::().ok().map(Value::Datetime)
+ }
+ _ => None,
+ }
+ .unwrap_or_else(|| value.clone());
+
+ match value {
+ Value::Str(s) => {
+ self.single_str.insert(s, ops);
+ return;
+ }
+ Value::I64(n) => {
+ self.single_u64
+ .insert(EntityKeyShape::tag_u64(VariantTag::I64, n as u64), ops);
+ return;
+ }
+ Value::F64(f) if !f.is_nan() => {
+ self.single_u64
+ .insert(EntityKeyShape::tag_u64(VariantTag::F64, f.to_bits()), ops);
+ return;
+ }
+ Value::Bool(b) => {
+ self.single_u64
+ .insert(EntityKeyShape::tag_u64(VariantTag::Bool, b as u64), ops);
+ return;
+ }
+ Value::Datetime(ms) => {
+ self.single_u64.insert(
+ EntityKeyShape::tag_u64(VariantTag::Datetime, ms as u64),
+ ops,
+ );
+ return;
+ }
+ _ => {}
+ }
+ }
+
self.multi.insert(key, ops);
}
@@ -568,8 +631,8 @@ impl AggStateTable {
match val {
Value::Str(s) => {
// single_str is the primary path for string-typed group keys.
- // Fall back to multi for entries inserted via insert_from_entity_key
- // (snapshot recovery path uses multi unconditionally).
+ // Fall back to multi for legacy entries inserted before
+ // snapshot recovery restored keys into their hot-path shape.
if let Some(ops) = self.single_str.get(s) {
ops
} else {
@@ -717,26 +780,23 @@ impl AggStateTable {
let key_name = self.group_keys.first().cloned().unwrap_or_default();
- // Reconstruct EntityKey from single_u64 entries.
+ // Reconstruct EntityKey from single_u64 entries. Preserve the typed
+ // Value variant so snapshot load restores into single_u64 again.
for (k, v) in &self.single_u64 {
let tag = k >> 60;
let payload = k & 0x0FFF_FFFF_FFFF_FFFF;
- let canonical: CompactString = match tag {
- 1 => (payload as i64).to_string().into(),
+ let value = match tag {
+ 1 => Value::I64(payload as i64),
2 => {
// F64: restore bits; upper 4 bits were replaced by tag=2.
- let f = f64::from_bits(payload | (2u64 << 60));
- format!("{:?}", f).into()
+ Value::F64(f64::from_bits(payload | (2u64 << 60)))
}
- 3 => (payload != 0).to_string().into(),
- 4 => (payload as i64).to_string().into(),
- _ => "unknown".into(),
+ 3 => Value::Bool(payload != 0),
+ 4 => Value::Datetime(payload as i64),
+ _ => Value::Str("unknown".into()),
};
let mut pairs: SmallVec<[(CompactString, Value); 2]> = SmallVec::new();
- pairs.push((
- CompactString::from(key_name.as_str()),
- Value::Str(canonical),
- ));
+ pairs.push((CompactString::from(key_name.as_str()), value));
entries.push((EntityKey(pairs), v));
}
@@ -936,6 +996,72 @@ mod tests {
}
}
+ #[test]
+ fn insert_from_entity_key_restores_single_str_hot_shape() {
+ let desc = make_descriptor("A", "S", &["user_id"], &[("cnt", count_op_desc())]);
+ let mut table = AggStateTable::new();
+ table.insert_from_entity_key(
+ make_user_key("alice"),
+ vec![AggOp::Count(crate::agg_state::CountState { n: 5 })],
+ );
+
+ assert_eq!(table.single_str.len(), 1);
+ assert!(table.multi.is_empty());
+
+ let event_row = Row::new().with_field("user_id", Value::Str("alice".into()));
+ let shape = EntityKeyShape::from_row(&desc.group_keys, &event_row).expect("shape");
+ let ops = table.get_or_init_by_shape(&shape, &desc);
+ ops[0].update(&event_row, 0, None, true);
+
+ assert_eq!(ops[0].query(0), Value::I64(6));
+ }
+
+ #[test]
+ fn snapshot_roundtrip_restores_single_u64_hot_shape() {
+ let desc = make_descriptor("A", "S", &["user_id"], &[("cnt", count_op_desc())]);
+ let event_row = Row::new().with_field("user_id", Value::I64(42));
+ let shape = EntityKeyShape::from_row(&desc.group_keys, &event_row).expect("shape");
+
+ let mut table = AggStateTable::new();
+ {
+ let ops = table.get_or_init_by_shape(&shape, &desc);
+ ops[0].update(&event_row, 0, None, true);
+ }
+
+ let mut restored = AggStateTable::new();
+ for (key, ops) in table.iter_sorted() {
+ restored.insert_from_entity_key(key, ops.clone());
+ }
+
+ assert_eq!(restored.single_u64.len(), 1);
+ assert!(restored.single_str.is_empty());
+ assert!(restored.multi.is_empty());
+
+ let ops = restored.get_or_init_by_shape(&shape, &desc);
+ ops[0].update(&event_row, 0, None, true);
+ assert_eq!(ops[0].query(0), Value::I64(2));
+ }
+
+ #[test]
+ fn insert_from_entity_key_with_types_coerces_legacy_i64_string() {
+ let desc = make_descriptor("A", "S", &["user_id"], &[("cnt", count_op_desc())]);
+ let mut table = AggStateTable::new();
+ table.insert_from_entity_key_with_types(
+ make_user_key_from_i64(42),
+ vec![AggOp::Count(crate::agg_state::CountState { n: 5 })],
+ Some(&[FieldType::I64]),
+ );
+
+ assert_eq!(table.single_u64.len(), 1);
+ assert!(table.single_str.is_empty());
+
+ let event_row = Row::new().with_field("user_id", Value::I64(42));
+ let shape = EntityKeyShape::from_row(&desc.group_keys, &event_row).expect("shape");
+ let ops = table.get_or_init_by_shape(&shape, &desc);
+ ops[0].update(&event_row, 0, None, true);
+ assert_eq!(ops[0].query(0), Value::I64(6));
+ }
+
#[test]
fn agg_state_table_entity_count_counts_distinct_keys() {
let desc = make_descriptor("A", "S", &["user_id"], &[("cnt", count_op_desc())]);
diff --git a/crates/beava-core/tests/snapshot_lock_hold_repro.rs b/crates/beava-core/tests/snapshot_lock_hold_repro.rs
new file mode 100644
index 00000000..2981cdce
--- /dev/null
+++ b/crates/beava-core/tests/snapshot_lock_hold_repro.rs
@@ -0,0 +1,192 @@
+//! Local reproduction of the kalshi-pulse incident (2026-05-21):
+//! measure how long `state_tables.lock()` is held during snapshot encoding.
+//!
+//! The hot operation under the `parking_lot::Mutex` in
+//! `crates/beava-server/src/snapshot_task.rs::do_snapshot` is exactly:
+//!
+//! for (node_name, desc) in registry.compiled_aggregations.iter() {
+//! if let Some(table) = state_tables.get(agg_id) {
+//! let entries: Vec<(EntityKey, Vec)> = table
+//! .iter_sorted()
+//! .map(|(k, v)| (k.clone(), v.clone())) // ← clone every entry
+//! .collect();
+//! serialized_tables.insert(node_name.clone(), entries);
+//! }
+//! }
+//!
+//! We bypass the Registry plumbing and time the inner clone-collect directly
+//! on a populated `AggStateTable`. This is the SAME byte-for-byte operation
+//! that parks the apply thread in production, just without the registry
+//! enumeration overhead (which is negligible — BTreeMap of ~14 aggs).
+//!
+//! Run with:
+//! cargo test -p beava-core --release --test snapshot_lock_hold_repro -- --nocapture
+//!
+//! Use `--release` — debug-mode `AggOp::clone` is ~10× slower than release
+//! and would mislead the projection.
+
+use beava_core::agg_op::AggOp;
+use beava_core::agg_state::CountState;
+use beava_core::agg_state_table::{AggStateTable, EntityKey};
+use beava_core::row::Value;
+use compact_str::CompactString;
+use smallvec::smallvec;
+use std::time::Instant;
+
+/// What op shape to put in each entry. Count is the smallest variant;
+/// CountDistinct is a representative sketch (HLL with 1024 registers per
+/// entity).
+#[derive(Copy, Clone)]
+enum OpShape {
+ Count,
+ CountDistinctHll1024,
+}
+
+impl OpShape {
+ fn build(self, n: u64) -> AggOp {
+ match self {
+ OpShape::Count => AggOp::Count(CountState { n }),
+ OpShape::CountDistinctHll1024 => AggOp::CountDistinct(Box::default()),
+ }
+ }
+ fn label(self) -> &'static str {
+ match self {
+ OpShape::Count => "Count",
+ OpShape::CountDistinctHll1024 => "CountDistinct(HLL-1024)",
+ }
+ }
+}
+
+/// Populate one `AggStateTable` with N entities; every entry has a single
+/// op of the requested shape.
+fn build_table(n_entities: usize, shape: OpShape) -> AggStateTable {
+ let mut table = AggStateTable::new();
+ for ent in 0..n_entities {
+ let key_str = format!("user_{ent:09}");
+ let entity_key = EntityKey(smallvec![(
+ CompactString::from("user_id"),
+ Value::Str(CompactString::from(key_str.as_str())),
+ )]);
+ table.insert_from_entity_key(entity_key, vec![shape.build(ent as u64)]);
+ }
+ table
+}
+
+/// The exact clone-collect that runs under `state_tables.lock()` in
+/// `snapshot_task.rs::do_snapshot`. Returns (lock_held_ms, entries_collected,
+/// approx_bytes_cloned).
+fn measure_lock_hold(table: &AggStateTable) -> (f64, usize, usize) {
+ let t0 = Instant::now();
+ let entries: Vec<(EntityKey, Vec)> = table
+ .iter_sorted()
+ .map(|(k, v)| (k.clone(), v.clone()))
+ .collect();
+ let elapsed_ms = t0.elapsed().as_secs_f64() * 1000.0;
+
+ // Rough byte estimate: AggOp ≤ 80 B (CI tripwire). EntityKey is a SmallVec
+ // of (CompactString, Value::Str(CompactString)). Inline string fit varies;
+ // assume ~32 B per EntityKey on average for "user_NNNNNNNNN" keys.
+ let n = entries.len();
+ let approx_bytes = n * (80 + 32 + 24); // AggOp + EntityKey + Vec overhead
+ (elapsed_ms, n, approx_bytes)
+}
+
+fn run_shape(label: &str, shape: OpShape, sizes: &[usize]) -> Vec<(usize, f64)> {
+ println!();
+ println!("=== {label} ===");
+ println!(
+ "{:>12} {:>14} {:>18}",
+ "entities", "lock_held_ms", "ns_per_entry"
+ );
+ println!("{}", "-".repeat(48));
+
+ let mut out = Vec::new();
+ for &n in sizes {
+ let table = build_table(n, shape);
+ let _ = measure_lock_hold(&table); // warm-up
+ let mut samples: Vec = (0..3).map(|_| measure_lock_hold(&table).0).collect();
+ samples.sort_by(|a, b| a.partial_cmp(b).unwrap());
+ let median_ms = samples[1];
+ let ns_per_entry = median_ms * 1_000_000.0 / (n as f64);
+ out.push((n, ns_per_entry));
+ println!("{:>12} {:>11.1}ms {:>15.0}ns", n, median_ms, ns_per_entry);
+ }
+ out
+}
+
+#[test]
+#[ignore = "diagnostic repro: allocates millions of entries; run manually in release with --ignored --nocapture"]
+fn measure_snapshot_lock_hold_time() {
+ println!();
+ println!("=== state_tables.lock() hold-time measurement ===");
+ println!("Operation timed (the exact body of the parking_lot lock-hold in");
+ println!("crates/beava-server/src/snapshot_task.rs::do_snapshot):");
+ println!(" table.iter_sorted().map(|(k,v)| (k.clone(), v.clone())).collect()");
+
+ let small = &[
+ 10_000usize,
+ 100_000,
+ 500_000,
+ 1_000_000,
+ 2_000_000,
+ 4_000_000,
+ ];
+ let small_sketch = &[10_000usize, 100_000, 500_000];
+
+ let count_samples = run_shape(OpShape::Count.label(), OpShape::Count, small);
+ let hll_samples = run_shape(
+ OpShape::CountDistinctHll1024.label(),
+ OpShape::CountDistinctHll1024,
+ small_sketch,
+ );
+
+ // Average ns/entry at large-N (skip the noisy small ones).
+ let count_ns: f64 = {
+ let tail: Vec = count_samples
+ .iter()
+ .rev()
+ .take(2)
+ .map(|(_, n)| *n)
+ .collect();
+ tail.iter().sum::() / tail.len() as f64
+ };
+ let hll_ns: f64 = {
+ let tail: Vec = hll_samples.iter().rev().take(2).map(|(_, n)| *n).collect();
+ tail.iter().sum::() / tail.len() as f64
+ };
+
+ println!();
+ println!("=== Projection to production scale ===");
+ println!("Per-entry clone cost (large-N median):");
+ println!(" Count : {count_ns:>8.0} ns");
+ println!(
+ " CountDistinct(HLL-1024) : {hll_ns:>8.0} ns ({:.1}× Count)",
+ hll_ns / count_ns
+ );
+ println!();
+ println!("For the 507 MB encoded production snapshot:");
+ println!("(actual entry count depends on op-mix; we project two cases)");
+ for entries in &[1_000_000usize, 5_000_000usize, 10_000_000usize] {
+ let count_s = count_ns * (*entries as f64) / 1_000_000_000.0;
+ let hll_s = hll_ns * (*entries as f64) / 1_000_000_000.0;
+ println!(" {entries:>10} entries → Count: {count_s:>5.2}s HLL: {hll_s:>6.2}s lock-hold",);
+ }
+
+ println!();
+ println!("=== Interpretation ===");
+ println!("/ping is FIFO-queued behind any push waiting on state_tables.lock().");
+ println!("Once the lock is held by the snapshot task, no push can dispatch,");
+ println!("and /ping (handled on the same single-threaded apply loop) cannot");
+ println!("progress until the lock is released.");
+ println!();
+ println!("Even with all-Count workloads, multi-second lock-holds at 5-10M");
+ println!("entities are enough to blow past a 3s docker healthcheck timeout.");
+ println!("Real production workloads with sketches push this to tens of");
+ println!("seconds, which matches the incident's 60-90s observation.");
+ println!();
+ println!("Additional time accrues OUTSIDE the lock (encode + WAL truncate +");
+ println!("fsync of 507 MB) — those don't park the apply thread directly,");
+ println!("but they extend the snapshot task's wall-clock cycle, and once");
+ println!("CPU/IO bandwidth is saturated, the next snapshot tick fires before");
+ println!("the previous one has fully drained, compounding the problem.");
+}
diff --git a/crates/beava-persistence/src/fsync_worker.rs b/crates/beava-persistence/src/fsync_worker.rs
index 23cb0f9b..a9786836 100644
--- a/crates/beava-persistence/src/fsync_worker.rs
+++ b/crates/beava-persistence/src/fsync_worker.rs
@@ -67,6 +67,7 @@ pub enum SyncMode {
struct AppendRequest {
record_type: RecordType,
payload: Vec,
+ min_lsn: Lsn,
mode: SyncMode,
done: oneshot::Sender>,
}
@@ -130,6 +131,7 @@ impl WalSink {
req = append_rx.recv() => {
match req {
Some(req) => {
+ next_lsn = next_lsn.max(req.min_lsn);
let assigned = next_lsn;
next_lsn = next_lsn.saturating_add(1);
// Advance the durable watermark immediately —
@@ -194,7 +196,20 @@ impl WalSink {
record_type: RecordType,
payload: Vec,
) -> Result {
- self.append_record_with_mode(record_type, payload, SyncMode::PerEvent)
+ self.append_record_with_mode_at_least(record_type, payload, 0, SyncMode::PerEvent)
+ .await
+ }
+
+ /// Strict append that first raises the assigned LSN to at least
+ /// `min_lsn`. Used when another WAL stream has already advanced the
+ /// logical LSN namespace.
+ pub async fn append_record_at_least(
+ &self,
+ record_type: RecordType,
+ payload: Vec,
+ min_lsn: Lsn,
+ ) -> Result {
+ self.append_record_with_mode_at_least(record_type, payload, min_lsn, SyncMode::PerEvent)
.await
}
@@ -207,12 +222,24 @@ impl WalSink {
record_type: RecordType,
payload: Vec,
mode: SyncMode,
+ ) -> Result {
+ self.append_record_with_mode_at_least(record_type, payload, 0, mode)
+ .await
+ }
+
+ async fn append_record_with_mode_at_least(
+ &self,
+ record_type: RecordType,
+ payload: Vec,
+ min_lsn: Lsn,
+ mode: SyncMode,
) -> Result {
let (tx, rx) = oneshot::channel();
self.append_tx
.send(AppendRequest {
record_type,
payload,
+ min_lsn,
mode,
done: tx,
})
@@ -468,6 +495,7 @@ fn stage_request(
next_lsn: &mut Lsn,
staged_bytes: &mut u64,
) -> Option {
+ *next_lsn = (*next_lsn).max(req.min_lsn);
let lsn = *next_lsn;
*next_lsn += 1;
let payload_len = req.payload.len();
diff --git a/crates/beava-persistence/src/lib.rs b/crates/beava-persistence/src/lib.rs
index 0c39df7b..b5037643 100644
--- a/crates/beava-persistence/src/lib.rs
+++ b/crates/beava-persistence/src/lib.rs
@@ -74,7 +74,9 @@ pub use error::{PersistError, SnapshotError};
pub use fsync_worker::{SyncMode, WalSink, WalSinkConfig};
pub use reader::WalReader;
pub use record::FORMAT_VERSION;
-pub use snapshot::{list_snapshots, prune_old_snapshots, SnapshotReader, SnapshotWriter};
+pub use snapshot::{
+ list_snapshots, prune_old_snapshots, SnapshotReader, SnapshotWriteStats, SnapshotWriter,
+};
pub use snapshot_header::{
SnapshotHeader, SNAPSHOT_EXT, SNAPSHOT_FORMAT_VERSION, SNAPSHOT_HEADER_SIZE, SNAPSHOT_MAGIC,
};
diff --git a/crates/beava-persistence/src/snapshot.rs b/crates/beava-persistence/src/snapshot.rs
index 8c4d019d..4dd70c55 100644
--- a/crates/beava-persistence/src/snapshot.rs
+++ b/crates/beava-persistence/src/snapshot.rs
@@ -15,7 +15,7 @@
use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use crate::error::{PersistError, SnapshotError};
use crate::snapshot_header::{
@@ -34,6 +34,25 @@ use crate::Lsn;
/// that still hold a writer handle.
pub struct SnapshotWriter;
+/// Best-effort instrumentation returned by [`SnapshotWriter::write_with_stats`].
+#[derive(Debug, Clone)]
+pub struct SnapshotWriteStats {
+ pub path: PathBuf,
+ /// Header bytes plus body bytes written to the committed snapshot file.
+ pub bytes: u64,
+ /// Required file fsync duration.
+ pub file_fsync_duration: Duration,
+ /// Best-effort parent-directory fsync duration. `None` when unsupported or
+ /// when opening the directory failed.
+ pub dir_fsync_duration: Option,
+}
+
+impl SnapshotWriteStats {
+ pub fn total_fsync_duration(&self) -> Duration {
+ self.file_fsync_duration + self.dir_fsync_duration.unwrap_or_default()
+ }
+}
+
impl SnapshotWriter {
/// Construct a no-op snapshot writer for `Persistence::Memory` mode.
pub fn no_op() -> Self {
@@ -54,6 +73,16 @@ impl SnapshotWriter {
registry_version: u64,
body: &[u8],
) -> Result {
+ Self::write_with_stats(dir, snapshot_lsn, registry_version, body).map(|stats| stats.path)
+ }
+
+ /// Atomically write a snapshot file and return write/fsync stats.
+ pub fn write_with_stats(
+ dir: &Path,
+ snapshot_lsn: Lsn,
+ registry_version: u64,
+ body: &[u8],
+ ) -> Result {
std::fs::create_dir_all(dir).map_err(PersistError::Io)?;
let tmp_path = dir.join(format!("snapshot-{snapshot_lsn:016x}.tmp"));
let final_path = dir.join(format!("snapshot-{snapshot_lsn:016x}.{SNAPSHOT_EXT}"));
@@ -74,6 +103,7 @@ impl SnapshotWriter {
body_crc32c,
};
let header_bytes = header.encode();
+ let bytes = header_bytes.len() as u64 + body.len() as u64;
let mut f = OpenOptions::new()
.read(true)
@@ -85,21 +115,34 @@ impl SnapshotWriter {
f.write_all(&header_bytes).map_err(PersistError::Io)?;
f.write_all(body).map_err(PersistError::Io)?;
+ let file_fsync_started = Instant::now();
f.sync_all().map_err(PersistError::Io)?;
+ let file_fsync_duration = file_fsync_started.elapsed();
drop(f);
std::fs::rename(&tmp_path, &final_path).map_err(PersistError::Io)?;
+ #[cfg(unix)]
+ let mut dir_fsync_duration = None;
+ #[cfg(not(unix))]
+ let dir_fsync_duration = None;
// Best-effort parent-directory fsync makes the rename durable on
// filesystems that don't journal directory entries with the file.
#[cfg(unix)]
{
if let Ok(d) = File::open(dir) {
+ let dir_fsync_started = Instant::now();
let _ = d.sync_all();
+ dir_fsync_duration = Some(dir_fsync_started.elapsed());
}
}
- Ok(final_path)
+ Ok(SnapshotWriteStats {
+ path: final_path,
+ bytes,
+ file_fsync_duration,
+ dir_fsync_duration,
+ })
}
}
diff --git a/crates/beava-persistence/tests/fsync_worker.rs b/crates/beava-persistence/tests/fsync_worker.rs
index 0c9ed903..0aaaf624 100644
--- a/crates/beava-persistence/tests/fsync_worker.rs
+++ b/crates/beava-persistence/tests/fsync_worker.rs
@@ -49,6 +49,35 @@ async fn append_returns_durable_lsn() {
assert_eq!(recs[0].lsn, 1);
}
+#[tokio::test(flavor = "current_thread")]
+async fn append_record_at_least_raises_assigned_lsn() {
+ let dir = tempfile::tempdir().unwrap();
+ let (sink, handle) = WalSink::spawn(config_for_test(dir.path().to_path_buf())).unwrap();
+
+ let lsn = sink
+ .append_record_at_least(
+ beava_persistence::RecordType::RegistryBump,
+ b"x".to_vec(),
+ 100,
+ )
+ .await
+ .unwrap();
+ assert_eq!(lsn, 100);
+
+ let next = sink.append_event(b"p".to_vec()).await.unwrap();
+ assert_eq!(next, 101);
+
+ sink.shutdown().await.unwrap();
+ handle.await.unwrap();
+
+ let seg = find_segment(dir.path());
+ let r = WalReader::open(&seg).unwrap();
+ let recs: Vec<_> = r.collect::>().unwrap();
+ assert_eq!(recs.len(), 2);
+ assert_eq!(recs[0].lsn, 100);
+ assert_eq!(recs[1].lsn, 101);
+}
+
#[tokio::test(flavor = "current_thread")]
async fn concurrent_appends_get_distinct_lsns() {
let dir = tempfile::tempdir().unwrap();
diff --git a/crates/beava-persistence/tests/snapshot_roundtrip.rs b/crates/beava-persistence/tests/snapshot_roundtrip.rs
index 4e3b9a8b..a206be83 100644
--- a/crates/beava-persistence/tests/snapshot_roundtrip.rs
+++ b/crates/beava-persistence/tests/snapshot_roundtrip.rs
@@ -87,6 +87,22 @@ fn snapshot_write_then_read_roundtrip() {
assert_eq!(body_read, body);
}
+#[test]
+fn snapshot_write_with_stats_reports_bytes_and_fsync_timing() {
+ let tmp = tempfile::tempdir().expect("tempdir");
+ let body = b"instrumented payload".to_vec();
+ let stats = SnapshotWriter::write_with_stats(tmp.path(), 1001, 3, &body).expect("write");
+
+ assert!(stats.path.exists());
+ assert_eq!(stats.bytes, SNAPSHOT_HEADER_SIZE as u64 + body.len() as u64);
+ assert!(stats.total_fsync_duration() >= stats.file_fsync_duration);
+ #[cfg(unix)]
+ assert!(
+ stats.dir_fsync_duration.is_some(),
+ "unix snapshot writes should attempt parent-dir fsync"
+ );
+}
+
#[test]
fn snapshot_body_corruption_detected() {
let tmp = tempfile::tempdir().expect("tempdir");
diff --git a/crates/beava-runtime-core/src/lib.rs b/crates/beava-runtime-core/src/lib.rs
index 362f91ba..145a130f 100644
--- a/crates/beava-runtime-core/src/lib.rs
+++ b/crates/beava-runtime-core/src/lib.rs
@@ -42,5 +42,5 @@ pub use router::{Route, Router};
pub use tcp_listener::TcpListener;
pub use wal_buffer::{WalBuffer, WalBufferRing};
pub use wal_lsn::{Lsn, WalLsn};
-pub use wal_writer::WalWriter;
+pub use wal_writer::{WalReclaimHandle, WalWriter};
pub use wire_request::WireRequest;
diff --git a/crates/beava-runtime-core/src/tcp_listener.rs b/crates/beava-runtime-core/src/tcp_listener.rs
index aad8fdd7..8869983f 100644
--- a/crates/beava-runtime-core/src/tcp_listener.rs
+++ b/crates/beava-runtime-core/src/tcp_listener.rs
@@ -16,6 +16,7 @@ use beava_core::wire::{
OP_PUSH, OP_REGISTER, OP_RESET,
};
use bytes::BytesMut;
+use std::borrow::Cow;
use std::net::SocketAddr;
use crate::wire_request::WireRequest;
@@ -502,7 +503,7 @@ impl std::error::Error for JsonEnvelopeError {}
/// hand-rolled scanner walks key/value pairs and tracks string state + brace
/// depth for the body value range.
#[inline]
-pub fn parse_json_envelope(payload: &[u8]) -> Result<(&str, &[u8]), JsonEnvelopeError> {
+pub fn parse_json_envelope(payload: &[u8]) -> Result<(Cow<'_, str>, &[u8]), JsonEnvelopeError> {
// Skip optional leading whitespace, then the opening brace.
let mut p = skip_ws(payload, 0);
if p >= payload.len() || payload[p] != b'{' {
@@ -510,7 +511,7 @@ pub fn parse_json_envelope(payload: &[u8]) -> Result<(&str, &[u8]), JsonEnvelope
}
p += 1;
- let mut event_name: Option<&str> = None;
+ let mut event_name: Option> = None;
let mut body_slice: Option<&[u8]> = None;
loop {
@@ -546,10 +547,7 @@ pub fn parse_json_envelope(payload: &[u8]) -> Result<(&str, &[u8]), JsonEnvelope
}
let v_start = p + 1;
let v_end = scan_string_end(payload, v_start)?;
- event_name = Some(
- std::str::from_utf8(&payload[v_start..v_end])
- .map_err(|_| JsonEnvelopeError::Decode("invalid utf-8".to_string()))?,
- );
+ event_name = Some(decode_json_string_fragment(payload, v_start, v_end)?);
p = v_end + 1;
}
"body" => {
@@ -589,6 +587,26 @@ pub fn parse_json_envelope(payload: &[u8]) -> Result<(&str, &[u8]), JsonEnvelope
))
}
+fn decode_json_string_fragment(
+ payload: &[u8],
+ start: usize,
+ end: usize,
+) -> Result, JsonEnvelopeError> {
+ let raw = std::str::from_utf8(&payload[start..end])
+ .map_err(|_| JsonEnvelopeError::Decode("invalid utf-8".to_string()))?;
+ if !raw.as_bytes().contains(&b'\\') {
+ return Ok(Cow::Borrowed(raw));
+ }
+
+ let mut quoted = Vec::with_capacity(raw.len() + 2);
+ quoted.push(b'"');
+ quoted.extend_from_slice(raw.as_bytes());
+ quoted.push(b'"');
+ serde_json::from_slice::("ed)
+ .map(Cow::Owned)
+ .map_err(|e| JsonEnvelopeError::Decode(format!("invalid string escape: {e}")))
+}
+
/// Skip ASCII whitespace (space/tab/newline/CR).
#[inline(always)]
fn skip_ws(payload: &[u8], mut p: usize) -> usize {
@@ -1133,6 +1151,16 @@ mod tests {
assert_eq!(body_val["q"], "a\"b");
}
+ #[test]
+ fn parse_json_envelope_decodes_escaped_event_name() {
+ let payload = br#"{"event":"Txn\u0001","body":{"amount":99}}"#;
+ let (event, body_bytes) = parse_json_envelope(payload).expect("ok");
+ assert_eq!(event.as_ref(), "Txn\u{0001}");
+ let body_val: serde_json::Value =
+ sonic_rs::from_slice(body_bytes).expect("body parses as json");
+ assert_eq!(body_val["amount"], 99);
+ }
+
#[test]
fn parse_json_envelope_malformed_returns_err() {
// Missing closing brace.
diff --git a/crates/beava-runtime-core/src/wal_lsn.rs b/crates/beava-runtime-core/src/wal_lsn.rs
index b4ebcd22..d314b811 100644
--- a/crates/beava-runtime-core/src/wal_lsn.rs
+++ b/crates/beava-runtime-core/src/wal_lsn.rs
@@ -93,10 +93,18 @@ impl std::fmt::Debug for WalLsn {
impl WalLsn {
/// Create a new `WalLsn` with all watermarks at zero.
pub fn new() -> Self {
+ Self::new_at(0)
+ }
+
+ /// Create a new `WalLsn` with all watermarks already advanced to `lsn`.
+ ///
+ /// Used after recovery so the hand-rolled WAL ring continues from the
+ /// recovered high-water mark instead of reusing low LSNs after restart.
+ pub fn new_at(lsn: Lsn) -> Self {
Self {
- committed: AtomicU64::new(0),
- written: AtomicU64::new(0),
- synced: AtomicU64::new(0),
+ committed: AtomicU64::new(lsn),
+ written: AtomicU64::new(lsn),
+ synced: AtomicU64::new(lsn),
synced_condvar: Condvar::new(),
synced_mutex: Mutex::new(()),
}
@@ -135,6 +143,18 @@ impl WalLsn {
self.committed.fetch_add(n, Ordering::AcqRel) + n
}
+ /// Raise the committed watermark to at least `lsn` without appending
+ /// bytes to the hand-rolled WAL.
+ ///
+ /// The server uses one logical LSN namespace across the legacy
+ /// `WalSink` registry WAL and the data-plane ring WAL. When a registry
+ /// bump advances the legacy WAL first, the next data-plane append must
+ /// jump past that durable point so snapshots can gate both WAL streams
+ /// with a single LSN.
+ pub fn mark_committed_at_least(&self, lsn: Lsn) {
+ self.committed.fetch_max(lsn, Ordering::AcqRel);
+ }
+
/// Advance `written_lsn` to at least `lsn`.
///
/// Called by the writer thread after `write(fd, ...)` returns.
diff --git a/crates/beava-runtime-core/src/wal_writer.rs b/crates/beava-runtime-core/src/wal_writer.rs
index 5d545dcf..62b4b839 100644
--- a/crates/beava-runtime-core/src/wal_writer.rs
+++ b/crates/beava-runtime-core/src/wal_writer.rs
@@ -42,7 +42,7 @@ use crate::wal_lsn::WalLsn;
use std::fs::{File, OpenOptions};
use std::io::Write as IoWrite;
use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::Duration;
@@ -90,6 +90,8 @@ impl From for WalWriterError {
pub struct WalWriter {
/// Open WAL segment file (O_APPEND | O_WRONLY | O_CREAT).
file: File,
+ /// Stable WAL path. The writer owns all truncate/compact operations.
+ wal_path: PathBuf,
/// Shared buffer ring — writer pops sealed buffers from here.
ring: Arc,
/// Shared LSN watermarks — writer advances written + synced here.
@@ -98,6 +100,34 @@ pub struct WalWriter {
tick_ms: u64,
/// Set to `true` by `shutdown()` to ask the writer thread to drain and exit.
shutdown: Arc,
+ /// Snapshot-covered LSN requested by the snapshot task. The writer is the
+ /// only thread that acts on this request because it owns the append file.
+ reclaim: WalReclaimHandle,
+}
+
+/// Handle used by snapshot/checkpoint code to request hand-rolled WAL
+/// compaction. Requests are monotone; the writer thread observes them after a
+/// flush+fsync boundary and safely rewrites the WAL tail.
+#[derive(Clone, Debug)]
+pub struct WalReclaimHandle {
+ requested_lsn: Arc,
+}
+
+impl WalReclaimHandle {
+ fn new() -> Self {
+ Self {
+ requested_lsn: Arc::new(AtomicU64::new(0)),
+ }
+ }
+
+ /// Request reclamation of WAL records covered by `covered_lsn`.
+ pub fn request_reclaim_up_to(&self, covered_lsn: u64) {
+ self.requested_lsn.fetch_max(covered_lsn, Ordering::AcqRel);
+ }
+
+ fn requested_lsn(&self) -> u64 {
+ self.requested_lsn.load(Ordering::Acquire)
+ }
}
impl WalWriter {
@@ -121,6 +151,7 @@ impl WalWriter {
}
let wal_path = dir.join("wal-0000000000000000.wal");
+ repair_wal_file_tail(&wal_path)?;
let file = OpenOptions::new()
.create(true)
.append(true)
@@ -128,10 +159,12 @@ impl WalWriter {
Ok(Self {
file,
+ wal_path,
ring,
lsn,
tick_ms,
shutdown: Arc::new(AtomicBool::new(false)),
+ reclaim: WalReclaimHandle::new(),
})
}
@@ -140,23 +173,32 @@ impl WalWriter {
Arc::clone(&self.shutdown)
}
+ /// Return a handle for snapshot/checkpoint code to request WAL
+ /// reclamation. The writer performs the actual compaction on its own
+ /// thread after flushing the ring.
+ pub fn reclaim_handle(&self) -> WalReclaimHandle {
+ self.reclaim.clone()
+ }
+
/// Start the writer + fsync loop in a dedicated `std::thread`.
///
/// The returned `JoinHandle` can be awaited for clean shutdown.
pub fn spawn(self) -> JoinHandle<()> {
let WalWriter {
mut file,
+ wal_path,
ring,
lsn,
tick_ms,
shutdown,
+ reclaim,
} = self;
let tick = Duration::from_millis(tick_ms);
std::thread::Builder::new()
.name("beava-wal-writer".to_owned())
.spawn(move || {
- run_writer_loop(&mut file, &ring, &lsn, tick, &shutdown);
+ run_writer_loop(&mut file, &wal_path, &ring, &lsn, tick, &shutdown, &reclaim);
})
.expect("failed to spawn WAL writer thread")
}
@@ -164,11 +206,14 @@ impl WalWriter {
fn run_writer_loop(
file: &mut File,
+ wal_path: &Path,
ring: &WalBufferRing,
lsn: &WalLsn,
tick: Duration,
shutdown: &AtomicBool,
+ reclaim: &WalReclaimHandle,
) {
+ let mut reclaimed_lsn = 0u64;
loop {
std::thread::sleep(tick);
@@ -178,12 +223,14 @@ fn run_writer_loop(
// 2. Drain all sealed buffers: write → fsync → free.
flush_sealed_buffers(file, ring, lsn);
+ maybe_reclaim_wal_file(file, wal_path, lsn, reclaim, &mut reclaimed_lsn);
// 3. Check shutdown after draining.
if shutdown.load(Ordering::Acquire) {
// Final drain + fsync on shutdown.
ring.seal_active();
flush_sealed_buffers(file, ring, lsn);
+ maybe_reclaim_wal_file(file, wal_path, lsn, reclaim, &mut reclaimed_lsn);
break;
}
}
@@ -246,6 +293,256 @@ fn sync_file(file: &mut File) -> std::io::Result<()> {
}
}
+fn maybe_reclaim_wal_file(
+ file: &mut File,
+ wal_path: &Path,
+ lsn: &WalLsn,
+ reclaim: &WalReclaimHandle,
+ reclaimed_lsn: &mut u64,
+) {
+ let requested = reclaim.requested_lsn();
+ if requested == 0 || requested <= *reclaimed_lsn {
+ return;
+ }
+ let synced_lsn = lsn.synced();
+
+ match compact_wal_file(file, wal_path, requested) {
+ Ok(stats) => {
+ *reclaimed_lsn = requested;
+ tracing::info!(
+ target: "beava.wal",
+ kind = "wal.handrolled_reclaimed",
+ covered_lsn = requested,
+ data_plane_synced_lsn = synced_lsn,
+ before_bytes = stats.before_bytes,
+ after_bytes = stats.after_bytes,
+ removed_records = stats.removed_records,
+ retained_records = stats.retained_records,
+ "hand-rolled WAL reclaimed after durable snapshot"
+ );
+ }
+ Err(e) => {
+ tracing::warn!(
+ target: "beava.wal",
+ kind = "wal.handrolled_reclaim_failed",
+ covered_lsn = requested,
+ error = %e,
+ "hand-rolled WAL reclaim skipped"
+ );
+ }
+ }
+}
+
+#[derive(Debug)]
+struct WalCompactStats {
+ before_bytes: u64,
+ after_bytes: u64,
+ removed_records: usize,
+ retained_records: usize,
+}
+
+#[derive(Debug)]
+struct ParsedWalRecord {
+ lsn: u64,
+ start: usize,
+ end: usize,
+ version: u8,
+}
+
+fn compact_wal_file(
+ open_append_file: &mut File,
+ wal_path: &Path,
+ covered_lsn: u64,
+) -> std::io::Result {
+ open_append_file.sync_all()?;
+
+ let data = std::fs::read(wal_path)?;
+ let before_bytes = data.len() as u64;
+ if data.is_empty() {
+ return Ok(WalCompactStats {
+ before_bytes,
+ after_bytes: 0,
+ removed_records: 0,
+ retained_records: 0,
+ });
+ }
+
+ let parsed = parse_wal_record_bounds_with_prefix(&data)?;
+ let records = parsed.records;
+ let mut retained = Vec::new();
+ let mut removed_records = 0usize;
+ let mut retained_records = 0usize;
+ for rec in records {
+ if rec.lsn <= covered_lsn {
+ removed_records += 1;
+ continue;
+ }
+ if rec.version != 0x03 {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ "cannot compact retained legacy v2 WAL records without assigned LSNs",
+ ));
+ }
+ retained.extend_from_slice(&data[rec.start..rec.end]);
+ retained_records += 1;
+ }
+
+ if removed_records == 0 && parsed.valid_prefix_len == data.len() {
+ sync_parent_dir(wal_path)?;
+ return Ok(WalCompactStats {
+ before_bytes,
+ after_bytes: before_bytes,
+ removed_records,
+ retained_records,
+ });
+ }
+
+ let tmp_path = wal_path.with_extension("wal.compact.tmp");
+ match std::fs::remove_file(&tmp_path) {
+ Ok(()) => {}
+ Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+ Err(e) => return Err(e),
+ }
+ {
+ let mut tmp = OpenOptions::new()
+ .write(true)
+ .create_new(true)
+ .open(&tmp_path)?;
+ tmp.write_all(&retained)?;
+ tmp.sync_all()?;
+ }
+
+ // Open the replacement WAL before the rename. After rename succeeds there
+ // must be no fallible reopen step, otherwise a transient fd/open failure
+ // could leave the writer appending to the old unlinked inode.
+ let new_append_file = OpenOptions::new().append(true).open(&tmp_path)?;
+
+ std::fs::rename(&tmp_path, wal_path)?;
+
+ *open_append_file = new_append_file;
+ sync_parent_dir(wal_path)?;
+
+ Ok(WalCompactStats {
+ before_bytes,
+ after_bytes: retained.len() as u64,
+ removed_records,
+ retained_records,
+ })
+}
+
+#[cfg(test)]
+fn parse_wal_record_bounds(data: &[u8]) -> std::io::Result> {
+ Ok(parse_wal_record_bounds_with_prefix(data)?.records)
+}
+
+#[derive(Debug)]
+struct ParsedWalRecords {
+ records: Vec,
+ valid_prefix_len: usize,
+}
+
+fn parse_wal_record_bounds_with_prefix(data: &[u8]) -> std::io::Result {
+ let mut records = Vec::new();
+ let mut pos = 0usize;
+ let mut valid_prefix_len = 0usize;
+
+ while pos < data.len() {
+ let start = pos;
+ let version = data[pos];
+ if version != 0x02 && version != 0x03 {
+ break;
+ }
+ pos += 1;
+
+ let assigned_lsn = if version == 0x03 {
+ if pos + 8 > data.len() {
+ break;
+ }
+ let lsn = u64::from_be_bytes(data[pos..pos + 8].try_into().unwrap());
+ pos += 8;
+ Some(lsn)
+ } else {
+ None
+ };
+
+ if pos + 15 > data.len() {
+ break;
+ }
+ pos += 1; // body_format
+ pos += 4; // registry version
+ pos += 8; // event time
+
+ let name_len = u16::from_be_bytes(data[pos..pos + 2].try_into().unwrap()) as usize;
+ pos += 2;
+
+ if pos + name_len + 4 > data.len() {
+ break;
+ }
+ if std::str::from_utf8(&data[pos..pos + name_len]).is_err() {
+ break;
+ }
+ pos += name_len;
+
+ let body_len = u32::from_be_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
+ pos += 4;
+
+ if pos + body_len > data.len() {
+ break;
+ }
+ pos += body_len;
+
+ let end = pos;
+ records.push(ParsedWalRecord {
+ lsn: assigned_lsn.unwrap_or(end as u64),
+ start,
+ end,
+ version,
+ });
+ valid_prefix_len = end;
+ }
+
+ Ok(ParsedWalRecords {
+ records,
+ valid_prefix_len,
+ })
+}
+
+fn repair_wal_file_tail(wal_path: &Path) -> std::io::Result<()> {
+ let data = match std::fs::read(wal_path) {
+ Ok(data) => data,
+ Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+ Err(e) => return Err(e),
+ };
+ let parsed = parse_wal_record_bounds_with_prefix(&data)?;
+ if parsed.valid_prefix_len == data.len() {
+ return Ok(());
+ }
+
+ let file = OpenOptions::new().write(true).open(wal_path)?;
+ file.set_len(parsed.valid_prefix_len as u64)?;
+ file.sync_all()?;
+ sync_parent_dir(wal_path)?;
+ tracing::warn!(
+ target: "beava.wal",
+ kind = "wal.handrolled_tail_repaired",
+ path = %wal_path.display(),
+ before_bytes = data.len(),
+ after_bytes = parsed.valid_prefix_len,
+ "repaired hand-rolled WAL by truncating invalid tail before append"
+ );
+ Ok(())
+}
+
+fn sync_parent_dir(path: &Path) -> std::io::Result<()> {
+ let parent = path.parent().ok_or_else(|| {
+ std::io::Error::new(
+ std::io::ErrorKind::InvalidInput,
+ format!("path has no parent: {}", path.display()),
+ )
+ })?;
+ File::open(parent)?.sync_all()
+}
+
/// Return `true` if `path` is on a network filesystem (NFS, SMB, FUSE, etc.).
///
/// On macOS: uses `statfs` and checks `f_fstypename` for "nfs", "smbfs",
@@ -339,3 +636,233 @@ pub fn is_network_fs(path: &Path) -> bool {
false
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::time::{SystemTime, UNIX_EPOCH};
+
+ fn temp_dir(name: &str) -> PathBuf {
+ let nanos = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .unwrap()
+ .as_nanos();
+ let dir = std::env::temp_dir().join(format!(
+ "beava-wal-writer-{name}-{}-{nanos}",
+ std::process::id()
+ ));
+ std::fs::create_dir_all(&dir).unwrap();
+ dir
+ }
+
+ fn encode_v3(buf: &mut Vec, lsn: u64, event_name: &str, body: &[u8]) {
+ buf.push(0x03);
+ buf.extend_from_slice(&lsn.to_be_bytes());
+ buf.push(0x02); // JSON
+ buf.extend_from_slice(&1u32.to_be_bytes());
+ buf.extend_from_slice(&123u64.to_be_bytes());
+ let name_bytes = event_name.as_bytes();
+ buf.extend_from_slice(&(name_bytes.len() as u16).to_be_bytes());
+ buf.extend_from_slice(name_bytes);
+ buf.extend_from_slice(&(body.len() as u32).to_be_bytes());
+ buf.extend_from_slice(body);
+ }
+
+ #[test]
+ fn compact_wal_file_removes_snapshot_covered_v3_records() {
+ let dir = temp_dir("compact");
+ let path = dir.join("wal-0000000000000000.wal");
+ let mut bytes = Vec::new();
+ encode_v3(&mut bytes, 10, "Txn", br#"{"user_id":"a"}"#);
+ encode_v3(&mut bytes, 20, "Txn", br#"{"user_id":"b"}"#);
+ encode_v3(&mut bytes, 30, "Txn", br#"{"user_id":"c"}"#);
+ std::fs::write(&path, &bytes).unwrap();
+
+ let mut file = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&path)
+ .unwrap();
+ let stats = compact_wal_file(&mut file, &path, 20).unwrap();
+ let compacted = std::fs::read(&path).unwrap();
+ let records = parse_wal_record_bounds(&compacted).unwrap();
+
+ assert_eq!(stats.removed_records, 2);
+ assert_eq!(stats.retained_records, 1);
+ assert!(stats.after_bytes < stats.before_bytes);
+ assert_eq!(records.len(), 1);
+ assert_eq!(records[0].lsn, 30);
+
+ file.write_all(b"tail").unwrap();
+ file.sync_all().unwrap();
+ assert!(
+ std::fs::metadata(&path).unwrap().len() > stats.after_bytes,
+ "writer must keep appending to the compacted WAL path"
+ );
+
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+
+ #[test]
+ fn compact_wal_file_skips_when_no_records_are_covered() {
+ let dir = temp_dir("noop");
+ let path = dir.join("wal-0000000000000000.wal");
+ let mut bytes = Vec::new();
+ encode_v3(&mut bytes, 50, "Txn", br#"{"user_id":"a"}"#);
+ std::fs::write(&path, &bytes).unwrap();
+
+ let mut file = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&path)
+ .unwrap();
+ let stats = compact_wal_file(&mut file, &path, 20).unwrap();
+
+ assert_eq!(stats.removed_records, 0);
+ assert_eq!(stats.retained_records, 1);
+ assert_eq!(std::fs::read(&path).unwrap(), bytes);
+
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+
+ #[test]
+ fn compact_wal_file_ignores_truncated_tail_like_recovery() {
+ let dir = temp_dir("truncated-tail");
+ let path = dir.join("wal-0000000000000000.wal");
+ let mut bytes = Vec::new();
+ encode_v3(&mut bytes, 10, "Txn", br#"{"user_id":"covered"}"#);
+ encode_v3(&mut bytes, 30, "Txn", br#"{"user_id":"retained"}"#);
+ bytes.extend_from_slice(&[0x03, 0, 0, 0]); // torn v3 header
+ std::fs::write(&path, &bytes).unwrap();
+
+ let mut file = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&path)
+ .unwrap();
+ let stats = compact_wal_file(&mut file, &path, 10).unwrap();
+ let compacted = std::fs::read(&path).unwrap();
+ let records = parse_wal_record_bounds(&compacted).unwrap();
+
+ assert_eq!(stats.removed_records, 1);
+ assert_eq!(stats.retained_records, 1);
+ assert_eq!(records.len(), 1);
+ assert_eq!(records[0].lsn, 30);
+
+ file.write_all(b"tail").unwrap();
+ file.sync_all().unwrap();
+ assert!(
+ std::fs::metadata(&path).unwrap().len() > stats.after_bytes,
+ "writer must keep appending to the renamed compacted WAL"
+ );
+
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+
+ #[test]
+ fn compact_wal_file_repairs_truncated_tail_even_without_covered_records() {
+ let dir = temp_dir("truncated-tail-no-covered");
+ let path = dir.join("wal-0000000000000000.wal");
+ let mut bytes = Vec::new();
+ encode_v3(&mut bytes, 30, "Txn", br#"{"user_id":"retained"}"#);
+ let valid_len = bytes.len() as u64;
+ bytes.extend_from_slice(&[0x03, 0, 0, 0]); // torn v3 header
+ std::fs::write(&path, &bytes).unwrap();
+
+ let mut file = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&path)
+ .unwrap();
+ let stats = compact_wal_file(&mut file, &path, 10).unwrap();
+ let compacted = std::fs::read(&path).unwrap();
+ let records = parse_wal_record_bounds(&compacted).unwrap();
+
+ assert_eq!(stats.removed_records, 0);
+ assert_eq!(stats.retained_records, 1);
+ assert_eq!(stats.after_bytes, valid_len);
+ assert_eq!(records.len(), 1);
+ assert_eq!(records[0].lsn, 30);
+
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+
+ #[test]
+ fn wal_writer_new_repairs_invalid_tail_before_append() {
+ let dir = temp_dir("startup-repair");
+ let path = dir.join("wal-0000000000000000.wal");
+ let mut bytes = Vec::new();
+ encode_v3(&mut bytes, 10, "Txn", br#"{"user_id":"covered"}"#);
+ let valid_len = bytes.len();
+ bytes.extend_from_slice(b"stale-garbage-tail");
+ std::fs::write(&path, &bytes).unwrap();
+
+ let ring = Arc::new(WalBufferRing::new(2, 4096, Arc::new(WalLsn::new())));
+ let lsn = Arc::new(WalLsn::new());
+ let mut writer = WalWriter::new(&dir, ring, lsn, 1).expect("WalWriter::new repairs tail");
+
+ assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_len as u64);
+
+ let mut appended = Vec::new();
+ encode_v3(&mut appended, 20, "Txn", br#"{"user_id":"retained"}"#);
+ writer.file.write_all(&appended).unwrap();
+ writer.file.sync_all().unwrap();
+
+ let repaired = std::fs::read(&path).unwrap();
+ let records = parse_wal_record_bounds(&repaired).unwrap();
+ assert_eq!(records.len(), 2);
+ assert_eq!(records[0].lsn, 10);
+ assert_eq!(records[1].lsn, 20);
+
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+
+ #[test]
+ fn compact_wal_file_replaces_stale_tmp_from_prior_crash() {
+ let dir = temp_dir("stale-tmp");
+ let path = dir.join("wal-0000000000000000.wal");
+ let mut bytes = Vec::new();
+ encode_v3(&mut bytes, 10, "Txn", br#"{"user_id":"a"}"#);
+ std::fs::write(&path, &bytes).unwrap();
+ std::fs::write(path.with_extension("wal.compact.tmp"), b"stale").unwrap();
+
+ let mut file = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&path)
+ .unwrap();
+ let stats = compact_wal_file(&mut file, &path, 10).unwrap();
+
+ assert_eq!(stats.removed_records, 1);
+ assert_eq!(stats.after_bytes, 0);
+ assert_eq!(std::fs::read(&path).unwrap(), b"");
+
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+
+ #[test]
+ fn reclaim_request_can_exceed_data_plane_synced_lsn_for_registry_lsn_gaps() {
+ let dir = temp_dir("registry-gap");
+ let path = dir.join("wal-0000000000000000.wal");
+ let mut bytes = Vec::new();
+ encode_v3(&mut bytes, 10, "Txn", br#"{"user_id":"a"}"#);
+ std::fs::write(&path, &bytes).unwrap();
+
+ let mut file = OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&path)
+ .unwrap();
+ let lsn = WalLsn::new_at(10);
+ let reclaim = WalReclaimHandle::new();
+ reclaim.request_reclaim_up_to(100);
+ let mut reclaimed_lsn = 0;
+
+ maybe_reclaim_wal_file(&mut file, &path, &lsn, &reclaim, &mut reclaimed_lsn);
+
+ assert_eq!(reclaimed_lsn, 100);
+ assert_eq!(std::fs::read(&path).unwrap(), b"");
+
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+}
diff --git a/crates/beava-server/src/apply_shard.rs b/crates/beava-server/src/apply_shard.rs
index 5677ab71..8a9fe7f3 100644
--- a/crates/beava-server/src/apply_shard.rs
+++ b/crates/beava-server/src/apply_shard.rs
@@ -12,7 +12,7 @@
use crate::register::RegisterPayload;
use crate::runtime_core_glue::GlueResponse;
use crate::AppState;
-use beava_core::row::Row;
+use beava_core::row::{Row, Value};
use beava_runtime_core::wal_buffer::WalBufferRing;
use beava_runtime_core::wal_lsn::WalLsn;
use beava_runtime_core::wire_request::WireRequest;
@@ -214,11 +214,9 @@ impl ApplyShard {
// categorised-lists schema. `dry_run` fires first (returns
// the diff JSON without applying). The force gate runs
// second: destructive entries without `force=true` return
- // 409 + `force_required`; with `force=true` we eagerly
- // remove the conflicting descriptors so
- // `execute_register_with_wal` treats the payload as
- // additive (`registry_version` bumps, WAL records the
- // change).
+ // 409 + `force_required`; with `force=true` we compute the
+ // exact removal set but leave the live registry untouched
+ // until the WAL-backed register path has fsynced the bump.
let (force, dry_run) = match serde_json::from_slice::(&payload) {
Ok(v) => (
v.get("force").and_then(|x| x.as_bool()).unwrap_or(false),
@@ -263,15 +261,11 @@ impl ApplyShard {
// `force=true` carries the explicit "drop existing state and
// replace fully" intent for any descriptor that already
- // exists in the registry. We pre-remove every existing
- // descriptor that the new payload would change before
- // invoking `execute_register_with_wal`, which delegates
- // installation to `Registry::apply_registration` (silently
- // skips descriptors already present by name). Without
- // pre-removal, an additive change against an existing
- // descriptor (e.g. NewField on an existing event source)
- // would no-op silently. Two reasons we can't gate solely
- // on `diff.destructive`:
+ // exists in the registry. Compute every existing descriptor
+ // the new payload would change and record it in the durable
+ // `RegistryBump`; the WAL-backed path applies the removal
+ // after fsync. Two reasons we can't gate solely on
+ // `diff.destructive`:
//
// 1. `classify_register_diff` classifies new fields on
// an existing event source and new aggregations in
@@ -283,12 +277,11 @@ impl ApplyShard {
// skip — it's genuinely-new (not in current
// registry), so there's nothing to pre-remove.
//
- // Pre-removal drops compiled chains, aggregations, feature
- // index entries, and accumulated state for the named
- // descriptors. That state loss is the documented `force=true`
- // semantic; callers who want to add a field without losing
- // state should send the request without `force` (additive
- // path is allowed by `register_check_force_required`).
+ // Forced removal drops compiled chains, aggregations, feature
+ // index entries, and the old agg slots become unreachable.
+ // That state loss is the documented `force=true` semantic;
+ // callers who want to add a field without losing state should
+ // send the request without `force`.
let mut force_removed: Vec = Vec::new();
if force {
let mut to_remove: Vec = Vec::new();
@@ -385,12 +378,6 @@ impl ApplyShard {
to_remove.dedup();
}
- if !to_remove.is_empty() {
- self.state
- .dev_agg
- .registry
- .force_remove_descriptors(&to_remove);
- }
// Captured for the RegistryBump WAL record so recovery
// can replay the same force-removal closure BEFORE
// re-applying `payload_nodes` — without this, replay
@@ -402,6 +389,11 @@ impl ApplyShard {
// function on a one-shot single-threaded tokio runtime so
// we don't pull tokio into the hot path.
let state_clone = Arc::clone(&self.state);
+ let registry_bump_min_lsn = state_clone
+ .dev_agg
+ .next_event_id
+ .load(std::sync::atomic::Ordering::Acquire)
+ .saturating_add(1);
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
@@ -411,6 +403,7 @@ impl ApplyShard {
reg_payload,
&state_clone.wal_sink,
force_removed,
+ registry_bump_min_lsn,
));
// Grow `state_tables` so the hot path can index by
// `desc.agg_id` without bounds issues. Register is rare,
@@ -903,6 +896,13 @@ impl ApplyShard {
};
let t_parse = t0.map(|t| t.elapsed());
+ if string_has_control_chars(event_name) || row_has_control_chars(&row) {
+ return GlueResponse::PushError {
+ code: "control_character_in_string",
+ registry_version,
+ };
+ }
+
// 2. Lookup event descriptor (Arc-backed; refcount bump only).
let descriptor = match self.state.dev_agg.registry.get_event_descriptor(event_name) {
Some(d) => d,
@@ -987,9 +987,9 @@ impl ApplyShard {
let now_ms_i64: i64 = now_ms as i64;
let t_validate = t0.map(|t| t.elapsed());
- // 6. Serialize WAL payload — v=2 binary format.
+ // 6. Serialize WAL payload — v=3 binary format.
//
- // `[u8 v=2][u8 body_format][u32 rv BE][u64 et_ms BE]
+ // `[u8 v=3][u64 assigned_lsn BE][u8 body_format][u32 rv BE][u64 et_ms BE]
// [u16 event_name_len BE][N bytes name][u32 body_len BE][M bytes body]`
//
// `body` is the exact bytes received from the wire — zero-copy
@@ -999,9 +999,13 @@ impl ApplyShard {
let name_bytes = event_name.as_bytes();
let name_len = name_bytes.len() as u16;
let body_len = body.len() as u32;
- let mut payload_bytes =
- Vec::with_capacity(1 + 1 + 4 + 8 + 2 + name_bytes.len() + 4 + body.len());
- payload_bytes.push(0x02u8);
+ let payload_len = 1 + 8 + 1 + 4 + 8 + 2 + name_bytes.len() + 4 + body.len();
+ self.wal_lsn
+ .mark_committed_at_least(self.state.wal_sink.durable_lsn());
+ let expected_ack_lsn = self.wal_lsn.committed().saturating_add(payload_len as u64);
+ let mut payload_bytes = Vec::with_capacity(payload_len);
+ payload_bytes.push(0x03u8);
+ payload_bytes.extend_from_slice(&expected_ack_lsn.to_be_bytes());
payload_bytes.push(body_format);
payload_bytes.extend_from_slice(®istry_version.to_be_bytes());
payload_bytes.extend_from_slice(&now_ms.to_be_bytes());
@@ -1013,6 +1017,7 @@ impl ApplyShard {
// 7. WAL append — lock-free, no Mutex, no channel.
let ack_lsn = self.wal_ring.append(&payload_bytes);
+ debug_assert_eq!(ack_lsn, expected_ack_lsn);
let t_wal_append = t0.map(|t| t.elapsed());
// 8. Apply to aggregations under the table lock (uncontended on
@@ -1032,6 +1037,16 @@ impl ApplyShard {
&mut tables,
descriptor.cold_after_ms,
);
+ self.state
+ .dev_agg
+ .next_event_id
+ .fetch_max(ack_lsn, Ordering::Release);
+ if now_ms > 0 {
+ self.state
+ .dev_agg
+ .query_time_ms
+ .fetch_max(now_ms, Ordering::Release);
+ }
// Refresh the process-static `beava_entity_count_resident`
// gauge under the lock we already hold. O(N_tables) sum of
// three `HashMap.len()` reads; typically < 30 tables (one
@@ -1042,20 +1057,9 @@ impl ApplyShard {
}
let t_agg = t0.map(|t| t.elapsed());
- // 9. Bump monotonic event counters. `query_time_ms` is fed
- // `now_ms` (server wall-clock at apply); the GET path's
- // `compute_query_time_ms` reads this watermark to surface a
- // meaningful query time for windowed-op queries.
- self.state
- .dev_agg
- .next_event_id
- .fetch_max(ack_lsn, Ordering::Relaxed);
- if now_ms > 0 {
- self.state
- .dev_agg
- .query_time_ms
- .fetch_max(now_ms, Ordering::Relaxed);
- }
+ // 9. Monotonic counters were bumped while holding `state_tables`, so a
+ // snapshot cannot observe applied table state without the matching
+ // applied LSN/query-time watermark.
let t_bk_counters = t0.map(|t| t.elapsed());
// `t_bk_evid` keeps the trace shape stable; the per-stage delta
@@ -1198,6 +1202,38 @@ fn extract_dedupe_str_from_row(row: &beava_core::row::Row, key: &str) -> Option<
})
}
+fn row_has_control_chars(row: &Row) -> bool {
+ row.iter()
+ .any(|(field, value)| string_has_control_chars(field) || value_has_control_chars(value))
+}
+
+fn value_has_control_chars(value: &Value) -> bool {
+ match value {
+ Value::Str(s) => string_has_control_chars(s),
+ Value::Json(jv) => json_value_has_control_chars(jv),
+ Value::List(values) => values.iter().any(value_has_control_chars),
+ Value::Map(map) => map
+ .iter()
+ .any(|(key, value)| string_has_control_chars(key) || value_has_control_chars(value)),
+ _ => false,
+ }
+}
+
+fn json_value_has_control_chars(value: &serde_json::Value) -> bool {
+ match value {
+ serde_json::Value::String(s) => string_has_control_chars(s),
+ serde_json::Value::Array(values) => values.iter().any(json_value_has_control_chars),
+ serde_json::Value::Object(map) => map.iter().any(|(key, value)| {
+ string_has_control_chars(key) || json_value_has_control_chars(value)
+ }),
+ _ => false,
+ }
+}
+
+fn string_has_control_chars(value: &str) -> bool {
+ value.chars().any(char::is_control)
+}
+
/// One entry in the `OP_BATCH_GET` request payload.
///
/// - `table`: aggregation node name.
diff --git a/crates/beava-server/src/http_admin.rs b/crates/beava-server/src/http_admin.rs
index 8314f115..64a335bc 100644
--- a/crates/beava-server/src/http_admin.rs
+++ b/crates/beava-server/src/http_admin.rs
@@ -93,6 +93,9 @@ async fn metrics_handler(State(state): State) -> impl IntoResponse {
// `categories_capped`; top-k displacement and histogram bucket drops
// join when those operator internals expose hooks.
let op_cap_hits = entropy_capped;
+ let snapshot_metrics = crate::snapshot_metrics::snapshot();
+ let snapshot_duration_seconds = snapshot_metrics.last_duration_us as f64 / 1_000_000.0;
+ let snapshot_fsync_seconds = snapshot_metrics.last_fsync_us as f64 / 1_000_000.0;
let body = format!(
"# HELP beava_registry_version Registry version (monotonic).\n\
@@ -121,7 +124,16 @@ async fn metrics_handler(State(state): State) -> impl IntoResponse {
beava_bucket_reclaim_total {bucket_reclaims}\n\
# HELP beava_bytes_per_entity_p99 Static estimate of per-entity memory footprint (~7 KB for a rich 30-feature pack).\n\
# TYPE beava_bytes_per_entity_p99 gauge\n\
- beava_bytes_per_entity_p99 {bytes_per_entity_p99}\n",
+ beava_bytes_per_entity_p99 {bytes_per_entity_p99}\n\
+ # HELP beava_snapshot_last_duration_seconds Wall-clock duration of the last successful snapshot.\n\
+ # TYPE beava_snapshot_last_duration_seconds gauge\n\
+ beava_snapshot_last_duration_seconds {snapshot_duration_seconds:.6}\n\
+ # HELP beava_snapshot_last_bytes Bytes written by the last successful snapshot, including snapshot header and body.\n\
+ # TYPE beava_snapshot_last_bytes gauge\n\
+ beava_snapshot_last_bytes {snapshot_bytes}\n\
+ # HELP beava_snapshot_last_fsync_seconds File plus parent-directory fsync time for the last successful snapshot.\n\
+ # TYPE beava_snapshot_last_fsync_seconds gauge\n\
+ beava_snapshot_last_fsync_seconds {snapshot_fsync_seconds:.6}\n",
registry_version = snap.version,
node_count = snap.node_count,
entropy_capped = entropy_capped,
@@ -130,6 +142,9 @@ async fn metrics_handler(State(state): State) -> impl IntoResponse {
entity_count = entity_count,
bucket_reclaims = bucket_reclaims,
bytes_per_entity_p99 = BYTES_PER_ENTITY_P99_V0_PLACEHOLDER,
+ snapshot_duration_seconds = snapshot_duration_seconds,
+ snapshot_bytes = snapshot_metrics.last_bytes,
+ snapshot_fsync_seconds = snapshot_fsync_seconds,
);
(
diff --git a/crates/beava-server/src/lib.rs b/crates/beava-server/src/lib.rs
index 2f31f6b1..41662025 100644
--- a/crates/beava-server/src/lib.rs
+++ b/crates/beava-server/src/lib.rs
@@ -18,7 +18,10 @@ pub mod registry_debug;
pub mod runtime_core_glue;
pub mod server;
pub mod shutdown;
+pub mod snapshot_fork;
+pub(crate) mod snapshot_metrics;
pub mod snapshot_task;
+pub mod thp;
pub mod wal_config;
#[cfg(any(feature = "testing", test))]
diff --git a/crates/beava-server/src/recovery.rs b/crates/beava-server/src/recovery.rs
index 81e2e84a..dcc69665 100644
--- a/crates/beava-server/src/recovery.rs
+++ b/crates/beava-server/src/recovery.rs
@@ -2,7 +2,8 @@
//!
//! 1. `load_snapshot_if_any(dir, dev_agg)` — descending-LSN scan; first valid
//! snapshot wins; install registry descriptors + state tables; return its
-//! LSN. Empty dir or all-corrupt files → 0 (cold start).
+//! snapshot LSN plus the applied data-plane watermark stored in the body.
+//! Empty dir or all-corrupt files → 0 (cold start).
//! 2. `replay_wal_from_lsn(wal_dir, snapshot_lsn, dev_agg)` — replay every WAL
//! record with `lsn > snapshot_lsn` in LSN order: `Event` decodes its
//! payload and feeds `apply_event_to_aggregations`; `RegistryBump`
@@ -17,7 +18,7 @@ use beava_core::row::{Row, Value};
use beava_core::snapshot_body::SnapshotBody;
use beava_persistence::{list_snapshots, Lsn, PersistError, RecordType, SnapshotReader, WalReader};
use serde::Deserialize;
-use std::path::Path;
+use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering;
/// Outcome counters reported back from `replay_wal_from_lsn`.
@@ -27,16 +28,125 @@ pub struct RecoveryOutcome {
pub snapshot_lsn: Lsn,
pub replay_event_count: u64,
pub replay_registry_bumps: u64,
+ pub quarantined_records: u64,
+ pub applied_registry_bump_after_snapshot: bool,
pub last_lsn: Lsn,
}
+/// Snapshot recovery result. `snapshot_lsn` comes from the snapshot header and
+/// gates legacy persistence-WAL replay. `applied_lsn` comes from the snapshot
+/// body and gates hand-rolled data-plane WAL replay, because older snapshots
+/// can have a header LSN that does not match the data-plane state already
+/// serialized into the body.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub struct SnapshotLoadOutcome {
+ pub snapshot_lsn: Lsn,
+ pub applied_lsn: Lsn,
+}
+
+#[derive(Debug, Clone, Copy)]
+enum WalQuarantineKind {
+ HandrolledJsonBody,
+ HandrolledMsgpackBody,
+ PersistenceEventPayload,
+}
+
+impl WalQuarantineKind {
+ fn as_str(self) -> &'static str {
+ match self {
+ WalQuarantineKind::HandrolledJsonBody => "handrolled-json-body",
+ WalQuarantineKind::HandrolledMsgpackBody => "handrolled-msgpack-body",
+ WalQuarantineKind::PersistenceEventPayload => "persistence-event-payload",
+ }
+ }
+}
+
+fn wal_quarantine_marker_path(wal_dir: &Path, lsn: Lsn, kind: WalQuarantineKind) -> PathBuf {
+ wal_dir
+ .join("quarantine")
+ .join(format!("lsn-{lsn:016x}-{}.json", kind.as_str()))
+}
+
+fn wal_quarantine_marker_exists(wal_dir: &Path, lsn: Lsn, kind: WalQuarantineKind) -> bool {
+ wal_quarantine_marker_path(wal_dir, lsn, kind).exists()
+}
+
+fn quarantine_wal_decode_failure(
+ wal_dir: &Path,
+ lsn: Lsn,
+ kind: WalQuarantineKind,
+ error: &dyn std::fmt::Display,
+) {
+ let marker = wal_quarantine_marker_path(wal_dir, lsn, kind);
+ if marker.exists() {
+ return;
+ }
+
+ let Some(dir) = marker.parent() else {
+ return;
+ };
+ if let Err(e) = std::fs::create_dir_all(dir) {
+ tracing::warn!(
+ target: "beava.recovery",
+ kind = "recovery.wal_quarantine_write_failed",
+ lsn,
+ quarantine_kind = kind.as_str(),
+ error = %e,
+ "failed to create WAL quarantine directory"
+ );
+ return;
+ }
+
+ let body = serde_json::json!({
+ "lsn": lsn,
+ "kind": kind.as_str(),
+ "reason": error.to_string(),
+ });
+ let bytes = serde_json::to_vec_pretty(&body).unwrap_or_default();
+ let tmp = marker.with_extension(format!("json.tmp-{}", std::process::id()));
+ if let Err(e) = std::fs::write(&tmp, bytes) {
+ tracing::warn!(
+ target: "beava.recovery",
+ kind = "recovery.wal_quarantine_write_failed",
+ lsn,
+ quarantine_kind = kind.as_str(),
+ error = %e,
+ "failed to write WAL quarantine marker"
+ );
+ return;
+ }
+ if let Err(e) = std::fs::rename(&tmp, &marker) {
+ let _ = std::fs::remove_file(&tmp);
+ tracing::warn!(
+ target: "beava.recovery",
+ kind = "recovery.wal_quarantine_write_failed",
+ lsn,
+ quarantine_kind = kind.as_str(),
+ marker = %marker.display(),
+ error = %e,
+ "failed to commit WAL quarantine marker"
+ );
+ return;
+ }
+
+ tracing::warn!(
+ target: "beava.recovery",
+ kind = "recovery.wal_record_quarantined",
+ lsn,
+ quarantine_kind = kind.as_str(),
+ marker = %marker.display(),
+ error = %error,
+ "WAL record decode failed; quarantined for future recovery passes"
+ );
+}
+
/// Scan `snapshot_dir` for the highest-LSN valid snapshot; install its
/// registry descriptors + state tables into `app_state`. Returns the
/// snapshot's LSN, or 0 if no valid snapshot exists (cold start).
pub fn load_snapshot_if_any(
snapshot_dir: &Path,
dev_agg: &DevAggState,
-) -> Result {
+) -> Result {
let snaps = list_snapshots(snapshot_dir)?;
for (lsn, path) in snaps {
match SnapshotReader::open(&path) {
@@ -58,13 +168,28 @@ pub fn load_snapshot_if_any(
new_next_agg_id,
);
for (node_name, entries) in state_tables {
- let agg_id = match dev_agg.registry.compiled_aggregation(&node_name) {
- Some(d) => d.agg_id as usize,
+ let agg_desc = match dev_agg.registry.compiled_aggregation(&node_name) {
+ Some(d) => d,
None => continue,
};
+ let agg_id = agg_desc.agg_id as usize;
+ let group_key_types = dev_agg
+ .registry
+ .get_event_descriptor(&agg_desc.source_node_name)
+ .map(|event| {
+ agg_desc
+ .group_keys
+ .iter()
+ .filter_map(|key| event.schema.fields.get(key).cloned())
+ .collect::>()
+ });
let tbl = &mut tables[agg_id];
for (key, ops) in entries {
- tbl.insert_from_entity_key(key, ops);
+ tbl.insert_from_entity_key_with_types(
+ key,
+ ops,
+ group_key_types.as_deref(),
+ );
}
}
}
@@ -80,10 +205,14 @@ pub fn load_snapshot_if_any(
target: "beava.recovery",
kind = "recovery.snapshot_loaded",
snapshot_lsn = lsn,
+ applied_lsn = next_event_id,
registry_version = header.registry_version,
"loaded snapshot"
);
- return Ok(lsn);
+ return Ok(SnapshotLoadOutcome {
+ snapshot_lsn: lsn,
+ applied_lsn: next_event_id,
+ });
}
Err(e) => {
tracing::warn!(
@@ -108,7 +237,7 @@ pub fn load_snapshot_if_any(
}
}
}
- Ok(0)
+ Ok(SnapshotLoadOutcome::default())
}
/// JSON shape of a WAL Event record's payload (matches push.rs encoder).
@@ -127,8 +256,9 @@ struct WalEventPayload {
b: serde_json::Value,
}
-/// A single decoded v=2 record from the hand-rolled WAL file.
-struct V2Record {
+/// A single decoded record from the hand-rolled WAL file.
+struct HandrolledWalRecord {
+ lsn: Lsn,
body_format: u8,
// reason: parsed from the v=2 record header for completeness; recovery
// doesn't depend on the per-record registry version.
@@ -139,29 +269,49 @@ struct V2Record {
body: Vec,
}
-/// Parse all v=2 binary records from a contiguous byte slice.
+/// Parse all hand-rolled binary records from a contiguous byte slice.
///
-/// Format: `[u8 v=2][u8 body_format][u32 rv BE][u64 et_ms BE]
-/// [u16 name_len BE][N bytes name][u32 body_len BE][M bytes body]`
+/// v=2 format:
+/// `[u8 v=2][u8 body_format][u32 rv BE][u64 et_ms BE]
+/// [u16 name_len BE][N bytes name][u32 body_len BE][M bytes body]`
///
-/// Stops at first byte that is not 0x02 (unknown version) or if bytes are
+/// v=3 format:
+/// `[u8 v=3][u64 assigned_lsn BE][u8 body_format][u32 rv BE][u64 et_ms BE]
+/// [u16 name_len BE][N bytes name][u32 body_len BE][M bytes body]`
+///
+/// Stops at first byte that is not 0x02/0x03 (unknown version) or if bytes are
/// insufficient (truncated record — treat as EOF).
-fn parse_v2_records(data: &[u8]) -> Vec {
+fn parse_handrolled_records(data: &[u8], base_lsn: Lsn) -> Vec {
let mut records = Vec::new();
let mut pos = 0usize;
loop {
- // Fixed header is 1+1+4+8+2 = 16 bytes.
- if pos + 16 > data.len() {
+ if pos >= data.len() {
break;
}
let version = data[pos];
- if version != 0x02 {
+ if version != 0x02 && version != 0x03 {
break;
}
pos += 1;
+ let assigned_lsn = if version == 0x03 {
+ if pos + 8 > data.len() {
+ break;
+ }
+ let lsn = u64::from_be_bytes(data[pos..pos + 8].try_into().unwrap());
+ pos += 8;
+ Some(lsn)
+ } else {
+ None
+ };
+
+ // Remaining fixed header is 1+4+8+2 = 15 bytes.
+ if pos + 15 > data.len() {
+ break;
+ }
+
let body_format = data[pos];
pos += 1;
@@ -194,7 +344,8 @@ fn parse_v2_records(data: &[u8]) -> Vec {
let body = data[pos..pos + body_len].to_vec();
pos += body_len;
- records.push(V2Record {
+ records.push(HandrolledWalRecord {
+ lsn: assigned_lsn.unwrap_or_else(|| base_lsn.saturating_add(pos as u64)),
body_format,
rv,
et_ms,
@@ -209,17 +360,17 @@ fn parse_v2_records(data: &[u8]) -> Vec {
/// Replay hand-rolled WAL files (`*.wal`) from `wal_dir`.
///
/// Hand-rolled WAL files are written by `WalBufferRing` + `WalWriter` in the
-/// v=2 binary record format (see `dispatch_push_sync` in `apply_shard`),
+/// binary push-record format (see `dispatch_push_sync` in `apply_shard`),
/// distinct from the `beava-persistence` `WalSink` format (`*.log`). Returns
-/// recovery counters; assigns monotonic LSNs starting from `lsn_start`.
+/// recovery counters and replays only records with `lsn > from_lsn_exclusive`.
pub fn replay_handrolled_wal_dir(
wal_dir: &Path,
- lsn_start: Lsn,
+ from_lsn_exclusive: Lsn,
dev_agg: &DevAggState,
) -> Result {
use beava_core::wire::CT_MSGPACK;
let mut outcome = RecoveryOutcome {
- snapshot_lsn: lsn_start.saturating_sub(1),
+ snapshot_lsn: from_lsn_exclusive,
..Default::default()
};
@@ -235,27 +386,42 @@ pub fn replay_handrolled_wal_dir(
.collect();
wal_files.sort();
- let mut next_lsn = lsn_start;
+ let mut base_lsn = 0;
for wal_file in &wal_files {
let data = std::fs::read(wal_file)?;
- let records = parse_v2_records(&data);
+ let records = parse_handrolled_records(&data, base_lsn);
+ base_lsn = base_lsn.saturating_add(data.len() as u64);
for rec in records {
- let lsn = next_lsn;
- next_lsn += 1;
+ if rec.lsn <= from_lsn_exclusive {
+ continue;
+ }
+ outcome.last_lsn = outcome.last_lsn.max(rec.lsn);
+
+ let quarantine_kind = if rec.body_format == CT_MSGPACK {
+ WalQuarantineKind::HandrolledMsgpackBody
+ } else {
+ WalQuarantineKind::HandrolledJsonBody
+ };
+ if wal_quarantine_marker_exists(wal_dir, rec.lsn, quarantine_kind) {
+ outcome.quarantined_records += 1;
+ tracing::debug!(
+ target: "beava.recovery",
+ kind = "recovery.wal_quarantine_skip",
+ lsn = rec.lsn,
+ quarantine_kind = quarantine_kind.as_str(),
+ "skipping quarantined WAL record"
+ );
+ continue;
+ }
let row: Row = if rec.body_format == CT_MSGPACK {
match rmp_serde::from_slice::(&rec.body) {
Ok(r) => r,
Err(e) => {
- tracing::warn!(
- target: "beava.recovery",
- kind = "recovery.v2_msgpack_decode_failed",
- lsn = lsn,
- error = %e,
- "v=2 msgpack body decode failed; skipping"
- );
+ quarantine_wal_decode_failure(wal_dir, rec.lsn, quarantine_kind, &e);
+ outcome.quarantined_records += 1;
continue;
}
}
@@ -263,13 +429,8 @@ pub fn replay_handrolled_wal_dir(
match serde_json::from_slice::(&rec.body) {
Ok(r) => r,
Err(e) => {
- tracing::warn!(
- target: "beava.recovery",
- kind = "recovery.v2_json_decode_failed",
- lsn = lsn,
- error = %e,
- "v=2 JSON body decode failed; skipping"
- );
+ quarantine_wal_decode_failure(wal_dir, rec.lsn, quarantine_kind, &e);
+ outcome.quarantined_records += 1;
continue;
}
}
@@ -298,7 +459,7 @@ pub fn replay_handrolled_wal_dir(
&rec.event_name,
&row,
rec.et_ms,
- lsn,
+ rec.lsn,
rec.rv as u64,
&dev_agg.registry,
&mut tables,
@@ -306,14 +467,13 @@ pub fn replay_handrolled_wal_dir(
);
}
- dev_agg.next_event_id.fetch_max(lsn, Ordering::Relaxed);
+ dev_agg.next_event_id.fetch_max(rec.lsn, Ordering::Relaxed);
if rec.et_ms > 0 {
dev_agg
.query_time_ms
.fetch_max(rec.et_ms as u64, Ordering::Relaxed);
}
outcome.replay_event_count += 1;
- outcome.last_lsn = lsn;
}
}
@@ -340,22 +500,29 @@ pub fn replay_wal_from_lsn(
}
let records = WalReader::read_all(wal_dir)?;
for rec in records {
- if rec.lsn <= from_lsn_exclusive {
- continue;
- }
- outcome.last_lsn = outcome.last_lsn.max(rec.lsn);
match rec.record_type {
RecordType::Event => {
+ if rec.lsn <= from_lsn_exclusive {
+ continue;
+ }
+ outcome.last_lsn = outcome.last_lsn.max(rec.lsn);
+ let quarantine_kind = WalQuarantineKind::PersistenceEventPayload;
+ if wal_quarantine_marker_exists(wal_dir, rec.lsn, quarantine_kind) {
+ outcome.quarantined_records += 1;
+ tracing::debug!(
+ target: "beava.recovery",
+ kind = "recovery.wal_quarantine_skip",
+ lsn = rec.lsn,
+ quarantine_kind = quarantine_kind.as_str(),
+ "skipping quarantined WAL record"
+ );
+ continue;
+ }
let payload: WalEventPayload = match serde_json::from_slice(&rec.payload) {
Ok(p) => p,
Err(e) => {
- tracing::warn!(
- target: "beava.recovery",
- kind = "recovery.event_decode_failed",
- lsn = rec.lsn,
- error = %e,
- "event payload decode failed; skipping"
- );
+ quarantine_wal_decode_failure(wal_dir, rec.lsn, quarantine_kind, &e);
+ outcome.quarantined_records += 1;
continue;
}
};
@@ -389,27 +556,41 @@ pub fn replay_wal_from_lsn(
outcome.replay_event_count += 1;
}
RecordType::RegistryBump => match RegistryBumpPayload::decode(&rec.payload) {
- Ok(bump) => match crate::register::apply_registry_bump(&dev_agg.registry, bump) {
- Ok(()) => {
- outcome.replay_registry_bumps += 1;
+ Ok(bump) => {
+ outcome.last_lsn = outcome.last_lsn.max(rec.lsn);
+ if bump.new_version <= dev_agg.registry.version() {
+ continue;
}
- Err(e) => {
- // Apply-after-fsync invariant: a durable RegistryBump
- // that fails to apply is a hard recovery failure —
- // silently skipping would let durable corruption hide.
- tracing::error!(
- target: "beava.recovery",
- kind = "recovery.registry_bump_apply_failed",
- lsn = rec.lsn,
- error = %e,
- "RegistryBump apply failed during replay"
- );
- return Err(PersistError::Io(std::io::Error::other(format!(
- "RegistryBump apply failed at LSN {}: {e}",
- rec.lsn
- ))));
+ match crate::register::apply_registry_bump(&dev_agg.registry, bump) {
+ Ok(()) => {
+ {
+ let mut tables = dev_agg.state_tables.lock();
+ beava_core::agg_state_table::ensure_capacity_for(
+ &mut tables,
+ dev_agg.registry.next_agg_id() as usize,
+ );
+ }
+ outcome.replay_registry_bumps += 1;
+ outcome.applied_registry_bump_after_snapshot = true;
+ }
+ Err(e) => {
+ // Apply-after-fsync invariant: a durable RegistryBump
+ // that fails to apply is a hard recovery failure —
+ // silently skipping would let durable corruption hide.
+ tracing::error!(
+ target: "beava.recovery",
+ kind = "recovery.registry_bump_apply_failed",
+ lsn = rec.lsn,
+ error = %e,
+ "RegistryBump apply failed during replay"
+ );
+ return Err(PersistError::Io(std::io::Error::other(format!(
+ "RegistryBump apply failed at LSN {}: {e}",
+ rec.lsn
+ ))));
+ }
}
- },
+ }
Err(e) => {
tracing::error!(
target: "beava.recovery",
@@ -433,6 +614,7 @@ pub fn replay_wal_from_lsn(
snapshot_lsn = outcome.snapshot_lsn,
events_replayed = outcome.replay_event_count,
registry_bumps_replayed = outcome.replay_registry_bumps,
+ quarantined_records = outcome.quarantined_records,
last_lsn = outcome.last_lsn,
"recovery complete"
);
@@ -463,3 +645,310 @@ fn json_object_to_row(jv: &serde_json::Value) -> Row {
}
row
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use beava_core::agg_op::AggOp;
+ use beava_core::agg_state::CountState;
+ use beava_core::agg_state_table::{ensure_capacity_for, AggStateTable, EntityKey};
+ use beava_core::registry::Registry;
+ use beava_persistence::{RecordType, SnapshotWriter, WalRecord};
+ use compact_str::CompactString;
+ use serde_json::json;
+ use smallvec::smallvec;
+ use std::sync::Arc;
+
+ fn txn_register_payload() -> crate::register::RegisterPayload {
+ serde_json::from_value(json!({
+ "nodes": [
+ {
+ "kind": "event",
+ "name": "Txn",
+ "schema": {"fields": {
+ "event_time": "i64",
+ "user_id": "str",
+ "amount": "f64"
+ }, "optional_fields": []}
+ },
+ {
+ "kind": "derivation",
+ "name": "TxnAgg",
+ "output_kind": "table",
+ "upstreams": ["Txn"],
+ "ops": [{"op": "group_by", "keys": ["user_id"], "agg": {
+ "cnt": {"op": "count", "params": {}}
+ }}],
+ "schema": {"fields": {"user_id": "str", "cnt": "i64"}, "optional_fields": []},
+ "table_primary_key": ["user_id"]
+ }
+ ]
+ }))
+ .expect("valid register payload")
+ }
+
+ fn install_txn_registry(dev_agg: &DevAggState) {
+ let payload = txn_register_payload();
+ let bump = crate::register::RegistryBumpPayload {
+ new_version: 1,
+ payload_nodes: payload.nodes,
+ force_removed_descriptors: Vec::new(),
+ };
+ crate::register::apply_registry_bump(&dev_agg.registry, bump)
+ .expect("install txn registry");
+ let mut tables = dev_agg.state_tables.lock();
+ ensure_capacity_for(&mut tables, dev_agg.registry.next_agg_id() as usize);
+ }
+
+ fn put_alice_count(dev_agg: &DevAggState, count: u64) {
+ let mut tables = dev_agg.state_tables.lock();
+ ensure_capacity_for(&mut tables, 1);
+ let mut table = AggStateTable::new();
+ let entity_key = EntityKey(smallvec![(
+ CompactString::from("user_id"),
+ Value::Str(CompactString::from("alice")),
+ )]);
+ table.insert_from_entity_key(entity_key, vec![AggOp::Count(CountState { n: count })]);
+ tables[0] = table;
+ }
+
+ fn alice_count(dev_agg: &DevAggState) -> u64 {
+ let tables = dev_agg.state_tables.lock();
+ let Some(ops) = tables
+ .first()
+ .and_then(|table| table.single_str.get("alice"))
+ else {
+ return 0;
+ };
+ match ops.first() {
+ Some(AggOp::Count(count)) => count.n,
+ _ => 0,
+ }
+ }
+
+ fn encode_handrolled_v3(
+ buf: &mut Vec,
+ lsn: Lsn,
+ body_format: u8,
+ rv: u32,
+ event_name: &str,
+ body: &[u8],
+ ) {
+ buf.push(0x03);
+ buf.extend_from_slice(&lsn.to_be_bytes());
+ buf.push(body_format);
+ buf.extend_from_slice(&rv.to_be_bytes());
+ buf.extend_from_slice(&(123i64).to_be_bytes());
+ buf.extend_from_slice(&(event_name.len() as u16).to_be_bytes());
+ buf.extend_from_slice(event_name.as_bytes());
+ buf.extend_from_slice(&(body.len() as u32).to_be_bytes());
+ buf.extend_from_slice(body);
+ }
+
+ #[test]
+ fn handrolled_v3_records_use_persisted_lsn() {
+ let mut bytes = Vec::new();
+ let body = br#"{"user_id":"alice","amount":1.0}"#;
+ let name = b"Txn";
+
+ bytes.push(0x03);
+ bytes.extend_from_slice(&10_000u64.to_be_bytes());
+ bytes.push(beava_core::wire::CT_JSON);
+ bytes.extend_from_slice(&7u32.to_be_bytes());
+ bytes.extend_from_slice(&(123i64).to_be_bytes());
+ bytes.extend_from_slice(&(name.len() as u16).to_be_bytes());
+ bytes.extend_from_slice(name);
+ bytes.extend_from_slice(&(body.len() as u32).to_be_bytes());
+ bytes.extend_from_slice(body);
+
+ let records = parse_handrolled_records(&bytes, 0);
+ assert_eq!(records.len(), 1);
+ assert_eq!(records[0].lsn, 10_000);
+ assert_eq!(records[0].rv, 7);
+ assert_eq!(records[0].event_name, "Txn");
+ }
+
+ #[test]
+ fn handrolled_decode_failure_writes_quarantine_marker() {
+ let wal = tempfile::tempdir().unwrap();
+ let registry = Arc::new(Registry::new());
+ let dev_agg = DevAggState::new(registry);
+ let lsn: Lsn = 379_827;
+ let body = b"not-json";
+ let name = b"Txn";
+
+ let mut bytes = Vec::new();
+ bytes.push(0x03);
+ bytes.extend_from_slice(&lsn.to_be_bytes());
+ bytes.push(beava_core::wire::CT_JSON);
+ bytes.extend_from_slice(&1u32.to_be_bytes());
+ bytes.extend_from_slice(&(123i64).to_be_bytes());
+ bytes.extend_from_slice(&(name.len() as u16).to_be_bytes());
+ bytes.extend_from_slice(name);
+ bytes.extend_from_slice(&(body.len() as u32).to_be_bytes());
+ bytes.extend_from_slice(body);
+ std::fs::write(wal.path().join("wal-0000000000000000.wal"), bytes).unwrap();
+
+ let outcome = replay_handrolled_wal_dir(wal.path(), 0, &dev_agg).expect("replay");
+ assert_eq!(outcome.last_lsn, lsn);
+ assert_eq!(outcome.replay_event_count, 0);
+ assert_eq!(outcome.quarantined_records, 1);
+ assert!(wal_quarantine_marker_exists(
+ wal.path(),
+ lsn,
+ WalQuarantineKind::HandrolledJsonBody
+ ));
+
+ let outcome = replay_handrolled_wal_dir(wal.path(), 0, &dev_agg).expect("replay again");
+ assert_eq!(outcome.last_lsn, lsn);
+ assert_eq!(outcome.quarantined_records, 1);
+ }
+
+ #[test]
+ fn handrolled_msgpack_decode_failure_writes_quarantine_marker() {
+ let wal = tempfile::tempdir().unwrap();
+ let registry = Arc::new(Registry::new());
+ let dev_agg = DevAggState::new(registry);
+ let lsn: Lsn = 379_827;
+
+ let mut bytes = Vec::new();
+ encode_handrolled_v3(
+ &mut bytes,
+ lsn,
+ beava_core::wire::CT_MSGPACK,
+ 1,
+ "Txn",
+ &[0xc1],
+ );
+ std::fs::write(wal.path().join("wal-0000000000000000.wal"), bytes).unwrap();
+
+ let outcome = replay_handrolled_wal_dir(wal.path(), 0, &dev_agg).expect("replay");
+ assert_eq!(outcome.last_lsn, lsn);
+ assert_eq!(outcome.replay_event_count, 0);
+ assert_eq!(outcome.quarantined_records, 1);
+ assert!(wal_quarantine_marker_exists(
+ wal.path(),
+ lsn,
+ WalQuarantineKind::HandrolledMsgpackBody
+ ));
+ }
+
+ #[test]
+ fn snapshot_load_body_applied_lsn_gates_handrolled_replay() {
+ let wal = tempfile::tempdir().unwrap();
+ let snap = tempfile::tempdir().unwrap();
+ let registry = Arc::new(Registry::new());
+ let dev_agg = DevAggState::new(registry);
+
+ install_txn_registry(&dev_agg);
+ put_alice_count(&dev_agg, 1);
+ dev_agg.next_event_id.store(100, Ordering::Relaxed);
+
+ let body = {
+ let registry_snap = dev_agg.registry.snapshot();
+ let tables = dev_agg.state_tables.lock();
+ SnapshotBody::from_live(®istry_snap, &tables, 100, 123)
+ };
+ let encoded = body.encode().expect("encode snapshot");
+ SnapshotWriter::write_with_stats(snap.path(), 5, body.registry.version, &encoded)
+ .expect("write snapshot with older header LSN");
+
+ let mut wal_bytes = Vec::new();
+ encode_handrolled_v3(
+ &mut wal_bytes,
+ 90,
+ beava_core::wire::CT_JSON,
+ dev_agg.registry.version() as u32,
+ "Txn",
+ br#"{"user_id":"alice","amount":1.0}"#,
+ );
+ std::fs::write(wal.path().join("wal-0000000000000000.wal"), wal_bytes).unwrap();
+
+ let registry = Arc::new(Registry::new());
+ let recovered = DevAggState::new(registry);
+ let loaded = load_snapshot_if_any(snap.path(), &recovered).expect("load snapshot");
+ assert_eq!(loaded.snapshot_lsn, 5);
+ assert_eq!(loaded.applied_lsn, 100);
+ assert_eq!(alice_count(&recovered), 1);
+
+ let replay =
+ replay_handrolled_wal_dir(wal.path(), loaded.applied_lsn, &recovered).expect("replay");
+ assert_eq!(replay.replay_event_count, 0);
+ assert_eq!(
+ alice_count(&recovered),
+ 1,
+ "using the snapshot body watermark must not double-apply covered v3 WAL records"
+ );
+ }
+
+ #[test]
+ fn persistence_event_decode_failure_writes_quarantine_marker() {
+ let wal = tempfile::tempdir().unwrap();
+ let registry = Arc::new(Registry::new());
+ let dev_agg = DevAggState::new(registry);
+ let lsn: Lsn = 379_827;
+
+ let mut writer =
+ beava_persistence::WalWriter::open(wal.path(), 100, dev_agg.registry.version() as u32)
+ .expect("open wal writer");
+ writer
+ .append(&WalRecord {
+ lsn,
+ record_type: RecordType::Event,
+ payload: b"not-json".to_vec(),
+ })
+ .expect("append event");
+ writer.sync_data().expect("sync wal");
+ drop(writer);
+
+ let outcome = replay_wal_from_lsn(wal.path(), 0, &dev_agg).expect("replay wal");
+ assert_eq!(outcome.last_lsn, lsn);
+ assert_eq!(outcome.replay_event_count, 0);
+ assert_eq!(outcome.quarantined_records, 1);
+ assert!(wal_quarantine_marker_exists(
+ wal.path(),
+ lsn,
+ WalQuarantineKind::PersistenceEventPayload
+ ));
+
+ let outcome = replay_wal_from_lsn(wal.path(), 0, &dev_agg).expect("replay wal again");
+ assert_eq!(outcome.last_lsn, lsn);
+ assert_eq!(outcome.quarantined_records, 1);
+ }
+
+ #[test]
+ fn replay_skips_already_installed_registry_bump_even_past_snapshot_lsn() {
+ let wal = tempfile::tempdir().unwrap();
+ let registry = Arc::new(Registry::new());
+ let dev_agg = DevAggState::new(registry);
+ let payload = txn_register_payload();
+ let bump = crate::register::RegistryBumpPayload {
+ new_version: 1,
+ payload_nodes: payload.nodes,
+ force_removed_descriptors: Vec::new(),
+ };
+
+ crate::register::apply_registry_bump(&dev_agg.registry, bump.clone())
+ .expect("install bump before replay");
+ assert_eq!(dev_agg.registry.version(), 1);
+
+ let mut writer =
+ beava_persistence::WalWriter::open(wal.path(), 100, dev_agg.registry.version() as u32)
+ .expect("open wal writer");
+ writer
+ .append(&WalRecord {
+ lsn: 123,
+ record_type: RecordType::RegistryBump,
+ payload: bump.encode().expect("encode bump"),
+ })
+ .expect("append bump");
+ writer.sync_data().expect("sync wal");
+ drop(writer);
+
+ let outcome = replay_wal_from_lsn(wal.path(), 42, &dev_agg).expect("replay wal");
+ assert_eq!(outcome.last_lsn, 123);
+ assert_eq!(outcome.replay_registry_bumps, 0);
+ assert!(!outcome.applied_registry_bump_after_snapshot);
+ assert_eq!(dev_agg.registry.version(), 1);
+ }
+}
diff --git a/crates/beava-server/src/register.rs b/crates/beava-server/src/register.rs
index c22a2193..74fd3d47 100644
--- a/crates/beava-server/src/register.rs
+++ b/crates/beava-server/src/register.rs
@@ -5,10 +5,9 @@
//! 2. JSON parse → 400
//! 3. Snapshot current registry for validation + diff
//! 4. validate_payload → 400
-//! 5. classify_register_diff (apply_shard's pre-flight has already
-//! pre-removed any descriptor whose shape would change with force=true,
-//! or returned 409 force_required without force; destructive entries
-//! here are an invariant violation → 503)
+//! 5. classify_register_diff (apply_shard's pre-flight has either returned
+//! 409 force_required without force, or supplied the force-removal set
+//! that this module persists in the RegistryBump before applying)
//! 6. No-op (no NewDescriptor entries in additive) → 200 same version
//! 7. Additive install → 200 new version
@@ -222,8 +221,16 @@ pub(crate) async fn execute_register_with_wal(
payload: RegisterPayload,
wal_sink: &WalSink,
force_removed: Vec,
+ wal_min_lsn: u64,
) -> RegisterOutcome {
- execute_register_inner(registry, payload, Some(wal_sink), force_removed).await
+ execute_register_inner(
+ registry,
+ payload,
+ Some(wal_sink),
+ force_removed,
+ wal_min_lsn,
+ )
+ .await
}
async fn execute_register_inner(
@@ -231,6 +238,7 @@ async fn execute_register_inner(
payload: RegisterPayload,
wal_sink: Option<&WalSink>,
force_removed: Vec,
+ wal_min_lsn: u64,
) -> RegisterOutcome {
if payload.nodes.is_empty() {
let version = registry.version();
@@ -273,46 +281,45 @@ async fn execute_register_inner(
let registered_descriptors: Vec = nodes.iter().map(|n| n.name().to_string()).collect();
// Use the new (Phase 13.4 Plan 06) categorized diff. apply_shard's
- // force-handling block pre-removes every existing descriptor the
- // payload would change (destructive + additive-against-existing,
- // when force=true) BEFORE this function runs. By the time we get
- // here, classify against the post-pre-removal snapshot should never
- // produce destructive entries — every remaining diff is additive
- // (NewDescriptor) or already_present (exact match).
+ // force-handling block computes every existing descriptor the payload
+ // would replace (destructive + additive-against-existing) but does not
+ // mutate the live registry before the WAL append. By the time we apply
+ // below, the removal list is carried in the durable RegistryBump.
let diff = beava_core::register_validate::classify_register_diff(¤t_snapshot, &nodes);
- // Invariant check: apply_shard's force-handling block pre-removes every
- // descriptor whose shape would change before this function is called.
- // If destructive entries reach here, the caller bypassed apply_shard
- // (or its pre-flight has a bug) — surface as WalUnavailable (503) so
- // operators see a noisy server-side error rather than a silent no-op.
- if !diff.destructive.is_empty() {
+ // Invariant check: destructive entries are allowed only when apply_shard
+ // supplied the force-removal set that will be applied after fsync.
+ if !diff.destructive.is_empty() && force_removed.is_empty() {
warn!(
kind = "register.preflight_invariant_violated",
version = current_snapshot.version,
destructive = ?diff.destructive,
- "execute_register saw destructive entries — apply_shard pre-flight should have pre-removed them; refusing to mutate"
+ "execute_register saw destructive entries without a force-removal set; refusing to mutate"
);
return RegisterOutcome::WalUnavailable {
version: current_snapshot.version,
};
}
- // Net-new descriptors are entries with `kind: new_descriptor`. Other
- // additive variants (NewField, NewAgg) shouldn't appear here in normal
- // flow — apply_shard pre-removes their target descriptor with
- // force=true so they re-land here as NewDescriptor against the
- // post-removal snapshot.
- let added: Vec = diff
- .additive
- .iter()
- .filter_map(|e| match e {
- beava_core::registry_diff::DiffEntry::NewDescriptor { name, .. } => Some(name.clone()),
- _ => None,
- })
- .collect();
+ // Net-new descriptors are entries with `kind: new_descriptor`. With
+ // force=true, apply_shard supplies the descriptor-removal set and the
+ // durable RegistryBump applies it before re-installing the payload, so
+ // every registered descriptor is reported as added.
+ let added: Vec = if force_removed.is_empty() {
+ diff.additive
+ .iter()
+ .filter_map(|e| match e {
+ beava_core::registry_diff::DiffEntry::NewDescriptor { name, .. } => {
+ Some(name.clone())
+ }
+ _ => None,
+ })
+ .collect()
+ } else {
+ registered_descriptors.clone()
+ };
- if added.is_empty() {
+ if added.is_empty() && force_removed.is_empty() {
info!(
kind = "register.noop",
version = current_snapshot.version,
@@ -326,15 +333,16 @@ async fn execute_register_inner(
};
}
+ let bump = RegistryBumpPayload {
+ new_version: current_snapshot.version + 1,
+ payload_nodes: nodes.clone(),
+ force_removed_descriptors: force_removed.clone(),
+ };
+
// Apply-after-fsync: append + fsync the `RegistryBump` BEFORE mutating
// the live registry, so recovery only sees descriptors that survived
// the durability boundary.
if let Some(sink) = wal_sink {
- let bump = RegistryBumpPayload {
- new_version: current_snapshot.version + 1,
- payload_nodes: nodes.clone(),
- force_removed_descriptors: force_removed.clone(),
- };
let encoded = match bump.encode() {
Ok(b) => b,
Err(e) => {
@@ -348,7 +356,10 @@ async fn execute_register_inner(
};
}
};
- if let Err(e) = sink.append_record(RecordType::RegistryBump, encoded).await {
+ if let Err(e) = sink
+ .append_record_at_least(RecordType::RegistryBump, encoded, wal_min_lsn)
+ .await
+ {
warn!(
kind = "register.wal_append_failed",
error = %e,
@@ -360,12 +371,28 @@ async fn execute_register_inner(
}
}
- let new_version = registry.apply_registration(
- nodes,
- compiled_chains,
- propagated_schemas,
- compiled_aggregations,
- );
+ let new_version = if force_removed.is_empty() {
+ registry.apply_registration(
+ nodes,
+ compiled_chains,
+ propagated_schemas,
+ compiled_aggregations,
+ )
+ } else {
+ match crate::register::apply_registry_bump(registry, bump) {
+ Ok(()) => registry.version(),
+ Err(e) => {
+ warn!(
+ kind = "register.force_apply_failed_after_wal",
+ error = %e,
+ "force RegistryBump apply failed after WAL append"
+ );
+ return RegisterOutcome::WalUnavailable {
+ version: current_snapshot.version,
+ };
+ }
+ }
+ };
info!(
kind = "register.success",
version = new_version,
@@ -526,3 +553,151 @@ pub(crate) fn error_code_to_wire_str(code: ErrorCode) -> &'static str {
pub fn format_serde_error_public(e: &serde_json::Error) -> (String, String) {
("".to_string(), e.to_string())
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use beava_core::registry::Registry;
+ use serde_json::json;
+ use std::sync::Arc;
+
+ fn baseline_payload() -> RegisterPayload {
+ serde_json::from_value(json!({
+ "nodes": [
+ {
+ "kind": "event",
+ "name": "Tx",
+ "schema": {
+ "fields": {
+ "event_time": "i64",
+ "user_id": "str",
+ "amount": "f64"
+ },
+ "optional_fields": []
+ }
+ },
+ {
+ "kind": "derivation",
+ "name": "UserSpend",
+ "output_kind": "event",
+ "upstreams": ["Tx"],
+ "ops": [{
+ "op": "group_by",
+ "keys": ["user_id"],
+ "agg": {
+ "cnt": {"op": "count", "params": {"window": "1h"}},
+ "total": {"op": "sum", "params": {"field": "amount", "window": "1h"}}
+ }
+ }],
+ "schema": {
+ "fields": {"user_id": "str", "cnt": "i64", "total": "f64"},
+ "optional_fields": []
+ }
+ }
+ ]
+ }))
+ .expect("valid baseline register payload")
+ }
+
+ fn changed_window_payload() -> RegisterPayload {
+ serde_json::from_value(json!({
+ "nodes": [
+ {
+ "kind": "event",
+ "name": "Tx",
+ "schema": {
+ "fields": {
+ "event_time": "i64",
+ "user_id": "str",
+ "amount": "f64"
+ },
+ "optional_fields": []
+ }
+ },
+ {
+ "kind": "derivation",
+ "name": "UserSpend",
+ "output_kind": "event",
+ "upstreams": ["Tx"],
+ "ops": [{
+ "op": "group_by",
+ "keys": ["user_id"],
+ "agg": {
+ "cnt": {"op": "count", "params": {"window": "30m"}},
+ "total": {"op": "sum", "params": {"field": "amount", "window": "30m"}}
+ }
+ }],
+ "schema": {
+ "fields": {"user_id": "str", "cnt": "i64", "total": "f64"},
+ "optional_fields": []
+ }
+ }
+ ]
+ }))
+ .expect("valid changed-window register payload")
+ }
+
+ #[tokio::test(flavor = "current_thread")]
+ async fn force_register_wal_failure_does_not_mutate_registry() {
+ let registry = Arc::new(Registry::new());
+ apply_registry_bump(
+ ®istry,
+ RegistryBumpPayload {
+ new_version: 1,
+ payload_nodes: baseline_payload().nodes,
+ force_removed_descriptors: Vec::new(),
+ },
+ )
+ .expect("install baseline");
+
+ let before = registry.snapshot();
+ assert_eq!(before.version, 1);
+ assert!(before.events.contains_key("Tx"));
+ assert!(before.derivations.contains_key("UserSpend"));
+ assert!(before.compiled_aggregations.contains_key("UserSpend"));
+ assert_eq!(
+ registry
+ .compiled_aggregation("UserSpend")
+ .expect("baseline agg")
+ .features
+ .iter()
+ .find(|f| f.feature_name == "cnt")
+ .and_then(|f| f.descriptor.window_ms),
+ Some(3_600_000)
+ );
+
+ let (sink, handle) = beava_persistence::WalSink::spawn_no_op();
+ sink.clone().shutdown().await.expect("shutdown no-op sink");
+ handle.await.expect("join no-op sink");
+
+ let outcome = execute_register_with_wal(
+ ®istry,
+ changed_window_payload(),
+ &sink,
+ vec!["UserSpend".to_string()],
+ 2,
+ )
+ .await;
+ let (status, body) = map_outcome_to_response(outcome);
+ assert_eq!(status, 503);
+ assert_eq!(body["error"]["code"], "wal_unavailable");
+ assert_eq!(body["registry_version"], 1);
+
+ let after = registry.snapshot();
+ assert_eq!(after.version, before.version);
+ assert!(after.events.contains_key("Tx"));
+ assert!(after.derivations.contains_key("UserSpend"));
+ assert!(after.compiled_aggregations.contains_key("UserSpend"));
+ assert_eq!(
+ registry
+ .compiled_aggregation("UserSpend")
+ .expect("agg survives WAL failure")
+ .features
+ .iter()
+ .find(|f| f.feature_name == "cnt")
+ .and_then(|f| f.descriptor.window_ms),
+ Some(3_600_000),
+ "force removal must not apply before the RegistryBump is durable"
+ );
+ }
+}
diff --git a/crates/beava-server/src/server.rs b/crates/beava-server/src/server.rs
index 6b4ae137..f5e8a56c 100644
--- a/crates/beava-server/src/server.rs
+++ b/crates/beava-server/src/server.rs
@@ -287,6 +287,7 @@ struct ServerV18State {
wal_ring: Arc,
wal_lsn: Arc,
wal_writer_handle: std::thread::JoinHandle<()>,
+ wal_reclaim: Option,
/// TCP frame-size cap plumbed through `WorkerConfig` rather than read
/// from env per-frame, so parallel `TestServer` instances don't
/// contaminate each other.
@@ -322,6 +323,12 @@ impl ServerV18 {
tcp_addr: std::net::SocketAddr,
admin_addr: std::net::SocketAddr,
) -> Result {
+ // Opt this process out of Transparent Huge Pages (Linux), warn if
+ // the system default is `[always]`. THP makes fork()-snapshot COW
+ // granularity 2 MB instead of 4 KB — a 500× amplifier on COW
+ // memory overhead during BGSAVE-style work. See crate::thp.
+ crate::thp::detect_and_opt_out();
+
// Bind event-plane listeners (std::net — they'll be handed to mio later).
let http_listener =
std::net::TcpListener::bind(http_addr).map_err(|e| ServerError::Bind {
@@ -769,8 +776,7 @@ async fn build_runtime_state_with_persistence(
// 2. Replay `*.log` records with `lsn > snapshot_lsn` (RegistryBumps
// and any persistence-WAL events from the legacy `WalSink`
// path).
- // 3. Replay `*.wal` data-plane events (v=2 binary format from
- // `apply_shard`).
+ // 3. Replay `*.wal` data-plane events from `apply_shard`.
// Memory mode skips this whole block — state starts fresh.
let initial_start_lsn = if is_memory {
tracing::info!(
@@ -780,29 +786,34 @@ async fn build_runtime_state_with_persistence(
);
1
} else {
- let snapshot_lsn = crate::recovery::load_snapshot_if_any(&snapshot_dir, &dev_agg)
+ let snapshot_load = crate::recovery::load_snapshot_if_any(&snapshot_dir, &dev_agg)
.ok()
- .unwrap_or(0);
+ .unwrap_or_default();
+ let snapshot_lsn = snapshot_load.snapshot_lsn;
let persistence_lsn = if wal_dir.exists() {
match replay_wal_from_lsn(&wal_dir, snapshot_lsn, &dev_agg) {
- Ok(outcome) => outcome.last_lsn,
+ Ok(outcome) => outcome.last_lsn.max(snapshot_lsn),
Err(_) => snapshot_lsn,
}
} else {
snapshot_lsn
};
- // Replay hand-rolled `*.wal` data-plane events. Setting
- // `lsn_start = persistence_lsn + 1` keeps LSNs monotonic across
- // the snapshot, persistence, and hand-rolled paths.
- let handrolled_lsn_start = persistence_lsn + 1;
+ // Replay hand-rolled `*.wal` data-plane events past the state already
+ // covered by the snapshot body. The snapshot header LSN is a mixed
+ // legacy/data-plane checkpoint; older snapshots can have a header LSN
+ // lower than their serialized data-plane state, and registry `.log`
+ // records can have higher LSNs than post-snapshot data-plane records.
+ // The body watermark is the only safe floor for hand-rolled replay.
+ let handrolled_from_lsn = snapshot_load.applied_lsn;
let handrolled_outcome =
- replay_handrolled_wal_dir(&wal_dir, handrolled_lsn_start, &dev_agg).unwrap_or_default();
+ replay_handrolled_wal_dir(&wal_dir, handrolled_from_lsn, &dev_agg).unwrap_or_default();
let initial = handrolled_outcome
.last_lsn
.max(persistence_lsn)
.max(snapshot_lsn)
+ .max(snapshot_load.applied_lsn)
+ 1;
tracing::debug!(
@@ -866,7 +877,7 @@ async fn build_runtime_state_with_persistence(
);
}
- let wal_lsn = Arc::new(WalLsn::new());
+ let wal_lsn = Arc::new(WalLsn::new_at(initial_start_lsn.saturating_sub(1)));
// Resolve WAL config from explicit overrides — hot path never reads
// env. Production resolves from `ServerV18Config::from_env()` at
// boot; tests pass explicit values via `TestServerBuilder`. The
@@ -893,8 +904,10 @@ async fn build_runtime_state_with_persistence(
// fsync. Memory mode: drain sealed buffers back to the free pool
// with no file I/O so the apply hot path can't backpressure-block
// once buffers fill.
- let (wal_writer_shutdown, wal_writer_handle) = if is_memory {
- spawn_no_op_wal_writer(Arc::clone(&wal_ring), Arc::clone(&wal_lsn), wal_cfg.tick_ms)
+ let (wal_writer_shutdown, wal_writer_handle, wal_reclaim) = if is_memory {
+ let (shutdown, handle) =
+ spawn_no_op_wal_writer(Arc::clone(&wal_ring), Arc::clone(&wal_lsn), wal_cfg.tick_ms);
+ (shutdown, handle, None)
} else {
let wal_writer = WalWriter::new(
&wal_dir,
@@ -903,13 +916,14 @@ async fn build_runtime_state_with_persistence(
wal_cfg.tick_ms,
)
.map_err(|e| ServerError::WalSpawn(e.to_string()))?;
+ let reclaim = wal_writer.reclaim_handle();
// Capture the shutdown flag BEFORE `spawn()` consumes the writer.
// Without it, the writer loop would never see the shutdown signal
// and `JoinHandle` drop would detach the thread mid-tick, losing
// any active-buffer contents that hadn't been sealed yet.
let shutdown = wal_writer.shutdown_flag();
let handle = wal_writer.spawn();
- (shutdown, handle)
+ (shutdown, handle, Some(reclaim))
};
// Memory mode skips the snapshot task entirely (zero file I/O), but
@@ -932,9 +946,16 @@ async fn build_runtime_state_with_persistence(
interval: Duration::from_millis(snapshot_interval_ms.max(1)),
snapshot_dir: snapshot_dir.clone(),
retain: 2,
+ // Read env once at boot — these are the SOLE legitimate
+ // env-read sites. Tests pass config fields directly per
+ // `phase13_5_3_no_env_var_pokes_in_tests`.
+ min_events_per_snapshot: crate::snapshot_task::min_events_from_env(),
+ use_fork_snapshot: crate::snapshot_fork::fork_enabled(),
+ snapshot_lsn_capture_tx: None,
},
Arc::clone(&app_state),
wal_sink.clone(),
+ wal_reclaim.clone(),
snapshot_cancel.clone(),
);
(Some((snapshot_cancel, snapshot_worker)), snapshot_trigger)
@@ -949,6 +970,7 @@ async fn build_runtime_state_with_persistence(
wal_ring,
wal_lsn,
wal_writer_handle,
+ wal_reclaim,
wal_writer_shutdown,
snapshot_task,
snapshot_trigger,
@@ -1042,6 +1064,7 @@ where
wal_ring,
wal_lsn,
wal_writer_handle,
+ wal_reclaim: _wal_reclaim,
wal_writer_shutdown,
snapshot_task,
snapshot_trigger,
diff --git a/crates/beava-server/src/snapshot_fork.rs b/crates/beava-server/src/snapshot_fork.rs
new file mode 100644
index 00000000..8e4c1e89
--- /dev/null
+++ b/crates/beava-server/src/snapshot_fork.rs
@@ -0,0 +1,606 @@
+//! fork()+COW snapshot path — Valkey BGSAVE pattern adapted for beava.
+//!
+//! The parent acquires `state_tables.lock()` across the `fork()` syscall, then
+//! immediately releases it in the parent. The child inherits the held guard and
+//! reads through that guard without taking any new `parking_lot` locks. Apply-
+//! thread blocking drops from ~seconds to the fork syscall window.
+//!
+//! Default ON on unix (linux/macos). Set `BEAVA_SNAPSHOT_FORK=0` (or
+//! `false`/`no`) to opt back into the legacy in-process synchronous
+//! snapshot in `snapshot_task::do_snapshot`. On non-unix targets the
+//! fork path is unavailable and the in-process path is always used.
+//!
+//! ## Safety / fork-correctness notes
+//!
+//! The single `unsafe { libc::fork() }` call has these invariants:
+//!
+//! 1. **beava's tokio runtime is `new_current_thread`** (see
+//! `crates/beava-server/src/main.rs` + `quickstart.rs`). Total OS threads
+//! at fork time = the tokio main thread + the mio apply thread + the
+//! `beava-wal-writer-noop` tick thread + possibly a `spawn_blocking`
+//! worker. The forking thread is the tokio main thread (it runs
+//! `snapshot_task`). All other threads vanish in the child.
+//!
+//! 2. **Allocator is fork-safe.** beava uses the system allocator (glibc on
+//! Linux, libc on macOS). Both have `pthread_atfork` malloc handlers that
+//! take the malloc lock pre-fork and release it post-fork in both
+//! parent and child. `bincode::serialize` in the child therefore allocates
+//! safely.
+//!
+//! 3. **Locks held by vanished threads are irrelevant.** The parent captures
+//! the registry snapshot before `fork()`, so the child never takes the
+//! registry `RwLock` in its inherited address space. The child only
+//! touches: `app_state.dev_agg.state_tables` (read-only via the lock guard
+//! the forking thread holds), scalar counter copies captured pre-fork, and
+//! `std::fs` (writes the new snapshot file via its own fds). It does NOT
+//! touch WAL state, tokio runtime, the admin sidecar, or any
+//! `parking_lot::Mutex` it didn't already hold at fork time.
+//!
+//! 4. **Child never returns; calls `libc::_exit`.** `_exit` is async-signal-
+//! safe and skips Rust destructors / atexit handlers / tokio shutdown
+//! that could touch parent state. `std::process::exit` would run atexit
+//! handlers — unsafe in a forked child.
+//!
+//! 5. **Child error reporting via sidecar file.** Child writes
+//! `snapshot-.error` on failure, then `_exit(1)`. Parent reads this
+//! after `waitpid`.
+
+use crate::AppState;
+use beava_core::snapshot_body::SnapshotBody;
+use beava_persistence::{PersistError, SnapshotWriteStats};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::Ordering;
+#[cfg(unix)]
+use std::time::{Duration, Instant};
+
+/// Result of a fork-snapshot. The parent uses `ChildExit::Success` to decide
+/// whether to truncate the WAL.
+#[derive(Debug)]
+pub enum ChildExit {
+ /// Child exited with status 0 — snapshot file is durable.
+ Success {
+ snapshot_lsn: u64,
+ write_stats: Option,
+ },
+ /// Child exited non-zero or with a signal. Snapshot file may be partial
+ /// or absent.
+ Failure { code: i32, message: String },
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SnapshotForkError {
+ #[error("fork(2) failed: {0}")]
+ ForkFailed(std::io::Error),
+ #[error("child waitpid failed: {0}")]
+ WaitFailed(std::io::Error),
+ #[error("persistence: {0}")]
+ Persist(#[from] PersistError),
+}
+
+/// Whether the fork path is enabled. Default ON on unix (linux/macos);
+/// callers can opt out by setting `BEAVA_SNAPSHOT_FORK` to `0`, `false`,
+/// `no`, or empty. Any other value (or unset) keeps the fork path on.
+/// Reads the env on every call (cold path; cost is negligible vs. a
+/// snapshot cycle). Always `false` on non-unix targets — fork(2) is
+/// unavailable there.
+pub fn fork_enabled() -> bool {
+ if !cfg!(unix) {
+ return false;
+ }
+ !matches!(
+ std::env::var("BEAVA_SNAPSHOT_FORK").as_deref(),
+ Ok("0") | Ok("false") | Ok("FALSE") | Ok("no") | Ok("")
+ )
+}
+
+/// Perform a snapshot via `fork()` + COW. Returns the child's exit summary
+/// so the caller can gate WAL reclamation on success.
+///
+/// The caller is responsible for:
+/// - Passing the legacy `WalSink` LSN. This function combines it with the
+/// applied data-plane watermark while holding `state_tables.lock()`, so the
+/// snapshot LSN matches the state snapshot the child inherits.
+/// - Truncating the WAL up to `snapshot_lsn` only on `ChildExit::Success`.
+///
+/// `snapshot_dir` must exist (the in-process path creates it lazily; the
+/// child cannot afford a directory-creation failure mid-flight). Caller
+/// should `std::fs::create_dir_all(snapshot_dir)` before this call.
+#[cfg(unix)]
+pub async fn do_snapshot_via_fork(
+ snapshot_dir: &Path,
+ legacy_snapshot_lsn: u64,
+ app_state: &AppState,
+) -> Result {
+ do_snapshot_via_fork_with_wait_timeout(
+ snapshot_dir,
+ legacy_snapshot_lsn,
+ app_state,
+ snapshot_fork_wait_timeout(),
+ )
+ .await
+}
+
+#[cfg(unix)]
+pub async fn do_snapshot_via_fork_with_wait_timeout(
+ snapshot_dir: &Path,
+ legacy_snapshot_lsn: u64,
+ app_state: &AppState,
+ wait_timeout: Duration,
+) -> Result {
+ use beava_persistence::SnapshotWriter;
+
+ // Ensure the snapshot dir exists in the parent (cheap; idempotent). The
+ // child cannot afford a mkdir failure.
+ std::fs::create_dir_all(snapshot_dir)
+ .map_err(|e| SnapshotForkError::Persist(PersistError::Io(e)))?;
+
+ let snapshot_dir_owned = snapshot_dir.to_path_buf();
+ let registry_snap = app_state.dev_agg.registry.snapshot();
+
+ let (state_lock, next_event_id, query_time_ms, snapshot_lsn) = loop {
+ let candidate_next_event_id = app_state.dev_agg.next_event_id.load(Ordering::Acquire);
+ let candidate_snapshot_lsn = legacy_snapshot_lsn.max(candidate_next_event_id);
+ let _ = std::fs::remove_file(snapshot_stats_sidecar_path(
+ snapshot_dir,
+ candidate_snapshot_lsn,
+ ));
+ let _ = std::fs::remove_file(snapshot_error_sidecar_path(
+ snapshot_dir,
+ candidate_snapshot_lsn,
+ ));
+
+ let state_lock = app_state.dev_agg.state_tables.lock();
+ let next_event_id = app_state.dev_agg.next_event_id.load(Ordering::Acquire);
+ let query_time_ms = app_state.dev_agg.query_time_ms.load(Ordering::Acquire) as i64;
+ let snapshot_lsn = legacy_snapshot_lsn.max(next_event_id);
+ if snapshot_lsn == candidate_snapshot_lsn {
+ break (state_lock, next_event_id, query_time_ms, snapshot_lsn);
+ }
+ drop(state_lock);
+ };
+
+ // Briefly take the state_tables lock so the fork sees a quiescent state
+ // snapshot. The lock-hold spans two scalar loads plus the fork syscall
+ // (~µs); path setup, registry capture, and sidecar cleanup happened above.
+ //
+ // SAFETY:
+ // - beava's tokio runtime is `new_current_thread`; the forking thread is
+ // the tokio main thread. All other OS threads (mio apply, wal-writer-
+ // noop, spawn_blocking workers) vanish in the child per POSIX.
+ // - System malloc (glibc/libc) is fork-safe via pthread_atfork handlers,
+ // so `bincode::serialize` in the child allocates safely.
+ // - Child uses the pre-captured registry snapshot and the inherited
+ // `state_lock` guard; it takes no registry RwLock, no parking_lot Mutex,
+ // no tokio, no WAL, no admin sidecar.
+ // - Child calls `libc::_exit` (async-signal-safe; skips at_exit
+ // handlers) rather than `std::process::exit`.
+ let pid = unsafe { libc::fork() };
+
+ if pid < 0 {
+ return Err(SnapshotForkError::ForkFailed(
+ std::io::Error::last_os_error(),
+ ));
+ }
+
+ if pid == 0 {
+ // === CHILD ===
+ // Build snapshot from our (now-frozen via COW) view of app_state.
+ // Do not lock or unlock `state_tables` in the child. The forking
+ // thread already held this guard, and `_exit` skips its destructor.
+ let body =
+ SnapshotBody::from_live(®istry_snap, &state_lock, next_event_id, query_time_ms);
+
+ let encoded = match body.encode() {
+ Ok(b) => b,
+ Err(e) => child_fail(&snapshot_dir_owned, snapshot_lsn, &format!("encode: {e}")),
+ };
+ let registry_version = body.registry.version;
+
+ match SnapshotWriter::write_with_stats(
+ &snapshot_dir_owned,
+ snapshot_lsn,
+ registry_version,
+ &encoded,
+ ) {
+ Ok(stats) => {
+ if let Err(e) = write_stats_sidecar(&snapshot_dir_owned, snapshot_lsn, &stats) {
+ child_fail(&snapshot_dir_owned, snapshot_lsn, &format!("stats: {e}"));
+ }
+ unsafe {
+ libc::_exit(0);
+ }
+ }
+ Err(e) => child_fail(&snapshot_dir_owned, snapshot_lsn, &format!("write: {e}")),
+ }
+ }
+
+ // === PARENT ===
+ drop(state_lock);
+
+ // Wait on the child without blocking the tokio runtime. The blocking side
+ // polls with WNOHANG so a wedged fork child can be killed and reaped instead
+ // of leaving an uncancelable waitpid task behind.
+ let exit = tokio::task::spawn_blocking(move || {
+ wait_for_snapshot_child(pid, snapshot_dir_owned, snapshot_lsn, wait_timeout)
+ })
+ .await
+ .map_err(|e| SnapshotForkError::WaitFailed(std::io::Error::other(format!("join: {e}"))))?
+ .map_err(SnapshotForkError::WaitFailed)?;
+
+ Ok(exit)
+}
+
+#[cfg(unix)]
+const DEFAULT_SNAPSHOT_FORK_WAIT_TIMEOUT_SECS: u64 = 600;
+
+#[cfg(unix)]
+fn snapshot_fork_wait_timeout() -> Duration {
+ std::env::var("BEAVA_SNAPSHOT_FORK_WAIT_TIMEOUT_SECS")
+ .ok()
+ .and_then(|raw| raw.parse::().ok())
+ .filter(|secs| *secs > 0)
+ .map(Duration::from_secs)
+ .unwrap_or_else(|| Duration::from_secs(DEFAULT_SNAPSHOT_FORK_WAIT_TIMEOUT_SECS))
+}
+
+#[cfg(unix)]
+fn wait_for_snapshot_child(
+ pid: libc::pid_t,
+ snapshot_dir: PathBuf,
+ snapshot_lsn: u64,
+ timeout: Duration,
+) -> Result {
+ let deadline = Instant::now() + timeout;
+ loop {
+ let mut status: libc::c_int = 0;
+ let waited = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) };
+ if waited == pid {
+ return Ok(child_exit_from_status(status, &snapshot_dir, snapshot_lsn));
+ }
+ if waited < 0 {
+ let err = std::io::Error::last_os_error();
+ if err.raw_os_error() == Some(libc::EINTR) {
+ if Instant::now() >= deadline {
+ return terminate_timed_out_child(pid, timeout);
+ }
+ continue;
+ }
+ return Err(err);
+ }
+
+ if Instant::now() >= deadline {
+ return terminate_timed_out_child(pid, timeout);
+ }
+
+ std::thread::sleep(Duration::from_millis(10));
+ }
+}
+
+#[cfg(unix)]
+fn terminate_timed_out_child(
+ pid: libc::pid_t,
+ timeout: Duration,
+) -> Result {
+ let kill_error = if unsafe { libc::kill(pid, libc::SIGKILL) } == 0 {
+ None
+ } else {
+ Some(std::io::Error::last_os_error())
+ };
+ let reap_grace = Duration::from_secs(1);
+ let reap_deadline = Instant::now() + reap_grace;
+ let reaped = reap_child_until(pid, reap_deadline)?;
+
+ let mut message = format!(
+ "child exceeded fork snapshot wait timeout of {}s and was killed",
+ timeout.as_secs()
+ );
+ if let Some(err) = kill_error {
+ message.push_str(&format!("; SIGKILL failed: {err}"));
+ }
+
+ match reaped {
+ Some(status) => {
+ if libc::WIFSIGNALED(status) {
+ message.push_str(&format!("; reaped signal {}", libc::WTERMSIG(status)));
+ } else if libc::WIFEXITED(status) {
+ message.push_str(&format!(
+ "; reaped exit status {}",
+ libc::WEXITSTATUS(status)
+ ));
+ } else {
+ message.push_str(&format!("; reaped status {status}"));
+ }
+ Ok(ChildExit::Failure { code: -1, message })
+ }
+ None => Err(std::io::Error::new(
+ std::io::ErrorKind::TimedOut,
+ format!(
+ "child exceeded fork snapshot wait timeout of {}s, but was not reaped within {}ms",
+ timeout.as_secs(),
+ reap_grace.as_millis()
+ ),
+ )),
+ }
+}
+
+#[cfg(unix)]
+fn reap_child_until(
+ pid: libc::pid_t,
+ deadline: Instant,
+) -> Result