What if a repository method must be called within the same transaction?

[frederikhors]

Hi dear and amazing @AngusGMorrison,

a doubt:

In the repository function create_author I need to call another repository method (example: check_if_author_name_is_already_taken(req.name())) and I need to call it within the same transaction I'm creating on line 50.

How to?

Plus the same method can be called not in a transaction too!

Should I have a new argument on the repository method?

An argument to indicate both a db connection or a db transaction?

Something like:

impl BlogRepository for Sqlite {
    async fn create_author(&self, req: &CreateAuthorRequest) -> Result<Author, CreateAuthorError> {}

    async fn check_if_author_name_is_already_taken(&self, db_conn_or_tx: ??? what type here ???, name: &str) -> Result<bool, String> {}
}

I'm asking for you help too, @TmLev.

[TmLev]

If we permit leaking transactions outside of repositories' scope, sqlx has a trait called Executor which is implemented for pools, connections, and transactions (some of them via deref):

async fn check_unique_constraint(exec: impl sqlx::Executor) {
    sqlx::query!("...").fetch(exec).await
}

check_unique_constraint(&pool).await;
check_unique_constraint(&mut *conn).await;
check_unique_constraint(&mut *tx).await;

If you want check_unique_constraint to control commits and rollbacks, you can replace Executor with Acquire:

async fn check_unique_constraint(conn: impl sqlx::Acquire) {
    let mut tx = conn.acquire().await?;
    sqlx::query!("...").fetch(&mut *tx).await
}

Now the hard part is to decide how you can provide a concrete type to use it in a trait's signature. Since generics are not allowed in object-safe traits, I can think of two approaches on the spot:

1. Use an associated type:

   trait Repo {
       type Conn;
   
       async fn check_unique_constraint(c: Self::Conn);
   }
   
   type MyRepo = Arc<dyn Repo<Conn = MyConn>>;
   type MyConn = Box<dyn sqlx::Connection>;

   Now, this kinda beats the purpose of a repo, since we have to explicitly depend on sqlx::Connection.

2. Use type erasure:

   trait Repo {
       async fn connect(&self) -> RepoConn;
       async fn check_unique_constraint(&self, c: RepoConn);
   }

   Where

   struct RepoConn(Box<dyn Any>);
   
   impl RepoConn {
       fn new<T: Any>(value: T) -> Self {
           Self(Box::new(value))
       }
   
       fn downcast<T: Any>(self) -> Option<Box<T>> {
           self.0.downcast().ok()
       }
   }

   So that implementation can give out connections and get them back.

   impl Repo for MyRepo {
       async fn connect(&self) -> RepoConn {
           let tx = self.pg_pool.begin().await.unwrap();
           RepoConn::new(tx)
       }
   
       async fn check_unique_constraint(&self, c: RepoConn) {
           let tx = c.downcast::<sqlx::Transaction>().expect("connection should be downcastable to `sqlx::Transaction`");
           // Use `tx`.
       }
   }

   Error handling omitted.

Granted, both of them are kinda suboptimal: the first one requires us to explicitly state associated type (making us depend on it), and the second loses compile-time safety (forcing us to do it at runtime via type erasure).

If we don't leak transactions outside of repositories, then we don't have this problem :)

Please correct me if I'm wrong or didn't think of another approach.

P.S. You can also do associated type with a trait bound, passing dyn RepoConn around, but I don't think it's useful (?) since you need concrete type in the implementation itself.

[frederikhors]

Wow, thanks @TmLev. This is a lot to digest for a newbie like me!

I asked a very similar thing on SO a while back and this was the response, what do you two think?

[frederikhors]

If we permit leaking transactions outside of repositories' scope

I am studying this problem in depth because it is the last thing I have left to solve and it is FUNDAMENTAL above all to have a rapid and easy development (also considering testability).

Let's say there is a service with the following methods:

• create_team()

• create_player()

• create_team_with_players()

and a repo with the same methods:

• create_team()

• create_player()

• create_team_with_players()

(note each repo method know how to "persist" each model which is in turn composed of other models (complex structs) so I cannot simply repeat the persisting logic everywhere because is a lot)

I can, in order:

1. from service/create_team_with_players()

   1. call repo/create_team_with_players()

      1. call repo/create_team()

      2. for each player call repo/create_player()

and as you can see I need to call each repo method within a transaction (and I'm still not leaking transaction outside of repositories' scope).

But since each of that repo method can be called from a service too I need to leak in the repo method signature the optional transaction.

So leaking outside the scope of the repositories is inevitable!

• domain/player/ports.rs:

pub trait PlayerRepository: Send + Sync + Clone + 'static {
    fn create_player(
        &self,
        db_conn: Option<&mut sqlx::PgConnection>,
        req: &CreatePlayerRequest,
    ) -> impl Future<Output = Result<Player, CreatePlayerError>> + Send;
}

Am I right?

One way that I think can avoid the leak is to create repository methods without db_conn: Option<&mut sqlx::PgConnection> and the same methods to implement in outbound/pg.rs that can be called with a db_conn: Option<&mut sqlx::PgConnection>, like this:

pub trait PlayerRepository: Send + Sync + Clone + 'static {
    fn create_player(
        &self,
        req: &CreatePlayerRequest,
    ) -> impl Future<Output = Result<Player, CreatePlayerError>> + Send;
}

pub struct Pg {
    pool: PgPool,
}

impl Pg {
    async fn create_player(
        &self,
        db_conn: Option<&mut sqlx::PgConnection>,
        req: &CreatePlayerRequest,
    ) -> Result<Player, CreatePlayerError>> {
        // Save player using db_conn or self.pool
    }
}

