impl claude

engine/Cargo.toml

@@ -8,6 +8,34 @@ description = "A Blazingly fast distributed task system"
 [features]
 default = []
 dev = []
+
+[[bin]]
+name = "server"
+path = "src/bin/server.rs"
+
+[[bin]]
+name = "client"
+path = "src/bin/client.rs"
+
+[[bin]]
+name = "bench"
+path = "src/bin/bench.rs"
+
+[[bin]]
+name = "bench_server"
+path = "src/bin/bench_server.rs"
+
 [dependencies]
 
 enginelib = { path = "../enginelib" }
+tokio = { version = "1.50.0", features = ["rt-multi-thread", "macros", "sync", "signal", "time"] }
+tmq = "0.5.0"
+futures = "0.3"
+serde = { workspace = true }
+toml = { workspace = true }
+tracing = "0.1"
+
+[dev-dependencies]
+chrono = { version = "0.4.44", features = ["serde"] }
+dashmap = "6.2.1"
+async-channel = "2.5.0"

engine/src/bench_task.rs

@@ -0,0 +1,57 @@
+//! Small statically linked task used only by the process benchmark.
+
+use enginelib::Identifier;
+use enginelib::task::{Task, Verifiable};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct FibTask {
+    pub iter: u64,
+    pub result: u64,
+}
+
+impl Verifiable for FibTask {
+    fn verify(&self, bytes: &[u8]) -> bool {
+        enginelib::api::from_bytes::<Self>(bytes).is_ok()
+    }
+}
+
+impl Task for FibTask {
+    fn to_toml(&self) -> String {
+        toml::to_string(self).unwrap_or_default()
+    }
+
+    fn from_toml(&self, data: String) -> Box<dyn Task> {
+        Box::new(toml::from_str::<Self>(&data).unwrap_or_default())
+    }
+
+    fn get_id(&self) -> Identifier {
+        ("engine_mod".to_string(), "fib".to_string())
+    }
+
+    fn clone_box(&self) -> Box<dyn Task> {
+        Box::new(self.clone())
+    }
+
+    fn run_cpu(&mut self) {
+        // Keep work negligible: this benchmark measures engine and transport
+        // throughput rather than Fibonacci performance.
+        let iterations = self.iter.min(16);
+        let mut a = 0u64;
+        let mut b = 1u64;
+        for _ in 0..iterations {
+            let previous = a;
+            a = b;
+            b = b.wrapping_add(previous);
+        }
+        self.result = a;
+    }
+
+    fn from_bytes(&self, bytes: &[u8]) -> Box<dyn Task> {
+        Box::new(enginelib::api::from_bytes::<Self>(bytes).unwrap_or_default())
+    }
+
+    fn to_bytes(&self) -> Vec<u8> {
+        enginelib::api::to_allocvec(self).unwrap_or_default()
+    }
+}

engine/src/bin/bench.rs

