Rust Valtron Iterator Creation

Rust Valtron Iterator Creation

Overview

This skill documents the correct patterns for implementing TaskIterator and StreamIterator in Rust within the Valtron framework. The central principle is that next() must return immediately after processing ONE item - never loop to consume multiple items before returning.

Valtron's execution model is progress-driven: the executor polls iterators frequently to observe intermediate states (Pending, Delayed, Init) and make scheduling decisions. Blocking loops in next() defeat this model by consuming items invisibly before the executor can observe progress.

When to Use

Use these patterns when:

• Implementing custom TaskIterator combinators in backends/foundation_core/src/valtron/task_iterators.rs
• Implementing custom StreamIterator combinators in backends/foundation_core/src/valtron/stream_iterators.rs
• Debugging iterator implementations that block or skip progress reporting

Prerequisites

• Understanding of Rust's Iterator trait
• Familiarity with Valtron's TaskStatus and Stream enums
• Knowledge of the executor model in backends/foundation_core/src/valtron/executors/

Usage Type

EDUCATIONAL - Study these patterns and apply them when implementing new iterator combinators. Reference the fixed implementations in the files: field for working examples.

Core Principle

The Single-Item Return Rule

next() must return immediately after processing ONE item.

// CORRECT: Process one item, return immediately
fn next(&mut self) -> Option<Self::Item> {
    let item = self.inner.next()?;
    // Process this single item
    match item {
        Stream::Next(v) => {
            // Transform or filter this ONE value
            Some(Stream::Next(transform(v)))
        }
        _ => Some(item), // Pass through immediately
    }
}

// INCORRECT: Loop consumes multiple items invisibly
fn next(&mut self) -> Option<Self::Item> {
    loop {
        let item = self.inner.next()?;
        match item {
            Stream::Next(v) if should_skip(v) => continue, // BLOCKS!
            _ => return Some(item),
        }
    }
}

Why This Matters

Blocking loops in next() defeat Valtron's progress-driven execution model:

Acceptable Loop Pattern

The ONLY acceptable loop is a single bounded pass through sources that returns immediately on any meaningful result:

// ACCEPTABLE: Single bounded pass, returns immediately on result
fn next(&mut self) -> Option<Self::Item> {
    let start_index = self.current_index;
    
    loop {
        // Bounded by number of sources - will complete this pass
        let source = &mut self.sources[self.current_index];
        match source.next() {
            Some(Stream::Next(v)) => {
                // Found meaningful result - return IMMEDIATELY
                self.current_index = (self.current_index + 1) % self.sources.len();
                return Some(Stream::Next(v));
            }
            Some(Stream::Pending(p)) => {
                // Report progress immediately
                self.current_index = (self.current_index + 1) % self.sources.len();
                return Some(Stream::Pending(p));
            }
            None => {
                // Source exhausted, continue to next in SAME pass
                self.current_index = (self.current_index + 1) % self.sources.len();
                if self.current_index == start_index {
                    return None; // Completed full pass, nothing found
                }
            }
            _ => {
                // Ignore/Delayed/Init - continue to next source in same pass
                self.current_index = (self.current_index + 1) % self.sources.len();
            }
        }
    }
}

Key characteristics of acceptable loops:

1. Bounded iteration - Loop runs at most N times (where N = number of sources)
2. Immediate return on progress - Returns Next, Pending, Delayed, Init immediately when encountered
3. No invisible consumption - Only Ignore or None results trigger continuation
4. Progress reported - Intermediate states escape the loop

Code Examples by Pattern

1. Filtering (return Ignore vs loop to find match)

BAD - Blocking Loop:

fn next(&mut self) -> Option<Self::Item> {
    loop {
        let item = self.inner.next()?;
        match item {
            Stream::Next(v) if self.predicate(v) => return Some(Stream::Next(v)),
            Stream::Next(_) => continue, // BLOCKS: consumes invisibly
            _ => return Some(item),
        }
    }
}

GOOD - Return Ignore:

fn next(&mut self) -> Option<Self::Item> {
    let item = self.inner.next()?;
    match item {
        Stream::Next(v) if self.predicate(v) => Some(Stream::Next(v)),
        Stream::Next(_) => Some(Stream::Ignore), // Yield control back
        _ => Some(item),
    }
}

Reference: TFilterReady in task_iterators.rs, SFilterState in stream_iterators.rs

2. Accumulating/Fold (yield Ignore while accumulating vs loop to collect all)

BAD - Blocking Loop:

