bench

Cargo.toml

@@ -9,6 +9,7 @@ members = [
     "crates/sqlite",
     "examples/basic",
     "examples/redis",
+    "examples/bench",
 ]
 
 [workspace.package]

crates/core/worker.rs

@@ -118,6 +118,21 @@ where
     /// This will spawn worker tasks, scheduler, retrier, heartbeater, and reaper,
     /// and block until shutdown is requested and draining is complete.
     pub async fn run(&mut self) -> Result<()> {
+        self.run_until(async {
+            tokio::signal::ctrl_c().await.ok();
+        })
+        .await
+    }
+
+    /// Run the worker pool until the provided shutdown future completes.
+    ///
+    /// This will spawn worker tasks, scheduler, retrier, heartbeater, and reaper,
+    /// and block until `shutdown` resolves, then initiate graceful shutdown and
+    /// wait for draining to complete.
+    pub async fn run_until<S>(&mut self, shutdown: S) -> Result<()>
+    where
+        S: Future<Output = ()> + Send,
+    {
         let backend = self
             .backend
             .take()
@@ -217,9 +232,9 @@ where
             "Worker pool started"
         );
 
-        // Wait for shutdown signal
-        tokio::signal::ctrl_c().await.ok();
-        tracing::info!("Shutdown signal received, draining...");
+        // Wait for shutdown signal/future
+        shutdown.await;
+        tracing::info!("Shutdown requested, draining...");
 
         self.shutdown().await;
 

crates/redis/lib.rs

@@ -135,8 +135,16 @@ impl Backend for RedisBackend {
 
     async fn pop_job(&self, ns: &str, timeout: Duration) -> Result<Option<String>> {
         let mut conn = self.conn.clone();
+        // Redis BRPOP timeout is effectively second-granular; avoid accidental infinite blocks
+        // when callers pass sub-second timeouts (e.g. 100ms -> 0s).
+        let secs = timeout.as_secs();
+        let timeout_secs = if secs == 0 && timeout != Duration::from_secs(0) {
+            1u64
+        } else {
+            secs
+        };
         let result: Option<(String, String)> = conn
-            .brpop(Self::keys(ns).jobs(), timeout.as_secs() as f64)
+            .brpop(Self::keys(ns).jobs(), timeout_secs as f64)
             .await
             .map_err(|e| WgError::Backend(e.to_string()))?;
         Ok(result.map(|(_, json)| json))

docs/benchmark.md

@@ -0,0 +1,69 @@
+# Benchmark
+
+This repo doesn’t ship `cargo bench` / `criterion` benches yet. Instead, it provides an **end-to-end benchmark example** that enqueues **N jobs**, processes **N jobs**, and exits automatically.
+
+## Example benchmark binary
+
+Run via:
+
+```bash
+cargo run -p example-bench --release
+```
+
+Note: the benchmark enqueues first, then starts the worker pool. This avoids a slowdown on Redis where `BRPOP` can block a shared connection and starve enqueues.
+
+### Common env vars
+
+- `WG_BENCH_BACKEND=redis|postgres`
+- `WG_BENCH_JOBS` (default `20000`)
+- `WG_BENCH_WORKERS` (default `8`)
+- `WG_BENCH_ENQUEUE_CONCURRENCY` (default `1`)
+- `WG_BENCH_MAX_CONCURRENCY` (default `0` = unlimited)
+- `WG_BENCH_FETCH_TIMEOUT_MS` (default `1000`)
+- `WG_BENCH_SHUTDOWN_TIMEOUT_SECS` (default `2`)
+- `WG_BENCH_TICK_MS` (default `50`)
+- `WG_BENCH_ENABLE_REAPER` (default `false`)
+- `WG_BENCH_JOB_NAME` (default `bench_noop`)
+- `WG_BENCH_NAMESPACE` (default `wg_bench_{pid}`)
+  - For SQL backends the namespace is used in table names; prefer `[a-zA-Z0-9_]` only.
+
+### Redis
+
+Prereq: Redis running (example):
+
+```bash
+docker run -d -p 6379:6379 redis
+```
+
+Run:
+
+```bash
+WG_BENCH_BACKEND=redis \
+REDIS_URL=redis://localhost \
+WG_BENCH_JOBS=20000 \
+WG_BENCH_WORKERS=8 \
+WG_BENCH_ENQUEUE_CONCURRENCY=64 \
+cargo run -p example-bench --release
+```
+
+### Postgres
+
+Prereq: Postgres running and `DATABASE_URL` pointing at a database.
+
+Run:
+
+```bash
+WG_BENCH_BACKEND=postgres \
+DATABASE_URL=postgres://user:pass@localhost/db \
+WG_BENCH_JOBS=20000 \
+WG_BENCH_WORKERS=8 \
+WG_BENCH_ENQUEUE_CONCURRENCY=64 \
+cargo run -p example-bench --release
+```
+
+## Tips for stable results
+
+- Always use `--release`.
+- Avoid noisy logging: the benchmark sets `RUST_LOG=error` by default.
+- Run multiple times and compare medians.
+

examples/bench/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "example-bench"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[[bin]]
+name = "example-bench"
+path = "main.rs"
+
+[dependencies]
+wg-core = { path = "../../crates/core" }
+wg-redis = { path = "../../crates/redis" }
+wg-postgres = { path = "../../crates/postgres" }
+tokio = { version = "1", features = ["full"] }
+serde = { version = "1", features = ["derive"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+