@@ -0,0 +1,164 @@
+//! Deployment-faithful throughput benchmark (load generator).
+//!
+//! Spawns the `bench_server` binary as a **separate OS process** and drives it
+//! over real tcp: one submitter connection plus N worker connections, each a real
+//! libzmq DEALER. No in-process server, no unbounded channel — the server runs the
+//! real bounded pipeline, fed by persistent per-task-type loaders started by the
+//! benchmark server (submit writes to the DB; loaders stream records into the
+//! lease channel). Stall detection is a safety net: if progress stops for
+//! `BENCH_STALL_SECS`, the run reports `stalled` rather than hanging or lying.
+//!
+//! Usage: bench <n> [submit_batch] [lease_batch] [workers] [port]
+//! Emits one JSON line on stdout.
+
+use std::process::{Command, Stdio};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+use engine::bench_task::FibTask;
+use engine::client::Client;
+use enginelib::Identifier;
+use enginelib::task::Task;
+
+#[tokio::main]
+async fn main() {
+    let mut args = std::env::args().skip(1);
+    let n: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(100_000);
+    let submit_batch: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(1000);
+    let lease_batch: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
+    let workers: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(16);
+    let port: u16 = args.next().and_then(|s| s.parse().ok()).unwrap_or(55610);
+
+    let endpoint = format!("tcp://127.0.0.1:{port}");
+    let task_type: Identifier = ("engine_mod".to_string(), "fib".to_string());
+    let db_path = format!("target/bench_db_{port}_{}", std::process::id());
+
+    // Launch the real server as its own process.
+    let server_bin = std::env::current_exe()
+        .ok()
+        .and_then(|p| p.parent().map(|d| d.join("bench_server")))
+        .expect("locate bench_server next to bench");
+    let mut server = Command::new(&server_bin)
+        .arg(port.to_string())
+        .arg(&db_path)
+        .stdout(Stdio::null())
+        .spawn()
+        .expect("spawn bench_server");
+    // Give it time to bind the ROUTER.
+    tokio::time::sleep(Duration::from_millis(800)).await;
+
+    let completed = Arc::new(AtomicU64::new(0));
+
+    // Workers (separate DEALER connections): lease → decode + run → complete.
+    let mut worker_handles = Vec::new();
+    for _ in 0..workers {
+        let endpoint = endpoint.clone();
+        let task_type = task_type.clone();
+        let completed = completed.clone();
+        worker_handles.push(tokio::spawn(async move {
+            let mut client = match Client::connect(&endpoint, String::new()) {
+                Ok(c) => c,
+                Err(_) => return,
+            };
+            loop {
+                let leased = match client
+                    .lease(task_type.clone(), "w".into(), lease_batch)
+                    .await
+                {
+                    Ok(v) => v,
+                    Err(_) => break,
+                };
+                if leased.is_empty() {
+                    continue;
+                }
+                let mut results = Vec::with_capacity(leased.len());
+                for st in &leased {
+                    let mut task: FibTask =
+                        enginelib::api::from_bytes(&st.bytes).unwrap_or_default();
+                    task.run_cpu();
+                    results.push((st.task_id.clone(), task.to_bytes()));
+                }
+                match client.complete(task_type.clone(), results).await {
+                    Ok(ok) => {
+                        completed.fetch_add(ok as u64, Ordering::Relaxed);
+                    }
+                    Err(_) => break,
+                }
+            }
+        }));
+    }
+
+    let payload = FibTask {
+        iter: 20,
+        result: 0,
+    }
+    .to_bytes();
+
+    // Submitter: submit all N over its own connection, timed.
+    let e2e_start = Instant::now();
+    let submit_start = Instant::now();
+    let submit_task = {
+        let endpoint = endpoint.clone();
+        let task_type = task_type.clone();
+        let payload = payload.clone();
+        tokio::spawn(async move {
+            let mut client = Client::connect(&endpoint, String::new()).expect("submitter connect");
+            let mut sent = 0u64;
+            while sent < n {
+                let this = std::cmp::min(submit_batch as u64, n - sent) as usize;
+                if client
+                    .submit(task_type.clone(), vec![payload.clone(); this])
+                    .await
+                    .is_err()
+                {
+                    break;
+                }
+                sent += this as u64;
+            }
+            sent
+        })
+    };
+    let submitted = submit_task.await.unwrap_or(0);
+    let submit_secs = submit_start.elapsed().as_secs_f64();
+
+    // Wait for completion. Stall detection is a safety net (e.g. a wedged loader,
+    // a crashed worker, or a genuine bug) — it reports `stalled` instead of
+    // hanging forever, rather than describing any expected behavior.
+    let stall_secs: u64 = std::env::var("BENCH_STALL_SECS")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(20);
+    let mut last = 0u64;
+    let mut last_change = Instant::now();
+    let mut stalled = false;
+    loop {
+        let c = completed.load(Ordering::Relaxed);
+        if c >= n {
+            break;
+        }
+        if c != last {
+            last = c;
+            last_change = Instant::now();
+        } else if last_change.elapsed() > Duration::from_secs(stall_secs) {
+            stalled = true;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(2)).await;
+    }
+    let done = completed.load(Ordering::Relaxed);
+    let e2e_secs = e2e_start.elapsed().as_secs_f64();
+
+    for h in worker_handles {
+        h.abort();
+    }
+    let _ = server.kill();
+    let _ = server.wait();
+    let _ = std::fs::remove_dir_all(&db_path);
+
+    let submit_tps = submitted as f64 / submit_secs;
+    let e2e_tps = done as f64 / e2e_secs;
+    println!(
+        "{{\"n\":{n},\"submitted\":{submitted},\"completed\":{done},\"stalled\":{stalled},\"workers\":{workers},\"submit_batch\":{submit_batch},\"lease_batch\":{lease_batch},\"submit_secs\":{submit_secs:.4},\"submit_tps\":{submit_tps:.1},\"e2e_secs\":{e2e_secs:.4},\"e2e_tps\":{e2e_tps:.1}}}"
+    );
+}

engine/src/bin/bench_server.rs

