futures-util 0.3.31: FuturesOrdered polls queued futures concurrently while yielding completed items strictly in FIFO submission order, buffering out-of-order completions until predecessor futures complete and allowing resumption after exhaustion

# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 4

[[package]]
name = "cargo-futures-util-version-depth"
version = "0.1.0"
dependencies = [
 "futures-util",
]

[[package]]
name = "futures-core"
version = "0.3.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"

[[package]]
name = "futures-sink"
version = "0.3.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d"

[[package]]
name = "futures-task"
version = "0.3.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd"

[[package]]
name = "futures-util"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
dependencies = [
 "futures-core",
 "futures-sink",
 "futures-task",
 "pin-project-lite",
 "pin-utils",
]

[[package]]
name = "pin-project-lite"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"

[[package]]
name = "pin-utils"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

[package]
name = "cargo-futures-util-version-depth"
version = "0.1.0"
edition = "2021"

[dependencies]
futures-util = { version = "=0.3.31", default-features = false, features = ["alloc", "async-await", "sink"] }

# futures-util 0.3.31: FuturesOrdered FIFO Ordering, Out-of-Order Buffering, and Resumption

## Prior Network Solutions

A query to the network cache for `futures-util 0.3` returned three existing samples:
- `sha256:8dad9b00522c8ec7137956d2d7912d83e26f8020a34995bffe05df84d036a872` (stream stalling in `chunks` vs `ready_chunks`, sink feed vs send flushing, forward `poll_close`, and `take_while` item loss)
- `sha256:d6d1ba895636c28d95e369bf0c9d1d6907c4fe9f6602510ca5ca39c7444d2304` (stream `unfold` tuple ordering, `buffer_unordered`/`buffered` concurrency, `ready_chunks` batching, `select` loser preservation, and `try_join_all` cancellation)
- `sha256:2336755e4371002b1fe77659a664fa05c66c674b78f48b4f45d6cad20aa9c8f0` (`FuturesUnordered` empty poll returning `None` without terminating vs `Fuse` latching `is_terminated`)

This sample proves a distinct behaviour in `futures-util`: the execution and ordering guarantees of `futures_util::stream::FuturesOrdered`.

## Pinned Version

- Target Package: `futures-util`
- Version: `0.3.31` (Cargo lockfile pinned)
- Features: `alloc`, `async-await`, `sink`

## Observed Behaviour

1. **Strict FIFO Yield Order**:
   `FuturesOrdered` polls queued futures concurrently as wakeups occur. However, results are yielded strictly in the order tasks were queued (FIFO). If task #2 and task #3 complete while task #1 is still pending, `FuturesOrdered` returns `Poll::Pending` on poll, retaining the completed results of tasks #2 and #3 in its internal buffer.

2. **Immediate Availability of Buffered Successors**:
   When task #1 completes and is yielded on a poll, the subsequent polls yield the pre-buffered results of task #2 and task #3 immediately without requiring additional task polling or new wake events.

3. **Resumption After Exhaustion**:
   When all queued tasks have resolved, polling `FuturesOrdered` yields `Poll::Ready(None)`. The collection does not enter a permanently terminated state: pushing a new future afterwards reactivates the collection, and subsequent polls drive and yield the newly added future normally.

4. **Queue Size Accounting**:
   `len()` and `is_empty()` track the total sum of currently pending futures and unyielded buffered results in the collection.

{"case":{"caseId":"case:sha256:36bd8c6cde16e2f4c37177b11368a8b2134b30b612d97746cbe252ba16b83a6b","contract":["FuturesOrdered executes queued futures concurrently but yields completed items strictly in the order they were pushed, returning Poll::Pending when successor futures finish before preceding ones and buffering their results for immediate subsequent delivery once predecessor futures complete.","Polling an empty FuturesOrdered returns Poll::Ready(None) without permanently terminating the stream, allowing new futures pushed after exhaustion to be polled and resolved normally.","FuturesOrdered::len and is_empty accurately track the sum of currently pending futures and unyielded buffered results in the queue."],"goal":"FuturesOrdered polls queued futures concurrently while yielding completed items strictly in FIFO submission order, buffering out-of-order completions until predecessor futures complete and allowing resumption after exhaustion","kind":"HOW","packages":["pkg:cargo/futures-util@0.3.31"],"schemaVersion":1,"symbols":["futures_util::stream::FuturesOrdered"]},"contractCommand":["cargo","test","--offline"],"environment":{"arch":"x64","ecosystem":"cargo","executionContext":"rust","language":"rust","os":"linux","packageManager":"cargo","runtime":"rust","schemaVersion":1},"license":"MIT-0","packages":["pkg:cargo/futures-util@0.3.31"],"schemaVersion":1,"symbols":["futures_util::stream::FuturesOrdered"],"verifierAdapter":"cargo@1"}

use std::cell::{Cell, RefCell};
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use futures_util::stream::{FuturesOrdered, Stream};

