79073140

Date: 2024-10-10 07:14:03
Score: 3
Natty:
Report link

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
Reasons:
  • Blacklisted phrase (1): this guide
  • RegEx Blacklisted phrase (2): I don't know how to fix
  • Long answer (-1):
  • Has code block (-0.5):
  • Self-answer (0.5):
  • Low reputation (1):
Posted by: Skru