# Valtron
An experimental async runtime built on iterators that do not specifically rely on the async future paradigm nor follow the current `async/wait` paradigm most know for a more elaborate iterator and trait syntax that both provide a unified set of principle and control.
## Troubleshooting
### Multi-threaded Executor: Tests Hanging or Not Receiving Values
**Symptom:** Integration tests using `WebSocketClient`, `DrivenStreamIterator`, or other multi-threaded valtron-based iterators hang indefinitely or return `Stream::Ignore` immediately without processing any values.
**Cause:** The `PoolGuard` returned by `initialize_pool()` must be kept alive for the duration of test execution. Dropping it (e.g., `let _ = initialize_pool(...)`) immediately shuts down all worker threads, preventing any task execution.
In multi-threaded mode (`feature = "multi"`), tasks run on worker threads managed by the pool. When `PoolGuard` is dropped, the pool shuts down and all worker threads exit. This means queued tasks never execute, so the concurrent queue never receives any values. The iterator then yields `Stream::Ignore` or loops waiting for values that will never arrive.
**Fix:** Keep `PoolGuard` alive for the test duration:
```rust
// WRONG - guard dropped immediately, worker threads shut down
let _ = foundation_core::valtron::initialize_pool(42, None);
// CORRECT - guard kept alive for test duration
let _pool_guard: PoolGuard = foundation_core::valtron::initialize_pool(42, None);
```
**Note:** An initial fix attempt added a `thread::sleep` wait loop in `run_until_stream_has_value` for multi-threaded mode, but this was unnecessary. The real issue was the dropped `PoolGuard` - once the guard is kept alive, worker threads remain active and tasks execute normally.
**Files affected:**
- `tests/backends/foundation_core/websocket/echo_tests.rs` - Test fix
- `tests/backends/foundation_core/websocket/subprotocol_tests.rs` - Test fix
- `tests/backends/foundation_core/websocket/reconnection_tests.rs` - Test fix
- `tests/backends/foundation_core/event_source/consumer_integration_tests.rs` - Test fix
- `backends/foundation_core/src/valtron/executors/drivers.rs` - Removed unnecessary multi-threaded wait code
---
### Multi-threaded Executor: Parallel Test Interference
**Symptom:** Integration tests using `initialize_pool()` exhibit intermittent hangs or timeouts when running tests in parallel. Different tests fail on each run, with symptoms showing worker threads starting but never completing, or WebSocket connections established but messages never received.
**Root Cause:** The valtron executor uses global static registries (`REGISTRY` and `BG_REGISTRY`) to store the `ThreadRegistry` and `BackgroundJobRegistry`. When tests run in parallel:
1. Test A starts, calls `initialize_pool()` → sets `REGISTRY = Arc_A`
2. Test B starts, calls `initialize_pool()` → **overwrites** `REGISTRY = Arc_B`
3. Test A's worker threads now reference the wrong registry
4. When Test A's `PoolGuard` drops, it shuts down workers that may be serving Test B
This race condition causes intermittent failures that are non-deterministic and vary between test runs.
**Fix:** Tests that use `initialize_pool()` must be serialized using the `#[serial]` attribute with a named lock group. All tests within the same file share the same lock name:
```rust
use serial_test::serial;
// All tests in this file use the same serial lock to prevent PoolGuard interference
#[test]
#[traced_test]
#[ntest::timeout(60000)]
#[serial(websocket_pool)] // Named lock groups tests by feature area
fn test_text_message_echo() {
let _pool_guard: PoolGuard = foundation_core::valtron::initialize_pool(42, None);
// ... test code
}
```
**Named lock groups by test file:**
- `echo_tests.rs` → `#[serial(websocket_pool)]`
- `subprotocol_tests.rs` → `#[serial(subprotocol_tests)]`
- `reconnection_tests.rs` → `#[serial(reconnection_tests)]`
- `http_redirect_integration.rs` → `#[serial(http_redirect)]`
- `consumer_integration_tests.rs` → `#[serial(sse_consumer_tests)]`
**Why not fix the global state?** The global registry design allows for a simple single-instance executor model. Adding per-test isolation would require significant architectural changes. The `#[serial]` attribute provides a pragmatic solution that:
- Keeps tests isolated with their own clean pool
- Allows parallel execution across different test groups (websocket vs http vs event_source)
- Maintains test independence without shared state
**Additional safeguards:**
- `ntest::timeout` attribute on all integration tests prevents indefinite hangs
- `PoolGuard::drop()` now clears global registries to prevent state leakage between sequential tests
**Files updated:**
- `backends/foundation_core/src/valtron/executors/threads.rs` - Added cleanup callback to PoolGuard
- `backends/foundation_core/src/valtron/executors/multi/mod.rs` - PoolGuard::with_cleanup() clears registries on drop
- All integration test files in `tests/backends/foundation_core/` - Added `#[serial(...)]` attributes
---
## Tasks
Tasks in Valtron are a custom iterator and in another sense can also become regular interators, they specifically provide a way to represent a asynchronous task as a series of values that can be iterated over and also provide a clear distinction when they are busy or when they are ready to give you a value.
```rust
use foundation_core::{
valtron::{FnReady, TaskIterator},
};
struct Counter(usize, usize);
impl Counter {
pub fn new(val: usize) -> Self {
Self(0, val)
}
}
impl TaskIterator for Counter {
/// Pending means how we wish to communicate that
/// the task is not yet ready to send its next `Done`
/// value
type Pending = ();
/// Done means the actual value we really care about
/// it can be returend as many times as possible.
type Ready = usize;
fn next(
&mut self,
) -> Option<crate::valtron::TaskStatus<Self::Ready, Self::Pending>> {
let current = self.0;
let next = current + 1;
if next == self.0 {
return None;
}
self.0 = next;
Some(crate::valtron::TaskStatus::Ready(next))
}
}
```
## Executors
The crate provides two main means of driving execution:
### Mulit-Threaded Executor
Provided in the [multi](./executors/multi) module which allows you start a multi-threaded executor that utilizes as much threads as you allow it where work is distributed across these threads but one unique behaviour of the executor is that they never steal work, once a job has being allocated to a thread executor that job is forever owned by it till completion hence removing concerns on runtime cost to move work across threads and memory boundaries.
#### Example 1
```rust
use core::time;
use std::thread;
use rand::RngCore;
use foundation_core::valtron::FnReady;
use foundation_core::valtron::multi::{block_on, get_pool};
fn main() {
let seed = rand::thread_rng().next_u64();
let handler_kill = thread::spawn(move || {
thread::sleep(time::Duration::from_millis(3000));
get_pool().kill();
});
let _guard = initialize_pool(seed, None);
block_on(seed, None, |pool| {
pool.spawn()
.with_task(Counter::new(5))
.with_resolver(Box::new(FnReady::new(|item, _executor| {
// We can hook into the value returned by the task
// for us to perform something with it or spawn some other task
// from that value.
println!("Received next: {}", item);
})))
.schedule()
.expect("should deliver task");
});
handler_kill.join().expect("should finish");
}
```
#### Example 2
```rust
use core::time;
use std::thread;
use rand::RngCore;
use foundation_core::valtron::FnReady;
use foundation_core::valtron::multi::{block_on, get_pool};
fn main() {
let seed = rand::thread_rng().next_u64();
let handler_kill = thread::spawn(move || {
thread::sleep(time::Duration::from_millis(3000));
get_pool().kill();
});
let _guard = initialize_pool(seed, None);
// spawn a task and get back a iterator that returns the state
// this means it return immediately with Pending or Done, so it never
// blocks but lets you spin things or control polling of iterator.
for status in pool.spawn().with_task(Counter::new(5)).iter() {
// do something with the status
}
handler_kill.join().expect("should finish");
}
```
#### Example 3
```rust
use core::time;
use std::thread;
use rand::RngCore;
use foundation_core::valtron::FnReady;
use foundation_core::valtron::multi::{block_on, get_pool};
fn main() {
let seed = rand::thread_rng().next_u64();
let handler_kill = thread::spawn(move || {
thread::sleep(time::Duration::from_millis(3000));
get_pool().kill();
});
let _guard = initialize_pool(seed, None);
// spawn a task but block the thread till we get a Done
// item, so it consumes the pending and blocks till
// a `Done` is received.
//
// Its iterable until all the the None value is received that
// closes the iterator.
for item in pool.spawn().with_task(Counter::new(5)).blocking_iter() {
// do something with the item
}
handler_kill.join().expect("should finish");
}
```
### Single-Threaded Executor
Valtron also provides a single-threaded module that provides a more finegrained and controllable executor useful in environments like WebAssembly or embedded systems. We get great control to determine when a executor actually processes all queued up tasks to completion.
A benefit of single-threaded executor is the freedom to not be forced to think in multi-threaded contexts and require use of Send-safe smart pointers and wrappers but work with normal smart pointers like RefCell and Rc.
#### SingleExecutorSingleton (wasm32/wasm64 only)
On wasm32/wasm64 targets (when `multi` feature is off), valtron provides `SingleExecutorSingleton` — a convenience wrapper that combines initialization + guard ownership:
```rust
use foundation_core::valtron::SingleExecutorSingleton;
// Initialize executor and get guard in one call
// Later calls return the same guard (zero-sized, cheap clone)
let guard = SingleExecutorSingleton::guard();
// Schedule and run
spawn()
.with_task(Counter::new(5, Rc::new(RefCell::new(Vec::new()))))
.with_resolver(Box::new(FnReady::new(|item, _| {
println!("Got: {:?}", item);
})))
.schedule()
.expect("should deliver task");
run_until_complete();
```
The existing free functions (`initialize_pool`, `spawn`, `run_until_complete`) continue to work independently — `SingleExecutorSingleton` is optional.
#### Free function API (always available)
```rust
use std::{cell::RefCell, rc::Rc};
use rand::RngCore;
use tracing_test::traced_test;
use foundation_core::valtron::{
single::{initialize_pool, run_until_complete, spawn},
FnReady, TaskIterator,
};
struct Counter(usize, Rc<RefCell<Vec<usize>>>);
impl Counter {
pub fn new(val: usize, list: Rc<RefCell<Vec<usize>>>) -> Self {
Self(val, list)
}
}
impl TaskIterator for Counter {
type Pending = ();
type Ready = usize;
fn next_status(
&mut self,
_ctx: crate::valtron::Ctx,
) -> Option<crate::valtron::TaskStatus<Self::Ready, Self::Pending>> {
let item_size = self.1.borrow().len();
if item_size == self.0 {
return None;
}
self.1.borrow_mut().push(item_size);
Some(crate::valtron::TaskStatus::Ready(self.1.borrow().len()))
}
}
```
#### Example 1
```rust
fn main() {
let seed = rand::thread_rng().next_u64();
let shared_list = Rc::new(RefCell::new(Vec::new()));
let counter = Counter::new(5, shared_list.clone());
// initialize executor with a predictable seed
// your task can get a random number generator that
// can provide predictable random numbers every single
// time if the seed provided is the same.
let _guard = initialize_pool(seed, None);
// spawn a task.
spawn()
.with_task(counter)
.with_resolver(Box::new(FnReady::new(|item, _| {
tracing::info!("Received next: {:?}", item)
})))
.schedule()
.expect("should deliver task");
// run tasks queued to completion
run_until_complete();
// check output is what we expect
let items = shared_list.borrow().clone(); // should be: vec![0, 1, 2, 3, 4]);
}
```
#### Example 2
```rust
fn main() {
let seed = rand::thread_rng().next_u64();
let shared_list = Rc::new(RefCell::new(Vec::new()));
let counter = Counter::new(5, shared_list.clone());
// initialize executor with a predictable seed
let _guard = initialize_pool(seed, None);
// spawn a task and get back a iterator that returns the state
// this means it return immediately with Pending or Done, so it never
// blocks but lets you spin things or control polling of iterator.
for status in spawn().with_task(counter).schedule_iter(std::time::Duration::from_nanos(50))
.expect("should deliver task")
{
// do something with status
}
// check output is what we expect
let items = shared_list.borrow().clone(); // should be: vec![0, 1, 2, 3, 4]);
}
```
#### Example 3
```rust
fn main() {
let seed = rand::thread_rng().next_u64();
let shared_list = Rc::new(RefCell::new(Vec::new()));
let counter = Counter::new(5, shared_list.clone());
// initialize executor with a predictable seed
let _guard = initialize_pool(seed, None);
// spawn a task but block the thread till we get a Done
// item, so it consumes the pending and blocks till
// a `Done` is received.
//
// Its iterable until all the the None value is received that
// closes the iterator.
for status in spawn().with_task(counter).schedule_iter(std::time::Duration::from_nanos(50))
.expect("should deliver task")
{
// do something with status
}
// check output is what we expect
let items = shared_list.borrow().clone(); // should be: vec![0, 1, 2, 3, 4]);
}
```
---
## StreamIterator — Send vs Non-Send Variants
Valtron provides two parallel implementations of `StreamIteratorExt` (the combinator methods like `map_done`, `map_pending`, `map_circuit`):
| `multi` | `D: Send + 'static, P: Send + 'static` | `extensions/streams/sendable.rs` | Required |
| `not(multi)` | `D: 'static, P: 'static` | `extensions/streams/non_sendable.rs` | Not required |
The blanket `StreamIterator` impl is cfg-gated to match: on `multi` it requires `Send` bounds, on non-`multi` (wasm32, single-threaded native) it only requires `'static`. This means `DrivenStreamIterator<FutureTask<F>>` — whose `Ready` is `Result<T, E>` (not `Send` when `T`/`E` aren't) and whose `Pending` is `()` — correctly implements `StreamIterator` on the non-`multi` path and gets access to `StreamIteratorExt` combinators.
### What this means for downstream code
You use the same API regardless of target:
```rust
use foundation_core::valtron::{from_future, execute, StreamIteratorExt};
let task = from_future(my_async_fn());
let stream = execute(task, None)?;
// Works on both multi (Send) and non-multi (non-Send) targets.
// The correct StreamIteratorExt implementation is selected at compile time.
let mapped = stream
.map_done(|result| result.map_err(MyError::from))
.map_pending(|_| ());
```
No `#[cfg(target_arch = "wasm32")]` or feature-gated imports needed in downstream code. The executor (`execute`, `from_future`) and the combinator trait (`StreamIteratorExt`) are always exported from `foundation_core::valtron` — the cfg selection happens entirely within valtron itself.
### Why two implementations
The `multi` executor runs tasks on background worker threads, so all task outputs (`Ready`, `Pending`) must be `Send` to cross thread boundaries. The single-threaded executor (used on wasm32 and native without `multi`) runs everything on the current thread, so no `Send` bound is needed. The `StreamIteratorExt` combinator types (`MapDone`, `MapPending`, etc.) store boxed closures — on the `multi` path those closures must be `Send`, on the non-`multi` path they don't need to be.
---
## ⚠️ CRITICAL: TaskIterator/Iterator Recursion Trap
**Location:** `task.rs:311-401`
**Problem:** The `TaskIterator` trait has a blanket implementation for any type that implements `Iterator<Item = TaskStatus<...>>`. This creates bidirectional coherence:
```rust
// Any Iterator<TaskStatus<...>> automatically implements TaskIterator
impl<I, D, P> TaskIterator for I
where
I: Iterator<Item = TaskStatus<D, P>>,
...
```
**The Trap:** Wrapper types that implement BOTH `Iterator` and `TaskIterator` will infinitely recurse on `next_status()`:
```rust
// WRONG - implements both, causes infinite recursion!
pub struct BadWrapper<I: TaskIterator> {
inner: I,
}
impl<I: TaskIterator> Iterator for BadWrapper<I> {
type Item = TaskStatus<I::Ready, I::Pending>;
fn next(&mut self) -> Option<Self::Item> {
self.inner.next_status() // Calls Iterator::next() due to blanket impl
// Which calls next_status() again... infinite loop!
}
}
impl<I: TaskIterator> TaskIterator for BadWrapper<I> {
// Blanket impl provides this, causing the recursion
}
```
**The Solution:** Wrapper types should:
1. Implement `TaskIterator` with explicit `next_status()`
2. Delegate to `inner.next_status()` (not `Iterator::next()`)
3. Use `Box<dyn TaskIterator>` to break the cycle via `as_mut().next_status()`
```rust
// CORRECT - explicit TaskIterator impl
pub struct GoodWrapper<I: TaskIterator> {
inner: I,
}
impl<I: TaskIterator> TaskIterator for GoodWrapper<I> {
type Ready = I::Ready;
type Pending = I::Pending;
fn next_status(&mut self, ctx: Ctx) -> Option<TaskStatus<Self::Ready, Self::Pending>> {
self.inner.next_status(ctx) // Direct delegation, no recursion
}
}
```
**Bottom Line:** Never implement both `Iterator<Item = TaskStatus<...>>` and `TaskIterator` on the same type. The blanket impl makes them mutually exclusive. Use `Box<dyn TaskIterator>` or explicit delegation to break the cycle.