dtm

package module
v1.6.3 Latest Latest
Warning

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

 Go to latest Published: Jun 7, 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() now return preconfigured *dtmcli.Tcc / *dtmcli.Xa instances with the plugin's server URL, timeout, and retry settings applied.
  • Helper wrappers and DTM's native global transaction APIs remain the recommended orchestration path for full TCC / XA lifecycle management.

Documentation

Overview

Package dtm provides a DTM (Distributed Transaction Manager) plugin for the go-lynx framework. It wraps dtm-labs/client and supports SAGA, TCC, XA, and two-phase message patterns, with automatic transaction barrier handling, optional gRPC connectivity, Prometheus metrics, and a context-aware lifecycle.

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 any) (*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 {
	*plugins.BasePlugin
	// contains filtered or unexported fields
}

DTMClient is the Lynx plugin that manages the DTM distributed transaction client.

func NewDTMClient 

func NewDTMClient() *DTMClient

NewDTMClient creates a new DTM plugin instance.

func (*DTMClient) CallBranch 

func (d *DTMClient) CallBranch(_ context.Context, _ any, _, _, _ 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) InitializeContext added in v1.6.1 

func (d *DTMClient) InitializeContext(ctx context.Context, plugin plugins.Plugin, rt plugins.Runtime) error

func (*DTMClient) InitializeResources 

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

InitializeResources loads DTM configuration and resolves the server URL and gRPC address.

func (*DTMClient) IsContextAware added in v1.6.1 

func (d *DTMClient) IsContextAware() bool

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) PluginProtocol added in v1.6.1 

func (d *DTMClient) PluginProtocol() plugins.PluginProtocol

func (*DTMClient) StartContext added in v1.6.1 

func (d *DTMClient) StartContext(ctx context.Context, _ plugins.Plugin) error

func (*DTMClient) StartupTasks 

func (d *DTMClient) StartupTasks() error

StartupTasks starts the DTM client

func (*DTMClient) StopContext added in v1.6.1 

func (d *DTMClient) StopContext(ctx context.Context, _ plugins.Plugin) error

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]any) error

HandleTCCCancelExample TCC Cancel phase handling example

func (*ExampleService) HandleTCCConfirmExample 

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

HandleTCCConfirmExample TCC Confirm phase handling example

func (*ExampleService) HandleTCCTryExample 

func (s *ExampleService) HandleTCCTryExample(ctx context.Context, req map[string]any) 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   any    // Request data
}

MsgBranch message branch definition

type SAGABranch 

type SAGABranch struct {
	Action     string // Forward operation URL
	Compensate string // Compensation operation URL
	Data       any    // 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    any    // 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.