DB Transaction with Foreign Dependency(ies)

[mabdh]

Golang database/sql package has an easy-to-use interface to implement transcation. There are also a lot of packages that extend database/sql functionalities e.g. sqlx, gorm, goqu, pgx, etc. This discussion won't focus to any of that and should be sql-extension-agnostic enough to discuss. We can assume to use pure database/sql only since all extensions should have compatibility to that.

To implement a 'clean architecture', one should breakdown the code into some layers. Each layer should only be allowed to interact to the components one layer below or in the same layer.

We (in odpf) tend to name the layers to this

middleware > handler > service > repository

Transaction is an implementation specific that only exist in repository implementation. But not all storage system have or support transaction. Service layer, where the business logic happen, does not need to know whether the repository does transaction or not.

It is not that complicated to implement transaction in the code in repository layer. Once could make the dependencies between one repository and other and clubbed the each query/repository function into a single atomic task. However, it will become more challenging if an atomic task consists of interaction to DB (writing data to DB) and interaction to the foreign system (calling external API). Foreign system is any system outside the storage system that provide transaction. This writing covers how are the pattern to implement transactional DB implementation when there is a dependency to foreign system. In ODPF, there are some cases where we need this.

To ease our understanding, let's scope our problem to only solve the base scenario with 2 tasks. We need to wrap these 2 tasks into a transaction or these 2 tasks should be considered atomic (either all of them success or all of them are failing)

1. Write data to DB
2. Write data to External API

Let's define a hypothetical scenario. We have 2 domains, User and Audit. Everytime we create a new user, we need to push the data to external API (could be kafka or some HTTP API).

Each domain has its components.
User

1. User entity
2. UserService
3. UserRepository

Audit

1. Audit entity
2. AuditService
3. AuditClient

No Transaction

Here is the example code when we don't consider both tasks to be atomic, there is no transaction. Hence if user repository successfully write the data to DB and the external call failed, the data in DB and external audit system is inconsistent.

package first

import (
	"context"

	"github.com/mabdh/godbtx/store"
	"github.com/mabdh/godbtx/user"
)

type UserRepository struct {
	dbc *store.Client
}

func NewUserRepository(dbc *store.Client) *UserRepository {
	return &UserRepository{
		dbc: dbc,
	}
}

func (r *UserRepository) Create(ctx context.Context, u user.User) (string, error) {
	var id string
	err := r.dbc.DB.QueryRowContext(ctx, "INSERT INTO users (name,email) VALUES (?,?)", u.Name, u.Email).Scan(&id)
	if err != nil {
		return "", err
	}

	return id, nil
}

package audit

import "context"

type Service struct {
	client Client
}

func NewService(client Client) *Service {
	return &Service{
		client: client,
	}
}

func (s *Service) Create(ctx context.Context, a Audit) (string, error) {
	return s.client.Create(ctx, a)
}

package first

import (
	"context"

	"github.com/mabdh/godbtx/audit"
	"github.com/mabdh/godbtx/user"
)

type UserService struct {
	r            UserRepository
	auditService audit.Service
}

func NewUserService(r UserRepository, as audit.Service) *UserService {
	return &UserService{
		r:            r,
		auditService: as,
	}
}

func (s *UserService) Create(ctx context.Context, u user.User) (string, error) {
	id, err := s.r.Create(ctx, u)
	if err != nil {
		return "", err
	}

	if _, err := s.auditService.Create(ctx, audit.Audit{
		Action:   "create",
		Domain:   "user",
		DomainID: id,
	}); err != nil {
		return "", err
	}

	return id, nil
}

Here are the patterns to implement transaction

1. Post-process function

Transaction is initiated within repository layer and repository provide a post-process callback function that could execute any function on the layer above (service layer).

package second

import (
	"context"
	"database/sql"

	"github.com/mabdh/godbtx/store"
	"github.com/mabdh/godbtx/user"
)

type UserRepository struct {
	dbc *store.Client
}

func NewUserRepository(dbc *store.Client) *UserRepository {
	return &UserRepository{
		dbc: dbc,
	}
}

func (r *UserRepository) CreateWithFn(ctx context.Context, u user.User, postProcessFn func(id string) error) (string, error) {
	var id string

	if err := r.dbc.WithTransaction(ctx, func(tx *sql.Tx) error {
		if err := tx.QueryRowContext(ctx, "INSERT INTO users (name,email) VALUES (?,?)", u.Name, u.Email).Scan(&id); err != nil {
			return err
		}

		if err := postProcessFn(id); err != nil {
			return err
		}

		return nil
	}); err != nil {
		return "", err
	}

	return id, nil
}

package second

import (
	"context"

	"github.com/mabdh/godbtx/audit"
	"github.com/mabdh/godbtx/user"
)

