Function next_input_driving_persistence 

async fn next_input_driving_persistence(
    stream: &mut StreamReaderWithPause<false, TxnMsg>,
    persistence_futures: &mut FuturesOrdered<impl Future<Output = StreamExecutorResult<()>>>,
) -> StreamExecutorResult<Option<Either<Message, TxnMsg>>>

Poll stream for the next input message while concurrently driving persistence_futures. Any error from a completed persistence future is propagated immediately, causing the executor (and thus the actor) to fail and trigger recovery.