Function drain_sealed_input_chunked 

async fn drain_sealed_input_chunked<T, O>(
    sealed: impl Iterator<Item = Column<(UpsertKey, T, UpsertDiff<O>)>>,
    ineligible: &mut Vec<(UpsertKey, T, UpsertDiff<O>)>,
    output_handle: &AsyncOutputHandle<T, FueledBuilder<CapacityContainerBuilder<Vec<(UpsertValue, T, Diff)>>>>,
    output_cap: &Capability<T>,
    persist_upper: &Antichain<T>,
    trace: &mut TraceAgent<ChunkSpine<ColumnChunk<(UpsertKey, Row), T, Diff>>>,
    worker_id: usize,
    source_id: GlobalId,
) -> DrainStats
where
    T: Timestamp + TotalOrder + Lattice + Sync + Columnation + Columnar + Default,
    for<'a> Ref<'a, T>: Copy + Ord,
    O: Columnar,

Process sealed chunks from the batcher, classifying each entry by its timestamp relative to persist_upper:

• ts == persist_upper: eligible for processing now (bulk probe of the feedback trace + output).
• ts > persist_upper: not yet processable. Returned in ineligible for re-stashing until the feedback frontier catches up to it.
• ts < persist_upper: already persisted by some writer and not relevant anymore, so DROPPED. The downstream persist_sink would filter such updates out anyway since the shard upper is further ahead, and our state is already up-to-date to persist_upper so we could not emit correct retractions for it. Re-stashing it would strand the data forever (persist_upper only advances, so ts == persist_upper can never again hold) and pin the operator’s output frontier below the shard upper. This mirrors v1’s relevant = persist_upper.less_equal(ts).

The sealed chunks are already sorted and consolidated by the merge batcher, so each chunk’s eligible keys form a sorted, deduplicated probe set, and the prior state comes back through one bulk extract_into pass per trace batch. Resident fence metadata selects the touched trace chunks, and only those bodies are read back, copy-out, per chunk probed. Sealed chunks are pulled from the iterator one at a time and dropped before the next is requested, so at most one loaded stash chunk (plus the probe hits for its keys) is resident regardless of drain size. Only the re-stashed ineligible set is materialized.