type UserService struct {
	r            UserRepository
	auditService audit.Service
}

func NewUserService(r UserRepository, as audit.Service) *UserService {
	return &UserService{
		r:            r,
		auditService: as,
	}
}

func (s *UserService) Create(ctx context.Context, u user.User) (string, error) {
	return s.r.CreateWithFn(ctx, u, func(id string) error {
		if _, err := s.auditService.Create(ctx, audit.Audit{
			Action:   "create",
			Domain:   "user",
			DomainID: id,
		}); err != nil {
			return err
		}
		return nil
	})
}

Pros

1. Transaction management and execution is within the repository layer and hidden from the layer above. Service layer does not need to care about it but just need to care which line of code should be atomic. The function name could also be renamed to CreateAtomic to make it clearer.
2. Straightforward code and more readable

Cons

1. Calling CreateWithFn needs a post-process function that will be pass as a function closure argument. If we mock the repository, t is not straight forward to test function closure argument.

2. Transactor interface

package third

import (
	"context"

	"github.com/mabdh/godbtx/audit"
	"github.com/mabdh/godbtx/user"
)

type UserService struct {
	repo         UserRepository
	auditService audit.Service
}

func NewUserService(repo UserRepository, as audit.Service) *UserService {
	return &UserService{
		repo:         repo,
		auditService: as,
	}
}

func (s *UserService) Create(ctx context.Context, u user.User) (string, error) {
	ctxTx := s.repo.WithTransaction(ctx)

	id, err := s.Create(ctxTx, u)
	if err != nil {
		if err := s.repo.Rollback(ctxTx, err); err != nil {
			return "", err
		}
		return "", err
	}

	if _, err := s.auditService.Create(ctxTx, audit.Audit{
		Action:   "create",
		Domain:   "user",
		DomainID: id,
	}); err != nil {
		if err := s.repo.Rollback(ctxTx, err); err != nil {
			return "", err
		}
		return "", err
	}

	if err := s.repo.Commit(ctxTx); err != nil {
		return "", err
	}
	return id, nil
}

func (c *Client) SetContextWithTransaction(ctx context.Context) context.Context {
	tx, _ := c.DB.BeginTx(ctx, nil)
	return context.WithValue(ctx, transactionContextKey{}, tx)
}

func getContextFromTransaction(ctx context.Context) *sql.Tx {
	if tx, ok := ctx.Value(transactionContextKey{}).(*sql.Tx); !ok {
		return nil
	} else {
		return tx
	}
}

func (c *Client) Rollback(ctx context.Context) error {
	if tx := getContextFromTransaction(ctx); tx != nil {
		if err := tx.Rollback(); err != nil {
			return err
		}
		return nil
	}
	return errors.New("no transaction")
}

func (c *Client) Commit(ctx context.Context) error {
	if tx := getContextFromTransaction(ctx); tx != nil {
		if err := tx.Commit(); err != nil {
			return err
		}
		return nil
	}
	return errors.New("no transaction")
}

func (c *Client) GetDBFromCtx(ctx context.Context) (db QueryerContext) {
	db = c.DB
	if tx := getContextFromTransaction(ctx); tx != nil {
		db = tx
	}
	return db
}

Pros

1. More control to initiate and end transaction in service layer

Cons

1. Need to incorporate transaction management in business logic
2. More boilerplate code
3. Passing transaction via context would be risky. If the transaction took longer time rather than context timeout, the context has already been cancelled but transaction was not ended yet. It will left the transaction uncommited (probably). We can actually pass a context within transaction so transaction should be cancelled once context is cancelled but that is kinda make the cycle dependencies between context and transaction.

So What?

• I would like to gather some ideas if there are any other patterns to solve the same problem
• Figure out what would be the best approach to pick as ODPF standard (considering the use cases)
• Any other inputs are welcome

[krtkvrm]

For Post-process function, more cons:

• It will hold up the connection till the txn is done which may starve other processes of db conn
• We need to ensure that the external dependency being operated on in closure should be idempotent

[mabdh]

@krtkvrm agree with both of these. but aren't they also happened with transactor interface approach ?
(regardless the approach, both could happen with the scenario that has foreign dependencies)

[krtkvrm]

feel that its kinda covered in con point 3

[spy16]

why does it need to be idempotent?

[mabdh]

The worst case might be, transaction is started, data is written to the DB, external API is called, on the API side the data has been written but somehow network is down so client timed-out, transaction will be rolled back. In this case, DB has no new data but the external dependency has written the data.

[spy16]

+1 for the post/pre transaction function approach (Have used it with good results in many places)

[krtkvrm]

The other approach is 2PC, if the traffic is huge enough that we cannot do external call within txn block

[krtkvrm]

We can also think about outbox pattern

[mabdh]

@krtkvrm yes this is also one good approach