dtm

package module
v1.5.5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 8, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

README

DTM Distributed Transaction Manager Plugin

Introduction

The DTM plugin provides distributed transaction management capabilities for the Lynx framework, supporting multiple distributed transaction patterns:

  • SAGA: Long transaction solution
  • TCC: Try-Confirm-Cancel pattern
  • 2-phase message: Reliable message eventual consistency
  • XA: Two-phase commit protocol

Features

  • Supports both HTTP and gRPC protocols
  • Automatically handles transaction timeouts and retries
  • Provides branch barrier functionality to automatically handle idempotency, suspension, and empty compensation issues
  • Supports custom request header passthrough
  • Flexible timeout configuration

Configuration Guide

lynx:
  dtm:
    enabled: true                          # Whether to enable the plugin
    server_url: "http://localhost:36789/api/dtmsvr"  # DTM HTTP service address
    grpc_server: "localhost:36790"         # DTM gRPC service address (optional)
    timeout: 10                            # Request timeout (seconds)
    retry_interval: 10                     # Retry interval (seconds)
    transaction_timeout: 60                # Global transaction timeout (seconds)
    branch_timeout: 30                     # Branch transaction timeout (seconds)
    pass_through_headers:                  # Request headers that need to be passed through
      - "X-Request-ID"
      - "X-User-ID"

Usage Examples

SAGA Transaction
import (
    "github.com/go-lynx/lynx/app"
    "github.com/go-lynx/lynx/plugins/dtm/dtm"
)

func UseSaga() {
    // Get DTM plugin instance
    dtmPlugin := app.GetPlugin("dtm.server").(*dtm.DTMClient)
    
    // Generate global transaction ID
    gid := dtmPlugin.GenerateGid()
    
    // Create SAGA transaction
    saga := dtmPlugin.NewSaga(gid)
    
    // Add transaction branches
    saga.Add(
        "http://localhost:8080/api/TransOut",     // Forward operation
        "http://localhost:8080/api/TransOutRevert", // Compensation operation
        map[string]interface{}{"amount": 100},
    )
    saga.Add(
        "http://localhost:8080/api/TransIn",
        "http://localhost:8080/api/TransInRevert",
        map[string]interface{}{"amount": 100},
    )
    
    // Submit transaction
    err := saga.Submit()
    if err != nil {
        log.Errorf("SAGA transaction failed: %v", err)
    }
}

Using Helper wrapper (Recommended):

func UseTCCWithHelper(ctx context.Context) {
    dtmPlugin := app.GetPlugin("dtm.server").(*dtm.DTMClient)
    helper := dtm.NewTransactionHelper(dtmPlugin)
    gid := helper.MustGenGid()

    branches := []dtm.TCCBranch{
        { // Example branch 1
            Try:     "http://localhost:8081/api/inventory/try",
            Confirm: "http://localhost:8081/api/inventory/confirm",
            Cancel:  "http://localhost:8081/api/inventory/cancel",
            Data:    map[string]any{"product_id": "sku-1", "quantity": 2},
        },
    }
    opts := &dtm.TransactionOptions{TimeoutToFail: 60, BranchTimeout: 10}
    if err := helper.ExecuteTCC(ctx, gid, branches, opts); err != nil {
        log.Errorf("ExecuteTCC failed: %v", err)
    }
}

Using Global Transaction API (Native usage):

func UseTCCWithNative(ctx context.Context) {
    dtmPlugin := app.GetPlugin("dtm.server").(*dtm.DTMClient)
    gid := dtmPlugin.GenerateGid()
    _ = dtmcli.TccGlobalTransaction(dtmPlugin.GetServerURL(), gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
        _, err := tcc.CallBranch(
            map[string]any{"product_id": "sku-1", "quantity": 2},
            "http://localhost:8081/api/inventory/try",
            "http://localhost:8081/api/inventory/confirm",
            "http://localhost:8081/api/inventory/cancel",
        )
        return nil, err
    })
}
2-Phase Message
func UseMsg() {
    dtmPlugin := app.GetPlugin("dtm.server").(*dtm.DTMClient)
    gid := dtmPlugin.GenerateGid()
    
    // Create message transaction
    msg := dtmPlugin.NewMsg(gid)
    
    // Add transaction steps
    msg.Add(
        "http://localhost:8080/api/TransOut",
        map[string]interface{}{"amount": 100},
    )
    msg.Add(
        "http://localhost:8080/api/TransIn",
        map[string]interface{}{"amount": 100},
    )
    
    // Prepare message
    err := msg.Prepare("http://localhost:8080/api/QueryPrepared")
    if err != nil {
        log.Errorf("Message prepare failed: %v", err)
        return
    }
    
    // Submit message
    err = msg.Submit()
    if err != nil {
        log.Errorf("Message transaction failed: %v", err)
    }
}