fn next(&mut self) -> Option<Self::Item> {
    // BLOCKS: consumes ALL items before returning
    for item in &mut self.inner {
        if let Stream::Next(v) = item {
            self.acc = self.folder(self.acc.take().unwrap_or_default(), v);
        }
    }
    Some(Stream::Next(self.acc.take()))
}

GOOD - Yield Ignore while accumulating:

fn next(&mut self) -> Option<Self::Item> {
    if self.done {
        return None;
    }

    match self.inner.next() {
        Some(Stream::Next(v)) => {
            // Accumulate but return Ignore to signal "still working"
            if let Some(acc) = self.acc.take() {
                self.acc = Some(self.folder(acc, v));
            }
            Some(Stream::Ignore)
        }
        Some(Stream::Pending(p)) => Some(Stream::Pending(p)), // Pass through immediately
        Some(Stream::Delayed(d)) => Some(Stream::Delayed(d)),
        Some(Stream::Init) => Some(Stream::Init),
        None => {
            // Inner exhausted - yield final result
            self.done = true;
            self.acc.take().map(Stream::Next)
        }
    }
}

Reference: SFold (lines 2638-2678) in stream_iterators.rs, TFold (lines 2406-2445) in task_iterators.rs

3. Finding (track found state vs loop until found)

BAD - Blocking Loop:

fn next(&mut self) -> Option<Self::Item> {
    loop {
        let item = self.inner.next()?;
        if let Stream::Next(v) = item {
            if self.predicate(v) {
                return Some(Stream::Next(Some(v))); // Found!
            }
            // continue loops invisibly
        }
    }
}

GOOD - Track found state:

fn next(&mut self) -> Option<Self::Item> {
    if self.found {
        return None; // Already found, stop iteration
    }

    match self.inner.next()? {
        Stream::Next(d) => {
            if self.predicate(&d) {
                self.found = true;
                Some(Stream::Next(Some(d)))
            } else {
                Some(Stream::Ignore) // Yield control, executor will poll again
            }
        }
        Stream::Pending(p) => Some(Stream::Pending(p)), // Report progress
        Stream::Delayed(d) => Some(Stream::Delayed(d)),
        Stream::Init => Some(Stream::Init),
        Stream::Ignore => Some(Stream::Ignore),
    }
}

Reference: SFind (lines 2560-2592) in stream_iterators.rs, TFind (lines 2328-2362) in task_iterators.rs

4. Skip While / Take While (track state vs loop)

BAD - Blocking Loop:

fn next(&mut self) -> Option<Self::Item> {
    loop {
        let item = self.inner.next()?;
        if self.should_skip(&item) {
            continue; // BLOCKS: skips invisibly
        }
        return Some(item);
    }
}

GOOD - Track skipping state:

fn next(&mut self) -> Option<Self::Item> {
    let item = self.inner.next()?;

    if !self.done_skipping && self.predicate(&item) {
        // Still skipping, but yield Ignore (not blocking loop)
        return Some(Stream::Ignore);
    }

    // Done skipping - mark state and return item
    self.done_skipping = true;
    Some(item)
}

Reference: SSkipWhileState (lines 2456-2483) in stream_iterators.rs, TSkipWhileState (lines 2225-2252) in task_iterators.rs

5. Skip/Take with Count (decrement counter vs loop)

BAD - Blocking Loop:

fn next(&mut self) -> Option<Self::Item> {
    while self.to_skip > 0 {
        self.inner.next()?; // BLOCKS: consumes invisibly
        self.to_skip -= 1;
    }
    self.inner.next()
}

GOOD - Decrement counter per call:

fn next(&mut self) -> Option<Self::Item> {
    let item = self.inner.next()?;

    if self.to_skip > 0 && self.predicate(&item) {
        self.to_skip -= 1;
        return Some(Stream::Ignore); // Yield control
    }

    Some(item)
}

Reference: SSkipState (lines 2513-2540) in stream_iterators.rs, TSkipState (lines 2275-2302) in task_iterators.rs

6. All/Any (track result state, yield final at end vs loop to check all)

BAD - Blocking Loop for All:

fn next(&mut self) -> Option<Self::Item> {
    // BLOCKS: checks ALL items before returning
    for item in &mut self.inner {
        if let Stream::Next(v) = item {
            if !self.predicate(v) {
                return Some(Stream::Next(false));
            }
        }
    }
    Some(Stream::Next(true))
}

GOOD - Track state, yield Ignore during check:

// ALL - check if all items satisfy predicate
fn next(&mut self) -> Option<Self::Item> {
    if self.done {
        return None;
    }

    // Short-circuit: already found a failure
    if !self.all_true {
        return None;
    }

    match self.inner.next() {
        Some(Stream::Next(v)) => {
            if !self.predicate(v) {
                // Found failure - return immediately
                self.all_true = false;
                self.done = true;
                return Some(Stream::Next(false));
            }
            // Still good, return Ignore (still accumulating)
            Some(Stream::Ignore)
        }
        Some(Stream::Pending(p)) => Some(Stream::Pending(p)),
        Some(Stream::Delayed(d)) => Some(Stream::Delayed(d)),
        Some(Stream::Init) => Some(Stream::Init),
        None => {
            // Exhausted without failure - all true
            self.done = true;
            Some(Stream::Next(true))
        }
    }
}

GOOD - Any (short-circuit on first true):

fn next(&mut self) -> Option<Self::Item> {
    if self.done {
        return None;
    }

    // Short-circuit: already found a match
    if self.any_true {
        return None;
    }

    match self.inner.next() {
        Some(Stream::Next(v)) => {
            if self.predicate(v) {
                // Found match - return immediately
                self.any_true = true;
                self.done = true;
                Some(Stream::Next(true))
            } else {
                Some(Stream::Ignore)
            }
        }
        Some(Stream::Pending(p)) => Some(Stream::Pending(p)),
        Some(Stream::Delayed(d)) => Some(Stream::Delayed(d)),
        None => {
            // Exhausted without match
            self.done = true;
            Some(Stream::Next(false))
        }
    }
}

Reference: SAll (lines 2688-2730), SAny (lines 2734-2776) in stream_iterators.rs; TAll (lines 2455-2498), TAny (lines 2502-2545) in task_iterators.rs

7. FlatMap (the acceptable nested iterator exception)

FlatMap drains inner iterator before polling outer - this is the acceptable nested iterator pattern.

fn next(&mut self) -> Option<Self::Item> {
    // First drain current inner iterator
    if let Some(ref mut inner) = self.current_inner {
        if let Some(item) = inner.next() {
            return Some(item); // Return inner items one at a time
        }
        // Inner exhausted
        self.current_inner = None;
    }

    // Poll outer for next item
    match self.outer.next() {
        Some(item) => match item {
            Stream::Next(inner) => {
                // Create new inner iterator
                self.current_inner = Some(self.mapper(inner));
                // Return Ignore to signal "setting up next inner"
                Some(Stream::Ignore)
            }
            Stream::Pending(p) => Some(Stream::Pending(p)),
            Stream::Delayed(d) => Some(Stream::Delayed(d)),
            Stream::Init => Some(Stream::Init),
            Stream::Ignore => Some(Stream::Ignore),
        },
        None => None,
    }
}

Why this is acceptable:

