From c07673a9846b6c6c0455cf3d10f426809e74a04b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 08:30:19 +0000 Subject: [PATCH 1/8] docs: honest concurrency docs + panic=unwind gate (Phase 0 PR-0a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WFL web request handlers run one at a time today, but several docs claimed parallel / "don't block others" behavior. Rewrite them to distinguish the concurrent transport (accept / TLS) from serial application handlers, and prefer "concurrent" over "parallel". Add a panic-strategy gate so the runtime's future catch_unwind-based request-handler fault isolation (Phase 1) cannot be silently undermined: - src/lib.rs: `#[cfg(panic = "abort")] compile_error!` — reflects the crate's real panic strategy; Cargo force-unwinds test/bench harnesses, so it never trips `cargo test`. - Cargo.toml: pin `panic = "unwind"` on the release profile (explicit; the default was already unwind). - CI: a failing-first "assert panic=abort is rejected" step that forces abort via `--config` and fails the job if the build succeeds. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- .github/workflows/ci.yml | 13 +++++++ Cargo.toml | 3 ++ Docs/01-introduction/key-features.md | 2 +- .../04-advanced-features/async-programming.md | 35 ++++++++++++------- Docs/04-advanced-features/index.md | 4 +-- Docs/06-best-practices/performance-tips.md | 4 +-- Docs/Archive/README.md | 2 +- src/lib.rs | 16 +++++++++ 8 files changed, 61 insertions(+), 18 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cacd161e..ad28e529 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -50,6 +50,19 @@ jobs: - name: Build (Release) run: cargo build --release --verbose + # Concurrency Phase 0 (PR-0a) gate: prove the panic-strategy gate is live. + # The `#[cfg(panic = "abort")] compile_error!` in src/lib.rs must fail the + # build when panic=abort is forced (via --config so Cargo actually compiles + # the crate with -C panic=abort). If this build SUCCEEDS the gate is not + # enforced and catch_unwind fault isolation would be a phantom control. + - name: Assert panic=abort is rejected + run: | + if cargo build --release --config 'profile.release.panic="abort"' --quiet 2>/dev/null; then + echo "::error::panic=abort build succeeded — the panic=unwind gate (src/lib.rs compile_error) is not enforced" + exit 1 + fi + echo "panic=abort correctly rejected by the compile_error gate" + # Run tests (integration tests now have access to release binary) - name: Run Tests run: cargo test --verbose diff --git a/Cargo.toml b/Cargo.toml index 45cc31c9..30deac97 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,3 +97,6 @@ harness = false [profile.release] debug = true +# Concurrency Phase 0 (PR-0a): pin unwinding so the runtime's catch_unwind-based +# request-handler fault isolation (Phase 1) stays sound. Enforced by build.rs. +panic = "unwind" diff --git a/Docs/01-introduction/key-features.md b/Docs/01-introduction/key-features.md index a00a65bd..d545d87e 100644 --- a/Docs/01-introduction/key-features.md +++ b/Docs/01-introduction/key-features.md @@ -62,7 +62,7 @@ close file input_file display "File contents: " with file_data ``` -Non-blocking I/O is natural and easy to use. +Cooperative, non-blocking I/O is natural and easy to use. ## 4. Built-in Web Server diff --git a/Docs/04-advanced-features/async-programming.md b/Docs/04-advanced-features/async-programming.md index cbbbf880..e71ec148 100644 --- a/Docs/04-advanced-features/async-programming.md +++ b/Docs/04-advanced-features/async-programming.md @@ -1,12 +1,14 @@ # Async Programming -WFL supports asynchronous operations using natural language syntax. Handle multiple operations concurrently without blocking. +WFL supports asynchronous operations using natural language syntax. The `wait for` keyword lets a slow operation yield cooperatively — while it waits on I/O, the WFL runtime can make progress on other awaited work instead of the thread sitting idle. ## What is Async? -**Synchronous (blocking):** Operations run one at a time. If one is slow, everything waits. +**Synchronous (blocking):** Operations run one at a time, and while one runs the thread can do nothing else. -**Asynchronous (non-blocking):** Operations can run concurrently. Slow operations don't block others. +**Asynchronous (cooperative):** An awaited operation *yields* while it waits on I/O, so the runtime can drive other awaited work in the meantime. + +> **Concurrent, not parallel.** WFL's async today is cooperative and single-threaded: awaited work is *interleaved* on one thread, not run on multiple cores at once. In a plain script, statements — including `wait for` statements — still execute one after another. The payoff of `wait for` is that a waiting operation releases the thread to the runtime rather than hard-blocking it. Running independent operations so they actually overlap is a planned feature (see [Concurrent Async](#concurrent-async-future-feature) below). ## The `wait for` Keyword @@ -56,7 +58,7 @@ close file file2 // Total time: Time1 + Time2 ``` -### With Async (Non-Blocking) +### With Async (Cooperative) ```wfl // Prepare two sample files @@ -67,16 +69,20 @@ open file at "file2.txt" for writing as setup2 wait for write content "second file" into setup2 close file setup2 -// Operations can overlap +// Each `wait for` still completes before the next statement runs — these do +// not overlap. What `wait for` changes is that while an operation waits on +// I/O, the thread yields to the runtime instead of hard-blocking, so other +// runtime work (such as a web server's transport layer) keeps making progress. open file at "file1.txt" for reading as file1 -wait for store content1 as read content from file1 // Doesn't block +wait for store content1 as read content from file1 // Yields while waiting close file file1 open file at "file2.txt" for reading as file2 -wait for store content2 as read content from file2 // Can run concurrently +wait for store content2 as read content from file2 // Runs after the first close file file2 -// Total time: ~max(Time1, Time2) +// Total time today: Time1 + Time2. Overlapping independent operations is a +// planned feature (see "Concurrent Async" below) — it is not available yet. ``` ## Common Async Operations @@ -102,7 +108,8 @@ display "File read complete" ```wfl listen on port 8080 as web_server -// Async request handling: wait for a request without blocking other work +// Wait for the next request. The transport layer accepts connections +// concurrently, but your handler code below runs one request at a time. wait for request comes in on web_server as incoming respond to incoming with "Response" and content_type "text/plain" @@ -158,7 +165,10 @@ end check ## Async in Web Servers -Web servers naturally use async operations: +Web servers naturally use async operations. Note that request *handlers* run one +at a time today — the transport layer (accepting connections, TLS handshakes) is +concurrent, but your handler code is serial. See +[Web Servers → Limitations](web-servers.md#limitations--notes) for details. ```wfl listen on port 8081 as web_server @@ -202,7 +212,8 @@ display "All operations complete" ### Concurrent Async (Future Feature) -Planned syntax for running operations in parallel: +Planned syntax for running independent operations concurrently (so they actually +overlap instead of running one after another): ```wfl // This is planned for future versions @@ -317,7 +328,7 @@ WFL's async support is built on the Tokio runtime and includes: In this section, you learned: ✅ **The `wait for` keyword** - Async operation syntax -✅ **Why async matters** - Non-blocking operations +✅ **Why async matters** - Cooperative, non-blocking I/O (concurrent, not parallel) ✅ **Common async operations** - File I/O, web requests, directory listing ✅ **Error handling** - Try-catch with async ✅ **Async in web servers** - Request handling diff --git a/Docs/04-advanced-features/index.md b/Docs/04-advanced-features/index.md index 3d73c569..f59b7321 100644 --- a/Docs/04-advanced-features/index.md +++ b/Docs/04-advanced-features/index.md @@ -32,7 +32,7 @@ If you've completed those, you're ready for advanced features! These features are "advanced" not because they're complicated, but because they're **powerful**: - **Web Servers** - Build HTTP APIs and web applications -- **Async** - Handle multiple operations concurrently +- **Async** - Cooperative, non-blocking I/O with `wait for` - **File I/O** - Persist data and process files - **Pattern Matching** - Validate and extract data - **Containers** - Organize code with object-oriented programming @@ -130,7 +130,7 @@ Object-oriented programming with readable syntax. **Focus on:** 1. [Containers (OOP)](containers-oop.md) - Code organization 2. [Web Servers](web-servers.md) - Backend services -3. [Async Programming](async-programming.md) - Concurrent operations +3. [Async Programming](async-programming.md) - Cooperative async I/O 4. [File I/O](file-io.md) - Persistence ## Real-World Examples diff --git a/Docs/06-best-practices/performance-tips.md b/Docs/06-best-practices/performance-tips.md index c217973a..12823c2c 100644 --- a/Docs/06-best-practices/performance-tips.md +++ b/Docs/06-best-practices/performance-tips.md @@ -50,7 +50,7 @@ display file_content // wait for request comes in on web_server as req ``` -**Why:** Non-blocking I/O lets WFL handle other work while waiting. +**Why:** While an awaited I/O operation is pending, WFL yields the thread to the runtime instead of hard-blocking it, so runtime-level work keeps progressing. (Request handlers still run one at a time — see the web server docs.) ## Short-Circuit Evaluation @@ -368,7 +368,7 @@ end action ## Best Practices ✅ **Choose right algorithm** - Most important! -✅ **Use async for I/O** - Non-blocking operations +✅ **Use async for I/O** - Cooperative, non-blocking operations ✅ **Cache expensive results** - Don't recalculate ✅ **Compile patterns once** - Reuse across iterations ✅ **Short-circuit cleverly** - Cheap checks first diff --git a/Docs/Archive/README.md b/Docs/Archive/README.md index 56a4231b..2de87a2c 100644 --- a/Docs/Archive/README.md +++ b/Docs/Archive/README.md @@ -56,7 +56,7 @@ end count ## ✨ Key Features - **📖 Natural Language Syntax**: Write code that reads like English sentences -- **🚀 Modern Async Support**: Built-in async/await for concurrent operations +- **🚀 Modern Async Support**: Built-in async/await for cooperative (single-thread) concurrency - **🛡️ Type Safety**: Static type checking with intelligent inference - **🌐 Web-First Design**: Native HTTP and database support - **🌍 Built-in Web Server**: Create HTTP servers with `listen on port 8080` - no external frameworks needed diff --git a/src/lib.rs b/src/lib.rs index 895634bb..05e413a6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,21 @@ #![deny(clippy::await_holding_refcell_ref)] +// Concurrency Phase 0 (PR-0a) — panic-strategy gate. +// +// The runtime relies on `std::panic::catch_unwind` to contain a panicking +// request handler so its siblings survive (concurrency Phase 1). That fault +// isolation is UNSOUND under `panic = "abort"`: an abort tears the whole +// process down before the catch can run, turning it into a phantom control. +// Fail the build rather than ship that. `cfg(panic = ...)` reflects the panic +// strategy actually compiled into this crate; Cargo force-unwinds test/bench +// harnesses, so this never trips `cargo test`. +#[cfg(panic = "abort")] +compile_error!( + "WFL requires panic = \"unwind\"; the runtime's catch_unwind-based request-handler \ + fault isolation (concurrency Phase 1) is unsound under panic = \"abort\". \ + Remove the panic = \"abort\" override." +); + // Global allocator for dhat heap profiling #[cfg(feature = "dhat-heap")] #[global_allocator] From adc91da5632039bb8479a17b22a7b809d6dfe35e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 08:30:35 +0000 Subject: [PATCH 2/8] feat: offload heavy crypto and bound the request queue (Phase 0 PR-0b, PR-0c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two runtime DoS mitigations that share the web-server dispatch path in src/interpreter/mod.rs, landed together. PR-0b — spawn_blocking for CPU-heavy crypto: - New src/stdlib/crypto_async.rs routes the 11 deliberately-slow crypto builtins (argon2/scrypt/bcrypt/pbkdf2 hash+verify, hash_password, verify_password, pbkdf2_hmac_sha256) onto Tokio's blocking pool. - The two async native-dispatch arms check the route first and await it, else fall back to the synchronous native. - Heavy compute stays in pub(crate) plain-data helpers in crypto.rs; args are extracted to owned String/u64/usize before the hop, so only plain data crosses the boundary and the interpreter core stays !Send. zeroize and the constant-time compare paths are unchanged. - Chosen over a new Value::AsyncNativeFunction variant to avoid rippling a new arm through every exhaustive Value match. PR-0c — bound the transport->interpreter queue (OOM shed): - New .wflcfg key web_server_request_queue_bound (default 256, zero rejected). - The request channel is now mpsc::channel(bound) instead of unbounded. The warp handler try_sends; on Full it logs and returns a 503 (with Retry-After) via overloaded_response(), without blocking the transport task. Tests: in-crate crypto_async tests (exact routed set; a deterministic off-thread proof on a current_thread runtime; hash/verify round-trips; routed PBKDF2 == direct); in-crate queue_bound_tests (well-formed 503; deterministic over-cap shed); tests/web_queue_bound_test.rs (config parsing). Existing crypto and web-server suites pass unchanged through the new paths. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- Docs/04-advanced-features/web-servers.md | 3 +- Docs/reference/configuration-reference.md | 10 + src/config.rs | 44 ++++ src/interpreter/mod.rs | 163 ++++++++++-- src/stdlib/crypto.rs | 91 +++++-- src/stdlib/crypto_async.rs | 303 ++++++++++++++++++++++ src/stdlib/mod.rs | 1 + tests/web_queue_bound_test.rs | 56 ++++ 8 files changed, 619 insertions(+), 52 deletions(-) create mode 100644 src/stdlib/crypto_async.rs create mode 100644 tests/web_queue_bound_test.rs diff --git a/Docs/04-advanced-features/web-servers.md b/Docs/04-advanced-features/web-servers.md index e1267c5f..8f8a9987 100644 --- a/Docs/04-advanced-features/web-servers.md +++ b/Docs/04-advanced-features/web-servers.md @@ -1026,6 +1026,7 @@ end check - **Single request handling:** Each `wait for request` handles one request - **Blocking:** Server handles requests sequentially (TLS handshakes are concurrent, but your responses are serialized) +- **Bounded accept queue:** Because handlers are serial, incoming requests queue up behind the one being handled. That queue is bounded (default 256, configurable via `web_server_request_queue_bound`). When it is full, the server sheds new requests with a `503 Service Unavailable` (plus a `Retry-After` header) and logs a warning, rather than growing memory without bound. See [Configuration Reference](../reference/configuration-reference.md#web_server_request_queue_bound). - **No middleware system** (yet) - Implement manually - **No built-in session management** - Implement yourself @@ -1196,7 +1197,7 @@ Expand your web development skills: Learn to read and write files for data persistence. **[Async Programming →](async-programming.md)** -Handle multiple operations concurrently. +Cooperative async I/O with `wait for`. **[Pattern Matching →](pattern-matching.md)** Validate request data and extract parameters. diff --git a/Docs/reference/configuration-reference.md b/Docs/reference/configuration-reference.md index 5eebc1b8..5d1993ee 100644 --- a/Docs/reference/configuration-reference.md +++ b/Docs/reference/configuration-reference.md @@ -308,6 +308,16 @@ Maximum HTTP request body size accepted by `listen on port` servers, in bytes. R Raise this when accepting file uploads via `parse_multipart` or raw `body_bytes`. Keep it as small as practical for public-facing APIs. +#### web_server_request_queue_bound + +Maximum number of accepted-but-not-yet-handled HTTP requests held in the queue between the transport layer and your `wait for request` loop (DoS protection). + +- **Type:** Integer (at least 1) +- **Default:** `256` +- **Example:** `web_server_request_queue_bound = 512` + +Because request handlers run one at a time (see [Web Servers → Limitations](../04-advanced-features/web-servers.md#limitations--notes)), a burst of traffic queues up behind the handler. Without a bound, that queue could grow until the process runs out of memory. When the queue is full, the server **sheds** further requests with a `503 Service Unavailable` (and a `Retry-After` header) and logs a warning, instead of buffering unbounded work. Raise it to absorb larger bursts at the cost of more memory; lower it to shed sooner under load. A value of `0` is rejected (the default is kept). + ## Example Configuration Files ### Development Configuration diff --git a/src/config.rs b/src/config.rs index dade12f3..be505160 100644 --- a/src/config.rs +++ b/src/config.rs @@ -48,6 +48,11 @@ pub struct WflConfig { /// Maximum accepted HTTP request body size in bytes (DoS protection). /// Default 1 MiB; raise for media uploads via `.wflcfg`. pub web_server_max_body_size: usize, + /// Maximum number of accepted-but-not-yet-handled HTTP requests held in the + /// transport→interpreter queue (DoS protection). When the queue is full the + /// server sheds new requests with a 503 instead of growing memory without + /// bound. Default 256; must be at least 1. + pub web_server_request_queue_bound: usize, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -128,6 +133,9 @@ impl Default for WflConfig { web_server_tls_key_file: None, // 1 MiB default body limit (DoS protection); raise for uploads web_server_max_body_size: 1_048_576, + // Bound the accept queue so a flood sheds with 503 rather than + // growing memory without bound. Aligns with the Phase 1 in-flight cap. + web_server_request_queue_bound: 256, } } } @@ -657,6 +665,42 @@ fn parse_config_text(config: &mut WflConfig, text: &str, file: &Path) { ); } } + "web_server_request_queue_bound" => { + if let Ok(bound) = value.parse::() { + // Reject zero: `tokio::sync::mpsc::channel(0)` panics, and a + // zero-length queue could never accept a request. + if bound == 0 { + log::warn!( + "Invalid web_server_request_queue_bound '0' in {}: must be at least 1. Keeping {}", + file.display(), + config.web_server_request_queue_bound + ); + } else { + if config.web_server_request_queue_bound + != WflConfig::default().web_server_request_queue_bound + { + log::debug!( + "Overriding web_server_request_queue_bound: {} -> {} from {}", + config.web_server_request_queue_bound, + bound, + file.display() + ); + } + config.web_server_request_queue_bound = bound; + log::debug!( + "Loaded web_server_request_queue_bound: {} from {}", + config.web_server_request_queue_bound, + file.display() + ); + } + } else { + log::warn!( + "Invalid web_server_request_queue_bound '{}' in {}: expected a positive integer", + value, + file.display() + ); + } + } _ => { log::warn!("Unknown configuration key: {} in {}", key, file.display()); } diff --git a/src/interpreter/mod.rs b/src/interpreter/mod.rs index 0cab3f03..3fae944a 100644 --- a/src/interpreter/mod.rs +++ b/src/interpreter/mod.rs @@ -117,11 +117,28 @@ pub(crate) fn lookup_header_case_insensitive( #[derive(Debug)] pub struct WflWebServer { - pub request_receiver: Arc>>, - pub request_sender: mpsc::UnboundedSender, + // Bounded transport→interpreter queue (Phase 0, PR-0c): a full queue sheds + // new requests with 503 rather than growing memory without bound. + pub request_receiver: Arc>>, + pub request_sender: mpsc::Sender, pub server_handle: Option>, } +/// Build the 503 response returned when the transport→interpreter request queue +/// is full (Phase 0, PR-0c). A free function so the shed path can be unit-tested +/// without standing up a live server. +pub(crate) fn overloaded_response() -> warp::http::Response> { + let body = b"Service Unavailable: the server is overloaded, please retry later\n".to_vec(); + let content_length = body.len(); + warp::http::Response::builder() + .status(warp::http::StatusCode::SERVICE_UNAVAILABLE) + .header("Content-Type", "text/plain; charset=utf-8") + .header("Content-Length", content_length) + .header("Retry-After", "1") + .body(body) + .expect("static 503 response is always valid") +} + // --------------------------------------------------------------------------- // WebSocket support // @@ -5378,9 +5395,12 @@ impl Interpreter { } }; - // Create request/response channels + // Create request/response channels. The request queue is bounded + // (Phase 0, PR-0c) so a flood of accepted-but-unhandled requests + // sheds with 503 instead of growing memory without bound. + let queue_bound = self.config.web_server_request_queue_bound.max(1); let (request_sender, request_receiver) = - mpsc::unbounded_channel::(); + mpsc::channel::(queue_bound); let request_receiver = Arc::new(tokio::sync::Mutex::new(request_receiver)); // Create warp routes that handle all HTTP methods and paths. @@ -5476,11 +5496,28 @@ impl Interpreter { ))), }; - // Send request to WFL interpreter - if sender.send(wfl_request).is_err() { - return Err(warp::reject::custom(ServerError( - "Request channel closed".to_string(), - ))); + // Send request to WFL interpreter. The queue is + // bounded (Phase 0, PR-0c): a full queue means the + // interpreter is saturated, so shed with 503 rather + // than buffering unbounded work. `try_send` never + // blocks the transport task. + match sender.try_send(wfl_request) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(shed)) => { + log::warn!( + "web server request queue full (capacity {}); shedding {} {} from {} with 503", + sender.max_capacity(), + shed.method, + shed.path, + shed.client_ip + ); + return Ok(overloaded_response()); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + return Err(warp::reject::custom(ServerError( + "Request channel closed".to_string(), + ))); + } } // Wait for response @@ -8281,14 +8318,29 @@ impl Interpreter { Value::Function(func) => { self.call_function(&func, arg_values, *line, *column).await } - Value::NativeFunction(_, native_fn) => { - native_fn(arg_values.clone()).map_err(|e| { - RuntimeError::new( - format!("Error in native function: {e}"), - *line, - *column, - ) - }) + Value::NativeFunction(native_name, native_fn) => { + // CPU-heavy crypto builtins hop onto the blocking pool so + // they don't monopolize the interpreter thread (Phase 0, + // PR-0b). Everything else runs synchronously as before. + if let Some(fut) = + crate::stdlib::crypto_async::route(native_name, &arg_values) + { + fut.await.map_err(|e| { + RuntimeError::new( + format!("Error in native function: {e}"), + *line, + *column, + ) + }) + } else { + native_fn(arg_values.clone()).map_err(|e| { + RuntimeError::new( + format!("Error in native function: {e}"), + *line, + *column, + ) + }) + } } _ => Err(RuntimeError::new( format!("Cannot call {}", function_val.type_name()), @@ -8354,12 +8406,21 @@ impl Interpreter { // Preserve the native error's message and kind; only // point the location at the call site (natives report - // their position as 0,0). - native_fn(arg_values).map_err(|mut e| { - e.line = *line; - e.column = *column; - e - }) + // their position as 0,0). CPU-heavy crypto builtins are + // routed onto the blocking pool (Phase 0, PR-0b). + if let Some(fut) = crate::stdlib::crypto_async::route(name, &arg_values) { + fut.await.map_err(|mut e| { + e.line = *line; + e.column = *column; + e + }) + } else { + native_fn(arg_values).map_err(|mut e| { + e.line = *line; + e.column = *column; + e + }) + } } _ => Err(RuntimeError::new( format!("'{name}' is not callable"), @@ -9928,3 +9989,59 @@ mod process_tests { ); } } + +/// Phase 0 (PR-0c): the bounded request queue sheds with a well-formed 503 when +/// the interpreter is saturated, instead of buffering work without bound. +#[cfg(test)] +mod queue_bound_tests { + use super::*; + + #[test] + fn overloaded_response_is_a_well_formed_503() { + let resp = overloaded_response(); + assert_eq!(resp.status(), warp::http::StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + resp.headers() + .get("Content-Type") + .and_then(|v| v.to_str().ok()), + Some("text/plain; charset=utf-8") + ); + // Content-Length matches the actual body byte count. + let declared: usize = resp + .headers() + .get("Content-Length") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse().ok()) + .expect("Content-Length header present and numeric"); + assert_eq!(declared, resp.body().len()); + assert!(!resp.body().is_empty()); + } + + /// The over-cap decision is deterministic: once a bounded channel is full, + /// `try_send` reports `Full` (which the warp handler maps to `overloaded_response`), + /// and never blocks or grows the queue. + #[tokio::test] + async fn full_queue_sheds_deterministically() { + let bound = 4usize; + let (tx, _rx) = mpsc::channel::(bound); + + // Fill to capacity — every send within the bound succeeds. + for i in 0..bound { + tx.try_send(i as u32) + .expect("send within capacity succeeds"); + } + assert_eq!(tx.max_capacity(), bound); + + // The next send over capacity sheds rather than blocking or growing. + match tx.try_send(999) { + Err(mpsc::error::TrySendError::Full(v)) => assert_eq!(v, 999), + other => panic!("expected Full over capacity, got {other:?}"), + } + + // The shed maps to a 503. + assert_eq!( + overloaded_response().status(), + warp::http::StatusCode::SERVICE_UNAVAILABLE + ); + } +} diff --git a/src/stdlib/crypto.rs b/src/stdlib/crypto.rs index bf67b5d4..aaab2738 100644 --- a/src/stdlib/crypto.rs +++ b/src/stdlib/crypto.rs @@ -594,7 +594,7 @@ const MAX_SECURE_RANDOM_BYTES: usize = 4096; /// Convert a WFL number argument into a non-negative integer count, rejecting /// non-finite, negative, or fractional values with a clear message. -fn expect_count(func: &str, name: &str, value: &Value) -> Result { +pub(crate) fn expect_count(func: &str, name: &str, value: &Value) -> Result { let n = expect_number(value)?; if !n.is_finite() || n < 0.0 || n.fract() != 0.0 { return Err(RuntimeError::new( @@ -612,7 +612,10 @@ fn expect_count(func: &str, name: &str, value: &Value) -> Result) -> Result { @@ -623,6 +626,22 @@ pub fn native_pbkdf2_hmac_sha256(args: Vec) -> Result Result { if password.len() > MAX_INPUT_SIZE || salt.len() > MAX_INPUT_SIZE { return Err(RuntimeError::new( format!( @@ -670,7 +689,7 @@ pub fn native_pbkdf2_hmac_sha256(args: Vec) -> Result Result { salt } -fn argon2_hash_str(func: &str, password: &str) -> Result { +pub(crate) fn argon2_hash_str(func: &str, password: &str) -> Result { check_password_len(func, password)?; let salt = random_salt()?; Argon2::default() @@ -797,22 +816,26 @@ pub fn native_argon2_hash(args: Vec) -> Result { )?))) } +/// Verify a password against a stored Argon2 PHC string (plain-data core). +pub(crate) fn argon2_verify_str(password: &str, stored: &str) -> bool { + match PasswordHash::new(stored) { + Ok(parsed) => Argon2::default() + .verify_password(password.as_bytes(), &parsed) + .is_ok(), + Err(_) => false, + } +} + /// Verify a password against a stored Argon2 PHC string. Returns a boolean. /// Usage: argon2_verify of "my password" and stored_hash pub fn native_argon2_verify(args: Vec) -> Result { check_arg_count("argon2_verify", &args, 2)?; let password = expect_text(&args[0])?; let stored = expect_text(&args[1])?; - let ok = match PasswordHash::new(&stored) { - Ok(parsed) => Argon2::default() - .verify_password(password.as_bytes(), &parsed) - .is_ok(), - Err(_) => false, - }; - Ok(Value::Bool(ok)) + Ok(Value::Bool(argon2_verify_str(&password, &stored))) } -fn scrypt_hash_str(func: &str, password: &str) -> Result { +pub(crate) fn scrypt_hash_str(func: &str, password: &str) -> Result { check_password_len(func, password)?; let salt = random_salt()?; Scrypt @@ -832,20 +855,24 @@ pub fn native_scrypt_hash(args: Vec) -> Result { )?))) } +/// Verify a password against a stored scrypt PHC string (plain-data core). +pub(crate) fn scrypt_verify_str(password: &str, stored: &str) -> bool { + match PasswordHash::new(stored) { + Ok(parsed) => Scrypt.verify_password(password.as_bytes(), &parsed).is_ok(), + Err(_) => false, + } +} + /// Verify a password against a stored scrypt PHC string. Returns a boolean. /// Usage: scrypt_verify of "my password" and stored_hash pub fn native_scrypt_verify(args: Vec) -> Result { check_arg_count("scrypt_verify", &args, 2)?; let password = expect_text(&args[0])?; let stored = expect_text(&args[1])?; - let ok = match PasswordHash::new(&stored) { - Ok(parsed) => Scrypt.verify_password(password.as_bytes(), &parsed).is_ok(), - Err(_) => false, - }; - Ok(Value::Bool(ok)) + Ok(Value::Bool(scrypt_verify_str(&password, &stored))) } -fn pbkdf2_hash_str(func: &str, password: &str) -> Result { +pub(crate) fn pbkdf2_hash_str(func: &str, password: &str) -> Result { check_password_len(func, password)?; let salt = random_salt()?; // Override the weak default iteration count with OWASP's recommendation. @@ -870,20 +897,24 @@ pub fn native_pbkdf2_hash(args: Vec) -> Result { )?))) } +/// Verify a password against a stored PBKDF2 PHC string (plain-data core). +pub(crate) fn pbkdf2_verify_str(password: &str, stored: &str) -> bool { + match PasswordHash::new(stored) { + Ok(parsed) => Pbkdf2.verify_password(password.as_bytes(), &parsed).is_ok(), + Err(_) => false, + } +} + /// Verify a password against a stored PBKDF2 PHC string. Returns a boolean. /// Usage: pbkdf2_verify of "my password" and stored_hash pub fn native_pbkdf2_verify(args: Vec) -> Result { check_arg_count("pbkdf2_verify", &args, 2)?; let password = expect_text(&args[0])?; let stored = expect_text(&args[1])?; - let ok = match PasswordHash::new(&stored) { - Ok(parsed) => Pbkdf2.verify_password(password.as_bytes(), &parsed).is_ok(), - Err(_) => false, - }; - Ok(Value::Bool(ok)) + Ok(Value::Bool(pbkdf2_verify_str(&password, &stored))) } -fn bcrypt_hash_str(func: &str, password: &str) -> Result { +pub(crate) fn bcrypt_hash_str(func: &str, password: &str) -> Result { check_password_len(func, password)?; // bcrypt manages its own salt internally and returns an MCF `$2b$` string. bcrypt::hash(password.as_bytes(), bcrypt::DEFAULT_COST) @@ -903,20 +934,24 @@ pub fn native_bcrypt_hash(args: Vec) -> Result { )?))) } +/// Verify a password against a stored bcrypt hash (plain-data core). +/// A malformed stored hash simply fails verification rather than erroring. +pub(crate) fn bcrypt_verify_str(password: &str, stored: &str) -> bool { + bcrypt::verify(password.as_bytes(), stored).unwrap_or(false) +} + /// Verify a password against a stored bcrypt hash. Returns a boolean. /// Usage: bcrypt_verify of "my password" and stored_hash pub fn native_bcrypt_verify(args: Vec) -> Result { check_arg_count("bcrypt_verify", &args, 2)?; let password = expect_text(&args[0])?; let stored = expect_text(&args[1])?; - // A malformed stored hash simply fails verification rather than erroring. - let ok = bcrypt::verify(password.as_bytes(), stored.as_ref()).unwrap_or(false); - Ok(Value::Bool(ok)) + Ok(Value::Bool(bcrypt_verify_str(&password, &stored))) } /// Verify a password against any supported stored hash, auto-detecting the /// algorithm from the stored string. Returns a boolean. -fn verify_any_password(password: &str, stored: &str) -> bool { +pub(crate) fn verify_any_password(password: &str, stored: &str) -> bool { // bcrypt MCF strings are not PHC format; detect them by their version prefix. if stored.starts_with("$2a$") || stored.starts_with("$2b$") || stored.starts_with("$2y$") { return bcrypt::verify(password.as_bytes(), stored).unwrap_or(false); diff --git a/src/stdlib/crypto_async.rs b/src/stdlib/crypto_async.rs new file mode 100644 index 00000000..f4634953 --- /dev/null +++ b/src/stdlib/crypto_async.rs @@ -0,0 +1,303 @@ +//! Off-thread routing for CPU-heavy crypto builtins (WFL concurrency Phase 0, +//! PR-0b). +//! +//! The password-hashing / KDF builtins (`argon2_hash`, `bcrypt_hash`, +//! `scrypt_hash`, `pbkdf2_hash`, `pbkdf2_hmac_sha256`, `hash_password`, and the +//! verify counterparts) are *deliberately* slow — Argon2 is memory-hard, PBKDF2 +//! runs 600k rounds. Run inline on the single interpreter thread, one login-like +//! call stalls the whole process (a cooperative-scheduling DoS). +//! +//! This module hops that work onto Tokio's blocking pool with +//! [`tokio::task::spawn_blocking`], following the libuv / Node pattern. The +//! interpreter core stays `!Send`: arguments are extracted into owned plain data +//! (`String`, `u64`, `usize`) on the interpreter thread *before* the hop, only +//! that plain data crosses the boundary, and the resulting `Value` is built back +//! on the interpreter thread after the `.await`. No `Rc`/`RefCell`/`Value`/ +//! `Environment` ever crosses threads. +//! +//! The returned future is a `!Send` [`LocalBoxFuture`]; it is awaited from the +//! interpreter future, which runs via `block_on` (never `tokio::spawn`), so +//! awaiting a `Send` `JoinHandle` from inside it is sound. + +use crate::interpreter::error::RuntimeError; +use crate::interpreter::value::Value; +use crate::stdlib::crypto; +use crate::stdlib::helpers::{check_arg_count, expect_text}; +use futures_util::future::{FutureExt, LocalBoxFuture}; +use std::sync::Arc; + +/// Route a CPU-heavy crypto builtin onto the blocking pool. +/// +/// Returns `None` for any name that is not a heavy crypto builtin — the caller +/// then invokes the plain synchronous native as before. For the routed set +/// (see below), returns a future that performs argument validation on the +/// interpreter thread and the heavy computation on `spawn_blocking`. +/// +/// Routed set (11): `argon2_hash`, `argon2_verify`, `scrypt_hash`, +/// `scrypt_verify`, `bcrypt_hash`, `bcrypt_verify`, `pbkdf2_hash`, +/// `pbkdf2_verify`, `hash_password`, `verify_password`, `pbkdf2_hmac_sha256`. +/// +/// `constant_time_equals` and `secure_random_bytes` are intentionally *not* +/// routed: they are fast and routing them would only add a scheduling hop. +pub(crate) fn route( + name: &str, + args: &[Value], +) -> Option>> { + let fut = match name { + "argon2_hash" => hash_route("argon2_hash", args, crypto::argon2_hash_str), + "scrypt_hash" => hash_route("scrypt_hash", args, crypto::scrypt_hash_str), + "pbkdf2_hash" => hash_route("pbkdf2_hash", args, crypto::pbkdf2_hash_str), + "bcrypt_hash" => hash_route("bcrypt_hash", args, crypto::bcrypt_hash_str), + // hash_password is Argon2id under the hood (see native_hash_password). + "hash_password" => hash_route("hash_password", args, crypto::argon2_hash_str), + "argon2_verify" => verify_route("argon2_verify", args, crypto::argon2_verify_str), + "scrypt_verify" => verify_route("scrypt_verify", args, crypto::scrypt_verify_str), + "pbkdf2_verify" => verify_route("pbkdf2_verify", args, crypto::pbkdf2_verify_str), + "bcrypt_verify" => verify_route("bcrypt_verify", args, crypto::bcrypt_verify_str), + "verify_password" => verify_route("verify_password", args, crypto::verify_any_password), + "pbkdf2_hmac_sha256" => pbkdf2_hmac_route(args), + _ => return None, + }; + Some(fut) +} + +/// One-argument password-hash builtins: ` of "password"` → PHC/MCF string. +/// `compute` takes `(func_name, password)` to match the crypto helper shape. +fn hash_route( + func: &'static str, + args: &[Value], + compute: fn(&str, &str) -> Result, +) -> LocalBoxFuture<'static, Result> { + // Extract on the interpreter thread; only the owned `String` crosses the + // boundary below. + let extracted = extract_password(func, args); + async move { + let password = extracted?; + let hash = tokio::task::spawn_blocking(move || compute(func, &password)) + .await + .map_err(|e| join_error(func, e))??; + Ok(Value::Text(Arc::from(hash))) + } + .boxed_local() +} + +/// Two-argument verify builtins: ` of "password" and stored` → boolean. +fn verify_route( + func: &'static str, + args: &[Value], + compute: fn(&str, &str) -> bool, +) -> LocalBoxFuture<'static, Result> { + let extracted = extract_password_and_stored(func, args); + async move { + let (password, stored) = extracted?; + let ok = tokio::task::spawn_blocking(move || compute(&password, &stored)) + .await + .map_err(|e| join_error(func, e))?; + Ok(Value::Bool(ok)) + } + .boxed_local() +} + +/// Raw PBKDF2-HMAC-SHA256 KDF: `pbkdf2_hmac_sha256 of password and salt and +/// iterations and length` → hex string. +fn pbkdf2_hmac_route(args: &[Value]) -> LocalBoxFuture<'static, Result> { + const FUNC: &str = "pbkdf2_hmac_sha256"; + let extracted = (|| { + check_arg_count(FUNC, args, 4)?; + let password = expect_text(&args[0])?.to_string(); + let salt = expect_text(&args[1])?.to_string(); + let iterations = crypto::expect_count(FUNC, "iterations", &args[2])?; + let length = crypto::expect_count(FUNC, "length", &args[3])? as usize; + Ok::<_, RuntimeError>((password, salt, iterations, length)) + })(); + async move { + let (password, salt, iterations, length) = extracted?; + let hex = tokio::task::spawn_blocking(move || { + crypto::pbkdf2_hmac_sha256_str(&password, &salt, iterations, length) + }) + .await + .map_err(|e| join_error(FUNC, e))??; + Ok(Value::Text(Arc::from(hex))) + } + .boxed_local() +} + +/// Validate arity and pull out an owned password `String` on the interpreter +/// thread. Kept as a plain (non-async) fn so extraction errors surface before +/// any thread hop. +fn extract_password(func: &str, args: &[Value]) -> Result { + check_arg_count(func, args, 1)?; + Ok(expect_text(&args[0])?.to_string()) +} + +fn extract_password_and_stored( + func: &str, + args: &[Value], +) -> Result<(String, String), RuntimeError> { + check_arg_count(func, args, 2)?; + let password = expect_text(&args[0])?.to_string(); + let stored = expect_text(&args[1])?.to_string(); + Ok((password, stored)) +} + +/// Map a `spawn_blocking` join failure (panic/cancel in the blocking task) to a +/// `RuntimeError`. Panics inside the crypto helpers are not expected, but a +/// join error must not be silently swallowed. +fn join_error(func: &str, e: tokio::task::JoinError) -> RuntimeError { + RuntimeError::new(format!("{func}: crypto task failed: {e}"), 0, 0) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + fn text(s: &str) -> Value { + Value::Text(Arc::from(s)) + } + + /// Only the deliberately-slow crypto builtins route; everything else falls + /// through to the synchronous native. Guards against silent scope creep in + /// the routed set. + #[test] + fn route_map_covers_exactly_the_heavy_crypto_builtins() { + let one = [text("x")]; + let two = [text("x"), text("y")]; + let four = [ + text("x"), + text("y"), + Value::Number(1.0), + Value::Number(16.0), + ]; + + for name in [ + "argon2_hash", + "scrypt_hash", + "bcrypt_hash", + "pbkdf2_hash", + "hash_password", + ] { + assert!(route(name, &one).is_some(), "{name} should route"); + } + for name in [ + "argon2_verify", + "scrypt_verify", + "bcrypt_verify", + "pbkdf2_verify", + "verify_password", + ] { + assert!(route(name, &two).is_some(), "{name} should route"); + } + assert!(route("pbkdf2_hmac_sha256", &four).is_some()); + + // Fast / non-crypto builtins must NOT route. + for name in [ + "constant_time_equals", + "secure_random_bytes", + "sha256", + "wflhash256", + "print", + "length", + ] { + assert!(route(name, &two).is_none(), "{name} must not route"); + } + } + + /// The core Phase 0 property (plan U4): the heavy work runs off the + /// interpreter thread. On a single-threaded runtime, a concurrently-spawned + /// ticker can only make progress *during* the crypto `.await` if that work + /// was offloaded via `spawn_blocking`. Had the hash run inline, it would + /// have monopolized the one thread and the ticker would still read zero when + /// the await returns. This is deterministic and independent of core count. + #[tokio::test(flavor = "current_thread")] + async fn routed_crypto_frees_the_interpreter_thread() { + let counter = std::sync::Arc::new(AtomicUsize::new(0)); + let ticker_counter = counter.clone(); + let ticker = tokio::spawn(async move { + for _ in 0..5 { + tokio::task::yield_now().await; + ticker_counter.fetch_add(1, Ordering::SeqCst); + } + }); + + let args = [text("a-password-to-hash")]; + let out = route("argon2_hash", &args) + .expect("argon2_hash routes") + .await + .expect("hash succeeds"); + + // Read the ticker's progress the instant the await returns, before we + // yield again — so a passing result can only mean the runtime polled the + // ticker while the blocking pool computed the hash. + let progressed = counter.load(Ordering::SeqCst); + ticker.await.expect("ticker joins"); + + assert!( + progressed > 0, + "ticker never advanced during the crypto await — work was not offloaded" + ); + assert!(matches!(out, Value::Text(_))); + } + + /// The routed hash still produces a valid, verifiable credential. + #[tokio::test] + async fn routed_argon2_hash_roundtrips() { + let out = route("argon2_hash", &[text("correct horse")]) + .unwrap() + .await + .unwrap(); + let hash = match out { + Value::Text(h) => h, + other => panic!("expected text, got {other:?}"), + }; + assert!(crypto::argon2_verify_str("correct horse", &hash)); + assert!(!crypto::argon2_verify_str("wrong password", &hash)); + } + + /// bcrypt uses a distinct MCF (`$2b$`) format, exercising a different helper. + #[tokio::test] + async fn routed_bcrypt_hash_roundtrips() { + let out = route("bcrypt_hash", &[text("hunter2")]) + .unwrap() + .await + .unwrap(); + let hash = match out { + Value::Text(h) => h, + other => panic!("expected text, got {other:?}"), + }; + assert!(hash.starts_with("$2")); + assert!(crypto::bcrypt_verify_str("hunter2", &hash)); + assert!(!crypto::bcrypt_verify_str("nope", &hash)); + } + + /// The routed KDF output is byte-identical to the direct plain-data helper — + /// the thread hop changes nothing about the result (deterministic: no salt). + #[tokio::test] + async fn routed_pbkdf2_matches_direct() { + let args = [ + text("password"), + text("salt"), + Value::Number(4096.0), + Value::Number(32.0), + ]; + let out = route("pbkdf2_hmac_sha256", &args).unwrap().await.unwrap(); + let routed = match out { + Value::Text(h) => h.to_string(), + other => panic!("expected text, got {other:?}"), + }; + let direct = crypto::pbkdf2_hmac_sha256_str("password", "salt", 4096, 32).unwrap(); + assert_eq!(routed, direct); + } + + /// Argument errors still surface (as the routed future's error), so routing + /// does not mask misuse. + #[tokio::test] + async fn routed_argerror_is_reported() { + // argon2_hash expects exactly one argument. + let err = route("argon2_hash", &[text("a"), text("b")]) + .unwrap() + .await + .unwrap_err(); + assert!(err.message.contains("argon2_hash")); + } +} diff --git a/src/stdlib/mod.rs b/src/stdlib/mod.rs index d769c2c9..b583cff0 100644 --- a/src/stdlib/mod.rs +++ b/src/stdlib/mod.rs @@ -1,5 +1,6 @@ pub mod core; pub mod crypto; +pub mod crypto_async; pub mod filesystem; pub mod helpers; pub mod json; diff --git a/tests/web_queue_bound_test.rs b/tests/web_queue_bound_test.rs new file mode 100644 index 00000000..f0c0ee05 --- /dev/null +++ b/tests/web_queue_bound_test.rs @@ -0,0 +1,56 @@ +//! Phase 0 (PR-0c) — bounded request-queue configuration. +//! +//! The transport→interpreter request queue is bounded so a flood of accepted +//! requests sheds with 503 rather than growing memory without bound. These +//! tests exercise the `web_server_request_queue_bound` config key through the +//! public `load_config` API: a valid override applies, and invalid values +//! (zero, non-numeric) are rejected while keeping the safe default. +//! +//! The runtime shed decision itself (`try_send` full → `overloaded_response` +//! 503) is covered by the in-crate `queue_bound_tests` unit tests in +//! `src/interpreter/mod.rs`, since `overloaded_response` is crate-internal. + +use std::fs; +use wfl::config::load_config; + +const DEFAULT_BOUND: usize = 256; + +/// Write a `.wflcfg` with the given body into a fresh temp dir and load it. +fn load_with_cfg(body: &str) -> wfl::config::WflConfig { + let dir = tempfile::tempdir().expect("create temp dir"); + fs::write(dir.path().join(".wflcfg"), body).expect("write .wflcfg"); + load_config(dir.path()) +} + +#[test] +fn default_queue_bound_is_256() { + // No key present → the safe default applies. + let cfg = load_with_cfg("# empty config\n"); + assert_eq!(cfg.web_server_request_queue_bound, DEFAULT_BOUND); +} + +#[test] +fn valid_queue_bound_is_applied() { + let cfg = load_with_cfg("web_server_request_queue_bound = 8\n"); + assert_eq!(cfg.web_server_request_queue_bound, 8); +} + +#[test] +fn zero_queue_bound_is_rejected_and_default_kept() { + // A zero bound would panic tokio's `mpsc::channel(0)` and could never accept + // a request, so it must be rejected in favor of the default. + let cfg = load_with_cfg("web_server_request_queue_bound = 0\n"); + assert_eq!(cfg.web_server_request_queue_bound, DEFAULT_BOUND); +} + +#[test] +fn non_numeric_queue_bound_is_rejected_and_default_kept() { + let cfg = load_with_cfg("web_server_request_queue_bound = lots\n"); + assert_eq!(cfg.web_server_request_queue_bound, DEFAULT_BOUND); +} + +#[test] +fn large_queue_bound_is_applied() { + let cfg = load_with_cfg("web_server_request_queue_bound = 4096\n"); + assert_eq!(cfg.web_server_request_queue_bound, 4096); +} From d7762b7c56e2217398df641a43ed9dbe401181cb Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 08:30:45 +0000 Subject: [PATCH 3/8] docs: Phase 0 dev diary, tracker, and validate-docs policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Dev Diary entry for Phase 0 concurrency hardening (0a/0b/0c), including the build.rs -> compile_error! dead end and the residual-risk notes. - Flip the Phase 0 rows (0a/0b/0c) to Done in the concurrency phase plan tracker. - CLAUDE.md: add a binding "Docs Must Be Honest (validate docs)" policy — docs describe what actually ships, planned behavior is marked as such, and every user-visible change ships validated docs AND a Dev Diary entry in the same change. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- CLAUDE.md | 5 + ...026-07-11-phase-0-concurrency-hardening.md | 122 ++++++++++++++++++ Docs/development/concurrency-phase-plan.md | 6 +- 3 files changed, 130 insertions(+), 3 deletions(-) create mode 100644 Dev diary/2026-07-11-phase-0-concurrency-hardening.md diff --git a/CLAUDE.md b/CLAUDE.md index 170c3790..f8437ff6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -176,6 +176,11 @@ Source Code → Lexer → Parser → Analyzer → Type Checker → Interpreter - A working example (in `TestPrograms/`, validated with MCP) demonstrating the feature. - A Dev Diary entry in `Dev diary/` for any non-trivial feature or behavior change. - When a feature is removed or its syntax changes, remove or fix the now-stale docs and examples — don't leave contradictions. +- **Docs Must Be Honest — "validate docs" (MANDATORY)**: Documentation describes **what actually ships today**, not what is aspirational. This is a binding policy, not a preference: + - **No overclaiming runtime behavior.** Never describe behavior the runtime does not have (e.g. calling serial request handlers "parallel" or saying they "don't block others"). Prefer the precise word — say "concurrent" (interleaved on one thread) vs "parallel" (multiple cores) deliberately, and describe the transport/handler split accurately. + - **Mark planned/future behavior explicitly.** Anything not yet implemented must be labeled as planned/future so a reader never mistakes it for current behavior. + - **Validate, don't just assert.** Every user-visible change ships **validated** docs (MCP tools + `python scripts/validate_docs_examples.py` for any touched example) **and** a Dev Diary entry, in the **same change**. "Validate docs" means both: the examples run, and the prose matches the implementation. + - When behavior changes, fix the now-stale claims in the same change — a doc that contradicts the code is a bug. - **Location**: `Docs/` organized in 6 sections (Introduction, Getting Started, Language Basics, Advanced Features, Standard Library, Best Practices). - **Structure**: Follow `Docs/wfl-documentation-policy.md` and 19 principles in `Docs/wfl-foundation.md`. - **Reference Documentation**: Two-tiered system for keywords diff --git a/Dev diary/2026-07-11-phase-0-concurrency-hardening.md b/Dev diary/2026-07-11-phase-0-concurrency-hardening.md new file mode 100644 index 00000000..22dfd4c0 --- /dev/null +++ b/Dev diary/2026-07-11-phase-0-concurrency-hardening.md @@ -0,0 +1,122 @@ +# Dev Diary — 2026-07-11: Phase 0 concurrency hardening + +## Context + +`Docs/development/concurrency-phase-plan.md` lays out a phased plan to give WFL a +real concurrency model without a risky interpreter rewrite. **Phase 0 ("Kill the +live DoS")** is the first, lowest-risk slice: no interpreter-core redesign, no new +language surface, plain `main loop` stays serial. It closes three concrete gaps +that a public WFL web server exposes today: + +1. **CPU-bound crypto stalls the whole process.** Password hashing (Argon2, + scrypt, bcrypt, PBKDF2 at 600k rounds) is *deliberately* slow and ran inline on + the single interpreter thread, so one login-like call froze every other request + — a cooperative-scheduling DoS. +2. **The request queue was unbounded.** The transport→interpreter channel was an + `mpsc::unbounded_channel`, so a flood of accepted requests could grow memory + without bound. +3. **The docs claimed parallelism we don't have.** Several pages said handlers run + "concurrently" / "don't block others" while they actually run one at a time. + +This entry covers all three sub-PRs (0a/0b/0c) that landed together on +`claude/phase-0-implementation-rpbdjp`. + +## What changed + +### 0a — Docs honesty + `panic = "unwind"` gate + +- **Panic gate.** The runtime will rely on `catch_unwind` to contain a panicking + request handler so its siblings survive (Phase 1). That is unsound under + `panic = "abort"`. Enforced with `#[cfg(panic = "abort")] compile_error!(...)` + in `src/lib.rs` — evaluated with the crate's *actual* panic strategy, so it + fails a real abort build but never trips `cargo test` (Cargo force-unwinds test + harnesses). + - **Dead end worth recording:** the first attempt used a `build.rs` check on the + `CARGO_CFG_PANIC` env var. It does not work — build scripts run on the host + and always see `unwind`, even under `--config 'profile.release.panic="abort"'` + (verified empirically). A `#[test]` on `cfg!(panic = ...)` is likewise a + phantom control. The crate-level `compile_error!` is the only mechanism that + reflects the real target panic strategy. + - `Cargo.toml` `[profile.release]` now pins `panic = "unwind"` explicitly + (self-documenting; default was already unwind). + - `.github/workflows/ci.yml` `clippy-and-test` job gains an "Assert panic=abort + is rejected" step that forces abort via `--config` and **fails if the build + succeeds** — the failing-first control proving the gate is live. +- **Docs.** Rewrote overclaims to distinguish the concurrent *transport* (accept / + TLS) from *serial application handlers*, and preferred "concurrent" over + "parallel". Primary rewrite in `async-programming.md`; `web-servers.md` was + already honest and served as the model. + +### 0b — `spawn_blocking` for CPU-heavy crypto + +- New `src/stdlib/crypto_async.rs` with a name-keyed `route(name, args)` that hops + the 11 heavy crypto builtins onto Tokio's blocking pool. Chosen over a new + `Value::AsyncNativeFunction` variant to avoid rippling a new arm through every + exhaustive `Value` match for zero user-visible benefit. +- The interpreter's two async native-dispatch arms (`FunctionCall`, `ActionCall` + in `src/interpreter/mod.rs`) now check `route()` first and `.await` it, falling + back to the synchronous native otherwise. +- The heavy compute stayed in plain-data helpers in `src/stdlib/crypto.rs` + (`argon2_hash_str`, `*_verify_str`, `pbkdf2_hmac_sha256_str`, …), now + `pub(crate)`. Arguments are extracted into owned `String`/`u64`/`usize` on the + interpreter thread *before* the hop; only that plain data crosses into + `spawn_blocking`; the `Value` is rebuilt after the `.await`. +- **Interpreter core stays `!Send`.** No `Rc`/`RefCell`/`Value`/`Environment` ever + crosses a thread boundary (HARD RULE 9 not triggered). `zeroize` and the + `subtle` constant-time compare paths are untouched — no early-exit refactor. + +### 0c — Bounded request queue (OOM shed) + +- New `.wflcfg` key `web_server_request_queue_bound` (default `256`, zero + rejected) in `src/config.rs`. +- The transport→interpreter channel is now `mpsc::channel(bound)` instead of + `unbounded_channel`. The warp handler uses `try_send`: on `Full` it logs a + structured warning and returns a `503` (with `Retry-After`) via the new + `overloaded_response()` helper, without blocking the transport task or awaiting + the per-request oneshot; on `Closed` it keeps the existing rejection. +- WebSocket channels remain unbounded — out of scope, noted as a follow-up. + +## Tests + +- `src/stdlib/crypto_async.rs` unit tests (in-crate because `route` is + `pub(crate)`): exact routed-set map; a **deterministic off-thread proof** — on a + `current_thread` runtime a concurrently-spawned ticker only advances during the + crypto `.await` if the work was offloaded (independent of core count, no timing + thresholds); routed hash/verify round-trips (argon2, bcrypt); routed PBKDF2 is + byte-identical to the direct helper; argument errors still surface. +- Existing crypto suites (`crypto_kdf_test`, `password_hashing_test`, + `crypto_test` — 44 tests) pass unchanged through the new routed dispatch path. +- `src/interpreter/mod.rs` `queue_bound_tests`: `overloaded_response()` is a + well-formed 503 (Content-Length matches body); a full bounded channel sheds + deterministically via `try_send` → `Full`. +- `tests/web_queue_bound_test.rs`: `web_server_request_queue_bound` parsing — + default 256, valid override applied, zero and non-numeric rejected. +- Existing web-server tests (`http_request_runtime_test`, `respond_headers_test`, + `route_params_test`, `websocket_test` — 30 tests) pass with the bounded queue at + the default. + +## Docs + +- `Docs/04-advanced-features/async-programming.md`, `index.md`, + `Docs/01-introduction/key-features.md`, + `Docs/06-best-practices/performance-tips.md`, `Docs/Archive/README.md`, + `Docs/04-advanced-features/web-servers.md` — honesty pass (concurrent transport + vs serial handlers; concurrent ≠ parallel). +- `Docs/reference/configuration-reference.md` and `web-servers.md` — document + `web_server_request_queue_bound` and the 503 shed behavior. +- `Docs/development/concurrency-phase-plan.md` — flipped 0a/0b/0c to ✅. +- `CLAUDE.md` — added a **"Docs must be honest (validate docs)"** policy: docs + describe what actually ships (planned behavior marked as such), and every + user-visible change ships validated docs *and* a Dev Diary entry in the same + change. + +## Notes / Follow-ups + +- Phase 0 is cooperative only: CPU-bound *non-awaiting* WFL code still stalls the + process. Concurrent request handlers arrive in Phase 1 (`main loop concurrently:`). +- The crypto auto-call dispatch path (arity-0 natives) is intentionally not + routed; every heavy crypto builtin takes ≥ 1 argument. A future arity-0 heavy + builtin would also need routing. +- WebSocket outbound/event channels are still unbounded — a Phase 1 follow-up. +- `#![deny(clippy::await_holding_refcell_ref)]` already guards `src/lib.rs`, which + is the clippy backstop Phase 1 will lean on. diff --git a/Docs/development/concurrency-phase-plan.md b/Docs/development/concurrency-phase-plan.md index ae8dcc1b..ee0d8c02 100644 --- a/Docs/development/concurrency-phase-plan.md +++ b/Docs/development/concurrency-phase-plan.md @@ -43,9 +43,9 @@ HARD RULES: | Phase | PR | Title | Status | |-------|-----|--------|--------| -| 0 | 0a | Docs honesty + `panic=unwind` CI | ⬜ Not started | -| 0 | 0b | `spawn_blocking` for blocking crypto | ⬜ Not started | -| 0 | 0c | Bound accept/queue (OOM shed) | ⬜ Not started | +| 0 | 0a | Docs honesty + `panic=unwind` CI | ✅ Done | +| 0 | 0b | `spawn_blocking` for blocking crypto | ✅ Done | +| 0 | 0c | Bound accept/queue (OOM shed) | ✅ Done | | 1 | 1a | Runtime spike (bridge, no surface) | ⬜ Not started | | 1 | 1b | `main loop concurrently:` surface + ops defaults | ⬜ Not started | | 1 | 1c | Honesty docs for real concurrent model | ⬜ Not started | From a0f6d739db6dd68e69cac637136e5b814f5d1c87 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 10:11:36 +0000 Subject: [PATCH 4/8] fix: address CodeRabbit review on PR #607 (Phase 0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ci.yml: the panic=abort gate now asserts our specific compile_error message ('WFL requires panic = "unwind"') instead of accepting any non-zero build exit, so an unrelated build failure can't masquerade as the gate firing. - Cargo.toml: fix the stale "Enforced by build.rs" comment — the gate lives in src/lib.rs (`#[cfg(panic = "abort")] compile_error!`). - crypto_async.rs: wrap the transient password copies in Zeroizing so they are wiped when the blocking closure that owns them is dropped (verify keeps the non-secret stored hash as a plain String). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- .github/workflows/ci.yml | 11 +++++++++-- Cargo.toml | 3 ++- src/stdlib/crypto_async.rs | 26 +++++++++++++++----------- 3 files changed, 26 insertions(+), 14 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ad28e529..6c39dcfa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -57,8 +57,15 @@ jobs: # enforced and catch_unwind fault isolation would be a phantom control. - name: Assert panic=abort is rejected run: | - if cargo build --release --config 'profile.release.panic="abort"' --quiet 2>/dev/null; then - echo "::error::panic=abort build succeeded — the panic=unwind gate (src/lib.rs compile_error) is not enforced" + set +e + output=$(cargo build --release --config 'profile.release.panic="abort"' --quiet 2>&1) + status=$? + set -e + # The gate must fail the build *with our specific compile_error*, not + # merely error out for some unrelated reason. + if [ "$status" -eq 0 ] || ! grep -Fq 'WFL requires panic = "unwind"' <<<"$output"; then + printf '%s\n' "$output" + echo "::error::panic=abort was not rejected by the src/lib.rs compile_error gate" exit 1 fi echo "panic=abort correctly rejected by the compile_error gate" diff --git a/Cargo.toml b/Cargo.toml index bd42f1e4..071c5c44 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -98,5 +98,6 @@ harness = false [profile.release] debug = true # Concurrency Phase 0 (PR-0a): pin unwinding so the runtime's catch_unwind-based -# request-handler fault isolation (Phase 1) stays sound. Enforced by build.rs. +# request-handler fault isolation (Phase 1) stays sound. Enforced by the +# `#[cfg(panic = "abort")] compile_error!` in src/lib.rs. panic = "unwind" diff --git a/src/stdlib/crypto_async.rs b/src/stdlib/crypto_async.rs index f4634953..6ec7e7d6 100644 --- a/src/stdlib/crypto_async.rs +++ b/src/stdlib/crypto_async.rs @@ -25,6 +25,7 @@ use crate::stdlib::crypto; use crate::stdlib::helpers::{check_arg_count, expect_text}; use futures_util::future::{FutureExt, LocalBoxFuture}; use std::sync::Arc; +use zeroize::Zeroizing; /// Route a CPU-heavy crypto builtin onto the blocking pool. /// @@ -73,7 +74,7 @@ fn hash_route( let extracted = extract_password(func, args); async move { let password = extracted?; - let hash = tokio::task::spawn_blocking(move || compute(func, &password)) + let hash = tokio::task::spawn_blocking(move || compute(func, password.as_str())) .await .map_err(|e| join_error(func, e))??; Ok(Value::Text(Arc::from(hash))) @@ -90,7 +91,7 @@ fn verify_route( let extracted = extract_password_and_stored(func, args); async move { let (password, stored) = extracted?; - let ok = tokio::task::spawn_blocking(move || compute(&password, &stored)) + let ok = tokio::task::spawn_blocking(move || compute(password.as_str(), &stored)) .await .map_err(|e| join_error(func, e))?; Ok(Value::Bool(ok)) @@ -104,7 +105,7 @@ fn pbkdf2_hmac_route(args: &[Value]) -> LocalBoxFuture<'static, Result LocalBoxFuture<'static, Result LocalBoxFuture<'static, Result Result { +/// Validate arity and pull out an owned password on the interpreter thread. +/// Kept as a plain (non-async) fn so extraction errors surface before any thread +/// hop. The password copy is wrapped in `Zeroizing` so it is wiped when the +/// blocking closure that owns it is dropped. +fn extract_password(func: &str, args: &[Value]) -> Result, RuntimeError> { check_arg_count(func, args, 1)?; - Ok(expect_text(&args[0])?.to_string()) + Ok(Zeroizing::new(expect_text(&args[0])?.to_string())) } +/// Like [`extract_password`] but also returns the stored hash. Only the password +/// is secret, so only it is zeroized; the stored hash is left as a plain `String`. fn extract_password_and_stored( func: &str, args: &[Value], -) -> Result<(String, String), RuntimeError> { +) -> Result<(Zeroizing, String), RuntimeError> { check_arg_count(func, args, 2)?; - let password = expect_text(&args[0])?.to_string(); + let password = Zeroizing::new(expect_text(&args[0])?.to_string()); let stored = expect_text(&args[1])?.to_string(); Ok((password, stored)) } From 9ad6775bdf8e753063d348c60187c497de2966b6 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 10:31:27 +0000 Subject: [PATCH 5/8] refactor: body admission + relocate tests (CodeRabbit review, PR #607) Addresses the remaining CodeRabbit findings on PR #607. Body admission (Stability): the 503 shed previously ran only after warp had already buffered the request body, so N concurrent transport tasks could each allocate up to web_server_max_body_size first. Add a per-listener in-flight Semaphore (sized to web_server_request_queue_bound) and acquire a permit in a filter *before* warp::body::bytes(); on saturation reject with a new `Overloaded` rejection that `handle_overloaded` maps to 503, so the body is never buffered. The permit is held for the request's lifetime and released on drop. The existing try_send queue check is retained as a second gate. Test location (repo convention): move the inline `#[cfg(test)]` modules out of src/. `crypto_async::route` and `interpreter::overloaded_response` are now `pub`, and the tests live in tests/crypto_async_test.rs and tests/web_queue_bound_test.rs. The crypto round-trips now verify through the public route (hash then verify) instead of crate-internal helpers; a new semaphore test covers the admission gate's deterministic shed. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- src/interpreter/mod.rs | 111 +++++++++++----------- src/stdlib/crypto_async.rs | 160 +------------------------------ tests/crypto_async_test.rs | 171 ++++++++++++++++++++++++++++++++++ tests/web_queue_bound_test.rs | 96 +++++++++++++++++-- 4 files changed, 313 insertions(+), 225 deletions(-) create mode 100644 tests/crypto_async_test.rs diff --git a/src/interpreter/mod.rs b/src/interpreter/mod.rs index 3fae944a..30eeeb7b 100644 --- a/src/interpreter/mod.rs +++ b/src/interpreter/mod.rs @@ -125,9 +125,9 @@ pub struct WflWebServer { } /// Build the 503 response returned when the transport→interpreter request queue -/// is full (Phase 0, PR-0c). A free function so the shed path can be unit-tested +/// is full (Phase 0, PR-0c). A free function so the shed path can be tested /// without standing up a live server. -pub(crate) fn overloaded_response() -> warp::http::Response> { +pub fn overloaded_response() -> warp::http::Response> { let body = b"Service Unavailable: the server is overloaded, please retry later\n".to_vec(); let content_length = body.len(); warp::http::Response::builder() @@ -347,6 +347,27 @@ pub struct ServerError(String); impl warp::reject::Reject for ServerError {} +/// Rejection raised when the server is at its in-flight request capacity, so a +/// request is shed *before* its body is buffered (Phase 0, PR-0c). Mapped to a +/// 503 by [`handle_overloaded`]. +#[derive(Debug)] +struct Overloaded; + +impl warp::reject::Reject for Overloaded {} + +/// Warp recover handler: turn an [`Overloaded`] rejection into a 503, and +/// re-raise every other rejection so warp's default handling still applies +/// (e.g. oversized-body `ServerError`s are unaffected). +async fn handle_overloaded( + err: warp::Rejection, +) -> Result>, warp::Rejection> { + if err.find::().is_some() { + Ok(overloaded_response()) + } else { + Err(err) + } +} + /// Strips the port from an HTTP Host header value, preserving IPv6 brackets /// (`example.com:8080` -> `example.com`, `[::1]:8080` -> `[::1]`). fn strip_host_port(host: &str) -> &str { @@ -5403,6 +5424,14 @@ impl Interpreter { mpsc::channel::(queue_bound); let request_receiver = Arc::new(tokio::sync::Mutex::new(request_receiver)); + // Cap concurrently in-flight requests so a flood cannot allocate + // an unbounded number of request bodies before the queue check + // (Phase 0, PR-0c). A permit is acquired *before* `body::bytes()` + // buffers the payload and released when the request completes; + // when none is available the request is shed with 503 before any + // body is read. + let inflight = Arc::new(tokio::sync::Semaphore::new(queue_bound)); + // Create warp routes that handle all HTTP methods and paths. // Body size: reject oversized Content-Length *before* buffering // (when the header is present), then re-check after @@ -5435,6 +5464,21 @@ impl Interpreter { }, ), ) + // Admission control: acquire an in-flight permit *before* the + // body is buffered. If the server is saturated, shed with a + // 503 (via the `Overloaded` rejection + `handle_overloaded`) + // so we never allocate a body for a request we can't serve. + .and({ + let inflight = inflight.clone(); + warp::any().and_then(move || { + let inflight = inflight.clone(); + async move { + inflight + .try_acquire_owned() + .map_err(|_| warp::reject::custom(Overloaded)) + } + }) + }) .and(warp::body::bytes()) .and(warp::addr::remote()) .and_then( @@ -5443,10 +5487,14 @@ impl Interpreter { query: String, headers: warp::http::HeaderMap, (), + permit: tokio::sync::OwnedSemaphorePermit, body: bytes::Bytes, remote_addr: Option| { let sender = request_sender_clone.clone(); async move { + // Hold the admission permit until this request is + // fully processed, then release it on drop. + let _permit = permit; // Safety net when Content-Length was absent or lied: // still refuse after buffering so the limit holds. if body.len() > max_body_size { @@ -5557,7 +5605,8 @@ impl Interpreter { } } }, - ); + ) + .recover(handle_overloaded); // Parse the bind address from config let bind_addr: IpAddr = match self.config.web_server_bind_address.parse() { @@ -9989,59 +10038,3 @@ mod process_tests { ); } } - -/// Phase 0 (PR-0c): the bounded request queue sheds with a well-formed 503 when -/// the interpreter is saturated, instead of buffering work without bound. -#[cfg(test)] -mod queue_bound_tests { - use super::*; - - #[test] - fn overloaded_response_is_a_well_formed_503() { - let resp = overloaded_response(); - assert_eq!(resp.status(), warp::http::StatusCode::SERVICE_UNAVAILABLE); - assert_eq!( - resp.headers() - .get("Content-Type") - .and_then(|v| v.to_str().ok()), - Some("text/plain; charset=utf-8") - ); - // Content-Length matches the actual body byte count. - let declared: usize = resp - .headers() - .get("Content-Length") - .and_then(|v| v.to_str().ok()) - .and_then(|s| s.parse().ok()) - .expect("Content-Length header present and numeric"); - assert_eq!(declared, resp.body().len()); - assert!(!resp.body().is_empty()); - } - - /// The over-cap decision is deterministic: once a bounded channel is full, - /// `try_send` reports `Full` (which the warp handler maps to `overloaded_response`), - /// and never blocks or grows the queue. - #[tokio::test] - async fn full_queue_sheds_deterministically() { - let bound = 4usize; - let (tx, _rx) = mpsc::channel::(bound); - - // Fill to capacity — every send within the bound succeeds. - for i in 0..bound { - tx.try_send(i as u32) - .expect("send within capacity succeeds"); - } - assert_eq!(tx.max_capacity(), bound); - - // The next send over capacity sheds rather than blocking or growing. - match tx.try_send(999) { - Err(mpsc::error::TrySendError::Full(v)) => assert_eq!(v, 999), - other => panic!("expected Full over capacity, got {other:?}"), - } - - // The shed maps to a 503. - assert_eq!( - overloaded_response().status(), - warp::http::StatusCode::SERVICE_UNAVAILABLE - ); - } -} diff --git a/src/stdlib/crypto_async.rs b/src/stdlib/crypto_async.rs index 6ec7e7d6..39f94856 100644 --- a/src/stdlib/crypto_async.rs +++ b/src/stdlib/crypto_async.rs @@ -40,7 +40,10 @@ use zeroize::Zeroizing; /// /// `constant_time_equals` and `secure_random_bytes` are intentionally *not* /// routed: they are fast and routing them would only add a scheduling hop. -pub(crate) fn route( +/// +/// Public so the interpreter dispatch (and the `tests/crypto_async_test.rs` +/// integration tests) can drive it; not part of the stable language surface. +pub fn route( name: &str, args: &[Value], ) -> Option>> { @@ -150,158 +153,3 @@ fn extract_password_and_stored( fn join_error(func: &str, e: tokio::task::JoinError) -> RuntimeError { RuntimeError::new(format!("{func}: crypto task failed: {e}"), 0, 0) } - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::atomic::{AtomicUsize, Ordering}; - - fn text(s: &str) -> Value { - Value::Text(Arc::from(s)) - } - - /// Only the deliberately-slow crypto builtins route; everything else falls - /// through to the synchronous native. Guards against silent scope creep in - /// the routed set. - #[test] - fn route_map_covers_exactly_the_heavy_crypto_builtins() { - let one = [text("x")]; - let two = [text("x"), text("y")]; - let four = [ - text("x"), - text("y"), - Value::Number(1.0), - Value::Number(16.0), - ]; - - for name in [ - "argon2_hash", - "scrypt_hash", - "bcrypt_hash", - "pbkdf2_hash", - "hash_password", - ] { - assert!(route(name, &one).is_some(), "{name} should route"); - } - for name in [ - "argon2_verify", - "scrypt_verify", - "bcrypt_verify", - "pbkdf2_verify", - "verify_password", - ] { - assert!(route(name, &two).is_some(), "{name} should route"); - } - assert!(route("pbkdf2_hmac_sha256", &four).is_some()); - - // Fast / non-crypto builtins must NOT route. - for name in [ - "constant_time_equals", - "secure_random_bytes", - "sha256", - "wflhash256", - "print", - "length", - ] { - assert!(route(name, &two).is_none(), "{name} must not route"); - } - } - - /// The core Phase 0 property (plan U4): the heavy work runs off the - /// interpreter thread. On a single-threaded runtime, a concurrently-spawned - /// ticker can only make progress *during* the crypto `.await` if that work - /// was offloaded via `spawn_blocking`. Had the hash run inline, it would - /// have monopolized the one thread and the ticker would still read zero when - /// the await returns. This is deterministic and independent of core count. - #[tokio::test(flavor = "current_thread")] - async fn routed_crypto_frees_the_interpreter_thread() { - let counter = std::sync::Arc::new(AtomicUsize::new(0)); - let ticker_counter = counter.clone(); - let ticker = tokio::spawn(async move { - for _ in 0..5 { - tokio::task::yield_now().await; - ticker_counter.fetch_add(1, Ordering::SeqCst); - } - }); - - let args = [text("a-password-to-hash")]; - let out = route("argon2_hash", &args) - .expect("argon2_hash routes") - .await - .expect("hash succeeds"); - - // Read the ticker's progress the instant the await returns, before we - // yield again — so a passing result can only mean the runtime polled the - // ticker while the blocking pool computed the hash. - let progressed = counter.load(Ordering::SeqCst); - ticker.await.expect("ticker joins"); - - assert!( - progressed > 0, - "ticker never advanced during the crypto await — work was not offloaded" - ); - assert!(matches!(out, Value::Text(_))); - } - - /// The routed hash still produces a valid, verifiable credential. - #[tokio::test] - async fn routed_argon2_hash_roundtrips() { - let out = route("argon2_hash", &[text("correct horse")]) - .unwrap() - .await - .unwrap(); - let hash = match out { - Value::Text(h) => h, - other => panic!("expected text, got {other:?}"), - }; - assert!(crypto::argon2_verify_str("correct horse", &hash)); - assert!(!crypto::argon2_verify_str("wrong password", &hash)); - } - - /// bcrypt uses a distinct MCF (`$2b$`) format, exercising a different helper. - #[tokio::test] - async fn routed_bcrypt_hash_roundtrips() { - let out = route("bcrypt_hash", &[text("hunter2")]) - .unwrap() - .await - .unwrap(); - let hash = match out { - Value::Text(h) => h, - other => panic!("expected text, got {other:?}"), - }; - assert!(hash.starts_with("$2")); - assert!(crypto::bcrypt_verify_str("hunter2", &hash)); - assert!(!crypto::bcrypt_verify_str("nope", &hash)); - } - - /// The routed KDF output is byte-identical to the direct plain-data helper — - /// the thread hop changes nothing about the result (deterministic: no salt). - #[tokio::test] - async fn routed_pbkdf2_matches_direct() { - let args = [ - text("password"), - text("salt"), - Value::Number(4096.0), - Value::Number(32.0), - ]; - let out = route("pbkdf2_hmac_sha256", &args).unwrap().await.unwrap(); - let routed = match out { - Value::Text(h) => h.to_string(), - other => panic!("expected text, got {other:?}"), - }; - let direct = crypto::pbkdf2_hmac_sha256_str("password", "salt", 4096, 32).unwrap(); - assert_eq!(routed, direct); - } - - /// Argument errors still surface (as the routed future's error), so routing - /// does not mask misuse. - #[tokio::test] - async fn routed_argerror_is_reported() { - // argon2_hash expects exactly one argument. - let err = route("argon2_hash", &[text("a"), text("b")]) - .unwrap() - .await - .unwrap_err(); - assert!(err.message.contains("argon2_hash")); - } -} diff --git a/tests/crypto_async_test.rs b/tests/crypto_async_test.rs new file mode 100644 index 00000000..8ad48d53 --- /dev/null +++ b/tests/crypto_async_test.rs @@ -0,0 +1,171 @@ +//! Phase 0 (PR-0b) — off-thread crypto routing, exercised through the public +//! `crypto_async::route` seam. Lives under `tests/` per the repo convention; +//! round-trips go through routing (hash then verify) rather than reaching into +//! crate-internal helpers. + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use wfl::interpreter::value::Value; +use wfl::stdlib::crypto_async::route; + +fn text(s: &str) -> Value { + Value::Text(Arc::from(s)) +} + +fn as_text(v: Value) -> String { + match v { + Value::Text(t) => t.to_string(), + other => panic!("expected text, got {other:?}"), + } +} + +/// Route a two-argument verify builtin and unwrap its boolean result. +async fn verify(name: &str, password: &str, stored: &str) -> bool { + let out = route(name, &[text(password), text(stored)]) + .expect("verify routes") + .await + .expect("verify succeeds"); + match out { + Value::Bool(b) => b, + other => panic!("expected bool, got {other:?}"), + } +} + +/// Only the deliberately-slow crypto builtins route; everything else falls +/// through to the synchronous native. Guards against silent scope creep. +#[test] +fn route_map_covers_exactly_the_heavy_crypto_builtins() { + let one = [text("x")]; + let two = [text("x"), text("y")]; + let four = [ + text("x"), + text("y"), + Value::Number(1.0), + Value::Number(16.0), + ]; + + for name in [ + "argon2_hash", + "scrypt_hash", + "bcrypt_hash", + "pbkdf2_hash", + "hash_password", + ] { + assert!(route(name, &one).is_some(), "{name} should route"); + } + for name in [ + "argon2_verify", + "scrypt_verify", + "bcrypt_verify", + "pbkdf2_verify", + "verify_password", + ] { + assert!(route(name, &two).is_some(), "{name} should route"); + } + assert!(route("pbkdf2_hmac_sha256", &four).is_some()); + + // Fast / non-crypto builtins must NOT route. + for name in [ + "constant_time_equals", + "secure_random_bytes", + "sha256", + "wflhash256", + "print", + "length", + ] { + assert!(route(name, &two).is_none(), "{name} must not route"); + } +} + +/// The core Phase 0 property (plan U4): the heavy work runs off the interpreter +/// thread. On a single-threaded runtime, a concurrently-spawned ticker can only +/// make progress *during* the crypto `.await` if that work was offloaded via +/// `spawn_blocking`. Had the hash run inline, it would have monopolized the one +/// thread and the ticker would still read zero when the await returns. This is +/// deterministic and independent of core count. +#[tokio::test(flavor = "current_thread")] +async fn routed_crypto_frees_the_interpreter_thread() { + let counter = Arc::new(AtomicUsize::new(0)); + let ticker_counter = counter.clone(); + let ticker = tokio::spawn(async move { + for _ in 0..5 { + tokio::task::yield_now().await; + ticker_counter.fetch_add(1, Ordering::SeqCst); + } + }); + + let out = route("argon2_hash", &[text("a-password-to-hash")]) + .expect("argon2_hash routes") + .await + .expect("hash succeeds"); + + // Read the ticker's progress the instant the await returns, before we yield + // again — so a passing result can only mean the runtime polled the ticker + // while the blocking pool computed the hash. + let progressed = counter.load(Ordering::SeqCst); + ticker.await.expect("ticker joins"); + + assert!( + progressed > 0, + "ticker never advanced during the crypto await — work was not offloaded" + ); + assert!(matches!(out, Value::Text(_))); +} + +/// The routed hash still produces a valid, verifiable credential, and +/// `verify_password` auto-detects the algorithm. +#[tokio::test] +async fn routed_argon2_hash_roundtrips() { + let hash = as_text( + route("argon2_hash", &[text("correct horse")]) + .unwrap() + .await + .unwrap(), + ); + assert!(hash.starts_with("$argon2")); + assert!(verify("argon2_verify", "correct horse", &hash).await); + assert!(!verify("argon2_verify", "wrong password", &hash).await); + assert!(verify("verify_password", "correct horse", &hash).await); +} + +/// bcrypt uses a distinct MCF (`$2b$`) format, exercising a different path. +#[tokio::test] +async fn routed_bcrypt_hash_roundtrips() { + let hash = as_text( + route("bcrypt_hash", &[text("hunter2")]) + .unwrap() + .await + .unwrap(), + ); + assert!(hash.starts_with("$2")); + assert!(verify("bcrypt_verify", "hunter2", &hash).await); + assert!(!verify("bcrypt_verify", "nope", &hash).await); +} + +/// The routed KDF is deterministic (no salt) and correctly shaped: two routed +/// calls with identical inputs produce identical 32-byte (64-hex) output. +#[tokio::test] +async fn routed_pbkdf2_is_deterministic() { + let args = [ + text("password"), + text("salt"), + Value::Number(4096.0), + Value::Number(32.0), + ]; + let a = as_text(route("pbkdf2_hmac_sha256", &args).unwrap().await.unwrap()); + let b = as_text(route("pbkdf2_hmac_sha256", &args).unwrap().await.unwrap()); + assert_eq!(a, b); + assert_eq!(a.len(), 64, "32 bytes rendered as lowercase hex"); +} + +/// Argument errors still surface (as the routed future's error), so routing +/// does not mask misuse. +#[tokio::test] +async fn routed_argerror_is_reported() { + // argon2_hash expects exactly one argument. + let err = route("argon2_hash", &[text("a"), text("b")]) + .unwrap() + .await + .unwrap_err(); + assert!(err.message.contains("argon2_hash")); +} diff --git a/tests/web_queue_bound_test.rs b/tests/web_queue_bound_test.rs index f0c0ee05..21d9b47c 100644 --- a/tests/web_queue_bound_test.rs +++ b/tests/web_queue_bound_test.rs @@ -1,17 +1,18 @@ -//! Phase 0 (PR-0c) — bounded request-queue configuration. +//! Phase 0 (PR-0c) — bounded request queue + in-flight admission. //! -//! The transport→interpreter request queue is bounded so a flood of accepted -//! requests sheds with 503 rather than growing memory without bound. These -//! tests exercise the `web_server_request_queue_bound` config key through the -//! public `load_config` API: a valid override applies, and invalid values -//! (zero, non-numeric) are rejected while keeping the safe default. -//! -//! The runtime shed decision itself (`try_send` full → `overloaded_response` -//! 503) is covered by the in-crate `queue_bound_tests` unit tests in -//! `src/interpreter/mod.rs`, since `overloaded_response` is crate-internal. +//! The transport→interpreter request queue is bounded, and requests are admitted +//! against an in-flight semaphore *before* their bodies are buffered, so a flood +//! sheds with 503 rather than growing memory without bound. These tests cover: +//! - the `web_server_request_queue_bound` config key (default, override, +//! zero/garbage rejection) via the public `load_config` API, and +//! - the runtime shed decision: `overloaded_response()` is a well-formed 503, +//! and both a full bounded channel and a drained semaphore reject +//! deterministically (the two admission gates in the warp handler). use std::fs; +use tokio::sync::{Semaphore, mpsc}; use wfl::config::load_config; +use wfl::interpreter::overloaded_response; const DEFAULT_BOUND: usize = 256; @@ -54,3 +55,78 @@ fn large_queue_bound_is_applied() { let cfg = load_with_cfg("web_server_request_queue_bound = 4096\n"); assert_eq!(cfg.web_server_request_queue_bound, 4096); } + +// --- runtime shed decision ------------------------------------------------- + +#[test] +fn overloaded_response_is_a_well_formed_503() { + let resp = overloaded_response(); + assert_eq!(resp.status(), warp::http::StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + resp.headers() + .get("Content-Type") + .and_then(|v| v.to_str().ok()), + Some("text/plain; charset=utf-8") + ); + // Content-Length matches the actual body byte count. + let declared: usize = resp + .headers() + .get("Content-Length") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse().ok()) + .expect("Content-Length header present and numeric"); + assert_eq!(declared, resp.body().len()); + assert!(!resp.body().is_empty()); +} + +/// Queue gate: once the bounded channel is full, `try_send` reports `Full` +/// (which the warp handler maps to a 503) and never blocks or grows the queue. +#[tokio::test] +async fn full_queue_sheds_deterministically() { + let bound = 4usize; + let (tx, _rx) = mpsc::channel::(bound); + + for i in 0..bound { + tx.try_send(i as u32) + .expect("send within capacity succeeds"); + } + assert_eq!(tx.max_capacity(), bound); + + match tx.try_send(999) { + Err(mpsc::error::TrySendError::Full(v)) => assert_eq!(v, 999), + other => panic!("expected Full over capacity, got {other:?}"), + } + + assert_eq!( + overloaded_response().status(), + warp::http::StatusCode::SERVICE_UNAVAILABLE + ); +} + +/// Admission gate: once every in-flight permit is held, `try_acquire_owned` +/// fails, which the warp filter turns into an `Overloaded` rejection → 503, +/// shedding the request before its body is buffered. +#[test] +fn full_inflight_semaphore_sheds_deterministically() { + let bound = 4usize; + let sem = std::sync::Arc::new(Semaphore::new(bound)); + + // Hold every permit. + let mut permits: Vec<_> = (0..bound) + .map(|_| { + sem.clone() + .try_acquire_owned() + .expect("permit within capacity") + }) + .collect(); + assert_eq!(sem.available_permits(), 0); + + // The next admission attempt is refused deterministically. + assert!(sem.clone().try_acquire_owned().is_err()); + + // Releasing exactly one permit re-opens exactly one admission slot (the + // other three stay held in `permits`). + permits.pop(); + assert_eq!(sem.available_permits(), 1); + assert!(sem.try_acquire_owned().is_ok()); +} From 4322966ffe3b90f2f923a6af7adf1479cb489cef Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 11:21:19 +0000 Subject: [PATCH 6/8] fix: release admission permit at enqueue + std future type (review, PR #607) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review follow-ups. Availability (CodeRabbit): the in-flight admission permit was held across the untimed response wait, so a WFL handler that never calls `respond` would pin its permit forever — after `web_server_request_queue_bound` such stuck requests, the semaphore is exhausted and every new request is shed with 503, taking the server offline (my admission change had turned a memory leak into an availability failure). Release the permit right after `try_send` enqueues the request: it still bounds concurrent body buffering (bodies are buffered before enqueue), but a request awaiting a response holds neither a permit nor a body. A real per-request response timeout remains Phase 1 scope. Public API (Copilot): `crypto_async::route` returned `futures_util`'s `LocalBoxFuture`, leaking a third-party type into the public signature. Return a plain `std` `Pin>>` (type alias `RoutedFuture`) via `Box::pin`, dropping the `futures_util` import. Behavior is unchanged. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- src/interpreter/mod.rs | 23 ++++++++++++++++++---- src/stdlib/crypto_async.rs | 40 +++++++++++++++++++------------------- 2 files changed, 39 insertions(+), 24 deletions(-) diff --git a/src/interpreter/mod.rs b/src/interpreter/mod.rs index 3889bd37..f133b9fe 100644 --- a/src/interpreter/mod.rs +++ b/src/interpreter/mod.rs @@ -5487,9 +5487,15 @@ impl Interpreter { remote_addr: Option| { let sender = request_sender_clone.clone(); async move { - // Hold the admission permit until this request is - // fully processed, then release it on drop. - let _permit = permit; + // `permit` (acquired before the body was buffered) + // is released right after the request is enqueued + // below — see the `drop(permit)` in the `try_send` + // Ok arm. Holding it across the untimed response + // wait would let a handler that never calls + // `respond` pin the in-flight cap and take the + // server offline; an actual per-request response + // timeout is Phase 1 work. On the early returns + // below, `permit` is dropped when the future ends. // Safety net when Content-Length was absent or lied: // still refuse after buffering so the limit holds. if body.len() > max_body_size { @@ -5545,7 +5551,16 @@ impl Interpreter { // than buffering unbounded work. `try_send` never // blocks the transport task. match sender.try_send(wfl_request) { - Ok(()) => {} + Ok(()) => { + // Enqueued: the body is now owned by the + // bounded queue (then the serial + // interpreter), so release the admission + // permit before the response wait. Concurrent + // body buffering stays bounded without pinning + // a permit to a possibly-never-answered + // request. + drop(permit); + } Err(mpsc::error::TrySendError::Full(shed)) => { log::warn!( "web server request queue full (capacity {}); shedding {} {} from {} with 503", diff --git a/src/stdlib/crypto_async.rs b/src/stdlib/crypto_async.rs index 39f94856..75f16d46 100644 --- a/src/stdlib/crypto_async.rs +++ b/src/stdlib/crypto_async.rs @@ -15,18 +15,24 @@ //! on the interpreter thread after the `.await`. No `Rc`/`RefCell`/`Value`/ //! `Environment` ever crosses threads. //! -//! The returned future is a `!Send` [`LocalBoxFuture`]; it is awaited from the -//! interpreter future, which runs via `block_on` (never `tokio::spawn`), so -//! awaiting a `Send` `JoinHandle` from inside it is sound. +//! The returned future is a boxed `Pin>` ([`RoutedFuture`]); it +//! is awaited from the interpreter future, which runs via `block_on` (never +//! `tokio::spawn`), so awaiting a `Send` `JoinHandle` from inside it is sound. use crate::interpreter::error::RuntimeError; use crate::interpreter::value::Value; use crate::stdlib::crypto; use crate::stdlib::helpers::{check_arg_count, expect_text}; -use futures_util::future::{FutureExt, LocalBoxFuture}; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use zeroize::Zeroizing; +/// Boxed future produced by [`route`]. A plain `std` type (rather than a +/// `futures_util` alias) so the public signature doesn't leak a third-party type +/// into the crate's API. It is awaited only on the interpreter thread. +type RoutedFuture = Pin>>>; + /// Route a CPU-heavy crypto builtin onto the blocking pool. /// /// Returns `None` for any name that is not a heavy crypto builtin — the caller @@ -43,10 +49,7 @@ use zeroize::Zeroizing; /// /// Public so the interpreter dispatch (and the `tests/crypto_async_test.rs` /// integration tests) can drive it; not part of the stable language surface. -pub fn route( - name: &str, - args: &[Value], -) -> Option>> { +pub fn route(name: &str, args: &[Value]) -> Option { let fut = match name { "argon2_hash" => hash_route("argon2_hash", args, crypto::argon2_hash_str), "scrypt_hash" => hash_route("scrypt_hash", args, crypto::scrypt_hash_str), @@ -71,18 +74,17 @@ fn hash_route( func: &'static str, args: &[Value], compute: fn(&str, &str) -> Result, -) -> LocalBoxFuture<'static, Result> { +) -> RoutedFuture { // Extract on the interpreter thread; only the owned `String` crosses the // boundary below. let extracted = extract_password(func, args); - async move { + Box::pin(async move { let password = extracted?; let hash = tokio::task::spawn_blocking(move || compute(func, password.as_str())) .await .map_err(|e| join_error(func, e))??; Ok(Value::Text(Arc::from(hash))) - } - .boxed_local() + }) } /// Two-argument verify builtins: ` of "password" and stored` → boolean. @@ -90,21 +92,20 @@ fn verify_route( func: &'static str, args: &[Value], compute: fn(&str, &str) -> bool, -) -> LocalBoxFuture<'static, Result> { +) -> RoutedFuture { let extracted = extract_password_and_stored(func, args); - async move { + Box::pin(async move { let (password, stored) = extracted?; let ok = tokio::task::spawn_blocking(move || compute(password.as_str(), &stored)) .await .map_err(|e| join_error(func, e))?; Ok(Value::Bool(ok)) - } - .boxed_local() + }) } /// Raw PBKDF2-HMAC-SHA256 KDF: `pbkdf2_hmac_sha256 of password and salt and /// iterations and length` → hex string. -fn pbkdf2_hmac_route(args: &[Value]) -> LocalBoxFuture<'static, Result> { +fn pbkdf2_hmac_route(args: &[Value]) -> RoutedFuture { const FUNC: &str = "pbkdf2_hmac_sha256"; let extracted = (|| { check_arg_count(FUNC, args, 4)?; @@ -114,7 +115,7 @@ fn pbkdf2_hmac_route(args: &[Value]) -> LocalBoxFuture<'static, Result((password, salt, iterations, length)) })(); - async move { + Box::pin(async move { let (password, salt, iterations, length) = extracted?; let hex = tokio::task::spawn_blocking(move || { crypto::pbkdf2_hmac_sha256_str(password.as_str(), &salt, iterations, length) @@ -122,8 +123,7 @@ fn pbkdf2_hmac_route(args: &[Value]) -> LocalBoxFuture<'static, Result Date: Sat, 11 Jul 2026 11:28:13 +0000 Subject: [PATCH 7/8] style: make RoutedFuture pub to match route's visibility (review, PR #607) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `route` is `pub` and returns `RoutedFuture`, so the alias should share that visibility. (The private alias already compiled — E0446 does not fire for a private alias of fully-public std types, confirmed by CI/tests — but a public alias for a public function's return type is clearer and removes any doubt.) Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- src/stdlib/crypto_async.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/stdlib/crypto_async.rs b/src/stdlib/crypto_async.rs index 75f16d46..6dfe0c53 100644 --- a/src/stdlib/crypto_async.rs +++ b/src/stdlib/crypto_async.rs @@ -30,8 +30,9 @@ use zeroize::Zeroizing; /// Boxed future produced by [`route`]. A plain `std` type (rather than a /// `futures_util` alias) so the public signature doesn't leak a third-party type -/// into the crate's API. It is awaited only on the interpreter thread. -type RoutedFuture = Pin>>>; +/// into the crate's API. `pub` to match the visibility of `route`, which returns +/// it. It is awaited only on the interpreter thread. +pub type RoutedFuture = Pin>>>; /// Route a CPU-heavy crypto builtin onto the blocking pool. /// From e6933020ad629d4fb77f43a7af74d874c8e12e65 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 11 Jul 2026 11:35:33 +0000 Subject: [PATCH 8/8] fix(crypto): checked usize conversion for PBKDF2 length (review, PR #607) `length` was cast `u64 as usize` at the call sites, which truncates on 32-bit targets (e.g. wasm32): a value above usize::MAX could wrap to a small number and slip past the MAX_PBKDF2_KEY_LENGTH check, deriving a wrong-length key instead of erroring. Thread the raw u64 into pbkdf2_hmac_sha256_str and convert once with a checked usize::try_from that returns a clear RuntimeError on overflow. No change on 64-bit; existing KDF vector tests pass. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BPoNkFqfnjC1AcfcQjazky --- src/stdlib/crypto.rs | 14 ++++++++++++-- src/stdlib/crypto_async.rs | 4 +++- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/src/stdlib/crypto.rs b/src/stdlib/crypto.rs index aaab2738..a965a98c 100644 --- a/src/stdlib/crypto.rs +++ b/src/stdlib/crypto.rs @@ -624,7 +624,7 @@ pub fn native_pbkdf2_hmac_sha256(args: Vec) -> Result Result { if password.len() > MAX_INPUT_SIZE || salt.len() > MAX_INPUT_SIZE { return Err(RuntimeError::new( @@ -665,6 +665,16 @@ pub(crate) fn pbkdf2_hmac_sha256_str( 0, )); } + // Convert to `usize` with a checked cast so a value that doesn't fit (e.g. + // greater than u32::MAX on a 32-bit target such as wasm32) is rejected + // cleanly rather than silently truncating past the bound check below. + let length = usize::try_from(length).map_err(|_| { + RuntimeError::new( + format!("pbkdf2_hmac_sha256: length exceeds maximum ({MAX_PBKDF2_KEY_LENGTH} bytes)"), + 0, + 0, + ) + })?; if length < 1 { return Err(RuntimeError::new( "pbkdf2_hmac_sha256: length must be at least 1 byte".to_string(), diff --git a/src/stdlib/crypto_async.rs b/src/stdlib/crypto_async.rs index 6dfe0c53..846a42a6 100644 --- a/src/stdlib/crypto_async.rs +++ b/src/stdlib/crypto_async.rs @@ -113,7 +113,9 @@ fn pbkdf2_hmac_route(args: &[Value]) -> RoutedFuture { let password = Zeroizing::new(expect_text(&args[0])?.to_string()); let salt = expect_text(&args[1])?.to_string(); let iterations = crypto::expect_count(FUNC, "iterations", &args[2])?; - let length = crypto::expect_count(FUNC, "length", &args[3])? as usize; + // Pass the raw u64 through; `pbkdf2_hmac_sha256_str` does a checked + // usize conversion so a value that doesn't fit can't truncate. + let length = crypto::expect_count(FUNC, "length", &args[3])?; Ok::<_, RuntimeError>((password, salt, iterations, length)) })(); Box::pin(async move {