DEV Community: yedf2

Use workflow to handle distributed transactions

yedf2 — Thu, 28 Jul 2022 12:13:00 +0000

In the world of microservices a transaction is now distributed to multiple services that are called in a sequence to complete the entire transaction.

With the advent of microservice architecture we are losing the ACID nature of databases. Transactions may now span multiple microservices and therefore databases.

To address the problems of distributed transactions, the following list of approaches have been described:

Two-Phase Commit / XA
Eventual Consistency and Compensation / SAGA
Try, Confirm, Cancel / TCC.

In classic solutions, the developers should choose one of the three patterns to handle distributed transactions.

But in this article, we introduce a workflow pattern in github.com/dtm-labs/dtm. Under this pattern, a mixture of XA, SAGA and TCC can be applied to different branches in a single distributed transactions, allowing users to customize most of the contents of a distributed transaction, providing great flexibility.

workflow example

In Workflow mode of DTM, both HTTP and gRPC protocols can be used. The following is an example of the gRPC protocol, which is divided into the following steps.

Initialize the SDK
Register workflow
Execute workflow

Initialize the SDK

First you need to Initialize the SDK's workflow before you can use it.

import "github.com/dtm-labs/dtmgrpc/workflow"

// Initialize the workflow SDK with three parameters.
// the first parameter, the dtm server address
// the second parameter, the business server address
// The third parameter, grpcServer
// workflow needs to receive the dtm server callback from the "business server address" + "grpcServer"
workflow.InitGrpc(dtmGrpcServer, busi.BusiGrpc, gsvr)

Register workflow

Then you need to register workflow's handler function. The following is a saga workflow to do cross-bank transfer.

wfName := "wf_saga"
err := workflow.Register(wfName, func(wf *workflow.Workflow, data []byte) error {
  req := MustUnmarshalReqGrpc(data)
  wf.NewBranch().OnRollback(func(bb *dtmcli.BranchBarrier) error {
    _, err := busi.BusiCli.TransOutRevert(wf.Context, req)
    return err
  })
  _, err := busi.BusiCli.TransOut(wf.Context, req)
  if err ! = nil {
    return err
  }
  wf.NewBranch().OnRollback(func(bb *dtmcli.BranchBarrier) error {
    _, err := busi.BusiCli.TransInRevert(wf.Context, req)
    return err
  })
  _, err = busi.BusiCli.TransIn(wf.Context, req)
  return err
})

This registration operation needs to be executed after the business service is started, because when the process crashes, dtm will call back to the business server to continue the unfinished task
The above code NewBranch will create a transaction branch, one that will include a forward action and a callback on global transaction commit/rollback
OnRollback/OnCommit will register a callback on global transaction rollback/commit for the current transaction branch, in the above code, only OnRollback is specified, so it is in Saga mode
The busi.BusiCli in the above code needs to add a workflow interceptor which will automatically record the results of the rpc request to dtm as follows

conn1, err := grpc.Dial(busi.BusiGrpc, grpc.WithUnaryInterceptor(workflow.Interceptor), nossl)
busi.BusiCli = busi.NewBusiClient(conn1)

You can of course add workflow.Interceptor to all gRPC clients, this middleware will only handle requests under wf.Context and wf.NewBranchContext()

When the workflow function returns nil/ErrFailure, the global transaction enters the Commit/Rollback phase, calling the operations registered in OnCommit/OnRollback inside the function in reverse order

Execute workflow

Finally the workflow is executed

req := &busi.ReqGrpc{Amount: 30}
err = workflow.Execute(wfName, shortuuid.New(), dtmgimp.MustProtoMarshal(req))

When the result of Execute is nil/ErrFailure, the global transaction has succeeded/been rolled back.
When the result of Execute is other values, the dtm server will subsequently call back this workflow task to retry

Principle of workflow

How does workflow ensure data consistency in distributed transactions? When a business process has a crash or other problem, the dtm server will find that this workflow global transaction has timed out and not completed, then the dtm server will use an exponential retreat algorithm and retry the workflow transaction. When the workflow retry request reaches the business service, the SDK will query the progress of the global transaction from the dtm server, and for the completed branch, it will take the previously saved result and return the branch result directly through an interceptor such as gRPC/HTTP. Eventually the workflow will complete successfully.

Workflow functions need to be idempotent, i.e. the first call, or subsequent retries, should get the same result

Saga in Workflow

The core idea of the Saga pattern, derived from this paper SAGAS, is that long transactions are split into short transactions, coordinated by the Saga transaction coordinator, and if each short transaction operation successfully completes, then the global transaction completes normally, and if a step fails, the compensating operations are invoked one at a time in reverse order.

In Workflow mode, you can call the function for the operation directly in the function and then write the compensation operation to OnRollback of the branch, and then the compensation operation will be called automatically, achieving the effect of Saga mode

Tcc under Workflow

The Tcc pattern is derived from this paper Life beyond Distributed Transactions: an Apostate's Opinion, he divides a large transaction into multiple smaller transactions, each of which has three operations.

Try phase: attempts to execute, completes all business checks, set aside enough business resources
Confirm phase: if the Try operation succeeds on all branches, it goes to the Confirm phase, which actually executes the transaction without any business checks, using only the business resources set aside in the Try phase
Cancel phase: If one of the Try operations from all branches fails, we go to the Cancel phase, which releases the business resources reserved in the Try phase.

For our scenario of an inter-bank transfer from A to B, if SAGA is used and the balance is adjusted in the forward operation, and is adjusted reversely in the compensating operation, then the following scenario would occur.

A deducts the money successfully
A sees the balance decrease and tells B
The transfer of the amount to B fails and the whole transaction is rolled back
B never receives the funds

This causes great distress to both ABs. This situation cannot be avoided in SAGA, but it can be resolved by TCC with the following design technique.

Introduce a trading_balance field in addition to the balance field in the account
Try phase to check if the account is frozen, check if the account balance-trading_balance is sufficient, and then adjust the trading_balance (i.e. the funds frozen for business purposes)
Confirm phase, adjust balance, adjust trading_balance (i.e. unfrozen funds for the business)
Cancel phase, adjust trading_balance (i.e. unfrozen funds on the business)

In this case, once end user A sees his balance deducted, then B must be able to receive the funds

In Workflow mode, you can call the Try operation directly in the function, then register the Confirm operation to OnCommit in the branch and register the Cancel operation to OnRollback in the branch, achieving the effect of the Tcc mode

XA under Workflow

XA is a specification for distributed transactions proposed by the X/Open organization. The XA specification essentially defines the interface between a (global) Transaction Manager (TM) and a (local) Resource Manager (RM). Local databases such as mysql play the RM role in the XA

XA is divided into two phases.

Phase 1 (prepare): i.e. all participants RM prepare to execute the transaction and lock the required resources. When the participants are ready, they report to TM that they are ready.
Phase 2 (commit/rollback): When the transaction manager (TM) confirms that all participants (RMs) are ready, it sends a commit command to all participants.

Currently all major databases support XA transactions, including mysql, oracle, sqlserver, postgres

In Workflow mode, you can call NewBranch().DoXa in the workflow function to open your XA transaction branch.

Mixing multiple modes

In Workflow mode, Saga, Tcc and XA as described above are all patterns of branching transactions, so you can use one pattern for some branches and another pattern for others. The flexibility offered by this mixture of patterns allows for sub-patterns to be chosen according to the characteristics of the branch transaction, so the following is recommended.

