Incorrect "implementation not general enough" error in async block

[jacobsa]

Let's say we have this API for a push-based async stream of elements:

/// A sequence of zero or more asynchronously produced elements. Elements are accepted by an
/// `Acceptor`, which can push back on production of the stream while it processes each element.
pub trait Stream {
    type Element: Send;

    fn consume(
        self,
        acceptor: &mut impl Acceptor<Self::Element>,
    ) -> impl Future<Output = ()> + Send;
}

/// Knows how to accept values from a stream that produces elements of type `E`.
pub trait Acceptor<E>: Send {
    fn accept(&mut self, element: E) -> impl Future<Output = ()> + Send;
}

Then we can provide this async function to collect a stream into a vector:

/// Consume the supplied stream, returning a vector of all of its elements in order.
pub async fn collect_stream<S>(s: S) -> Vec<S::Element>
where
    S: Stream,
{
    struct VecAcceptor<E>(Vec<E>);

    impl<E> Acceptor<E> for VecAcceptor<E>
    where
        E: Send,
    {
        async fn accept(&mut self, element: E) {}
    }

    let mut acceptor = VecAcceptor(Vec::new());
    s.consume(&mut acceptor).await;
    acceptor.0
}

However, if we call that function in this very specific context (playground), then we get a nonsense error about trait implementations not being general enough:

fn accept_future<F>(future: F)
where
    F: Future,
    F: Send,
{
    todo!();
}

fn call_it() {
    struct S<'a> {
        element_0: &'a i32,
    }

    impl<'a> Stream for S<'a> {
        type Element = &'a i32;

        async fn consume(self, acceptor: &mut impl Acceptor<Self::Element>) {
            acceptor.accept(self.element_0).await;
        }
    }

    accept_future(async move {
        let n0 = 17;
        let _ = collect_stream(S { element_0: &n0 }).await;
    });
}

error: implementation of `Acceptor` is not general enough
  --> src/lib.rs:61:5
   |
61 | /     accept_future(async move {
62 | |         let n0 = 17;
63 | |         let _ = collect_stream(S { element_0: &n0 }).await;
64 | |     });
   | |______^ implementation of `Acceptor` is not general enough
   |
   = note: `VecAcceptor<&'0 i32>` must implement `Acceptor<&'1 i32>`, for any two lifetimes `'0` and `'1`...
   = note: ...but it actually implements `Acceptor<&i32>`

error: implementation of `Stream` is not general enough
  --> src/lib.rs:61:5
   |
61 | /     accept_future(async move {
62 | |         let n0 = 17;
63 | |         let _ = collect_stream(S { element_0: &n0 }).await;
64 | |     });
   | |______^ implementation of `Stream` is not general enough
   |
   = note: `Stream` would have to be implemented for the type `S<'_>`
   = note: ...but `Stream` is actually implemented for the type `S<'0>`, for some specific lifetime `'0`

This is very specific to the calling context. For example if we remove the Send bound on accept_future then it works, and it also works if you send the stream through a generic identity function before it goes to collect_stream. It seems to also require the presence of an async block; an async function is not enough to trigger it.

[jacobsa]

Any ideas what's going on here? I've seen this a few times in different contexts, sometimes where the workaround above doesn't apply.

[tmandry]

Based on discussions with @lcnr I expect this to be fixed as part of the assumptions on binders project goal.