Bogus higher-ranked lifetime error in an async block

[sfackler]

Running stable 1.64.0.

Unfortunately this happened in a complex bit of code that I can't easily reduce :(

            task::spawn({
                let handle_service = handle_service.clone();
                async move {
                    if let Err(e) = handle_service.call(connection).await {
                        debug!("http connection terminated", error: e);
                    }
                }
            });

error: higher-ranked lifetime error
   --> witchcraft-server/src/server.rs:118:13
    |
118 | /             task::spawn({
119 | |                 let handle_service = handle_service.clone();
120 | |                 async move {
121 | |                     if let Err(e) = handle_service.call(connection).await {
...   |
124 | |                 }
125 | |             });
    | |______________^
    |
    = note: could not prove `for<'r> impl for<'r> futures_util::Future<Output = ()>: std::marker::Send`

The future returned by handle_service.call(connection) is definitely Send, and I can work around the failure by boxing it:

            task::spawn({
                let handle_service = handle_service.clone();
                async move {
                    // The compiler hits a `higher-ranked lifetime error` if we don't box this future :/
                    let f: Pin<Box<dyn Future<Output = Result<(), Error>> + Send>> =
                        Box::pin(handle_service.call(connection));
                    if let Err(e) = f.await {
                        debug!("http connection terminated", error: e);
                    }
                }
            });

[eholk]

We discussed this in the wg-async triage today. We think the best course of action is to try to get a minimal test case for this. Afterwards, we'll probably want to get T-Types input on this. @vincenzopalazzo is going to try to reproduce this and minimize the test case.

[eholk]

@rustbot label AsyncAwait-Triaged

[vincenzopalazzo]

@rustbot claim

[danielhenrymantilla]

I wonder if the following snippet wouldn't be a reduction:

struct Type<'a, 'b>(&'a &'b (), ::core::marker::PhantomData<*mut ()>);

// Note: if this had an *implicit* `'b : 'a` things would work.
unsafe impl<'b : 'a, 'a> Send for Type<'a, 'b> {}

fn foo() -> impl Send { async {
    let local = ();
    let r = &&local;
    async {}.await;
    let it = Type(r, <_>::default());
    async {}.await;
}}

• from Discord

[eholk]

@sfackler - Would it be possible to share more of the surrounding code for this example so we can see more of what's going on?

[sfackler]

I'm working on minimizing it.

[sfackler]

Here is a semi-minimized repro. It still depends on Hyper, but it could probably be replaced with some PhantomData inserts with some more work: https://gist.github.com/sfackler/a7f3aaf2ce3f5ff4c99e1a3d67f828ed

error: higher-ranked lifetime error
   --> witchcraft-server/src/lib.rs:162:5
    |
162 | /     spawn(async move {
163 | |         let _ = handle_service.call(stream).await;
164 | |     });
    | |______^
    |
    = note: could not prove `for<'r> impl for<'r> futures_util::Future<Output = ()>: Send`

[sfackler]

For whatever reason, the presence of the AuditLogService in the future seems to be a trigger. If you change line 160 to just let request_service = HandlerService it compiles.

[vincenzopalazzo]

Thank @sfackler,

I think the core concept missed in your previous report are:

Using a wrapper around Future like

pub struct AdaptorFuture<F> {
    inner: F,
}

impl<F, T> Future for AdaptorFuture<F>
where
    F: Future<Output = T>,
{
    type Output = Result<T, io::Error>;

    fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
        panic!()
    }
}

and your handle_service implementation

impl<S, R> hyper::service::Service<R> for AdaptorService<S>
where
    S: Service<R>,
{
    type Response = S::Response;

    type Error = io::Error;

    type Future = AdaptorFuture<S::Future>;

    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: R) -> Self::Future {
        AdaptorFuture {
            inner: self.inner.call(req),
        }
    }
}

[vincenzopalazzo]

@sfackler I'm trying to reproduce your error without hyper, and till now I had no luck.

Maybe you can help with some tips? I had the code there vincenzopalazzo/rio#13

