Hello.

A total Rust noobie here, coming from Java world.

I am trying to build my first backend project with Rust and I'm looking for a way to chain multiple async operations with a single error-checking point to avoid tons of let x = match f() { Ok()...; Err()... } boilerplate code.

Specifically, my code should run within a sqlx transaction and it must be rolled back on error.

So, in Java, I would write something like this using the Mutiny library (or its alternative RxJava):

dbPool.beginTransaction()
  .map((tx) -> {
    runDbQuery1()
      .map(() -> callExternalService1())
      .map(() -> runDbQuery2(tx))
      .map(() -> callExternalService2())
      .map(() -> tx.commit())                
  })
  .onFailure().retry().withBackOff(Duration.ofMillis(1_000), Duration.ofMillis(1_000))
  .subscribe.with(() -> { ... });

So, here is some naive code I wrote after two days of googling and countless rewrites:

use std::time::Duration;
use sqlx::{PgPool, Postgres, Transaction};
use sqlx::postgres::PgPoolOptions;

async fn run_db_query1(tx: &mut Transaction<'_, Postgres>) -> anyhow::Result<()> {
    Ok(())
}

async fn run_db_query2(tx: &mut Transaction<'_, Postgres>) -> anyhow::Result<()> {
    Ok(())
}

async fn call_external_service1() -> anyhow::Result<()> {
    Ok(())
}

async fn call_external_service2() -> anyhow::Result<()> {
    Ok(())
}

#[tokio::main]
async fn main() {
    let db_pool = PgPoolOptions::new()
        .connect("postgres://...")
        .await
        .unwrap();

    loop {
        let result = db_pool.begin().await
            .and_then(|mut tx| async move {
                run_db_query1(&mut tx).await
                    .and_then(|_| async move {
                        call_external_service1().await
                    })
                    .and_then(|_| async move {
                        run_db_query2(run_db_query2(&mut tx)).await
                    })
                    .and_then(|_| async move {
                        call_external_service2().await
                    })
                    .and_then(|_| async move {
                        tx.commit().await
                    })
            });

        match result {
            Ok(_) => break,
            Err(error) => println!("Error: {}", error)
        }

        tokio::time::sleep(Duration::from_millis(1_000)).await;
    }
}

Unfortunately, this code doesn't compile because of the following compiler error:

error[E0308]: mismatched types
  --> src/main.rs:28:35
   |
28 |                       .and_then(|_| async move {
   |  ___________________________________^
29 | |                         call_external_service().await
30 | |                     })
   | |_____________________^ expected `Result<_, Error>`, found `async` block
   |
   = note:       expected enum `Result<_, anyhow::Error>`
           found `async` block `{async block@src/main.rs:28:35: 28:45}`
help: try wrapping the expression in `Ok`
   |
28 ~                     .and_then(|_| Ok(async move {
29 |                         call_external_service().await
30 ~                     }))

Wrapping each call with Ok() leads to another compiler error:

error[E0382]: use of moved value: `tx`
  --> src/main.rs:34:31
   |
31 |                       .and_then(|_| Ok(async move {
   |                                 --- value moved into closure here
32 |                           run_db_query2(&mut tx).await
   |                                              -- variable moved due to use in closure
33 |                       }))
34 |                       .and_then(|_| Ok(async move {
   |  _______________________________^
35 | |                         tx.commit().await
36 | |                     }))
   | |______________________^ value used here after move
   |
   = note: move occurs because `tx` has type `Transaction<'_, Postgres>`, which does not implement the `Copy` trait

I managed to find suprisingly few Rust examples that use this reactive-style approach for multiple async operations, and none of them helped me to resolve compiler errors I described above.

I would be happy to hear what I do wrong and to receive a few suggestions on what to read on proper error management when calling a chain of async operations.
If the only way is checking each function result individually, then let it be, but I feel just wrong.

3 posts - 2 participants

Read full topic