Using Helper wrapper (Recommended):

func UseXAWithHelper(ctx context.Context) {
    dtmPlugin := app.GetPlugin("dtm.server").(*dtm.DTMClient)
    helper := dtm.NewTransactionHelper(dtmPlugin)
    gid := helper.MustGenGid()

    branches := []dtm.XABranch{
        { // XABranch.Data uses string (e.g., JSON)
            Action: "http://localhost:8080/api/TransOut",
            Data:   `{"amount": 100}`,
        },
    }
    opts := &dtm.TransactionOptions{TimeoutToFail: 60, BranchTimeout: 10}
    if err := helper.ExecuteXA(ctx, gid, branches, opts); err != nil {
        log.Errorf("ExecuteXA failed: %v", err)
    }
}

Using Global Transaction API (Native usage):

func UseXAWithNative(ctx context.Context) {
    dtmPlugin := app.GetPlugin("dtm.server").(*dtm.DTMClient)
    gid := dtmPlugin.GenerateGid()
    _ = dtmcli.XaGlobalTransaction(dtmPlugin.GetServerURL(), gid, func(xa *dtmcli.Xa) (*resty.Response, error) {
        _, err := xa.CallBranch("http://localhost:8080/api/TransOut", `{"amount": 100}`)
        return nil, err
    })
}

Installing DTM Server

Before using the plugin, you need to install and run the DTM server:

# Run using Docker
docker run -itd --name dtm -p 36789:36789 -p 36790:36790 yedf/dtm:latest

# Or use binary file
wget https://github.com/dtm-labs/dtm/releases/download/v1.17.0/dtm_1.17.0_linux_amd64.tar.gz
tar -xzvf dtm_1.17.0_linux_amd64.tar.gz
./dtm -c conf.yml

References

Notes

  • NewTcc() and NewXa() do not provide direct instance return implementations in the current version. Please use the Helper wrapper or DTM's native Global Transaction API as shown above.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotImplemented indicates the feature is not yet implemented
	ErrNotImplemented = errors.New("lynx-dtm: feature not implemented")
)

Functions

func CreateGrpcContext 

func CreateGrpcContext(ctx context.Context, gid string, transType string, branchID string, op string) context.Context

CreateGrpcContext create gRPC Context containing transaction information

func ExtractGrpcTransInfo 

func ExtractGrpcTransInfo(ctx context.Context) (*dtmcli.BranchBarrier, error)

ExtractGrpcTransInfo extract transaction information from gRPC Context

func ValidateConfig added in v1.5.4 

func ValidateConfig(c *conf.DTM) error

ValidateConfig validates DTM configuration. Returns error if invalid.

Types

type BarrierHandler 

type BarrierHandler struct {
	// contains filtered or unexported fields
}

BarrierHandler transaction barrier handler

func NewBarrierHandler 

func NewBarrierHandler(client *DTMClient) *BarrierHandler

NewBarrierHandler create transaction barrier handler

func (*BarrierHandler) CallWithDB 

func (b *BarrierHandler) CallWithDB(ctx context.Context, db *sql.DB, req *dtmcli.BranchBarrier, fn dtmcli.BarrierBusiFunc) error

CallWithDB execute branch barrier within database transaction

func (*BarrierHandler) CallWithTx 

func (b *BarrierHandler) CallWithTx(ctx context.Context, tx *sql.Tx, req *dtmcli.BranchBarrier, fn dtmcli.BarrierBusiFunc) error

CallWithTx execute branch barrier within existing transaction

func (*BarrierHandler) CreateBarrierFromGin 

