Add reshard_try_stream function

This adds support for fallible streams in resharding, so it can operate on `Result<T, E>` rather than just `T`.

Any error that occurs in the input stream is propagated and resharding terminates after that.