• Inner iterator is itself a StreamIterator that reports Pending/Delayed
• Each inner.next() call returns immediately (doesn't block loop)
• Outer progress (Pending/Delayed) passes through immediately
• The nesting is explicit in the type system

Reference: MapIter (lines 2117-2156), MapIterDone (lines 2064-2109), MapIterPending (lines 2013-2056) in stream_iterators.rs

8. Multi-Source Round-Robin (single pass, sort indices descending before swap_remove)

fn next(&mut self) -> Option<Self::Item> {
    if self.done {
        return None;
    }

    let start_index = self.current_index;
    let mut exhausted_indices = Vec::new();

    // Single bounded pass through all sources
    loop {
        let idx = self.current_index;
        let source = &mut self.sources[idx];

        match source.next() {
            Some(Stream::Next(v)) => {
                // Found value - return immediately
                self.collected.push(v);
                self.current_index = (self.current_index + 1) % self.sources.len();
                return Some(Stream::Pending(self.collected.len()));
            }
            Some(Stream::Pending(p)) => {
                // Report progress immediately
                self.current_index = (self.current_index + 1) % self.sources.len();
                return Some(Stream::Pending(p));
            }
            Some(Stream::Delayed(d)) => {
                self.current_index = (self.current_index + 1) % self.sources.len();
                return Some(Stream::Delayed(d));
            }
            Some(Stream::Init) => {
                self.current_index = (self.current_index + 1) % self.sources.len();
                return Some(Stream::Init);
            }
            Some(Stream::Ignore) => {
                // Continue to next source in same pass
                self.current_index = (self.current_index + 1) % self.sources.len();
            }
            None => {
                // Source exhausted - mark for removal AFTER pass
                exhausted_indices.push(idx);
                self.current_index = (self.current_index + 1) % self.sources.len();
            }
        }

        // Check if we completed a full pass
        if self.current_index == start_index {
            break;
        }
    }

    // Remove exhausted sources AFTER the pass
    // CRITICAL: Sort descending so swap_remove doesn't invalidate indices
    exhausted_indices.sort_by(|a, b| b.cmp(a));
    for idx in exhausted_indices {
        self.sources.swap_remove(idx);
    }

    // Adjust current_index if it now points beyond new length
    if !self.sources.is_empty() && self.current_index >= self.sources.len() {
        self.current_index = 0;
    }

    if self.sources.is_empty() {
        self.done = true;
        if self.collected.is_empty() {
            return None;
        }
        return Some(Stream::Next(std::mem::take(&mut self.collected)));
    }

    Some(Stream::Pending(self.collected.len()))
}

Key patterns:

1. Single bounded pass - Loop runs at most sources.len() times
2. Immediate return on progress - Next, Pending, Delayed, Init all escape immediately
3. Deferred removal - Exhausted sources removed AFTER pass completes
4. Sort descending before swap_remove - Prevents index invalidation
5. Adjust current_index after removal - Handles index shift from swap_remove

Reference: CollectNextFromStreams in stream_iterators.rs

Checklist for Iterator Implementation

Before submitting an iterator implementation, verify:

• No blocking loop {} - next() returns after processing one item
• State tracking - Uses struct fields (found, done, acc) to track progress across calls
• Immediate progress reporting - Pending, Delayed, Init pass through without delay
• Proper termination - Returns None when exhausted, sets done flag when appropriate
• Ignore for intermediate - Returns Stream::Ignore (or TaskStatus::Ignore) while working, not continue
• Short-circuit logic - For find/all/any, stops polling after result determined
• Multi-source safety - If removing sources, sort indices descending before swap_remove

Summary Table: Do's and Don'ts

References

Fixed Implementations (StreamIterator)

• SAll - lines 2688-2730
• SAny - lines 2734-2776
• SFold - lines 2638-2678
• SCountAll - lines 2809-2845
• SFind / SFindMap - lines 2560-2630
• SSkipState / SSkipWhileState - lines 2456-2540
• CollectNextFromStreams - multi-source round-robin
• MapIter / MapIterDone / MapIterPending - lines 2013-2156 (flatmap pattern)

Fixed Implementations (TaskIterator)

• TAll - lines 2455-2498
• TAny - lines 2502-2545
• TFold - lines 2406-2445
• TCount / TCountAll - lines 2549-2610
• TFind / TFindMap - lines 2328-2402
• TSkipState / TSkipWhileState - lines 2225-2302

Error Handling: foundation_errstacks Required

All types used with VfsResult<T> MUST implement Debug.

• VfsResult<T> is Result<T, ErrorTrace<VfsError>> — never raw Result<T, VfsError>
• All sync bridge wrappers (SyncFile, SyncFs, SyncDirectory, SyncDynDirectory, LocalSeekableFile, SyncLibsqlDelta, etc.) implement Debug using finish_non_exhaustive() for inner async types that may not be Debug
• Tests use err.current_context() to access the typed &VfsError — never downcast_ref
• Import foundation_errstacks::ErrorTraceResultExt when you need change_context() or attach_printable()

// CORRECT: current_context() returns &VfsError
let err = sync.stat("/missing").unwrap_err();
assert!(matches!(err.current_context(), VfsError::NotFound { .. }));

// WRONG: downcast_ref bypasses foundation_errstacks API
let err = sync.stat("/missing").unwrap_err();
assert!(err.downcast_ref::<VfsError>().is_some()); // DON'T DO THIS

Test Annotations: Required for All Valtron Tests

Every test that uses the valtron pool MUST have these annotations:

#[test]
#[ntest::timeout(60_000)]           // 60s timeout — prevents hung tests
#[serial_test::serial]              // Global serialization (pool is global state)
#[tracing_test::traced_test]        // Log visibility in test output
fn test_name() {
    let _guard = init_pool();       // MUST be first line
    // ...
}

Named serialization — use when specific test groups conflict with each other but not with all other tests:

#[serial_test::serial(vfs_sqlite)]  // Only serializes with other vfs_sqlite tests

All valtron pool tests use unnamed #[serial_test::serial] because the pool is global state. Named groups (#[serial_test::serial(name)]) are useful when non-pool resources also conflict.

Version: 1.0 - Created: 2026-04-04