Function with_ready_batches 

async fn with_ready_batches<L: Layout, W, Write, Close>(
    input_frontier: Antichain<Timestamp>,
    input_batches: &mut VecDeque<Rc<OrdValBatch<L>>>,
    output_frontier: Antichain<Timestamp>,
    output_batches: &mut VecDeque<((Antichain<Timestamp>, Antichain<Timestamp>), W)>,
    write_rows: Write,
    close_batch: Close,
) -> Result<(), Error>
where
    L::TimeContainer: BatchContainer<Owned = Timestamp>,
    Write: AsyncFnMut(&OrdValBatch<L>, (Antichain<Timestamp>, Antichain<Timestamp>), &mut W) -> Result<(), Error>,
    Close: AsyncFnMut((Antichain<Timestamp>, Antichain<Timestamp>), &mut W) -> Result<(), Error>,

Write out as much of the input as we can.

Drop input batches when:

• their contents have all been written out
• no possible future output batch could need their contents

Close and drop output batches when:

• no possible future input batch could overlap with their time window

Invariant: We assume the batches in each stream (input vs output) are in order and non-overlapping.