When executing async database inserts how do you ensure the query data being inserted lives long enough?

user_13991179 0
Viewed 292

I am using these crates: tokio-postgres, rust_decimal, chrono and bb8 (connection pooling library).

I give the quick summary up top here and then a detailed explanation underneath. Basically with my query data:

let data_array = [
    &data.id,
    &data.currency_id,
    &data.price,
    &data.date_time,
];

I put it into a future query:

let the_query = client.execute(&statement, &data_array)
query_futures.push(the_query);

Then later on I do a futures::future::join_all(query_futures).await; which executes all the inserts together. But this means the data may not live long enough, causing the well known borrowed value does not live long enough error.

How can I fix this? I have tried many things!

Here is the full code:

#[derive(Clone)]
struct MyStruct {
id: i32,
currency_id: i64,
price: Decimal,
date_time: chrono::NaiveDateTime,
}

let mut query_futures = Vec::with_capacity(trades.len());

while (some_condition) {

    //these three lines work fine:
    let pool = pg_connection.clone().connection_pool.unwrap().clone();
    let client = pool.get().await.unwrap();
    let statement = client
    .prepare("INSERT INTO table (id, currency_id, price, date_time) VALUES ($1, $2, $3, $4").await.unwrap();

    //this for loop batches together the query futures into a single future
    for (another_condition) {

        let currency_id = 34 as i64;
        let price = Decimal::new(0, 3);
        let mut data  = MyStruct {
            id: 23 as i32,
            currency_id: currency_id,
            price: price,
            date_time: NaiveDateTime::from_timestamp(
                SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_secs() as i64,
                0,
            )
        }

     //HERE IS WHERE I HAVE A PROBLEM:

        let data_array = [
            &data.id as &(dyn ToSql + Sync),
            &data.currency_id,
            &data.price,
            &data.date_time,
        ];

        let the_query = client.execute(&statement, &data_array);
        query_futures.push(the_query);

    }
    let queries = futures::future::join_all(query_futures);
    queries.await;
}

Note I use the as &(dyn ToSql + Sync) trait to ensure each item of the slice is of the same type.

rust
1 Answers

One way to solve this is with an async move closure. That will take ownership of the parameters you're trying to pass lend into the array, which can then be lent to execute:

let data = MyStruct { ... };   
// Need to explicitly borrow `statement` so that the
// future doesn't try to take ownership of it
let statement = &statement;

query_futures.push(async move {
     let data_array = [
        &data.id as &(dyn ToSql + Sync),
        &data.currency_id,
        &data.price,
        &data.date_time,
    ];

    client.execute(statement, &data_array).await
})

It's tangential to your question, but unless you need these insertions to specifically be ordered, I'm going to recommend you use FuturesUnordered instead of join_all to handle the concurrency for these tasks.

user_8644060
Related