/// Creates a minimal non-waking root context waker.
fn create_test_waker() -> Waker {
    fn noop_clone(_: *const ()) -> RawWaker {
        raw_waker()
    }
    fn noop(_: *const ()) {}
    static VTABLE: RawWakerVTable = RawWakerVTable::new(noop_clone, noop, noop, noop);
    fn raw_waker() -> RawWaker {
        RawWaker::new(std::ptr::null(), &VTABLE)
    }
    unsafe { Waker::from_raw(raw_waker()) }
}

/// A controllable future that records poll counts and captures its task waker.
struct ControllableFuture {
    id: usize,
    ready: Rc<Cell<bool>>,
    poll_count: Rc<Cell<usize>>,
    waker: Rc<RefCell<Option<Waker>>>,
}

impl ControllableFuture {
    fn new(id: usize) -> (Self, Rc<Cell<bool>>, Rc<Cell<usize>>, Rc<RefCell<Option<Waker>>>) {
        let ready = Rc::new(Cell::new(false));
        let poll_count = Rc::new(Cell::new(0));
        let waker = Rc::new(RefCell::new(None));
        let fut = Self {
            id,
            ready: Rc::clone(&ready),
            poll_count: Rc::clone(&poll_count),
            waker: Rc::clone(&waker),
        };
        (fut, ready, poll_count, waker)
    }
}

impl std::future::Future for ControllableFuture {
    type Output = usize;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.poll_count.set(self.poll_count.get() + 1);
        if self.ready.get() {
            Poll::Ready(self.id)
        } else {
            *self.waker.borrow_mut() = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_fifo_ordering_with_out_of_order_completions() {
        let root_waker = create_test_waker();
        let mut cx = Context::from_waker(&root_waker);

        let mut ordered = FuturesOrdered::new();
        assert!(ordered.is_empty());
        assert_eq!(ordered.len(), 0);

        let (fut1, ready1, polls1, waker1) = ControllableFuture::new(10);
        let (fut2, ready2, polls2, waker2) = ControllableFuture::new(20);
        let (fut3, ready3, polls3, waker3) = ControllableFuture::new(30);

        ordered.push_back(fut1);
        ordered.push_back(fut2);
        ordered.push_back(fut3);

        assert_eq!(ordered.len(), 3);
        assert!(!ordered.is_empty());

        // Initial poll: all 3 futures are polled once and register their wakers
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Pending);
        assert_eq!(polls1.get(), 1);
        assert_eq!(polls2.get(), 1);
        assert_eq!(polls3.get(), 1);

        // Future 3 and Future 2 complete before Future 1
        ready3.set(true);
        if let Some(w) = waker3.borrow().as_ref() {
            w.wake_by_ref();
        }
        ready2.set(true);
        if let Some(w) = waker2.borrow().as_ref() {
            w.wake_by_ref();
        }

        // Poll when 2 & 3 are ready but head (1) is pending:
        // FuturesOrdered polls woken tasks internally, buffers their ready values,
        // but returns Pending because the head future has not yet resolved.
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Pending);
        assert_eq!(polls1.get(), 1); // Head was not re-polled (no wake event)
        assert_eq!(polls2.get(), 2); // Woken and completed
        assert_eq!(polls3.get(), 2); // Woken and completed
        assert_eq!(ordered.len(), 3); // All 3 still in queue / buffer

        // Now Future 1 completes
        ready1.set(true);
        if let Some(w) = waker1.borrow().as_ref() {
            w.wake_by_ref();
        }

        // Subsequent polls yield results strictly in FIFO order: 10, then 20, then 30
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(Some(10)));
        assert_eq!(ordered.len(), 2);

        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(Some(20)));
        assert_eq!(ordered.len(), 1);

        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(Some(30)));
        assert_eq!(ordered.len(), 0);
        assert!(ordered.is_empty());

        // Exhausted stream yields None
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(None));
    }

    #[test]
    fn test_resumption_after_empty_poll() {
        let root_waker = create_test_waker();
        let mut cx = Context::from_waker(&root_waker);

        let mut ordered = FuturesOrdered::new();
        // Polling an empty collection immediately returns None without latching closed
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(None));

        // Push new task into previously-exhausted collection
        let (fut_a, ready_a, _polls_a, _waker_a) = ControllableFuture::new(100);
        ready_a.set(true);
        ordered.push_back(fut_a);

        assert_eq!(ordered.len(), 1);
        assert!(!ordered.is_empty());

        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(Some(100)));
        assert_eq!(ordered.len(), 0);
        assert!(ordered.is_empty());
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(None));
    }

    #[test]
    fn test_from_iterator_preserves_submission_order() {
        let root_waker = create_test_waker();
        let mut cx = Context::from_waker(&root_waker);

        let (fut1, ready1, _, _) = ControllableFuture::new(1);
        let (fut2, ready2, _, _) = ControllableFuture::new(2);
        ready1.set(true);
        ready2.set(true);

        let mut ordered: FuturesOrdered<_> = vec![fut1, fut2].into_iter().collect();
        assert_eq!(ordered.len(), 2);

        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(Some(1)));
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(Some(2)));
        assert_eq!(Pin::new(&mut ordered).poll_next(&mut cx), Poll::Ready(None));
    }
}