seata

README

Seata Distributed Transaction Plugin for Lynx Framework

The Seata Plugin provides comprehensive distributed transaction management for the Lynx framework using Alibaba's Seata (Simple Extensible Autonomous Transaction Architecture). It supports multiple transaction patterns including AT, TCC, SAGA, and XA modes.

Features

Core Transaction Support

• AT Mode: Automatic compensation transaction mode (recommended)
• TCC Mode: Try-Confirm-Cancel transaction mode
• SAGA Mode: Long-running business process transaction mode
• XA Mode: X/Open XA distributed transaction protocol
• Mixed Mode: Support for multiple transaction modes in the same application

Advanced Features

• Global Transaction Management: Centralized transaction coordination
• Branch Transaction Support: Local transaction management
• Compensation Mechanisms: Automatic rollback and compensation
• Transaction Recovery: Automatic transaction recovery after failures
• Performance Optimization: High-performance transaction processing
• Monitoring Integration: Comprehensive transaction monitoring

Security & Reliability

• ACID Compliance: Full ACID transaction properties
• Fault Tolerance: Automatic failure detection and recovery
• Data Consistency: Strong consistency guarantees
• Rollback Support: Comprehensive rollback mechanisms
• Timeout Management: Configurable transaction timeouts

Architecture

The plugin follows the Lynx framework's layered architecture:

┌─────────────────────────────────────────────────────────────┐
│                    Application Layer                        │
├─────────────────────────────────────────────────────────────┤
│                    Seata Plugin Layer                       │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │   Client    │  │   Manager   │  │   Configuration    │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│                    Transaction Layer                        │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │    AT       │  │    TCC      │  │        SAGA         │ │
│  │   Mode      │  │   Mode      │  │       Mode          │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│                    Registry Layer                           │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │   Nacos     │  │   Eureka    │  │       Consul        │ │
│  └─────────────┘  └─────────────┘  └─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

Configuration

Basic Configuration

lynx:
  seata:
    enabled: true
    application_id: "lynx-seata-client"
    tx_service_group: "my_test_tx_group"
    service:
      vgroup_mapping:
        my_test_tx_group: "default"
      grouplist:
        default: "127.0.0.1:8091"
      enable_degrade: false
      disable_global_transaction: false
    
    config:
      type: "file"
      file:
        name: "file.conf"
      nacos:
        server_addr: "127.0.0.1:8848"
        namespace: ""
        group: "SEATA_GROUP"
        username: ""
        password: ""
    
    registry:
      type: "file"
      file:
        name: "registry.conf"
      nacos:
        application: "seata-server"
        server_addr: "127.0.0.1:8848"
        group: "SEATA_GROUP"
        namespace: ""
        username: ""
        password: ""

Advanced Configuration

lynx:
  seata:
    enabled: true
    application_id: "lynx-seata-client"
    tx_service_group: "my_test_tx_group"
    
    # Transaction service configuration
    service:
      vgroup_mapping:
        my_test_tx_group: "default"
      grouplist:
        default: "127.0.0.1:8091"
      enable_degrade: false
      disable_global_transaction: false
      enable_auto_data_source_proxy: true
    
    # Configuration center
    config:
      type: "nacos"
      nacos:
        server_addr: "127.0.0.1:8848"
        namespace: "seata"
        group: "SEATA_GROUP"
        username: "nacos"
        password: "nacos"
        data_id: "seata.properties"
    
    # Registry center
    registry:
      type: "nacos"
      nacos:
        application: "seata-server"
        server_addr: "127.0.0.1:8848"
        group: "SEATA_GROUP"
        namespace: "seata"
        username: "nacos"
        password: "nacos"
    
    # Client configuration
    client:
      rm:
        async_commit_buffer_limit: 10000
        report_retry_count: 5
        table_meta_check_enable: false
        report_success_enable: false
        saga_branch_register_enable: false
        saga_json_parser: "fastjson"
        saga_retry_persist_mode_update: false
        saga_retry_persist_period: 1000
        lock_retry_policy_branch_rollback_on_conflict: true
      tm:
        commit_retry_count: 5
        rollback_retry_count: 5
        default_global_transaction_timeout: 60000
        degrade_check: false
        degrade_check_allow_times: 10
        degrade_check_period: 2000
        interceptor_order: -2147482648
      undo:
        data_validation: true
        log_serialization: "jackson"
        log_table: "undo_log"
        only_care_update_columns: true
      log:
        exception_rate: 100

Usage

Basic Usage

package main

