Correct Way to Achieve Query Pipelining with Vec of Futures

[ahmed-said-jax]

I'm looking to implement an "Upsert" trait, and then implement it for Vec<T> where T: Upsert. Here's a naive implementation:

use diesel_async::AsyncConnection;
use futures::FutureExt;

pub trait Upsert {
    type Returns;

    async fn upsert(&self, conn: &mut AsyncPgConnection) -> Self::Returns;
}

impl<T: Upsert> Upsert for Vec<T> {
    type Returns = Vec<T::Returns>;

    async fn upsert(&self, conn: &mut AsyncPgConnection) -> Self::Returns {
        let upserts = self.iter().map(|item| item.upsert(conn));
        futures::future::join_all(upserts).await
    }
}

obviously this doesn't compile, as conn is mutable:

error: captured variable cannot escape `FnMut` closure body
  --> src/db.rs:53:46
   |
52 |     async fn upsert(&self, conn: &mut AsyncPgConnection) -> Self::Returns {
   |                            ---- variable defined here
53 |         let upserts = self.iter().map(|item| item.upsert(conn));
   |                                            - ^^^^^^^^^^^^----^
   |                                            | |           |
   |                                            | |           variable captured here
   |                                            | returns a reference to a captured variable which escapes the closure body
   |                                            inferred to be a `FnMut` closure
   |
   = note: `FnMut` closures only have access to their captured variables while they are executing...
   = note: ...therefore, they cannot allow references to captured variables to escape

In this discussion #190, I found the recommendation to use futures::future::BoxFuture, which compiles:

pub trait Upsert2 {
    type Returns: Send;
    fn upsert(&self, conn: &mut AsyncPgConnection) -> BoxFuture<Self::Returns>;
}

impl<T: Upsert2> Upsert2 for Vec<T> {
    type Returns = Vec<T::Returns>;
    fn upsert(&self, conn: &mut AsyncPgConnection) -> BoxFuture<Self::Returns> {
        let upserts = self.iter().map(|item| item.upsert(conn));
        futures::future::join_all(upserts).boxed()
    }
}

note that I had to add Send to Returns. Anyways, I'm wondering if this "workaround" is correct and/or the simplest, or is there some simple API I'm failing to see? Also, I'd be happy to create a PR to update the documentation in the query pipelining section to illustrate how to do it with any number of futures rather than just two.

[weiznich]

I wouldn't call this "workaround" but expected behavior. It's one of rusts core values that you cannot have more than one mutable reference at the same time. You have exactly that by using an async function there as the compiler generated future contains any function argument.

Also, I'd be happy to create a PR to update the documentation in the query pipelining section to illustrate how to do it with any number of futures rather than just two.

That would be great, especially if you can point out that requirement to return a future that does not borrow the connection (e.g. by using BoxFuture with a lifetime decoupled from the connection) explicitly.

[ahmed-said-jax]

Makes sense! Didn't feel like a "workaround", hence the quotation marks around it. I'll get started on that PR when I get a chance.

Thanks for your help!