XA: If the business has no row lock contention, and the global transaction will not last long, then XA can be used. This pattern requires less additional development and Commit/Rollback is done automatically by the database. For example, this pattern is suitable for order creation business, where different orders lock different order rows and have no effect on concurrency between each other; it is not suitable for deducting inventory, because orders involving the same item will all compete for the row lock of this item, which will lead to low concurrency.
Saga: common business that is not suitable for XA can use this model, this model has less extra development than Tcc, only need to develop forward operation and compensation operation
Tcc: suitable for high consistency requirements, such as the transfer described earlier, this pattern has the most additional development and requires the development of operations including Try/Confirm/Cancel

gRPC support

Support for gRPC, consists of two aspects:

communicating with the dtm server by gRPC protocol, e.g. in the above example, after calling workflow.InitGrpc above to initialize, the dtm SDK will take the gRPC interface to interact with the dtm server
In the above example, reply, err = busi.BusiCli.TransIn(wf.Context, req), this gRPC call will automatically save the result of the call to the dtm server through the gRPC interceptor. When workflow is executed again, the result saved by the dtm server will be returned directly to the caller.

HTTP support

Support for HTTP, consists of two aspects:

Communicating with the dtm server by means of the HTTP protocol. After initialization of workflow.InitHTTP, the dtm SDK will take the HTTP interface to interact with the dtm server.
Accessing the transaction branch by HTTP protocol, e.g. resp, err := wf.NewBranch().NewRequest().SetBody(req).Post(Busi + "/TransOut"), in this HTTP call, the result of the call will be automatically saved to the dtm server and, when workflow is executed again, the result saved by the dtm server is automatically returned directly to the caller.

Local Operation Support

In Workflow mode, your transaction branch, not only can use remote branches such as HTTP/gRPC, but can also be a local operation. The following code demonstrates a local transactional operation.

wf.Do(func(bb *dtmcli.BranchBarrier) ([]byte, error) {
  return nil, bb.CallWithDB(dbGet(), func(tx *sql.Tx) error {
    return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "")
  })
})

You can perform local transaction operations as well as other operations, as flexibly as possible.

gRPC/HTTP/Local Mixed Usage

Within a distributed transaction, you can also use a mix of gRPC/HTTP/local. These approaches to your transaction branch can give you a great deal of flexibility. It provides a very good solution to introducing distributed transactions for multiple technology stacks, for various scenarios such as the coexistence of multiple protocols, and for existing legacy systems.

Idempotency Requirements

In the Workflow pattern, when a crash occurs, a retry is performed, so the individual operations are required to support idempotency, i.e. the result of the first call is the same as the next tries, returning the same result. In business, the unique key of the database is usually used to achieve idempotency, specifically insert ignore "unique-key", if the insert fails, it means that this operation has been completed, this time directly ignored to return; if the insert succeeds, it means that this is the first operation, continue with the subsequent business operations.

If your business itself is idempotent, then you can operate your business directly; if your business does not provides idempotent functionality, then dtm provides a BranchBarrier helper class, based on the above unique-key principle, which can easily help developers implement idempotent operations for Mysql/Mongo/Redis.

Please note that the following two are typical non-idempotent operations.

Timeout rollback: If you have an operation in your business that may take a long time, and you want your global transaction to roll back after waiting for the timeout to return a failure. Then this is not an idempotent operation because in the extreme case of two processes calling the operation at the same time, one returns a timeout failure and the other a success, resulting in different results
Rollback after reaching the retry limit: the analysis process is the same as above.

Workflow mode does not support the above timeout rollback and retry limit rollback at the moment, if you have a relevant scenario, please send us the specific scenario, we will actively consider whether to add this kind of support

Branch Operation Results

Branching operations will return the following results.

Success: the branch operation returns HTTP-200/gRPC-nil
Business failure: the branch operation returns HTTP-409/gRPC-Aborted, no more retries, and the global transaction needs to be rolled back
In progress: branch operation returns HTTP-425/gRPC-FailPrecondition, this result indicates that the transaction is in progress normally and requires the dtm to retry not with the exponential retreat algorithm but with fixed interval retries
Unknown error: the branch operation returns other results, indicating an unknown error, and dtm will retry this workflow, using the exponential retreat algorithm

If your existing service has a different result to the above, then you can customise this part of the result with workflow.Options.HTTPResp2DtmError/GRPCError2DtmError

Saga's Compensation operation and Tcc's Confirm/Cancel operation are not allowed to return business failures according to the Saga and Tcc protocols, because when in the second stage of the workflow, Commit/Rollback, is neither successful nor allowed to retry, then the global transaction cannot be completed, so please take care to avoid this when designing

Transaction Completion Notification

Some business scenarios where you want to be notified of the completion of a transaction can be achieved by setting an OnFinish callback on the first transaction branch. By the time the callback is called, all business operations have been performed and the global transaction is therefore substantially complete. The callback function can determine whether the global transaction has finally been committed or rolled back based on the isCommit passed in.

One thing to note is that when the OnFinish callback is called, the state of the transaction has not yet been modified to final state on the dtm server, so if you use a mixture of transaction completion notifications and querying global transaction results, the results of the two may not be consistent, and it is recommended that users use only one of these methods rather than a mixture.

Conclusion

We introduce a workflow pattern to support the mixed usage of Saga, XA and Tcc. This pattern also support HTTP, gRPC and local transactions.

We also illustrate the usage of workflow in Golang with runnable examples.

You are welcome to visit https://github.com/dtm-labs/dtm

How to elegantly implement a multi-database outbox pattern

yedf2 — Tue, 12 Jul 2022 08:00:52 +0000

Introduction to the Outbox pattern

A microservice may need to perform two steps, "save data" and "send events". For example, after publishing an article, the author's posting statistics need to be updated. The business requirement is that both operations fail or succeed at the same time, rather than one succeeding and one failing. If the article is eventually published and the update of posting statistics fails, the data will be inconsistent.

The outbox pattern is the most common pattern used to solve this problem and works as follows.

the local business runs as a transaction, writing events to the message table before committing the transaction; when the transaction is committed, it commits both the business, and the events
the events are sent to the message queue by polling the message table or listening to the binlog
- Polling: retrieve events from the message table every 1s or 0.2s, send them to the message queue, and then delete them
- Listening to the binlog: using a database tool such as Debezium, listen to the database binlog, fetch the events and send them to the message queue
Write consumers and process events

As in 1, the business and the event are updated in the same transaction, ensuring that both will be committed at the same time.
In steps 2,3, both are operations that will not fail and will be retried and eventually succeed if a downtime event occurs in between, etc.

For the aforementioned post-commit statistics scenario, the above solution ensures that the statistics are finally updated and the data will reach eventual consistency

Problems with multiple databases

With today's popular microservices architecture, a single database is usually used for a microservice. When multiple services need to use the outbox model, then the traditional outbox architecture is more difficult to maintain.

Polling for events: multiple database polling tasks need to be written in the polling process
Listening to binlog for events: you need to listen to multiple database binlogs

Both of the above methods of fetching events are not very maintainable when dealing with a large number of databases. The architecture is not very resilient and if there are many databases and few events generated at any one time, the load on the architecture will be high and resources will be wasted. The ideal architecture load is one that is only related to the number of events sent, and not to other factors.

Solution

The open source distributed transaction framework https://github.com/dtm-labs/dtm has a two-stage message pattern inside that handles this problem very well. The following is an example of the use of an interbank transfer operation.


 Go
msg := dtmcli.NewMsg(DtmServer, gid).
    Add(busi.Busi+"/TransIn", &TransReq{Amount: 30})
err := msg.DoAndSubmitDB(busi.Busi+"/QueryPreparedB", db, func(tx *sql.Tx) error {
    return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "SUCCESS")
})

In this part of the code

