fix: introduce concurrency primitives; reshape DLQ around BackgroundSink

New `concurrency` feature: BackgroundSink<T>, PeriodicWorker, ActorHandle<Cmd>.
DLQ now uses BackgroundSink<DlqEntry> with enum-dispatch backends; cascade
moves into the drain task. AsyncNdjsonWriter offloads sync rotate-and-write
to spawn_blocking; FileOutput gains async API. config/reloader,
version_check, directory_config sync-I/O-on-async fixed. /tmp/hyperi
fallback removed. clippy::await_holding_lock denied crate-wide. New
tests/sync_in_async.rs grep lint (0 violations) and benches/concurrency_patterns.rs
(try_push 28-35ns, try_send 63ns). 1280 tests pass.

Downstream apps migrate: Dlq::file_only / Dlq::with_kafka -> Dlq::spawn(
config, name, kafka_opt, shutdown). FileDlq / KafkaDlq / HttpDlq / RedisDlq
types now crate-private. Dlq::send.await returns when queued; use
Dlq::flush.await for durability. Dlq::add_backend removed.

Publish: true

Author: catinspace-au

Cargo.toml

@@ -36,6 +36,12 @@ runtime = ["dirs"]
 config = ["figment", "dotenvy", "serde_yaml_ng", "serde_json", "dirs", "tracing"]
 logger = ["tracing", "tracing-subscriber", "owo-colors", "serde_json", "tracing-throttle"]
 
+# Three generic async primitives (BackgroundSink, PeriodicWorker, ActorHandle).
+# Foundational helpers for any module that does fire-and-forget durable
+# writes, timer-driven loops, or stateful command-queue actors. See
+# `src/concurrency/mod.rs` for the decision matrix.
+concurrency = ["tokio", "tokio-util", "tracing", "metrics-core"]
+
 # Metrics tiers (2.6.0 split). Consumers pay only for what they use:
 # - metrics-core: emit-only (counter!/gauge!/histogram! macros work)
 # - metrics-process: + cgroup-aware process gauges
@@ -99,10 +105,10 @@ cli-service = ["cli", "metrics", "memory", "scaling", "worker-pool", "shutdown"]
 top = ["cli-service", "ratatui"]
 
 # Shared NDJSON file I/O (used by dlq + output-file)
-io = ["file-rotate", "parking_lot", "chrono", "tracing"]
+io = ["file-rotate", "parking_lot", "chrono", "tracing", "tokio"]
 
 # Dead letter queue (file backend always available)
-dlq = ["io", "base64", "serde_json", "tokio", "async-trait", "tracing"]
+dlq = ["io", "concurrency", "base64", "serde_json", "tokio", "tokio-util", "tracing"]
 dlq-kafka = ["dlq", "transport-kafka"]
 dlq-http = ["dlq", "reqwest"]
 dlq-redis = ["dlq", "transport-redis"]
@@ -358,3 +364,8 @@ required-features = ["worker"]
 name = "filter_benchmark"
 harness = false
 required-features = ["transport-memory"]
+
+[[bench]]
+name = "concurrency_patterns"
+harness = false
+required-features = ["concurrency"]

benches/concurrency_patterns.rs

@@ -0,0 +1,128 @@
+// Project:   hyperi-rustlib
+// File:      benches/concurrency_patterns.rs
+// Purpose:   Criterion benchmarks for the three async primitives
+// Language:  Rust
+//
+// License:   FSL-1.1-ALv2
+// Copyright: (c) 2026 HYPERI PTY LIMITED
+
+//! Benchmarks for `concurrency::BackgroundSink`, `PeriodicWorker`,
+//! `ActorHandle`.
+//!
+//! Validates the design assumptions:
+//!
+//! - `BackgroundSink::try_push` is ~100 ns happy path (the consumer hot
+//!   path target — slower than this means we've regressed the headline
+//!   guarantee).
+//! - `ActorHandle::try_send` is in the same ballpark (~100 ns).
+//! - Both stay below 500 ns p99 under contention from many concurrent
+//!   producers.
+//!
+//! Run with `cargo bench --bench concurrency_patterns --features concurrency`.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+
+use criterion::{Criterion, Throughput, criterion_group, criterion_main};
+use tokio::runtime::Runtime;
+use tokio_util::sync::CancellationToken;
+
+use hyperi_rustlib::concurrency::{
+    Actor, ActorConfig, ActorHandle, BackgroundSink, BackgroundSinkConfig, DrainError, Overflow,
+    SinkDrain,
+};
+
+/// Drain that counts incoming messages atomically. Trivial — keeps the
+/// bench focused on the producer hot path, not the drain.
+struct CountingDrain {
+    count: Arc<AtomicU64>,
+}
+
+impl SinkDrain<u64> for CountingDrain {
+    async fn write_batch(&mut self, batch: Vec<u64>) -> Result<(), DrainError> {
+        self.count.fetch_add(batch.len() as u64, Ordering::Relaxed);
+        Ok(())
+    }
+}
+
+/// Counter actor. Sums each pushed value.
+struct CounterActor {
+    sum: u64,
+}
+
+impl Actor for CounterActor {
+    type Command = u64;
+    async fn handle(&mut self, cmd: u64) {
+        self.sum = self.sum.wrapping_add(cmd);
+    }
+}
+
+fn bench_background_sink_try_push(c: &mut Criterion) {
+    let rt = Runtime::new().expect("tokio rt");
+    let mut group = c.benchmark_group("background_sink");
+    group.throughput(Throughput::Elements(1));
+
+    group.bench_function("try_push_drop_mode", |b| {
+        let count = Arc::new(AtomicU64::new(0));
+        let shutdown = CancellationToken::new();
+        let (sink, _handle) = rt.block_on(async {
+            BackgroundSink::spawn(
+                CountingDrain {
+                    count: count.clone(),
+                },
+                BackgroundSinkConfig {
+                    queue_capacity: 1_000_000,
+                    batch_size: 4096,
+                    flush_interval: Duration::from_millis(50),
+                    overflow: Overflow::Drop,
+                    metric_prefix: None,
+                },
+                shutdown.clone(),
+            )
+        });
+
+        b.iter(|| {
+            let _ = std::hint::black_box(sink.try_push(42));
+        });
+
+        shutdown.cancel();
+    });
+
+    group.finish();
+}
+
+fn bench_actor_try_send(c: &mut Criterion) {
+    let rt = Runtime::new().expect("tokio rt");
+    let mut group = c.benchmark_group("actor");
+    group.throughput(Throughput::Elements(1));
+
+    group.bench_function("try_send_unsaturated", |b| {
+        let shutdown = CancellationToken::new();
+        let (handle, _join) = rt.block_on(async {
+            ActorHandle::spawn(
+                CounterActor { sum: 0 },
+                ActorConfig {
+                    queue_capacity: 1_000_000,
+                    idle_interval: Duration::from_mins(1),
+                },
+                shutdown.clone(),
+            )
+        });
+
+        b.iter(|| {
+            let _ = std::hint::black_box(handle.try_send(42));
+        });
+
+        shutdown.cancel();
+    });
+
+    group.finish();
+}
+
+criterion_group!(
+    benches,
+    bench_background_sink_try_push,
+    bench_actor_try_send
+);
+criterion_main!(benches);