impl PlayerRepository for Pg {
    fn create_player(
        &self,
        req: &CreatePlayerRequest,
    ) -> impl Future<Output = Result<Player, CreatePlayerError>> + Send {
        self.create_player(None, req)
    }
}

What do you think?

And @AngusGMorrison, is everything ok? Are you there? Are you writing juicy new episodes? Give us a silent sign of your presence! LOL 🤣

[musjj]

IMO, repository methods should be simple functions for retrieving data from a store. Anything containing more complicated application logic should live in the service. I'm also new to this though so not 100% sure.

[ghistrash]

Hi @frederikhors !

I agree with the previous comment @musjj .. But going back to your code example..

impl BlogRepository for Sqlite {
    async fn create_author(&self, req: &CreateAuthorRequest) -> Result<Author, CreateAuthorError> {}

    async fn check_if_author_name_is_already_taken(&self, db_conn_or_tx: ??? what type here ???, name: &str) -> Result<bool, String> {}
}

Why not instead define a method interface that receives all the necessary context?

impl BlogRepository for Sqlite {
    async fn create_author(&self, req: &CreateAuthorRequest, name: &str) -> Result<Author, CreateAuthorError> {}
}

And implement additional checks inside the type impl block..

impl Sqlite {
    async fn check_if_author_name_is_already_taken(&self, name: &str) -> Result<bool, String> {}
}

Essentially, repository methods will have to do all the work necessary to complete a database-related unit of work.

Why is this not the first thing we think of?
Maybe because we are used to expecting such code at the level above -- the service (with several repository methods called)?

Maybe there are downsides that haven't occurred to me right now..

I have read your repository on the topic and the following pseudocode snippet:

https://github.com/frederikhors/rust-clean-architecture-with-db-transactions

// Start a new DB transaction now to use with the below methods

let transaction = [DONT_KNOW_HOW_PLEASE_START_A_NEW_DB_TRANSACTION]();

let team = self.repo.team_by_id(transaction, team_id).await?;

if !team.has_free_places() { return };

let mut player = self.repo.player_by_id(transaction, player_id).await?;

player.team_id = team.id;

let player = self.repo.player_update(player).await?;

Ok(player)

Given the following context: team_id, player_id, "player_new_name** -- we could define the following interface:

impl TeamPlayerRepository for Sqlite {
    async fn create_player_or_player_with_team_or_team(&self, team_id: &Id, player_id: &Id, player_new_name: &str) -> Result<Player, UpdatePlayerError> {}
}

Let's try to start with the code from your repository:

impl TeamPlayerRepository for Sqlite {
    async fn create_player_or_player_with_team_or_team(&self, team_id: &Id, player_id: &Id, player_new_name: &str) -> Result<Player, UpdatePlayerError> {
    // it is pseudocode..
    let conn = self.pool.get_connection().unwrap();
    let team = conn
        .query_team_by_id(&team_id)
        .await
        .unwrap();

    if let Some(team) = team {
        if team.missing_players == 0 {
            return Err(UpdatePlayerError::NoPlace("no place for your player!".to_string()));
        }
    }

    let obj = Player {
        id: player_id.to_owned(),
        name: player_new_name.to_owned(),
        team_id: team_id.to_owned(),
    };

    let tx = conn.transaction().unwrap();
    let res = tx.prepaer_and_execute_with_params("insert into players (id, name, team_id) values (:id, :name, :team_id)", 
        &obj).await.unwrap();
    tx.commit().await.unwrap();

    Ok(obj)
    }
}

You called this code mess -- I don't quite understand why you think so.. What confuses you?

• do you want to define other transaction boundaries? Lock the player row from potential changes?

• or that here we deal with both players and teams, which potentially would like to be separeted in two different repositories (which will cause the problem we are discussing -- sharing the transaction between 2 repositories)?

• the method body is large, potentially many repeated checks are performed?

What i think..

• there seem to be no obstacles to defining the necessary transaction boundaries? having a tx object -- we can configure/start as we need..

• I think, despite the fact that interaction with both players and teams in one repository may seem like a violation of the boundary of responsibility or something else.. -- this is more than adequate, since it naturally reflects the requirements of the real world: both entities are important for us at the same time -- Team, Player make up our aggregate (DDD) -- an element that defines the invariants of the our system consistency. Which of them will be AggregateRoot is either not so important or will depend on other factors not visible in the context under consideration..

• most of the helper methods/checks (probably most of them will require read-only access) can be implemented outside the TeamPlayerRepository interface -- only inside the impl block for the Sqlite type -- and used in the interface impl block.

Maybe I still didn't get you, ha-ha)
Basically.. my point is that the entire unit of work should/can happen within a single repository method (let it be large, interacting with several entities) -- but I don't have much experience, I can't say how this will happen in practice.

I'm currently working on a similar project and trying to implement @AngusGMorrison's great approach -- we'll see! =)

[frederikhors]

I was talking about using a transaction across many services. What's the best way.

[VCCICCV]

I can provide some solutions: Both of these two methods offer weak consistency and eventual consistency. Implementing them in Rust is too complex.

When saving an aggregate, store and send events with version and type. Other services subscribe to these events and determine the steps and status of the events through the version and type. However, this may lead to data inconsistency.
Outbox pattern. When saving an aggregate, store events with version and type in the database simultaneously. Publish these events to Kafka via Debezium. Other services subscribe to these events and determine the steps and status of the events through the version and type.