First generate a msg global transaction for the DTM, passing the server address of the DTM and the global transaction id
Add a branch business logic to the msg, here the business logic is the transfer operation TransIn
Then call msg's DoAndSubmitDB, a function that ensures that the business is executed successfully and the msg global transaction is committed, either succeeding or failing at the same time
1. the first parameter for the check-back URL, the detailed meaning will be described later
2. the second parameter is sql.DB, which is the database object accessed by the business
3. the third parameter is the business function, the business in our example is to deduct $30 from the balance of A

Success Flow

How does DoAndSubmitDB ensure the atomicity of successful business execution and msg submission? Consider the following timing diagram.

In general, the 5 steps in the timing diagram will complete normally, the whole business proceeds as expected and the global transaction completes. There is a new element that needs to be explained here, which is that the commit of a msg is initiated in two phases, the first phase calls Prepare and the second phase calls Commit. when DTM receives the Prepare call, it does not call the branch transaction, but waits for the subsequent Submit. only when it receives the Submit, it starts the branch call and finally completes the global transaction.

Exceptions

In a distributed system, all types of downtime and network exceptions need to be considered, so let's look at the following possible problems.

First of all the most important goal we want to achieve is that the business executes successfully and the msg transaction is an atomic operation, so what if in the previous timing diagram, after the Prepare message is sent successfully and before the Submit message is sent successfully, what happens if there is an abnormal downtime? At this point dtm will detect that the transaction has timed out and will check back. For developers, this check-back is as simple as pasting in the following code.


 Go
app.GET(BusiAPI+"/QueryPreparedB", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).QueryPrepared(dbGet())
}))

If you are using something other than the go framework gin, then you will need to make some minor modifications to suit your framework, but the code is generic and suitable for each of your businesses.

The main principle of check-back is mainly through the message table, but dtm's check-back has been carefully proved to be able to handle the following situations.

The local transaction has not started at the time of the check-back
The local transaction is still in progress at the time of the check-back
The local transaction has been rolled back at the time of the check-back
The local transaction has been committed t the time of the check-back

The detailed check-back principle is somewhat complex and has been patented, so it will not be described in detail here, for more information you can refer to https://en.dtm.pub/practice/msg.html

Multi-database support

With this solution, if you need to handle multiple databases, all you need to do is create the event tables for each database, and pass in different database connections where you check back.

Compared to the original polling table and listening binlog solution, the cost of operation and maintenance is greatly reduced. The load of the architecture is only related to the number of events and not to other factors such as the number of databases, making it more resilient.

More storage engine support

dtm's two-stage messages provide not only database support for DoAndSubmitDB, but also NoSQL support

Mongo support

The following code ensures that both business and messages are committed simultaneously under Mongo


 Go
err := msg.DoAndSubmit(busi.Busi+"/RedisQueryPrepared", func(bb *dtmcli.BranchBarrier) error {
    return bb.MongoCall(MongoGet(), func(sc mongo.SessionContext) error {
        return SagaMongoAdjustBalance(sc, sc.Client(), TransOutUID, -reqFrom(c).Amount, reqFrom(c).TransOutResult)
    })
})

Redis support

The following code ensures that both the business and the message are committed simultaneously under Redis


 Go
err := msg.DoAndSubmit(busi.Busi+"/RedisQueryPrepared", func(bb *dtmcli.BranchBarrier) error {
    return bb.RedisCheckAdjustAmount(busi.RedisGet(), busi.GetRedisAccountKey(busi.TransOutUID), -30, 86400)
})