import (
    "context"
    "time"

    seata "github.com/go-lynx/lynx-seata"
)

func main() {
    // Get the Seata plugin instance (requires Lynx app to be initialized)
    plugin := seata.GetPlugin()
    if plugin == nil || !plugin.IsEnabled() {
        // Seata is disabled or plugin not loaded
        return
    }

    // Execute business logic within a global transaction
    ctx := context.Background()
    err := plugin.WithGlobalTx(ctx, "business-service", 60*time.Second, func(ctx context.Context) error {
        // Your business logic here - use seata-go AT mode with DB operations
        // or TCC/SAGA as needed. The ctx carries the XID for propagation.
        return executeBusinessLogic(ctx)
    })
    if err != nil {
        panic(err)
    }
}

func executeBusinessLogic(ctx context.Context) error {
    // Business operations - use Seata AT driver for DB or TCC/SAGA APIs
    return nil
}

TCC Mode Usage

// TCC mode - Try-Confirm-Cancel
func executeTCCBusiness(ctx context.Context, tx *seata.Transaction) error {
    // Try phase
    err := tryReserveInventory(ctx, "product-1", 10)
    if err != nil {
        return err
    }
    
    err = tryCreateOrder(ctx, "order-123")
    if err != nil {
        // Cancel phase
        cancelReserveInventory(ctx, "product-1", 10)
        return err
    }
    
    // Confirm phase
    err = confirmReserveInventory(ctx, "product-1", 10)
    if err != nil {
        return err
    }
    
    err = confirmCreateOrder(ctx, "order-123")
    if err != nil {
        return err
    }
    
    return nil
}

SAGA Mode Usage

// SAGA mode - Long-running business process
func executeSAGABusiness(ctx context.Context) error {
    saga := seataClient.NewSaga("order-process")
    
    // Add saga steps
    saga.AddStep("reserve-inventory", 
        func(ctx context.Context) error {
            return reserveInventory(ctx, "product-1", 10)
        },
        func(ctx context.Context) error {
            return releaseInventory(ctx, "product-1", 10)
        })
    
    saga.AddStep("create-order",
        func(ctx context.Context) error {
            return createOrder(ctx, "order-123")
        },
        func(ctx context.Context) error {
            return cancelOrder(ctx, "order-123")
        })
    
    // Execute saga
    return saga.Execute(ctx)
}

XA Mode Usage

// XA mode - X/Open XA protocol
func executeXABusiness(ctx context.Context) error {
    xa := seataClient.NewXA("xa-transaction")
    
    // Add XA resources
    err := xa.AddResource("mysql", "jdbc:mysql://localhost:3306/db1")
    if err != nil {
        return err
    }
    
    err = xa.AddResource("mysql", "jdbc:mysql://localhost:3306/db2")
    if err != nil {
        return err
    }
    
    // Execute XA transaction
    return xa.Execute(ctx, func(ctx context.Context) error {
        // Business logic using XA resources
        return executeBusinessLogic(ctx)
    })
}

API Reference

GetPlugin

Returns the Seata plugin instance from the Lynx application. Returns nil if the plugin is not loaded or Lynx is not initialized.

TxSeataClient Methods

• GetConfig() *conf.Seata - Returns the current configuration
• GetConfigFilePath() string - Returns the Seata config file path
• IsEnabled() bool - Checks if Seata is enabled
• WithGlobalTx(ctx, name, timeout, business) error - Executes business logic within a global transaction
• CheckHealth() error - Performs health check

Transaction Patterns

1. AT Mode (Automatic Compensation)

AT mode is the most commonly used mode in Seata. It automatically generates reverse SQL for compensation.