[sfackler]

Not sure I have a better idea of how to minimize than you, unfortunately.

[brandonros]

https://play.rust-lang.org/?version=stable&mode=debug&edition=2021&gist=b247265803ef4fbc2223a4ddcec31432

I think I was able to recreate it?

It took me like 35 minutes to figure out where it was because the compiler only gives this message with nothing else and no extra details:

error: higher-ranked lifetime error
   --> bot/src/main.rs:141:23
    |
141 |         let handle2 = tokio::task::spawn(robinhood_task());
    |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |
    = note: could not prove `impl Future<Output = ()>: Send`

[vincenzopalazzo]

This looks promising @brandonros! I will give it a shot and also try to reproduce inside vincenzopalazzo/rio#13

Looks like that when the runtiime try to span the following task there is an error like the one reported from you

async fn robinhood_task() {
    let robinhood = Robinhood::new();
    let token = String::from("fake");
    let instrument_ids = vec![String::from("fake")];
    robinhood.get_options_market_data(token, instrument_ids).await;
}

In particulat I play a little bit with it and I found the following example to reproduce the error

use futures::StreamExt;
use log::info;
use serde::{Deserialize, Serialize};

pub struct Robinhood;

impl Robinhood {
    pub fn new() -> Robinhood {
        return Robinhood {};
    }

    async fn foo(&self, t: String, b: Vec<String>) -> Vec<String> {
        // Looks like that the problem is generated from the following code
        let b: Vec<&[String]> = b.chunks(2).collect();
        let futures = b.into_iter().map(|it| {
            return async { vec![] };
        });

        let results: Vec<String> = futures::stream::iter(futures)
            .buffer_unordered(4)
            .collect::<Vec<_>>()
            .await
            .into_iter()
            .flatten()
            .collect();
        results
    }
}

async fn foo_ok(t: String, b: Vec<String>) -> Vec<String> {
    let b: Vec<&[String]> = b.chunks(2).collect();
    let futures = b.into_iter().map(|it| {
        return async { vec![] };
    });

    let results: Vec<String> = futures::stream::iter(futures)
        .buffer_unordered(4)
        .collect::<Vec<_>>()
        .await
        .into_iter()
        .flatten()
        .collect();
    results
}

async fn robinhood_task() {
    let robinhood = Robinhood::new();
    foo_ok(String::new(), vec![]).await;
    robinhood.foo(String::new(), vec![]).await;
}

fn main() {
    async {
        let handle1 = tokio::task::spawn(robinhood_task());
        let _ = tokio::join!(handle1);
    };
}

[vincenzopalazzo]

@rustbot label +A-async-await

our penguing triage bot it is not smart enough to reassign this in the next triage meeting so I need to do this trick!

[rustbot]

Error: Label AsyncAwait-Triaged can only be set by Rust team members

Please file an issue on GitHub at triagebot if there's a problem with this bot, or reach out on #t-infra on Zulip.

[brandonros]

I'm sure this is obvious to you, I just want to make sure it's documented somewhere:

switching let b: Vec<&[String]> = b.chunks(2).collect(); to let b: Vec<Vec<String>> = b.chunks(2).map(|c| c.to_vec()).collect(); works around the issue.

[awesomelemonade]

Unsure if this is the same exact issue, but I ran into a "higher-ranked lifetime error" when running https://play.rust-lang.org/?version=nightly&mode=debug&edition=2021&gist=e8d2668ceca75a8ebf4f0c16cee04958. I tried to work around the failure by boxing it like suggested, but it still doesn't seem to work.
The full error is

error: higher-ranked lifetime error
  --> src/lib.rs:25:18
   |