```''
The dtm check-back pattern can easily be extended to a wide variety of other transaction-enabled storage engines

## Solution features
The following features are available under two-phase messaging.
- Elegant support for multiple databases
- Support for not only SQL databases, but also NoSQL such as Mongo and Redis
- Short code, significantly less code than the usual outbox pattern
- The entire architecture and development process does not involve message queues, only api, making it easier to get started
- Load is only related to the volume of messages, not the number of databases involved

## Compare to RocketMQ transactional messages
Check-back was first proposed in RocketMQ transaction messages, but I has searched for examples of check-backs and various case studies, but has not found a check-back solution that handles all kinds of exceptions well. None of the solutions found correctly handle the "local transaction is still in progress" scenario, and all have corner cases that lead to inconsistent data, see [https://en.dtm.pub/practice/msg.html](https://en.dtm.pub/practice/msg.html) for more information .

In addition, dtm's two-stage messages do not require the introduction of a queue, or can be used in conjunction with other message queues, so they are more widely available

## Summary
The dtm two-stage messaging presented in this article is better suited to multi-database situations. The architecture solution, with its many advantages, is a perfect alternative to the outbox pattern and gives developers a simpler and easier to use architecture.

You are welcomed to visit [https://github.com/dtm-labs/dtm](https://github.com/dtm-labs/dtm)

Three Pitfalls and Best Practice for Saga Pattern in Microservices

yedf2 — Thu, 26 May 2022 03:15:28 +0000

Saga is the most common pattern for ensuring cross-service data consistency. It splits a global transaction into multiple sub-transactions, and if all sub-transaction operations succeed, then the global transaction succeeds, and if one sub-transaction still fails after retries, then all executed sub-transactions are compensated.

The above process is simple and clear, but in a distributed environment, there are anomalies that can occur. It is not an easy job to ensure that a Saga transaction is working correctly and to keep the data in consistent. We will discuss the various anomalies that can occur here and how to manage them.

For the purposes of discussion, we will refer to the forward action of a branch transaction in Saga as Action and the compensating action as Compensation.

Three Types of Anomalies

Normally, the Action is executed before the Compensation in Saga transactions if rollback happens. But due to network delays, or process pauses, it is possible that the Compensation is executed first and the Action is executed later or not at all.

This scenarios introduces two anomalies in distributed transactions:

Null Compensation: When Compensation is executing, the corresponding Action has not been executed, so the Compensation needs to determine that the Action has not been executed, ignore the business data updates, and return directly.
Hanging Action: When the Action is executing, the Compensation has been executed, so the Action needs to determine that the Compensation has been executed, ignore the business data updates, and return directly

Another common anomaly that needs to be dealt with in distributed transactions is Duplicated Requests. When a process crashes and Saga retries the service, the service may be called multiple times, so idempotent processing is needed.

These three types of anomalies need to be managed carefully, otherwise data inconsistencies can occur and disrupt the business.

Best Practice

https://github.com/dtm-labs/dtm pioneered a technique name Sub-transaction Barrier for managing these three types of anomalies at the same time. For Saga pattern, the principle is as follows.

Create a table dtm_barrier.barrier in the local database, with a unique index of gid-branchid-branchop
Begin a local transaction
In Action and Compensation, insert ignore a row gid-branchid-action|compensation, if the number of affected rows is 0 (in case of Duplicate Requests, Hanging Action), commit directly and return
In Compensation, insert ignore an additional row gid-branchid-action, if the number of rows affected is 1 (in case of Null Compensation), commit directly and return
Execute business logic and commit, or roll back if errors occur

Duplicated Requests

Because of the unique index above, insertion in duplicated requests are guaranteed to be ignored and business processing is skipped

Null Compensation

When a null compensation occurs, a successful insertion of step 3 and an ignored insertion of step 4 will be committed directly and returned, skipping the business logic.

Hanging Action

When Hanging Action occurs, the compensation has been executed and a row gid-branchid-action has been inserted, so in Action, it will find that the insertion at step 3 was ignored and then return directly, skipping the business logic

Action and Compensation Overlap

If Action and Compensation overlap in execution time, both Compensation and Action will insert the same row gid-branchid-action. Due to a unique index conflict, only one of the two operations will succeed, while the other will wait for the transaction holding the lock to complete and then return.

Scenario 1: The Action fails and the Compensation succeeds in inserting gid-branchid-action. This is a typical Null Compensation and Hanging Action scenario and both Action and Compensation will ignore the business logic and return directly according to the above algorithm
Scenario 2: The Action succees and the Compensation fails in inserting gid-branchid-action. The order of transaction execution is Action before Compensation, and no anomaly occur. According to the above algorithm, the business in Action and Compensation will be executed orderly
Scenario 3, If the database go down during the overlap, the operations will be retried and will eventually go to scenario 1 or 2.

In summary of the above scenarios, the Sub-transaction Barrier is able to properly manage the anomalies and ensure the data consistency.

The algorithm described above also applies to TCC distributed transactions. Even if you are using a workflow engine, such as Candence, Camunda, to handle Saga transactions, the algorithm also applies.

Code

The above Sub-transaction Barrier technique, when used in conjunction with the distributed transaction framework https://github.com/dtm-labs/dtm, has been made available in several language SDKs, with the following example code in Go.

app.POST(BusiAPI+"/SagaBTransOut", dtmutil.WrapHandler(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransOutUID, -reqFrom(c).Amount, "")
  })
}))
app.POST(BusiAPI+"/SagaBTransOutCom", dtmutil.WrapHandler(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransOutUID, reqFrom(c).Amount, "")
  })
}))

In TransOut service, the first line is to create a barrier from http request, and the next several lines call the business inside the barrier.

When we put business processing inside a sub-transaction barrier, the various anomalies described earlier, are filtered and the developer only needs to consider the business processing under normal flow.

The complete example can be found here: https://github.com/dtm-labs/dtm-examples. After dtm setup, you can run a complete example by following command:
go run main.go http_saga_barrier

Conclusion

This article propose an algorithm to handle anomalies in Saga pattern for microservices. It is elegant and efficient, which can apply to TCC pattern, and also apply to many workflows.

You are welcomed to visit https://github.com/dtm-labs/dtm. It is an opensourced project dedicated to ease the development of distributed transactions.

Understanding XA Transactions With Practical Examples in Go

yedf2 — Tue, 17 May 2022 08:36:23 +0000

What is XA

XA is a specification for distributed transactions proposed by the X/Open organization.
The X/Open Distributed Transaction Processing (DTP) model envisages three software
components:

An application program (AP) defines transaction boundaries and specifies actions that constitute a transaction.
Resource managers (RMs, such as databases or file access systems) provide access to shared resources.
A separate component called a transaction manager (TM) assigns identifiers to transactions, monitors their progress, and takes responsibility for transaction completion and for failure recovery.

The following figure illustrates the interfaces defined by the X/Open DTP model.

XA is divided into two phases.

Phase 1 (prepare): All participating RMs prepare to execute their transactions and lock the required resources.
When each participant is ready, it report to TM.
Phase 2 (commit/rollback): When the transaction manager (TM) receives that all participants (RM) are ready, it sends commit commands to all participants. Otherwise, it sends rollback commands to all participants.

At present, almost all popular databases support XA transactions, including Mysql, Oracle, SqlServer, and Postgres

XA in Mysql

Let's see how database Mysql supports XA.

XA start '4fPqCNTYeSG' -- start a xa transaction
UPDATE `user_account` SET `balance`=balance + 30,`update_time`='2021-06-09 11:50:42.438' WHERE user_id = 1
XA end '4fPqCNTYeSG'
-- if connection closed before `prepare`, then the transaction is rolled back automatically
XA prepare '4fPqCNTYeSG'

-- When all participants have all prepared, call commit in phase 2
xa commit '4fPqCNTYeSG'

-- When any participants have failed to prepare, call rollback in phase 2
-- xa rollback '4fPqCNTYeSG'

Business Scenario

An inter-bank transfer is a typical distributed transaction scenario, where A needs to transfer money across a bank to B. Both the withdraw and the deposit may succeed or fail, and it is required that the sum of balance of A and B should not change after the transfer finished, regardless of any errors that occur.

Implement a Distributed XA Transaction

Distributed XA transactions can solve the above business problem. This article presents a solution based on dtm-labs/dtm. DTM is a popular distributed transaction framework which supports XA, Saga, OutBox, and TCC patterns.

A successful global transaction is illustrated by the following figure:

The code to implement it in Go is quite simple:

gid := dtmcli.MustGenGid(dtmutil.DefaultHTTPServer)
err := dtmcli.XaGlobalTransaction(dtmutil.DefaultHTTPServer, gid, func(xa *dtmcli.Xa) (*resty.Response, error) {
  resp, err := xa.CallBranch(&busi.TransReq{Amount: 30}, busi.Busi+"/TransOutXa")
  if err != nil {
    return resp, err
  }
  return xa.CallBranch(&busi.TransReq{Amount: 30}, busi.Busi+"/TransInXa")
})

app.POST(BusiAPI+"/TransInXa", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  return dtmcli.XaLocalTransaction(c.Request.URL.Query(), BusiConf, func(db *sql.DB, xa *dtmcli.Xa) error {
    return AdjustBalance(db, TransInUID, reqFrom(c).Amount, reqFrom(c).TransInResult)
  })
}))
app.POST(BusiAPI+"/TransOutXa", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  return dtmcli.XaLocalTransaction(c.Request.URL.Query(), BusiConf, func(db *sql.DB, xa *dtmcli.Xa) error {
    return AdjustBalance(db, TransOutUID, reqFrom(c).Amount, reqFrom(c).TransOutResult)
  })
}))

The above code first registers a global XA transaction, and then calls two sub-transactions TransOut, TransIn.
After all the sub-transactions are executed successfully, the global XA transaction is committed to DTM.
DTM receives the commitment of the XA global transaction, then calls the XA commit of all the sub-transactions, and finally change the status of global transaction to succeeded.

Code samples in other languages can be found here: SDKs

Run It

You can run the above example by running the following commands.

Run DTM

git clone https://github.com/dtm-labs/dtm && cd dtm
go run main.go

Run Example

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_xa

Rollback upon Failure

If any prepare operation fails, DTM will call XA rollback of each sub-transaction to roll back, and finally change the status of the global transaction to failed.

Let's pass TransInResult="FAILURE" in the request payload of XaFireRequest to trigger a failure.

req := &busi.TransReq{Amount: 30, TransInResult: "FAILURE"}

The timing diagram for failure is as follows:

Notices

The command of second phase, is also sent to the API BusiAPI+"/TransOutXa", and within this service, dtmcli.XaLocalTransaction will automatically call XA commit | XA rollback. So the body of the request is nil, and parsing body operations, such as the previous reqFrom, need to be placed inside XaLocalTransaction, otherwise the body parsing will result in errors.

TM Failure

The complexity of two-phase commit comes from all the failure scenarios that can arise. The most annoying failure happens after a participant has acknowledged prepared and before it receives the decision, such as a failure of the coordinator.

Our solution to this failure is quite simple and robust. Since most of cloud vendors provide highly-available databases, we can store the progress of sub-transactions in these databases, run multiple instance of TMs. Each instance will poll the paused transactions and continue to process them.

The cloud vendor takes care of the database failure, and may use Paxos/Raft to robustly elect a healthy instance.

Advantages and Disadvantages

Compared to other patterns like SAGA and TCC, the advantages of XA global transactions are:

Simple and easy to understand
Automatic rollback of business, no need to write compensation manually

The disadvantages of XA are:

Need the XA transaction from underlying database
Data is locked from data modification until the commitment, much longer than other patterns. It not suitable for highly concurrent business.

Conclusion

In this article, I've introduce the basic principle of XA transaction, and presents a practical example of it. Readers can easily follow the example to handle their own businesses.

Welcome to visit github.com/dtm-labs/dtm. It is a dedicated project to make distributed transactions in microservices easier. It supports multiple languages, and multiple patterns like a 2-phase message, Saga, TCC, and XA.

How to Implement a Distributed Transaction Across Mysql, Redis, and Mongo

yedf2 — Thu, 05 May 2022 02:27:34 +0000

Mysql, Redis and Mongo are all very popular stores, and each has its own advantages. In practical applications, it is common to use multiple stores at the same time and ensuring data consistency across multiple stores becomes a requirement.

This article gives an example of implementing a distributed transaction across multiple store engines, Mysql, Redis and Mongo. This example is based on the Distributed Transaction Framework https://github.com/dtm-labs/dtm and will hopefully help to solve your problems in data consistency across microservices.

The ability to flexibly combine multiple storage engines to form a distributed transaction is firstly proposed by DTM, and no other distributed transaction framework has stated the ability like this.

Problem scenarios

Let's look at the problem scenario first. Suppose that a user now participates in an promotion: he or she have a balance, recharge the phone bill, and the promotion will give away mall points. The balance is stored in Mysql, the bill is stored in Redis, the mall points is stored in Mongo. Because the promotion is limited in time, there is a possibility that participation may fail, so rollback support is required.

For the above problem scenario, you can use DTM's Saga transaction, and we will explain the solution in detail below.

Preparing the Data

The first step is to prepare the data. To make it easier for users to quickly get started with the examples, we have prepared the relevant data at en.dtm.pub, which includes Mysql, Redis and Mongo, and the specific connection username and password can be found at https://github.com/dtm-labs/dtm-examples.

If you want to prepare the data environment locally yourself, you can use https://github.com/dtm-labs/dtm/blob/main/helper/compose.store.yml to start Mysql, Redis, Mongo; and then execute scripts in https://github.com/dtm-labs/dtm/tree/main/sqls to prepare the data for this example, where busi.* is the business data and barrier.* is the auxiliary table used by DTM

Writing the Business Code

Let's start with the business code for the most familiar Mysql.

The following code is in Golang. Other languages such as C#, PHP, Java can be found here: DTM SDKs

func SagaAdjustBalance(db dtmcli.DB, uid int, amount int) error {
    _, err := dtmimp.DBExec(db, "update dtm_busi.user_account set balance = balance + ? where user_id = ?" , amount, uid)
    return err
}

This code mainly performs the adjustment of the user's balance in the database. In our example, this part of the code is used not only for Saga's forward operation, but also for the compensation operation, where only a negative amount needs to be passed in for compensation.

For Redis and Mongo, the business code is handled similarly, just incrementing or decrementing the corresponding balances.

How to Ensure Idempotency

For the Saga transaction pattern, when we have a temporary failure in the sub-transaction service, the failed operation will be retried. This failure may occur before or after the sub-transaction commits, so the sub-transaction operation needs to be idempotent.

DTM provides helper tables and helper functions to help users achieve idempotency quickly. For Mysql, it will create an auxiliary table barrier in the business database, when the user start a transaction to adjust the balance, it will first insert Gid in the barrier table. If there is a duplicate row, then the insertion will fail, and then skip the balance adjustment to ensure the idempotent. The code using the helper function is as follows:

app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).Call(txGet(), func(tx *sql.Tx) error {
        return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, reqFrom(c).TransInResult)
    })
}))

Mongo handles idempotency in a similar way to Mysql, so I won't go into detail again.

Redis handles idempotency differently than Mysql, mainly because of the difference in the principle of transactions. Redis transactions are mainly ensured by atomic execution of Lua. the DTM helper function will adjust the balance via a Lua script. Before adjusting the balance, it will query Gid in Redis. If Gid exists, it will skip the balance adjustment; if not, it will record Gid and perform the balance adjustment. The code used for the helper function is as follows:

app.POST(BusiAPI+"/SagaRedisTransOut", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), -reqFrom(c).Amount, 7*86400)
}))

How to do Compensation

For Saga, we also need to deal with the compensation operation, but the compensation is not simply a reverse adjustment, and there are many pitfalls that should be aware of.

On the one hand, compensation needs to take idempotency into account, because the failure and retries described in previous subsection also exists in compensation. On the other hand, compensation also needs to take "null compensation" into account, since the forward operation of Saga may returns a failure, which may have happened before or after the data adjustment. For failures where the adjustment has been committed we need to perform the reverse adjustment; but for failures where the adjustment has not been committed we need to skip the reverse operation.

In the helper table and helper functions provided by DTM, on the one hand, it will determine whether the compensation is a null compensation based on the Gid inserted by the forward operation, and on the other hand, it will insert Gid+'compensate' again to determine whether the compensation is a duplicate operation. If there is a normal compensation operation, then it will execute the data adjustment on the business; if there is a null compensation or duplicate compensation, it will skip the adjustment on the business.

The Mysql code is as follows.

app.POST(BusiAPI+"/SagaBTransInCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).Call(txGet(), func(tx *sql.Tx) error {
        return SagaAdjustBalance(tx, TransInUID, -reqFrom(c).Amount, "")
    })
}))

The code for Redis is as follows.

app.POST(BusiAPI+"/SagaRedisTransOutCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).RedisCheckAdjustAmount(RedisGet(), GetRedisAccountKey(TransOutUID), reqFrom(c).Amount, 7*86400)
}))

The compensation service code is almost identical to the previous code of the forward operation, except that the amount is multiplied by -1. The DTM helper function automatically handles idempotency and null compensation properly.

Other exceptions

When writing forward operations and compensation operations, there is actually another exception called "Suspension". A global transaction will roll back when it is timeout or retries has reach the configured limit. The normal case is that the forward operation is performed before the compensation, but in case of process suspension the compensation may be performed before the forward operation. So the forward operation also needs to determine whether the compensation has been executed, and in the case that it has, the data adjustment needs to be skipped as well.

For DTM users, these exceptions have been handled gracefully and properly and you, as a user, need only follow the MustBarrierFromGin(c).Call call described above and do not need to care about them at all. The principle for DTM handling these exceptions is described in detail here: Exceptions and sub-transaction barriers

Initiating a Distributed Transaction

After writing the individual sub-transaction services, the following codes of the code initiates a Saga global transaction.

saga := dtmcli.NewSaga(dtmutil.DefaultHTTPServer, dtmcli.MustGenGid(dtmutil.DefaultHTTPServer)).
  Add(busi.Busi+"/SagaBTransOut", busi.Busi+"/SagaBTransOutCom", &busi.TransReq{Amount: 50}).
  Add(busi.Busi+"/SagaMongoTransIn", busi.Busi+"/SagaMongoTransInCom", &busi.TransReq{Amount: 30}).
  Add(busi.Busi+"/SagaRedisTransIn", busi.Busi+"/SagaRedisTransOutIn", &busi.TransReq{Amount: 20})
err := saga.Submit()

In this part of the code, a Saga global transaction is created which consists of 3 sub-transactions.

Transfer out 50 from Mysql
Transfer in 30 to Mongo
Transfer in 20 to Redis

Throughout the transaction, if all the sub-transactions complete successfully, then the global transaction succeeds; if one of the sub-transactions returns a business failure, then the global transaction rolls back.

Run

If you want to run a complete example of the above, the steps are as follows.

Run DTM

git clone https://github.com/dtm-labs/dtm && cd dtm
go run main.go

Run a successful example

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_saga_multidb

Run a failed example

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_saga_multidb_rollback

You can modify the example to simulate various temporary failures, null compensation situations, and various other exceptions where the data is consistent when the entire global transaction is finished.

Summary

This article gives an example of a distributed transaction across Mysql, Redis and Mongo. It describes in detail the problems that need to be dealt with, and the solutions.

The principles in this article are suitable for all storage engines that support ACID transactions, and you can quickly extend it for other engines such as TiKV.

A Better Pattern than OutBox - 2-Phase Message

yedf2 — Tue, 12 Apr 2022 12:26:03 +0000

This article proposes an alternative pattern to OutBox: 2-phase message. It is not based on message queue, but based on github.com/dtm-labs/dtm, a highly available distributed transaction framework.

An inter-bank transfer is a typical distributed transaction scenario, where A needs to transfer money across a bank to B. The balances of A and B are not in the same bank so that they are not stored in a single database. This transfer is typically crossing micro-services also.

The main problem is that the transfer must update two systems simultaneously -- the increment of A' balance and the decrement of B's balance. This is called well-known "dual writes". A process crash between the two updates leaves the entire system in an inconsistent state.

This "dual writes" problem can be solved by OutBox pattern. The principal of OutBox pattern can be found here Transactional OutBox

2-Phase Message

First let's take a glance at how to accomplish the above transfer task using the new pattern. The following codes is in Go, other languages like C#, PHP can be found here: dtm SDKs

msg := dtmcli.NewMsg(DtmServer, gid).
    Add(busi.Busi+"/TransIn", &TransReq{Amount: 30})
err := msg.DoAndSubmitDB(busi.Busi+"/QueryPrepared", db, func(tx *sql.Tx) error {
    return AdjustBalance(tx, busi.TransOutUID, -req.Amount)
})

In the above codes:

First new a DTM msg global transaction, passing the dtm server address and the global transaction id
Add to the msg a branch business, which is the transfer operation TransIn, together with the data that needs to be passed to this service, the amount 30$
Then call msg's DoAndSubmitDB. This function will ensure the atomic execution of both the business and submission of msg, either both succeeded, or both failed. There are three parameters for this function:
1. The check-back URL, will be explained later
2. DB, is the database object for the business
3. The business function, here in our example is to debit 30$ for A's balance

What will happen when the process crashed immediately after the success of decrement for A's balance? After a timeout, DTM will call the check-back URL to query whether the decrement is successful or unsuccessful. We can accomplish the check-back service by pasting the following code:

    app.GET(BusiAPI+"/QueryPrepared", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
        return MustBarrierFromGin(c).QueryPrepared(db)
    }))

After writing these two pieces of codes, a 2-phase message is accomplished, much easier to use than OutBox.

Run It

You can run the above example by running the following commands.

Run DTM

git clone https://github.com/dtm-labs/dtm && cd dtm
go run main.go

Run Example

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_msg_doAndCommit

Successful Process

How does DoAndSubmitDB ensure the atomicity of successful business execution and msg submission? Please see the following timing diagram.

In general, the 5 steps in the timing diagram will complete normally, and the global transaction completes. There is something needed to explain here: the commitment of msg is done in two phases, first Prepare, then Submit. After DTM receives the Prepare request, it does not call the branch transaction, but waits for the subsequent Submit. Only when it receives the Submit request, it starts the branch call and finally completes the global transaction.

Crash After Commitment

In a distributed system, all kinds of downtime and network exceptions need to be considered, so let's take a look at what can happen.

The most important goal we want to achieve is that both the business execution and the message submission compose an atomic operation. So let's first look at what will happen if there is a downtime failure after the business execution and before the message submission, and how the new pattern will ensure the atomicity.

Let's take a look at the timing diagram in this case.

In this case, DTM will poll the messages that is only Prepared but not Submitted after a certain timeout and call the check-back service specified by the message to query whether the business execution is successful.

This check-back service goes inside the message table and queries whether the local transaction for business has been committed.

Committed: Returns success, dtm submits the global transaction and proceeds to the next sub-transaction call
Rolled back: Failure is returned, dtm terminates the global transaction and no more sub-transaction calls are made
In progress: This check-back will wait for the final result and then proceeds to the previous committed/rollbacked case
Not Started: This check-back will insert data to ensure that the local transaction for business eventually fails

Crash Before Commitment

Let's take a look at the timing diagram of a local transaction being rolled back.

If the process is crashed immediately after the dtm receives the Prepare call and before the transaction commitment, the local database will detect the process's disconnection and rollback the local transaction automatically.

Subsequently, dtm polls for the global transactions that have timed out, that is only Prepared but not Submitted, and checks back. The check-back service finds that the local transaction has been rollbacked and returns the result to dtm. dtm receives the result indicating rollbacked, and then marks the global transaction as failed, and finally ends the global transaction.

2-Phase Message VS OutBox

The OutBox pattern can also ensure the eventual consistency of the data. As far as OutBox pattern is used, the work required includes

Executing the local business logic in the local transaction, inserting the messages into the message table and committing them at last.
Writing polling tasks to take messages from the local message table and send them to the message queue. Instead of periodically executing SQL to poll, this step may use another technique Log-based Change Data Capture.
Consuming messages.

Compared with OutBox, 2-phase message has the following advantages.

No need to learn or maintain any message queues
No polling tasks to handle
No need to consume messages

2-phase message only need DTM, which is much easier to learn or to maintain than message queues. All skills involved are function calls and services calls, which are familiar things to all developers.

The exposed interfaces of 2-phase messages are completely independent of the queue and are only related to the actual business and service calls, making it more developer-friendly
2-phase messages do not have to consider the message stacking and other failures, because 2-phase messages depend only on dtm. Developers can think of dtm as being the same as any other ordinary stateless service in the system, relying only on the storage behind it, Mysql/Redis.
The message queue is asynchronous, while 2-phase messages support both asynchronous and synchronous. The default behaviour is asynchronous, and you can wait for the downstream service to complete synchronously just by setting msg.WaitResult=true.
2-phase messages also support specifying multiple downstream services at the same time

Application of 2-Phase Message

2-phase messages can significantly reduce the difficulty of the eventual consistency solution and have been widely used, here are two typical applications.

flash-sale system: this architecture can easily carry tens of thousands of order requests on a single machine, and ensure that the number of inventory deducted and the number of orders are accurately matched
cache consistency: this architecture can easily ensure the consistency of DB and cache through 2-phase message, which is much better than queue or subscription binlog solution

Example of using redis, Mongo storage engine in combination with 2-phase messages can be found in dtm-examples

Check-back Principle

The check-back service appears in the previous timing diagram, as well as in the interface. This check-back design firstly existed in RocketMQ, and the implementation is left to developers to handle manually. In the 2-phase messages, it is handled automatically by copy-and-paste code. So what is the principle of automatic processing?

To perform a check-back, we firstly create a separate table in the business database instance where the gid(global transaction id) is stored. Gid is written to this table when the business transaction is processed.

When we check back with the gid, if we find gid in the table, then it means the local transaction has been committed, so we can return to dtm the result that the local transaction has been committed.

When we check back with the gid, if we don't find gid in the table, then it means the local transaction has not been committed. There are three possible results:

The transaction is still in progress.
The transaction has been rolled back.
The transaction has not started.

I have searched a lot of information about RocketMQ's check-back, but have not found a error-free solution. Most suggestions is that if the gid is not found, then do nothing and wait for the next check-back in next 10 seconds. If the check-back has lasted 2 minutes or longer and still cannot find the gid, then the local transaction is considered rollbacked.

There are problems in the following cases.

In the extreme case, a database failure (such as a process pause or disk jam) may occur, lasting longer than 2 minutes, and finally the data is committed. But RocketMQ assume the transaction is rolled back, and cancel the global transaction, leaving the data in inconsistent state.
If a local transaction, has been rollbacked, but the check-back service, within two minutes, will constantly polling every 10 seconds, causing unnecessary load on the server.

These problems are completely solved by dtm's 2-phase message solution. It works as follows.

When a local transaction is processed, gid is inserted into the table dtm_barrier.barrier with an insert reason of COMMITTED. Table dtm_barrier.barrier has a unique index on gid.
When checking back, the 2-phase message does not directly query whether gid exists, but instead insert ignore a row with the same gid, together with the reason ROLLBACKED. At this time, if there is already a record with gid in the table, then the new insert operation will be ignored, otherwise the row will be inserted.
Query the records in the table with gid, if the reason of the record is COMMITTED, then the local transaction has been committed; if the reason of the record is ROLLBACKED, then the local transaction has been rolled back or will be rolled back.

So how do 2-phase message distinguish between in-progress and rolled back messages? The trick lies in the data inserted during the check-back. If the database transaction is still in progress at the time of the check-back, then the insert operation will be blocked by the in-progress transaction, because the insert operation in check-back will wait for the row lock held by the in-progress transaction. If the insert operation returns normally, then the local transaction in the database, which must have ended.

Normal messages

2-phase messages can replace not only OutBox, but also the normal message pattern. If you call Submit directly, then it is similar to the normal message pattern, but provides a more flexible and simple interface.

Suppose an application scenario where there is a button on the UI to participate in an activity that grants permanent access to two eBooks. In this case, the server side can be handled like this:

msg := dtmcli.NewMsg(DtmServer, gid).
    Add(busi.Busi+"/AuthBook", &Req{UID: 1, BookID: 5}).
    Add(busi.Busi+"/AuthBook", &Req{UID: 1, BookID: 6})
err := msg.Submit()

This approach also provides an asynchronous interface without relying on a message message queue.

Summary

The 2-phase message proposed in this article has a simple and elegant interface that brings a more elegant pattern than OutBox.

Welcome to visit github.com/dtm-labs/dtm. It is a dedicated project to make distributed transactions in micro-services easier. It support multiple languages, and multiple patterns like 2-phase message, Saga, Tcc and Xa.

How to Implement Saga Pattern in Microservices

yedf2 — Mon, 04 Apr 2022 12:54:13 +0000

The Microservice architecture allows us to maintain applications services and their databases separately. As a sequence, it is common to have data changes that need to propagate across multiple services. Database transactions can not go across multiple services, so the data integrity becomes a challenge.

SAGA

Saga is a distributed transaction pattern mentioned in this paper SAGAS. The core idea is to split long transactions into multiple short local transactions, which are coordinated by the Saga transaction coordinator, so that if each local transaction completes successfully then it completes normally, and if anyone step fails then the compensating operations are invoked one at a time in reverse order.

This article will present a complete SAGA example to give the reader an accurate understanding of SAGA transactions

Business Scenario

An inter-bank transfer is a typical distributed transaction scenario, where A needs to transfer money across a bank to B. Both the withdraw and the deposit may succeed of fail, and it is required that the sum of balance of A and B should not change after the transfer finished, regardless of any errors that occur.

SAGA transaction

We will present you a detailed runable example base on dtm, a distributed transaction framework.

Suppose that you have finished your implementation of the business for transfer and rollback.

"/api/SagaBTransOut"
"/api/SagaBTransOutCom" compensation for TransOut
"/api/SagaBTransIn"
"/api/SagaBTransInCom" compensation for TransIn

The following code will orchestrate these operations to a Sagas distributed transaction. The transaction will finished when both TransIn and TransOut succeed, or both rolled back. In both cases, the sum of balance of A and B remains the same.

    req := &gin.H{"amount": 30} // load of microservice
    // DtmServer is the address of the DTM service
    saga := dtmcli.NewSaga(DtmServer, dtmcli.MustGenGid(DtmServer)).
        // Add a child transaction of TransOut with url: qsBusi+"/TransOut" for the forward operation and url: qsBusi+"/TransOutCom" for the reverse operation
        Add(qsBusi+"/SagaBTransOut", qsBusi+"/SagaBTransOutCom", req).
        // Add a subtransaction of TransIn with url: qsBusi+"/TransOut" for the forward action and url: qsBusi+"/TransInCom" for the reverse action
        Add(qsBusi+"/SagaBTransIn", qsBusi+"/SagaBTransInCom", req)
    // commit saga transaction, dtm will complete all subtransactions or rollback all subtransactions
    err := saga.Submit()

A successful transaction timing diagram is as follows.

Core operations

The adjustment and compensation of user balance should be handled carefully. Here we dive into the detail of the adjustment. For the example of the bank transfer we are going to perform, we will do the TransOut and TransIn in the action operation and the opposite adjustment in the compensation operation.

First we create the account balance table.

CREATE TABLE dtm_busi.`user_account` (
  `id` int(11) AUTO_INCREMENT PRIMARY KEY,
  `user_id` int(11) not NULL UNIQUE ,
  `balance` decimal(10,2) NOT NULL DEFAULT '0.00',
  `trading_balance` decimal(10,2) NOT NULL DEFAULT '0.00',
  `create_time` datetime DEFAULT now(),
  `update_time` datetime DEFAULT now()
);

Then write the core business code to adjust the user's account balance

func SagaAdjustBalance(db dtmcli.DB, uid int, amount int, result string) error {
    _, err := dtmimp.DBExec(db, "update dtm_busi.user_account set balance = balance + ? where user_id = ?" , amount, uid)
    return err
}

Then write the specific processing function for the action/compensation operation

app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, "")
  })
}))
app.POST(BusiAPI+"/SagaBTransInCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransInUID, -reqFrom(c).Amount, "")
  })
}))
app.POST(BusiAPI+"/SagaBTransOut", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransOutUID, -reqFrom(c).Amount, "")
  })
}))
app.POST(BusiAPI+"/SagaBTransOutCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  return barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransOutUID, reqFrom(c).Amount, "")
  })
}))

The core logic of these processing functions is to adjust the balance, for which the role of barrier.Call will be explained in detail later

Run

Follow these steps to run a successful example.

Run dtm which manage the distributed transactions

git clone https://github.com/dtm-labs/dtm && cd dtm
go run main.go

Run the example

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_saga_barrier

Handling network exceptions

Suppose a transaction committed to dtm has a transient fault when an operation is invoked. dtm will retry the incomplete operation, which requires the subtransactions of the global transaction to be idempotent. dtm framework pioneered the sub-transaction barrier technique, providing the BranchBarrier tool class to help users handle idempotency easily. It provides a function Call that guarantees that the operation inside this function will be commited at most once:

func (bb *BranchBarrier) Call(tx *sql.Tx, busiCall BarrierBusiFunc) error

This BranchBarrier can automatically handle not only idempotency, but also null-compensation and hanging issues, see exceptions and solutions for details.

Handling rollbacks

What happens if the bank is preparing to transfer the amount to user B and finds that user B's account is abnormal and returns a failure? We update the handler function so that the transfer operation returns a failure

app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  return dtmcli.ErrFailure
}))

We give the timing diagram for the transaction failure interaction

The action of TransIn branch did nothing and returned a failure. Will compensation of TransIn branch cause the reverse adjustment to go wrong?

Don't worry, the preceding sub-transaction barrier technique ensures that the TransIn failure is compensated as a null operation if error occurs before the commit, and is compensated to do opposite adjustment if error occurs after the commit

You can change the TransIn that returns an error after commit.

app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  barrier := MustBarrierFromGin(c)
  barrier.Call(txGet(), func(tx *sql.Tx) error {
    return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, "")
  })
  return dtmcli.ErrFailure
}))

The final result balance will still be right, see Exceptions and Solutions for details.

Summary

This article gives a complete SAGA transaction solution, a working SAGA, which you can use to solve your real problems with a few simple modifications to this example

A detail description of SAGA can be found here SAGA

You are welcomed to visit https://github.com/dtm-labs/dtm

Best Practice for TCC Distributed Transaction In Go

yedf2 — Fri, 01 Apr 2022 13:15:11 +0000

This article will present a complete TCC example to give the reader an accurate understanding of TCC-type transactions

Business Scenario

A typical distributed transaction scenario is inter-bank transfers, where A needs to transfer funds across a bank to B. The hypothetical requirement scenario is that both transfers out of A and into B may succeed and fail, and that both transfers in and out will eventually succeed or fail.

There is also a requirement that if there is a rollback, the SAGA mode will result in A discovering that its balance has been deducted, but the recipient, B, is late in receiving the balance, which will cause A great distress. The business would prefer to avoid this situation

TCC Components

The TCC is divided into 3 part

Try part: attempts to execute, completes all business checks (consistency), reserve necessary business resources.
Confirm part: if all branches succeed in the Try phase, then we move to the Confirm phase, where Confirm actually executes the business without any business checks, using only the business resources reserved in the Try phase
Cancel part: If one of the Trys in all branches fails, we go to the Cancel phase, which releases the business resources reserved in the Try phase.

If we were to perform a transaction similar to a bank interbank transfer, with TransOut and TransIn in separate micro-services, a typical timing diagram for a successfully completed TCC transaction would be as follows.

Our solution is based on distributed transaction framework dtm, an excellent project dedicated to distributed transaction solution.

Core Operations

First we create the account balance table, where trading_balance indicates the amount that has been frozen.



create table if not exists dtm_busi.user_account(
  id int(11) PRIMARY KEY AUTO_INCREMENT,
  user_id int(11) UNIQUE,
  balance DECIMAL(10, 2) not null default '0',
  trading_balance DECIMAL(10, 2) not null default '0',
  create_time datetime DEFAULT now(),
  update_time datetime DEFAULT now(),
  key(create_time),
  key(update_time)
);

Let's write the core code first, the freeze/unfreeze funds operation will check the constraint balance+trading_balance >= 0, if the constraint is not valid, the execution fails


 go
func tccAdjustTrading(db dtmcli.DB, uid int, amount int) error {
    affected, err := dtmimp.DBExec(db, "update dtm_busi.user_account set trading_balance=trading_balance+?       where user_id=? and trading_balance + ? + balance >= 0", amount, uid, amount)
    if err == nil && affected == 0 {
        return fmt.Errorf("update error, maybe balance not enough")
    }
    return err
}

func tccAdjustBalance(db dtmcli.DB, uid int, amount int) error {
    affected, err := dtmimp.DBExec(db, "update dtm_busi.user_account set trading_balance=trading_balance-?,          balance=balance+? where user_id=? ", amount, amount, uid)
    if err == nil && affected == 0 {
        return fmt.Errorf("update user_account 0 rows")
    }
    return err
}

Let's write the specific Try/Confirm/Cancel handler functions


 go
app.POST(BusiAPI+"/TccBTransOutTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  bb := MustBarrierFromGin(c)
  return bb.Call(txGet(), func(tx *sql.Tx) error {
    return tccAdjustTrading(tx, TransOutUID, -req.Amount)
  })
}))
app.POST(BusiAPI+"/TccBTransOutConfirm", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  bb := MustBarrierFromGin(c)
  return bb.Call(txGet(), func(tx *sql.Tx) error {
    return tccAdjustBalance(tx, TransOutUID, -reqFrom(c).Amount)
  })
}))
app.POST(BusiAPI+"/TccBTransOutCancel", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  bb := MustBarrierFromGin(c)
  return bb.Call(txGet(), func(tx *sql.Tx) error {
    return tccAdjustTrading(tx, TransOutUID, req.Amount)
  })
}))
app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  bb := MustBarrierFromGin(c)
  return bb.Call(txGet(), func(tx *sql.Tx) error {
    return tccAdjustTrading(tx, TransInUID, req.Amount)
  })
}))
app.POST(BusiAPI+"/TccBTransOutConfirm", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  bb := MustBarrierFromGin(c)
  return bb.Call(txGet(), func(tx *sql.Tx) error {
    return tccAdjustBalance(tx, TransInUID, reqFrom(c).Amount)
  })
}))
app.POST(BusiAPI+"/TccBTransInCancel", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
  bb := MustBarrierFromGin(c)
  return bb.Call(txGet(), func(tx *sql.Tx) error {
    return tccAdjustTrading(tx, TransInUID, -req.Amount)
  })
}))

The core logic of these functions is to freeze and adjust the balance, the role of bb.Call in this will be explained in detail later

TCC transactions

Then the TCC transaction is created and branch calls are made


 go
// TccGlobalTransaction will open a global transaction
_, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
  // CallBranch will register the Confirm/Cancel of the transaction branch to the global transaction, and then call Try directly
  res1, rerr := tcc.CallBranch(&TransReq{Amount: 30}, host+"/api/TccBTransOutTry", host+"/api/TccBTransOutConfirm", host+"/api/ TccBTransOutCancel"
  if err ! = nil {
    return resp, err
  }
  return tcc.CallBranch(&TransReq{Amount: 30}, host+"/api/TccBTransInTry", host+"/api/TccBTransInConfirm", host+"/api/TccBTransInCancel")
})

At this point, a complete TCC distributed transaction is finished.

Run

If you want to run a successful example entirety, the steps are as follows.

run dtm ``` bash git clone https://github.com/dtm-labs/dtm && cd dtm go run main.go


