The code finally compiles!
The partial solution in the comment by @trueequalsfalse uses transmute() to turn the mutable reference to the diesel connection into a ConnectionToSpecificDbType without the necessity of adding a lifetime to ConnectionToSpecificDbType.
The transaction() function in diesel-async uses boxed futures and the async-trait crate to encapsulate the lifetimes of the transaction there.
I combined both approaches to wrap the diesel transaction into a generic transaction.
I also added a mock for the diesel connection at the end, so that the code can be executed without db.
For completeness, the reason I wanted to use generic connection is that I wanted to try out this guide on Hexagonal Architecture, but needed transactions in my Service impl, which were not covered in the guide.
# Cargo.toml
[dependencies]
tokio = { version = "1.0.0", features = ["rt", "rt-multi-thread", "macros"] }
diesel = { version = "2.2.4", features = ["postgres"] }
diesel-async = { version = "0.5.0", features = ["postgres"] }
async-trait = "0.1.83"
scoped-futures = "0.1.3"
All changes to the original code in the question are marked with ***.
#[deny(unsafe_code)] // *** added
use std::future::Future;
use scoped_futures::{ScopedBoxFuture, ScopedFutureExt}; // ** changed
use crate::diesel_async_mock::{AsyncConnection, AsyncPgConnection}; // *** changed to use diesel mock
// Business logic as async functions that can be shared between threads
trait Service: Send + Sync + 'static {
/// Update value of item if new value is larger than current value.
/// Example functino that needs multiple repo functions
fn maximize_item(&self, id: i32, value: i32) -> impl Future<Output=Result<i32, String>> + Send;
}
// Persist the data in e.g. a database
// The repository provides a new connection and expects that repo methods are called with
trait Repository: Send + Sync + 'static {
type Conn: Connection;
/// Connect to data storage, or throw error if attempt failed
fn connect(&self) -> impl Future<Output=Result<Self::Conn, String>> + Send;
/// Get current value for item, or throw error if item doesn't exist or value is not retrievable
fn get_item_value(&self, conn: &mut Self::Conn, id: i32) -> impl Future<Output=Result<i32, String>> + Send;
/// Update value for item, or throw error if item doesn't exist or can't be updated
fn update_item_value(&self, conn: &mut Self::Conn, id: i32, value: i32) -> impl Future<Output=Result<i32, String>> + Send;
}
// Connection to some data storage
#[async_trait::async_trait] // *** added
trait Connection: AsMut<Self::SpecificConn> + Send + Sync + 'static {
type SpecificConn: Send + Sync;
/// Provide a generic transaction as closure that is agnostic to the underlying db or ORM.
/// Commit if closure returns Ok`, rollback on `Err`
// THe lifetimes are wrong, and possibly more, but I don't know how to fix this method
async fn transaction<'a, R, F>(&mut self, callback: F) -> Result<R, String> // *** changed
where
F: for<'r> FnOnce(&'r mut Self) -> ScopedBoxFuture<'a, 'r, Result<R, String>> + Send + 'a, // *** changed
R: Send + 'a; // *** changed
}
struct ServiceImpl<R: Repository> {
repo: R, // service holds a generic repository to persist date
}
// implementation for Service while keeping the repository generic
impl<R: Repository> Service for ServiceImpl<R> {
// The service methods sometimes need to call multiple repository functions, but need
// to ensure this is done as an atomic transaction. Instead of writing business logic to roll
// back failed attempts on my own, I rather want to use a database transaction for this
async fn maximize_item(&self, id: i32, value: i32) -> Result<i32, String> {
// get new db connection wrapped into a generic object
let mut conn = self.repo.connect().await?;
// wrap repository tasks into an atomic but generic transaction. The repo needs
// to ensure a transaction is atomic
conn.transaction(|conn: &mut R::Conn| async move {
// get value and update value are different methods, but
// should be executed as atomic transaction here
let current = self.repo.get_item_value(conn, id).await?;
if value > current {
let updated = self.repo.update_item_value(conn, id, value).await?;
println!("Maximized item [{id}] from {current} to {updated}"); // *** added
Ok::<i32, String>(updated)
} else {
println!("Kept item [{id}] at {current}, because {value} is smaller"); // *** added
Ok::<i32, String>(current)
}
}.scope_boxed()).await // *** added scope_boxed
}
}
struct RepositoryForSpecificDbType;
// implement the repository for a specific database and ORM (or a mock for testing).
// In this case it is Postgres with Diesel
impl Repository for RepositoryForSpecificDbType { // *** lifetime removed
type Conn = ConnectionToSpecificDbType; // *** lifetime removed
// wrap diesel connection to postgres db into a generic connection object
async fn connect(&self) -> Result<ConnectionToSpecificDbType, String> {
let dsn = "postgres:://root:****@localhost:my_db";
let conn = AsyncPgConnection::establish(dsn).await.map_err(|_| "Connection failed".to_string())?;
Ok(ConnectionToSpecificDbType::new(conn))
}
async fn get_item_value(&self, _conn: &mut Self::Conn, id: i32) -> Result<i32, String> {
// mocking some database lookup
Ok(id * 2)
}
async fn update_item_value(&self, _conn: &mut Self::Conn, _id: i32, value: i32) -> Result<i32, String> {
// mocking some database update
Ok(value)
}
}
#[repr(transparent)] // *** added
struct ConnectionToSpecificDbType(AsyncPgConnection); // *** struct def changed
impl ConnectionToSpecificDbType { // *** lifetime removed
fn new(conn: AsyncPgConnection) -> Self { // *** function changed
Self(conn)
}
fn new_from_mut(conn: &mut AsyncPgConnection) -> &mut Self { // *** function changed
#[allow(unsafe_code)]
unsafe {
// Transmutes a diesel connection into a ConnectionToSpecificDbType.
// Source and result are both mutable references, lifetime of source is copied (I guess)
// SAFETY: transparent repr is used for Self
core::mem::transmute::<&mut AsyncPgConnection, &mut Self>(conn)
}
}
}
// get mutable reference to connection
impl AsMut<AsyncPgConnection> for ConnectionToSpecificDbType { // *** lifetime removed
fn as_mut(&mut self) -> &mut AsyncPgConnection { // *** function changed
&mut self.0
}
}
#[derive(Debug)]
enum TransactionError<E: Send + Sync> {
UserCaused(E),
#[allow(dead_code)] // *** added: DieselCaused is not used currently
DieselCaused(diesel::result::Error),
}
// A diesel transaction requires that a closure error implements `From<Error>`.
// To stay generic, handle errors in the closure and translate them into a generic error
impl<E: Send + Sync> From<diesel::result::Error> for TransactionError<E> {
fn from(value: diesel::result::Error) -> Self {
TransactionError::DieselCaused(value)
}
}
#[async_trait::async_trait] // *** added
impl Connection for ConnectionToSpecificDbType { // *** lifetime removed
type SpecificConn = AsyncPgConnection;
// This function doesn't work
// it should provide a generic abstraction of a transaction
async fn transaction<'a, R, F>(&mut self, callback: F) -> Result<R, String> // *** changed
where
F: for<'r> FnOnce(&'r mut Self) -> ScopedBoxFuture<'a, 'r, Result<R, String>> + Send + 'a, // *** changed
R: Send + 'a, // *** changed
{
// *** inner_conn def removed
let result = self.as_mut().transaction(|diesel_conn| async move {
println!("Entered transaction"); // *** added
// wrap mut ref to diesel connection into connection object that implements Connection
let inner_conn: &mut ConnectionToSpecificDbType = ConnectionToSpecificDbType::new_from_mut(diesel_conn); // *** changed
// call closure
let result = callback(inner_conn).await; // *** changed
// translate into generic intermediate error type that implements From<Error>
result.map_err(|e| TransactionError::UserCaused(e))
}.scope_boxed()).await;
println!("Left transaction"); // *** added
// return result
result.map_err(|e| format!("{e:?}").to_string())
}
}
mod diesel_async_mock { // *** mod added
use diesel::ConnectionResult;
use scoped_futures::ScopedBoxFuture;
pub struct AsyncPgConnection;
#[async_trait::async_trait]
pub trait AsyncConnection: Sized + Send {
async fn establish(database_url: &str) -> ConnectionResult<Self>;
async fn transaction<'a, R, E, F>(&mut self, callback: F) -> Result<R, E>
where
F: for<'r> FnOnce(&'r mut Self) -> ScopedBoxFuture<'a, 'r, Result<R, E>> + Send + 'a,
E: From<diesel::result::Error> + Send + 'a,
R: Send + 'a,
{
callback(self).await
}
}
#[async_trait::async_trait]
impl AsyncConnection for AsyncPgConnection {
async fn establish(_database_url: &str) -> ConnectionResult<Self> {
Ok(Self)
}
}
}
#[tokio::main]
async fn main() {
// create a new service instance and define which repository implementation is used
let service = ServiceImpl {
repo: RepositoryForSpecificDbType
};
assert_eq!(service.maximize_item(5, 3).await.unwrap(), 10);
assert_eq!(service.maximize_item(5, 8).await.unwrap(), 10);
assert_eq!(service.maximize_item(5, 13).await.unwrap(), 13);
assert_eq!(service.maximize_item(5, 21).await.unwrap(), 21);
}
Output:
Entered transaction
Kept item [5] at 10, because 3 is smaller
Left transaction
Entered transaction
Kept item [5] at 10, because 8 is smaller
Left transaction
Entered transaction
Maximized item [5] from 10 to 13
Left transaction
Entered transaction
Maximized item [5] from 10 to 21
Left transaction
