Why it is so difficult to pass this &mut db transaction to this closure?

help
1

I'm using the below code in my real project. (Obviously this is simplified and one file only for playground.)

Working playground here.

As you can see I'm trying to use a sqlx Transaction from one function to a closure.

But I'm stucked.

I don't even know if this (so common Golang pattern) is the best wat to do in Rust. But at least it should work now.

use std::{future::Future, pin::Pin, sync::Arc};

pub trait Trait: Send + Sync + Player + Shirt {}

impl<T: Player + Shirt> Trait for T {}

pub type Lambda<'a, ArgT, ResT> =
    dyn Fn(ArgT) -> Pin<Box<dyn Future<Output = Result<ResT, String>> + Send + 'a>> + Sync + 'a;

#[async_trait::async_trait]
pub trait Player: Send + Sync {
    async fn player_create<'a>(
        &'a self,
        _input: &PlayerInput,
        lambda: &Lambda<'_, PlayerCreateLambdaArgs<'a>, DomainPlayer>,
    ) -> Result<DomainPlayer, String>;
}

#[async_trait::async_trait]
pub trait Shirt: Send + Sync {
    async fn shirt_get_next_and_increase<'a>(
        &'a self,
        tx: &'a mut sqlx::PgConnection,
        model: String,
    ) -> Result<i64, String>;
}

pub struct Repo {
    pub pool: Arc<sqlx::PgPool>,
}

impl Repo {
    pub fn new(pool: Arc<sqlx::PgPool>) -> Self {
        Self { pool }
    }
}

#[async_trait::async_trait]
impl Player for Repo {
    async fn player_create<'a>(
        &'a self,
        _input: &PlayerInput,
        lambda: &Lambda<'_, PlayerCreateLambdaArgs<'a>, DomainPlayer>,
    ) -> Result<DomainPlayer, String> {
        let mut tx = self.pool.begin().await.unwrap();

        // use _input here

        let shirt_next_value = Box::new(|model: String| {
            self::Shirt::shirt_get_next_and_increase(self, &mut tx, model)
        });

        let domain_player = lambda(PlayerCreateLambdaArgs { shirt_next_value }).await?;

        let res =
            sqlx::query_as::<_, DomainPlayer>("INSERT INTO player (...) VALUES (...) RETURNING *")
                .bind(domain_player.id)
                .bind(domain_player.shirt_number)
                .fetch_one(&mut *tx)
                .await
                .unwrap();

        Ok(res)
    }
}

#[async_trait::async_trait]
impl Shirt for Repo {
    async fn shirt_get_next_and_increase<'a>(
        &'a self,
        _tx: &'a mut sqlx::PgConnection,
        _model: String,
    ) -> Result<i64, String> {
        // Here I'm awaiting an async call for DB operations using the same DB transacion of the caller (_tx)...

        // use _tx here...

        let res = 123;

        Ok(res)
    }
}

pub struct Needs {
    pub command_pg_repo: Arc<dyn Trait>,
}

#[derive(Default)]
pub struct PlayerInput {
    pub id: String,
}

#[derive(Debug, Default, Clone, sqlx::FromRow)]
pub struct DomainPlayer {
    pub id: String,
    pub shirt_number: i64,
}

pub struct PlayerCreateLambdaArgs<'a> {
    // other needed fields here
    pub shirt_next_value: Box<
        dyn Fn(String) -> Pin<Box<dyn Future<Output = Result<i64, String>> + Send + 'a>>
            + Send
            + Sync
            + 'a,
    >,
}

pub struct Handler {
    needs: Arc<Needs>,
}

impl Handler {
    pub fn new(needs: Arc<Needs>) -> Self {
        Self { needs }
    }

    pub async fn handle(&self, input: &PlayerInput) -> Result<DomainPlayer, String> {
        let res = self
            .needs
            .command_pg_repo
            .player_create(&input, &|args| {
                let input = input;

                Box::pin(async move {
                    let shirt_number = (args.shirt_next_value)("player".to_string()).await?;

                    let o = DomainPlayer {
                        id: input.id.to_string(),
                        shirt_number,
                    };

                    Ok(o)
                })
            })
            .await?;

        Ok(res)
    }
}

#[tokio::main]
async fn main() -> Result<(), String> {
    let db_conn = sqlx::PgPool::connect("fake_url").await.unwrap();

    let pg_repo = Arc::new(Repo::new(Arc::new(db_conn)));

    let needs = Arc::new(Needs {
        command_pg_repo: pg_repo,
    });

    let handler = Handler::new(needs);

    let new_player_input = PlayerInput {
        id: "abc".to_string(),
    };

    let player = handler.handle(&new_player_input).await?;

    dbg!(player);

    Ok(())
}

The error:

error[E0525]: expected a closure that implements the `Fn` trait, but this closure only implements `FnMut`
  --> src/main.rs:61:41
   |
61 |         let shirt_next_value = Box::new(|model: String| {
   |                                         ^^^^^^^^^^^^^^^ this closure implements `FnMut`, not `Fn`
62 |             self::Shirt::shirt_get_next_and_increase(self, &mut tx, model)
   |                                                                 -- closure is `FnMut` because it mutates the variable `tx` here
...
65 |         let domain_player = lambda(PlayerCreateLambdaArgs { shirt_next_value }).await?;
   |                                                             ---------------- the requirement to implement `Fn` derives from here
   |
   = note: required for the cast from `[closure@src/main.rs:61:41: 61:56]` to the object type `dyn Fn(std::string::String) -> Pin<Box<dyn Future<Output = Result<i64, std::string::String>> + Send>> + Send + Sync`

For more information about this error, try `rustc --explain E0525`.
2

The error is exactly what the compiler says. If a closure mutates a captured variable, then it only implements FnMut and not Fn. It appears to me that you control the internal API, so you could just require FnMut instead of Fn.

3

Thanks @H2CO3. I tried with FnMut too, but the error then becomes:

error: captured variable cannot escape `FnMut` closure body
  --> src\main.rs:50:13
   |
45 |         let mut tx = self.pool.begin().await.unwrap();
   |             ------ variable defined here
...
49 |         let shirt_next_value = Box::new(|model: String| {
   |                                                       - inferred to be a `FnMut` closure
50 |             self::Shirt::shirt_get_next_and_increase(self, &mut tx, model)
   |             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^--^^^^^^^^
   |             |                                                   |
   |             |                                                   variable captured here
   |             returns a reference to a captured variable which escapes the closure body
   |
   = 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
4

Rust explorer with the FnMut

5

@vague I tried to solve this with your previous explanations but I cannot fix it. Can you help me?

6

Aside from the FnMut, this time the closure captures &'a self which comes out of the method and &mut tx which comes from the method. So the signature is wrong:

lambda: &Lambda<'_, PlayerCreateLambdaArgs<'a>, DomainPlayer> should be a HRTB due to &mut tx.

I tried but didn't figure out a solution.

7

And there are quirks when using (async) closures.

8

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.