Help with adding a transaction to async bulk insert

[tjallingt]

I'm trying to insert many values into a database as quickly as possible.
These values are requested from an external API.

In order to efficiently request the data from the external API, I'm using futures::Stream to perform the many requests in parallel.
I then use the same stream to also efficiently (I believe) insert the data into the database.

I would also like to wrap this in a transaction so that the data is not updated when an error occurs, however when I try to use the Transaction I run into issues.

Please see the following example code:

fn example_stream() -> impl futures::Stream<Item = anyhow::Result<String>> {
    futures::stream::iter(std::iter::repeat_with(|| Ok(String::from("hello world"))))
}

async fn example_bulk_insert(
    conn: impl sqlx::MySqlExecutor<'_>,
    chunk: Vec<String>,
) -> anyhow::Result<()> {
    let mut query_builder = sqlx::QueryBuilder::new(
        r#"
            INSERT INTO test(value)
        "#,
    );

    query_builder.push_values(chunk, |mut b, value| {
        b.push_bind(value);
    });

    let query = query_builder.build();

    query.execute(conn).await.context("bulk insert failed")?;

    Ok(())
}

// Working example (without transaction)
async fn example_working(pool: MySqlPool) -> anyhow::Result<()> {
    example_stream()
        .try_chunks(1000)
        .map_err(|error| anyhow::anyhow!(error))
        .try_for_each(|chunk| example_bulk_insert(&pool, chunk))
        .await?;

  Ok(())
}

// Non working example (with transaction)
async fn example_not_working(pool: MySqlPool) -> anyhow::Result<()> {
    let mut transaction = pool.begin().await?;

    example_stream()
        .try_chunks(1000)
        .map_err(|error| anyhow::anyhow!(error))
        // error: captured variable cannot escape `FnMut` closure body
        .try_for_each(|chunk| example_bulk_insert(&mut *transaction, chunk))
        .await?;

    transaction.commit().await?;

    Ok(())
}

Is there a way to make this work with a transaction?
Do I need to restructure my code for this?