2. Run the example

bash
git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples
go run main.go http_tcc_barrier


## Handling network exceptions

Suppose a transaction committed to dtm fails briefly at one of these steps. dtm will retry the incomplete operation, requiring the subtransactions of the global transaction to be idempotent. dtm framework pioneered the subtransaction barrier technique, providing the BranchBarrier utility class to help users handle idempotency easily. It provides a function Call which guarantees that the operation inside this function will be called at most once:

go
func (bb *BranchBarrier) Call(tx *sql.Tx, busiCall BarrierBusiFunc) error


This BranchBarrier can automatically handle not only idempotency, but also null-compensation and hanging issues, see [exceptions and solutions](https://en.dtm.pub/practice/barrier) for details.

### Rollback of TCC
What happens if the bank is preparing to transfer the amount to user 2 and finds that user 2's account is abnormal and returns a failure? We modify the code to simulate this situation.

go
app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
return dtmcli.ErrFailure
}))

This is the timing diagram for a transaction failure interaction
![tcc-rollback](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/8e10v4r8lrr0gx1sxjig.png)


The difference between this and a successful TCC is that when a child transaction returns a failure, the global transaction is subsequently rolled back, calling the Cancel operation of each child transaction to ensure that the global transaction is all rolled back.

The forward operation of TransInTry returned a failure without doing anything, at this point calling the TransInCancel compensation operation will cause the reverse adjustment to go wrong?

Don't worry, the preceding subtransaction barrier technique ensures that the TransInTry error is compensated as a null operation if it occurs before the commit, and that the compensation operation commits the data if the TransInTry error occurs after the commit.

You can change `TccBTransInTry` to

go
app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
bb := MustBarrierFromGin(c)
bb.Call(txGet(), func(tx *sql.Tx) error {
return tccAdjustTrading(tx, TransInUID, req.Amount)
})
return dtmcli.ErrFailure
}))


The final result balance will still be right, see [Exceptions and Solutions](https://en.dtm.pub/practice/barrier) for details.

### Summary

This article gives a complete TCC transaction solution. You can use it to solve your real problems with a few simple modifications to this example

For more information on the principles of TCC, see [TCC](https://en.dtm.pub/practice/tcc)

Welcome to visit the [https://github.com/dtm-labs/dtm](https://github.com/dtm-labs/dtm) project