25 |     let handle = tokio::spawn(accept_connection(rpc));
   |                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   |
   = note: could not prove `impl futures::Future<Output = Result<(), anyhow::Error>>: std::marker::Send

EDIT: if I pin & box the lol future, it works: https://play.rust-lang.org/?version=nightly&mode=debug&edition=2021&gist=5d7d8b1dd6542571425b37b5e2cd6ee9

[Tails]

Had the same problem, but @awesomelemonade's example helped. It worked by pinning the try_join_all call and explicitly specifying the type in the assignment. Note that the Send bound made the difference in 'dyn Future<> + Send'.

        // let objects = futures::future::try_join_all(futs)
        //     .await?
        //     .into_iter()
        //     .collect::<HashMap<Self::Address, C>>();

        let f: Pin<
            Box<
                dyn Future<Output = anyhow::Result<Vec<(<Self as StorageProvider>::Address, C)>>>
                    + Send,
            >,
        > = Box::pin(futures::future::try_join_all(futs));

        f.await;

[danielhenrymantilla]

@Tails @awesomelemonade you can skip the heap allocation with a helper function:

fn assert_send<'u, R>(fut: impl 'u + Send + Future<Output = R>)
  -> impl 'u + Send + Future<Output = R>
{
    fut
}

let f = assert_send(futures::future::try_join_all(futs));
f.await

All this, and more is present in that Discord discussion I mentioned:

• from Discord

[mcronce]

I was running into this problem in a project tonight (which prompted me to go through a bunch of its dependencies and enable warn(clippy::future_not_send) 😂); the assert_send() (and, indeed, Box::pin) solution didn't work, but switching to stable did. Not sure whether or not it's well known that this is a nightly-specific problem, or if I'm just hitting it in a nightly-specific way, so take that for what it's worth.

I did update my nightly to latest, but I'm not sure what version I was on before. Not something more than a few days old, though, I'm pretty sure.

[yuri-rs]

I face this issue with futures::stream::Buffered (and BufferUnordered) wrappers.
The following code reproduce the issue:

use futures::stream::StreamExt;

fn higher_ranked_lifetime_error(
) -> std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = ()> + Send>> {
    Box::pin(async {
        let vec_with_readers: Vec<Box<dyn std::io::Read + Send>> = Vec::new();

        let _ = futures::stream::iter(
            vec_with_readers
                .into_iter()
                .map(|_| futures::future::ready::<()>(())),
        )
        .buffered(1)
        .count()
        .await;

        ()
    })
}

while removing .buffered(1) allow the code to compile.
Playground: https://play.rust-lang.org/?version=stable&mode=debug&edition=2021&gist=9cd1b46b17239d28184e14251910e8f7

Could anyone point me to the workaround for the issue please?

Workaround from #102211 (comment) will require changes for futures::stream::Buffered, so it is not what I'm looking for.

[zertyz]

I'm sure this is obvious to you, I just want to make sure it's documented somewhere:

switching let b: Vec<&[String]> = b.chunks(2).collect(); to let b: Vec<Vec<String>> = b.chunks(2).map(|c| c.to_vec()).collect(); works around the issue.

Or better yet, using @frxstrem's suggestion of replacing the closure for a function pointer and skipping copying things around unnecessarily:
https://play.rust-lang.org/?version=stable&mode=debug&edition=2021&gist=a4f501590c778b0dbe78ca620af23b4d

[veeshi]

Could anyone point me to the workaround for the issue please?

Workaround from #102211 (comment) will require changes for futures::stream::Buffered, so it is not what I'm looking for.

I'm facing the same issue after introducing BufferUnordered, attempted to Pin and Box the future that is causing the error but still getting the same error.

Anyone know another workaround?

[Palmik]

In my case the issue was when registering a route handler with axum for foo below.
Note that foo itself compiles when used on its own, it's only when used as an axum handler that it runs into this issue.

To fix this, I had to change do_something_with_xs to take xs: Vec<X> instead of xs: &[X]. This is unintuitive to me, since it's not clear to me why the lifetimes from do_something_with_xs would be able to "escape" into the type of foo.

Another fix would be to not use buffered, but that's not acceptable in my case.

async fn foo() -> Result<()> {
  let xs = vec![...];
  do_something_with_xs(&xs).await
}

async fn do_something_with_xs(xs: &[X]) -> Result<()> {
  // Broken version:
  let ys = futures::stream::iter(xs.iter().map(|x| do_something_with_x(x)));

  // Working version when using xs: Vec<X>
  // let ys = futures::stream::iter(xs.into_iter().map(|x| async move { do_something_with_x(&x).await }));

  // Working version when using xs: &[X] without buffered
  // let ys = futures::stream::iter(xs.iter()).then(|x| do_something_with_x(x));

  tokio::pin!(stream);

  while let Some(y) = ys.next().await {
    // Do something with y
  }
}

// You can keep the reference here
async fn do_something_with_x(x: &X) -> Result<Y> {
  // ...
}

[UVUUUVUU]

i have a same problem.

pub async fn select_paginate() {
    match get_sql_db_conn().await.unwrap() {
        DBPool::AsyncPgPool(pool) => {
            let mut conn = pool.get().await.unwrap();
            conn.transaction::<(), diesel::result::Error, _>(|conn_mutex| async move {
                    let sql = "SELECT * FROM tableone WHERE id = 1".to_string();
                    diesel::sql_query(&sql).execute(conn_mutex).await.unwrap();
                    let mut db_query = tableone.into_boxed();
                    db_query = db_query.filter(title.eq("shenhu"));
                    db_query = db_query.limit(1).offset(1);
                    let result = db_query.paginate(1, 1).load_and_total::<Tableone>(conn_mutex).await.unwrap();
                    println!("{:?}", result);
                    Ok(())
                }.scope_boxed())
            .await;
        }
        _ => {}   
    };
}

the error:

higher-ranked lifetime error
could not prove `[async block@src/utils/select_async.rs:84:75: 93:18]: std::marker::Send`

[LetMut1]

Sorry, I do not understand. Is the problem being solved?

[alilleybrinker]

Looks like @danielhenrymantilla's comment is the most up-to-date look into what's wrong here. It also looks like @vincenzopalazzo was previously assigned this issue, but has removed themself. It's now unassigned, but is part of a tracking issue for lifetime errors in async code.

Based on Daniel's explanation, it looks like this is due to a bug. As he explained, auto traits (like Send or Sync) are checked later than other trait bounds, in a context where lifetimes have been erased. The specific requirements for triggering this bug appear to be:

• An auto-trait bound like Send or Sync
• Async (really, generators, which async uses under the hood)
• An associated type
• A lifetime

I'm not part of the team handling this, so I have no idea where this sits in their priority, but I do think this describes the current state. I highly recommend looking at Daniel's post (which I linked to at the start of this comment), which details a workaround.

[westonpace]

Could anyone point me to the workaround for the issue please?

One potential workaround I've discovered for the buffered/buffer_unordered issue detailed in this comment appears to be collecting the futures into a Vec before calling buffered / buffer_unordered. Rust playground example.

This does introduce some copying and is only really practical when the source of the futures is synchronous (though the futures themselves can be asynchronous). However, this workaround has worked for me in a real world example and I figured I would share it in case it helps someone else.

[williameric87]

Could anyone point me to the workaround for the issue please?

One potential workaround I've discovered for the buffered/buffer_unordered issue detailed in this comment appears to be collecting the futures into a Vec before calling buffered / buffer_unordered. Rust playground example.

This does introduce some copying and is only really practical when the source of the futures is synchronous (though the futures themselves can be asynchronous). However, this workaround has worked for me in a real world example and I figured I would share it in case it helps someone else.

Instead of collecting the futures into a Vec like this, one could also just call boxed() before buffered(), which achieves a similar effect with less code, and is implemented by Box::pinning each future. Using the same Rust playground example would then look like this, or similarly, using just an async fn.

[steffahn]

I’ve come up with a zero-overhead workaround, as a drop-in replacement of the .boxed() approach; but without boxing and without dyn: one can just wrap the Future / Stream (etc…) in a newtype wrapper type with an unconditional Send impl.

I’ve packed this up in a tiny helper crate in order to offer a sound API :-) - The wrapper type AlwaysSend<T> implements Send for all types T; so in order to stay sound, when you construct the wrapper, T: Send is enforced.