@@ -0,0 +1,48 @@
+//! Standalone server process for the deployment-faithful benchmark.
+//!
+//! Mirrors `bin/server.rs` (the deployed pipeline): `populate()` builds the real
+//! `bounded(8192)` channels, a startup `load()` recovers any persisted backlog,
+//! then `serve()` runs the ROUTER over tcp. The only deviation from a real deploy
+//! is that `FibTask` is statically linked here instead of dynamically loaded from
+//! a `.rf` mod — that changes registration, not per-task runtime cost.
+//!
+//! Usage: bench_server <port> <db_path>
+
+use std::sync::Arc;
+
+use engine::bench_task::FibTask;
+use enginelib::api::ServerAPI;
+use enginelib::task::Task;
+
+#[tokio::main]
+async fn main() {
+    let mut args = std::env::args().skip(1);
+    let port: u16 = args.next().and_then(|s| s.parse().ok()).unwrap_or(55610);
+    let db_path = args.next().unwrap_or_else(|| "target/bench_db".to_string());
+
+    let task_type = ("engine_mod".to_string(), "fib".to_string());
+
+    let api = ServerAPI::with_path(&db_path);
+    let fib: Arc<dyn Task> = Arc::new(FibTask::default());
+    api.task_registry.tasks.insert(task_type.clone(), fib);
+    let api = Arc::new(api);
+    // Real bounded queue + dedup set, as the deployed server builds. Depth is
+    // configurable (ENGINE_QUEUE_SIZE, default 8192) for the sweep.
+    let queue_size: usize = std::env::var("ENGINE_QUEUE_SIZE")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(8192);
+    ServerAPI::populate_with(&api, queue_size);
+    // Match init()'s deployment behavior: run the lease reaper. (At the 3600s TTL
+    // it never fires within a bench, but it's free and keeps this faithful.)
+    ServerAPI::spawn_reaper(&api);
+    // Submit only writes to the DB; one loader per task type feeds the bounded
+    // lease queue. Transport shards must not create duplicate loaders.
+    ServerAPI::spawn_loaders(&api);
+
+    let endpoint = format!("tcp://127.0.0.1:{port}");
+    eprintln!("bench_server listening on {endpoint} (db={db_path})");
+    if let Err(e) = engine::server::serve(api, &endpoint).await {
+        eprintln!("bench_server exited: {e}");
+    }
+}

engine/src/bin/client.rs

@@ -0,0 +1,85 @@
+//! Thin CLI over [`engine::client::Client`] for driving the engine by hand.
+//!
+//! Usage:
+//!   client submit   <ns> <name> <payload>          submit one task
+//!   client lease    <ns> <name> [max]              lease up to max (default 1)
+//!   client complete <ns> <name> <task_id> <result> complete one leased task
+//!   client cancel   <ns> <name> <task_id>          cancel (requeue) a lease
+//!
+//! Endpoint and auth token come from config.toml (same as the server).
+
+use engine::client::Client;
+use enginelib::config::Config;
+
+#[tokio::main]
+async fn main() {
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    if args.is_empty() {
+        eprintln!("usage: client <submit|lease|complete|cancel> ...");
+        std::process::exit(2);
+    }
+
+    let cfg = Config::new();
+    let endpoint = format!("tcp://{}", cfg.config_toml.host);
+    let auth = cfg.config_toml.auth_token.clone().unwrap_or_default();
+
+    let mut client = match Client::connect(&endpoint, auth) {
+        Ok(c) => c,
+        Err(e) => {
+            eprintln!("connect failed: {e}");
+            std::process::exit(1);
+        }
+    };
+
+    let result = match args[0].as_str() {
+        "submit" if args.len() >= 4 => {
+            let task_type = (args[1].clone(), args[2].clone());
+            let payload = args[3].clone().into_bytes();
+            client
+                .submit(task_type, vec![payload])
+                .await
+                .map(|ids| format!("submitted: {}", ids.join(", ")))
+        }
+        "lease" if args.len() >= 3 => {
+            let task_type = (args[1].clone(), args[2].clone());
+            let max: u32 = args.get(3).and_then(|s| s.parse().ok()).unwrap_or(1);
+            client
+                .lease(task_type, "cli".to_string(), max)
+                .await
+                .map(|tasks| {
+                    let lines: Vec<String> = tasks
+                        .iter()
+                        .map(|t| format!("  {} -> {} bytes", t.task_id, t.bytes.len()))
+                        .collect();
+                    format!("leased {}:\n{}", tasks.len(), lines.join("\n"))
+                })
+        }
+        "complete" if args.len() >= 5 => {
+            let task_type = (args[1].clone(), args[2].clone());
+            let results = vec![(args[3].clone(), args[4].clone().into_bytes())];
+            client
+                .complete(task_type, results)
+                .await
+                .map(|ok| format!("completed: {ok}"))
+        }
+        "cancel" if args.len() >= 4 => {
+            let task_type = (args[1].clone(), args[2].clone());
+            client
+                .cancel(task_type, args[3].clone())
+                .await
+                .map(|()| "cancelled".to_string())
+        }
+        other => {
+            eprintln!("unknown or malformed command: {other}");
+            std::process::exit(2);
+        }
+    };
+
+    match result {
+        Ok(msg) => println!("{msg}"),
+        Err(e) => {
+            eprintln!("error: {e}");
+            std::process::exit(1);
+        }
+    }
+}