func (b *BarrierHandler) CreateBarrierFromGin(c interface{}) (*dtmcli.BranchBarrier, error)

CreateBarrierFromGin creates a branch barrier from Gin's *gin.Context. Pass c as *gin.Context and extract query params: dtmcli.BarrierFromQuery(c.Request.URL.Query()). Returns ErrNotImplemented as placeholder - implement based on your HTTP framework.

func (*BarrierHandler) CreateBarrierFromGrpc 

func (b *BarrierHandler) CreateBarrierFromGrpc(ctx context.Context) (*dtmcli.BranchBarrier, error)

CreateBarrierFromGrpc create branch barrier from gRPC request

func (*BarrierHandler) HandleMsg 

func (b *BarrierHandler) HandleMsg(ctx context.Context, bb *dtmcli.BranchBarrier, db *sql.DB, busiCall dtmcli.BarrierBusiFunc) error

HandleMsg handle 2-phase message

func (*BarrierHandler) HandleSAGA 

func (b *BarrierHandler) HandleSAGA(ctx context.Context, bb *dtmcli.BranchBarrier, db *sql.DB, busiCall dtmcli.BarrierBusiFunc) error

HandleSAGA handle SAGA transaction

func (*BarrierHandler) HandleTCCCancel 

func (b *BarrierHandler) HandleTCCCancel(ctx context.Context, bb *dtmcli.BranchBarrier, db *sql.DB, busiCall dtmcli.BarrierBusiFunc) error

HandleTCCCancel handle TCC Cancel phase

func (*BarrierHandler) HandleTCCConfirm 

func (b *BarrierHandler) HandleTCCConfirm(ctx context.Context, bb *dtmcli.BranchBarrier, db *sql.DB, busiCall dtmcli.BarrierBusiFunc) error

HandleTCCConfirm handle TCC Confirm phase

func (*BarrierHandler) HandleTCCTry 

func (b *BarrierHandler) HandleTCCTry(ctx context.Context, bb *dtmcli.BranchBarrier, db *sql.DB, busiCall dtmcli.BarrierBusiFunc) error

HandleTCCTry handle TCC Try phase

type DTMClient 

type DTMClient struct {
	// Embed base plugin, inherit common properties and methods of the plugin
	*plugins.BasePlugin
	// contains filtered or unexported fields
}

DTMClient represents the DTM client plugin

func NewDTMClient 

func NewDTMClient() *DTMClient

NewDTMClient creates a new DTM plugin instance

func (*DTMClient) CallBranch 

func (d *DTMClient) CallBranch(_ context.Context, _ interface{}, _, _, _ string) (*dtmcli.BranchBarrier, error)

CallBranch is deprecated. It does not perform actual TCC branch calls. Use dtmcli.TccGlobalTransaction with tcc.CallBranch, or TransactionHelper.ExecuteTCC instead.

func (*DTMClient) CheckHealth added in v1.5.4 

func (d *DTMClient) CheckHealth() error

CheckHealth performs a health check of the DTM client. When disabled, returns nil. When enabled, probes DTM server via /newGid or /query.

func (*DTMClient) CleanupTasks 

func (d *DTMClient) CleanupTasks() error

CleanupTasks cleans up DTM client resources

func (*DTMClient) Configure added in v1.5.4 

func (d *DTMClient) Configure(c any) error

Configure updates the plugin configuration at runtime

func (*DTMClient) GenerateGid 

func (d *DTMClient) GenerateGid() string

GenerateGid generates a new global transaction ID. Returns empty string if DTM server is unreachable (avoids panic from dtmcli.MustGenGid).

func (*DTMClient) GetConfig 

func (d *DTMClient) GetConfig() *conf.DTM

GetConfig returns the DTM configuration

func (*DTMClient) GetGRPCServer 

func (d *DTMClient) GetGRPCServer() string

GetGRPCServer returns the gRPC server address

func (*DTMClient) GetServerURL 

func (d *DTMClient) GetServerURL() string

GetServerURL returns the DTM server URL

func (*DTMClient) InitializeResources 

func (d *DTMClient) InitializeResources(rt plugins.Runtime) error

InitializeResources method is used to load and initialize the DTM plugin

func (*DTMClient) IsEnabled added in v1.5.4 

func (d *DTMClient) IsEnabled() bool

