Store jobs in a transactional way

[enerdgumen]

Feature request: when using postgres as storage, it could be useful to be able to insert new jobs using the same applicative transaction, to ensure consistency.

Currently the apalis storage own the connection pool and there is no way to pass a connection or a transaction to the insert method. The only way to achieve this is to manually insert the job, bypassing the apalis storage.

[geofmureithi]

Sure, this can be considered in a future release.
But a workaround exists
Postgres offers a way to get the underlying pool:
https://docs.rs/apalis-sql/latest/apalis_sql/postgres/struct.PostgresStorage.html#method.pool

So something like:

trait PushWithTx {
    async fn push_with_tx(&self, job: Self::Job);
}

impl PushWithTx for PostgresStorage {
    // .....
}

[enerdgumen]

Thanks for considering this.

I didn't understand the possible workaround. If I get the pool used by apalis, I cannot push a job using the same connection used by my application, because apalis takes a new connection from the pool on every push.

[geofmureithi]

What do you mean by takes a new connection from...?
If you have a pool, you dont have to create a connection, you can create a new storage with the same pool.
Unless I dont understand you correctly.

[enerdgumen]

I mean, when apalis insert a new job, it passes the pool reference to sqlx and then sqlx runs the statement taking a connection from the pool. So the client can't control the connection used by apalis.
This is important when working with transactions because the transaction is bound to a specific connection.
So if the application already has a transaction and wants to insert a new job, the application should pass the specific connection or the related transaction to the apalis storage.

[geofmureithi]

I am not sure how to help here, but I think you should be able to use sqlx transactions using begin. I am not the biggest sql expert but I thought this should work. Am I missing something?