// AT mode automatically handles compensation
func atModeExample(ctx context.Context) error {
    tx, err := seataClient.Begin(ctx, "at-transaction")
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // These operations will be automatically compensated if transaction fails
    err = updateInventory(ctx, "product-1", -10)
    if err != nil {
        return err
    }
    
    err = createOrder(ctx, "order-123")
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

2. TCC Mode (Try-Confirm-Cancel)

TCC mode requires manual implementation of Try, Confirm, and Cancel methods.

// TCC mode requires manual compensation logic
func tccModeExample(ctx context.Context) error {
    tx, err := seataClient.Begin(ctx, "tcc-transaction")
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // Try phase
    err = tryReserveInventory(ctx, "product-1", 10)
    if err != nil {
        return err
    }
    
    // Confirm phase (called on success)
    err = confirmReserveInventory(ctx, "product-1", 10)
    if err != nil {
        // Cancel phase (called on failure)
        cancelReserveInventory(ctx, "product-1", 10)
        return err
    }
    
    return tx.Commit()
}

3. SAGA Mode (Long-running Process)

SAGA mode is suitable for long-running business processes.

// SAGA mode for long-running processes
func sagaModeExample(ctx context.Context) error {
    saga := seataClient.NewSaga("order-process")
    
    // Define saga steps with compensation
    saga.AddStep("reserve-inventory",
        func(ctx context.Context) error {
            return reserveInventory(ctx, "product-1", 10)
        },
        func(ctx context.Context) error {
            return releaseInventory(ctx, "product-1", 10)
        })
    
    saga.AddStep("create-order",
        func(ctx context.Context) error {
            return createOrder(ctx, "order-123")
        },
        func(ctx context.Context) error {
            return cancelOrder(ctx, "order-123")
        })
    
    saga.AddStep("send-notification",
        func(ctx context.Context) error {
            return sendOrderNotification(ctx, "order-123")
        },
        func(ctx context.Context) error {
            return cancelOrderNotification(ctx, "order-123")
        })
    
    return saga.Execute(ctx)
}

Monitoring and Metrics

Health Checks

plugin := seata.GetPlugin()
if plugin != nil && plugin.IsEnabled() {
    err := plugin.CheckHealth()
    if err != nil {
        log.Printf("Seata health check failed: %v", err)
    }
}

Prometheus Metrics

The plugin exposes comprehensive Prometheus metrics:

Transaction Metrics

• lynx_seata_transactions_total - Total transactions
• lynx_seata_transactions_active - Active transactions
• lynx_seata_transactions_committed_total - Committed transactions
• lynx_seata_transactions_rolled_back_total - Rolled back transactions
• lynx_seata_transaction_duration_seconds - Transaction duration

Branch Metrics

• lynx_seata_branches_total - Total branch transactions
• lynx_seata_branches_committed_total - Committed branches
• lynx_seata_branches_rolled_back_total - Rolled back branches

Error Metrics

• lynx_seata_errors_total - Total errors
• lynx_seata_timeout_errors_total - Timeout errors
• lynx_seata_network_errors_total - Network errors

Deployment

Seata Server Setup

1. Download Seata Server

wget https://github.com/seata/seata/releases/download/v1.8.0/seata-server-1.8.0.zip
unzip seata-server-1.8.0.zip
cd seata

1. Configure Seata Server

# Edit conf/application.yml
vim conf/application.yml

1. Start Seata Server

./bin/seata-server.sh -p 8091 -h 127.0.0.1 -m file

Docker Deployment

version: '3.8'
services:
  seata-server:
    image: seataio/seata-server:1.8.0
    ports:
      - "8091:8091"
    environment:
      - SEATA_PORT=8091
      - STORE_MODE=file
    volumes:
      - ./seata:/opt/seata-server/conf

Kubernetes Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: seata-server
spec:
  replicas: 1
  selector:
    matchLabels:
      app: seata-server
  template:
    metadata:
      labels:
        app: seata-server
    spec:
      containers:
      - name: seata-server
        image: seataio/seata-server:1.8.0
        ports:
        - containerPort: 8091
        env:
        - name: SEATA_PORT
          value: "8091"
        - name: STORE_MODE
          value: "file"

Troubleshooting

Common Issues

1. Transaction Not Starting

   • Check Seata server connectivity
   • Verify configuration settings
   • Check network connectivity

2. Compensation Failures

   • Verify compensation logic
   • Check database connectivity
   • Review transaction logs

3. Performance Issues

   • Monitor transaction duration
   • Check resource utilization
   • Review configuration settings

4. Configuration Errors

   • Validate configuration files
   • Check registry and config center settings
   • Verify service group mappings

Debug Mode

Enable debug logging for detailed troubleshooting:

lynx:
  seata:
    client:
      log:
        exception_rate: 100
    logging:
      level: "DEBUG"

Best Practices

Transaction Design

• Keep transactions short and focused
• Avoid long-running transactions
• Design for compensation
• Use appropriate transaction modes

Performance

• Monitor transaction performance
• Optimize compensation logic
• Use connection pooling
• Implement proper timeout handling

Monitoring

• Set up comprehensive monitoring
• Monitor transaction success rates
• Track compensation performance
• Alert on transaction failures

Security

• Secure transaction data
• Implement proper authentication
• Use encrypted connections
• Regular security audits

Contributing

Contributions are welcome! Please see the main Lynx framework contribution guidelines.

License

This plugin is part of the Lynx framework and follows the same license terms.

Support

For support and questions:

• GitHub Issues: Lynx Framework Issues
• Documentation: Lynx Documentation
• Community: Lynx Community
• Seata Documentation: Seata Official Docs

Documentation

Index

• type SeataMetrics
  • func GetMetrics() *SeataMetrics
  • func (m *SeataMetrics) RecordHealthCheck(status string)
  • func (m *SeataMetrics) RecordTransaction(status string)
  • func (m *SeataMetrics) RecordTransactionDuration(status string, duration float64)
  • func (m *SeataMetrics) SetActiveTransactions(count float64)
  • func (m *SeataMetrics) SetBranchTransactions(count float64)
• type TxSeataClient
  • func GetPlugin() *TxSeataClient
  • func NewTxSeataClient() *TxSeataClient
  • func (t *TxSeataClient) CheckHealth() error
  • func (t *TxSeataClient) CleanupTasks() error
  • func (t *TxSeataClient) Configure(cfg any) error
  • func (t *TxSeataClient) GetConfig() *conf.Seata
  • func (t *TxSeataClient) GetConfigFilePath() string
  • func (t *TxSeataClient) InitializeResources(rt plugins.Runtime) error
  • func (t *TxSeataClient) IsEnabled() bool
  • func (t *TxSeataClient) StartupTasks() error
  • func (t *TxSeataClient) WithGlobalTx(ctx context.Context, name string, timeout time.Duration, ...) error

Types

type SeataMetrics

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

SeataMetrics defines Seata-related monitoring metrics.

func GetMetrics

func GetMetrics() *SeataMetrics

GetMetrics returns the Seata metrics instance for external use.

func (*SeataMetrics) RecordHealthCheck

func (m *SeataMetrics) RecordHealthCheck(status string)

RecordHealthCheck records a health check.

func (*SeataMetrics) RecordTransaction

func (m *SeataMetrics) RecordTransaction(status string)

RecordTransaction records a transaction completion.

func (*SeataMetrics) RecordTransactionDuration

func (m *SeataMetrics) RecordTransactionDuration(status string, duration float64)

RecordTransactionDuration records transaction duration.

func (*SeataMetrics) SetActiveTransactions

func (m *SeataMetrics) SetActiveTransactions(count float64)

SetActiveTransactions sets the number of active transactions.

func (*SeataMetrics) SetBranchTransactions

func (m *SeataMetrics) SetBranchTransactions(count float64)

SetBranchTransactions sets the number of branch transactions.

type TxSeataClient

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

func GetPlugin

func GetPlugin() *TxSeataClient

GetPlugin obtains the TxSeataClient plugin instance from the application's plugin manager.

func NewTxSeataClient

func NewTxSeataClient() *TxSeataClient

NewTxSeataClient creates a new Seata plugin instance.

func (*TxSeataClient) CheckHealth

func (t *TxSeataClient) CheckHealth() error

CheckHealth performs a health check of the Seata client. When disabled, returns nil. When enabled, verifies the plugin is in a valid state.

func (*TxSeataClient) CleanupTasks

func (t *TxSeataClient) CleanupTasks() error

CleanupTasks performs cleanup during plugin shutdown. Seata-go does not expose a public shutdown API; connections will be released when the process exits.

func (*TxSeataClient) Configure

func (t *TxSeataClient) Configure(cfg any) error

Configure updates the plugin configuration. Overrides base to apply *conf.Seata.

func (*TxSeataClient) GetConfig

func (t *TxSeataClient) GetConfig() *conf.Seata

GetConfig returns the Seata configuration.

func (*TxSeataClient) GetConfigFilePath

func (t *TxSeataClient) GetConfigFilePath() string

GetConfigFilePath returns the Seata configuration file path.

func (*TxSeataClient) InitializeResources

func (t *TxSeataClient) InitializeResources(rt plugins.Runtime) error

InitializeResources loads and initializes the Seata plugin configuration.

func (*TxSeataClient) IsEnabled

func (t *TxSeataClient) IsEnabled() bool

IsEnabled returns whether the Seata plugin is enabled.

func (*TxSeataClient) StartupTasks

func (t *TxSeataClient) StartupTasks() error

StartupTasks initializes the Seata client when enabled.

func (*TxSeataClient) WithGlobalTx

func (t *TxSeataClient) WithGlobalTx(ctx context.Context, name string, timeout time.Duration, business func(context.Context) error) error

WithGlobalTx executes the given business function within a global transaction. It wraps seata-go's tm.WithGlobalTx with sensible defaults.