IsEnabled returns whether the DTM plugin is enabled.

func (*DTMClient) NewMsg 

func (d *DTMClient) NewMsg(gid string) *dtmcli.Msg

NewMsg creates a new 2-phase message transaction

func (*DTMClient) NewSaga 

func (d *DTMClient) NewSaga(gid string) *dtmcli.Saga

NewSaga creates a new SAGA transaction

func (*DTMClient) NewTcc 

func (d *DTMClient) NewTcc(gid string) *dtmcli.Tcc

NewTcc creates a new TCC transaction

func (*DTMClient) NewXa 

func (d *DTMClient) NewXa(gid string) *dtmcli.Xa

NewXa creates a new XA transaction

func (*DTMClient) StartupTasks 

func (d *DTMClient) StartupTasks() error

StartupTasks starts the DTM client

type DtmMetrics added in v1.5.4 

type DtmMetrics struct {
	// contains filtered or unexported fields
}

DtmMetrics defines DTM-related monitoring metrics for production observability.

func GetDtmMetrics added in v1.5.4 

func GetDtmMetrics() *DtmMetrics

GetDtmMetrics returns the DTM metrics instance for external use.

func (*DtmMetrics) DecActiveTransactions added in v1.5.4 

func (m *DtmMetrics) DecActiveTransactions()

DecActiveTransactions decrements active transaction count.

func (*DtmMetrics) IncActiveTransactions added in v1.5.4 

func (m *DtmMetrics) IncActiveTransactions()

IncActiveTransactions increments active transaction count.

func (*DtmMetrics) IncBarrierOperations added in v1.5.4 

func (m *DtmMetrics) IncBarrierOperations()

IncBarrierOperations increments barrier operation count.

func (*DtmMetrics) RecordGidRequest added in v1.5.4 

func (m *DtmMetrics) RecordGidRequest(status string)

RecordGidRequest records a GID generation request.

func (*DtmMetrics) RecordHealthCheck added in v1.5.4 

func (m *DtmMetrics) RecordHealthCheck(status string)

RecordHealthCheck records a health check.

func (*DtmMetrics) RecordTransaction added in v1.5.4 

func (m *DtmMetrics) RecordTransaction(txType, status string)

RecordTransaction records a transaction completion.

func (*DtmMetrics) RecordTransactionDuration added in v1.5.4 

func (m *DtmMetrics) RecordTransactionDuration(txType, status string, duration float64)

RecordTransactionDuration records transaction duration.

type ExampleService 

type ExampleService struct {
	// contains filtered or unexported fields
}

ExampleService example service, demonstrating how to use the DTM plugin

func NewExampleService 

func NewExampleService(dtmClient *DTMClient, db *sql.DB) *ExampleService

NewExampleService create example service

func (*ExampleService) BarrierExample 

func (s *ExampleService) BarrierExample(ctx context.Context, bb *dtmcli.BranchBarrier) error

BarrierExample branch barrier example - handling idempotency, suspension, and empty compensation issues

func (*ExampleService) HandleTCCCancelExample 

func (s *ExampleService) HandleTCCCancelExample(ctx context.Context, req map[string]interface{}) error

HandleTCCCancelExample TCC Cancel phase handling example

func (*ExampleService) HandleTCCConfirmExample 

func (s *ExampleService) HandleTCCConfirmExample(ctx context.Context, req map[string]interface{}) error

HandleTCCConfirmExample TCC Confirm phase handling example

func (*ExampleService) HandleTCCTryExample 

func (s *ExampleService) HandleTCCTryExample(ctx context.Context, req map[string]interface{}) error

HandleTCCTryExample TCC Try phase handling example

func (*ExampleService) MessageExample 

func (s *ExampleService) MessageExample(ctx context.Context, messageID string, content string) error

MessageExample message example - using 2-phase message pattern

func (*ExampleService) OrderExample 

func (s *ExampleService) OrderExample(ctx context.Context, orderID string, userID string, productID string, quantity int) error

OrderExample order example - using TCC pattern

func (*ExampleService) TransferExample 

func (s *ExampleService) TransferExample(ctx context.Context, fromAccount, toAccount string, amount float64) error

TransferExample transfer example - using SAGA pattern

