futures::sink::unfold easily causes panics if function returns errors

[ijackson]

I found some strange behaviour and have minimised it to this:

    #[derive(Debug,Eq,PartialEq)]
    struct TestError(char);

    #[async_test]
    async fn unfold_crash() {
        let mut unfold = Box::pin(futures::sink::unfold((), |(), c| async move {
            Err(TestError(c))
        }));
        for v in ['1','2','3'] {
            let _ = dbg!(unfold.send(v).await);
        }
    }

Which crashes:

---- futures::test::unfold_crash stdout ----
[crates/tor-basic-utils/src/futures.rs:158] unfold.send(v).await = Err(
    TestError(
        '1',
    ),
)
thread 'futures::test::unfold_crash' panicked at '`async fn` resumed after completion', crates/tor-basic-utils/src/futures.rs:154:80

I think what is happening is that (from the outside) the Sink impl for Unfold panics in some situations when it is called after returning an error. (The Sink trait docs say "In most cases, if the sink encounters an error, the sink will permanently be unable to receive items" which I think suggests it might keep returning Err, but doesn't seem to me to encompass panicking.)

On the inside, evidently it is trying to reuse a returned future R, after it got Ready(Err), when it ought to call F again to get a fresh future.

[taiki-e]

Fixed in #2686