func (*ExampleService) WorkflowExample 

func (s *ExampleService) WorkflowExample(ctx context.Context) error

WorkflowExample workflow example - using helper tools

type MsgBranch 

type MsgBranch struct {
	Action string      // Operation URL
	Data   interface{} // Request data
}

MsgBranch message branch definition

type SAGABranch 

type SAGABranch struct {
	Action     string      // Forward operation URL
	Compensate string      // Compensation operation URL
	Data       interface{} // Request data
}

SAGABranch SAGA branch definition

type TCCBranch 

type TCCBranch struct {
	Try     string      // Try phase URL
	Confirm string      // Confirm phase URL
	Cancel  string      // Cancel phase URL
	Data    interface{} // Request data
}

TCCBranch TCC branch definition

type TransactionHelper 

type TransactionHelper struct {
	// contains filtered or unexported fields
}

TransactionHelper transaction helper tool

func NewTransactionHelper 

func NewTransactionHelper(client *DTMClient) *TransactionHelper

NewTransactionHelper create transaction helper tool

func (*TransactionHelper) CheckTransactionStatus 

func (h *TransactionHelper) CheckTransactionStatus(gid string) (string, error)

CheckTransactionStatus queries DTM server for transaction status. Returns status: "succeed", "failed", "prepared", "submitted", "aborting", or "ongoing". Returns ErrNotImplemented if DTM query API is unavailable or returns unexpected format.

func (*TransactionHelper) ExecuteMsg 

func (h *TransactionHelper) ExecuteMsg(ctx context.Context, gid string, queryPrepared string, branches []MsgBranch, opts *TransactionOptions) error

ExecuteMsg execute 2-phase message transaction

func (*TransactionHelper) ExecuteSAGA 

func (h *TransactionHelper) ExecuteSAGA(ctx context.Context, gid string, branches []SAGABranch, opts *TransactionOptions) error

ExecuteSAGA execute SAGA transaction

func (*TransactionHelper) ExecuteTCC 

func (h *TransactionHelper) ExecuteTCC(ctx context.Context, gid string, branches []TCCBranch, opts *TransactionOptions) error

ExecuteTCC execute TCC transaction

func (*TransactionHelper) ExecuteXA 

func (h *TransactionHelper) ExecuteXA(ctx context.Context, gid string, branches []XABranch, opts *TransactionOptions) error

ExecuteXA execute XA transaction

func (*TransactionHelper) GenGid 

func (h *TransactionHelper) GenGid() (string, error)

GenGid generates a transaction GID with error handling

func (*TransactionHelper) MustGenGid 

func (h *TransactionHelper) MustGenGid() string

MustGenGid generates a global transaction ID. Returns empty string if generation fails (e.g. DTM server unreachable). Caller should check for empty and handle accordingly.

func (*TransactionHelper) RegisterGrpcService 

func (h *TransactionHelper) RegisterGrpcService(serviceName string, endpoint string) error

RegisterGrpcService registers gRPC service to DTM. Not yet implemented.

type TransactionOptions 

type TransactionOptions struct {
	// Transaction timeout (seconds)
	TimeoutToFail int64
	// Branch timeout (seconds)
	BranchTimeout int64
	// Retry interval (seconds)
	RetryInterval int64
	// Custom request headers
	CustomHeaders map[string]string
	// Whether to wait for result
	WaitResult bool
	// Concurrent execution of branches
	Concurrent bool
}

TransactionOptions transaction options

func DefaultTransactionOptions 

func DefaultTransactionOptions() *TransactionOptions

DefaultTransactionOptions returns default transaction options

type TransactionType 

type TransactionType string

TransactionType transaction type 

const (
	// TransTypeSAGA SAGA transaction type
	TransTypeSAGA TransactionType = "saga"
	// TransTypeTCC TCC transaction type
	TransTypeTCC TransactionType = "tcc"
	// TransTypeMsg 2-phase message transaction type
	TransTypeMsg TransactionType = "msg"
	// TransTypeXA XA transaction type
	TransTypeXA TransactionType = "xa"
)

type XABranch 

type XABranch struct {
	Action string // Operation URL
	Data   string // Request data as serialized string (e.g., JSON)
}

XABranch XA branch definition

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.