0x0B-a Internal Transfer Architecture (Strict FSM)

🇺🇸 English    |    🇨🇳 中文

🇺🇸 English

📦 Code Changes: View Diff

1. Problem Statement

1.1 System Topology

1.2 The Core Constraint

These two systems cannot share a transaction. There is no XA/2PC database protocol. Therefore: We must build our own 2-Phase Commit using an external FSM Coordinator.

1.5 Security Pre-Validation (MANDATORY)

Caution

Defense-in-Depth All checks below MUST be performed at every independent module, not just API layer.

• API Layer: First line of defense, reject obviously invalid requests
• Coordinator: Re-validate, prevent internal calls bypassing API
• Adapters: Final defense, each adapter must independently validate parameters
• UBSCore: Last check before in-memory operations

Safety > Performance. The cost of redundant checks is acceptable; security vulnerabilities are not.

1.5.1 Identity & Authorization Checks

1.5.2 Account Type Checks

1.5.3 Amount Checks

1.5.4 Asset Checks

1.5.5 Account Status Checks

Account Initialization Rules (Overview)

Note

• Specific initialization behaviors and business rules for each account type are defined in their dedicated documents.
• Each account has its own state definitions (e.g., whether transfer is allowed); not detailed here.
• Default State: On account initialization, transfer is allowed by default.

Account Status Check Table

1.5.6 Rate Limiting - [P2 Future Optimization]

Note

This is a V2 optimization. V1 may skip this.

1.5.7 Idempotency Check

1.5.8 Check Order (Recommended)

1. Authentication (JWT valid?)
2. Authorization (user_id match?)
3. Request Format (from/to/amount valid?)
4. Account Type (from != to, type supported?)
5. Asset Check (exists? enabled? transferable?)
6. Amount Check (range? precision? overflow?)
7. Rate Limiting (exceeded?)
8. Idempotency (duplicate?)
9. Balance Check (sufficient?) ← Check last, avoid unnecessary queries

2. FSM Design (The State Machine)

2.0 Library Choice: rust-fsm

We use the rust-fsm library, providing:

• ✅ Compile-time validation - Illegal state transitions cause compile errors.
• ✅ Declarative DSL - Clearly defined states and transitions.
• ✅ Type Safety - Prevents missing match arms.

Cargo.toml:

[dependencies]
rust-fsm = "0.7"

DSL Definition:

#![allow(unused)]
fn main() {
use rust_fsm::*;

state_machine! {
    derive(Debug, Clone, Copy, PartialEq, Eq)
    
    TransferFsm(Init)  // Initial State
    
    // State Definitions
    Init => {
        SourceWithdrawOk => SourceDone,
        SourceWithdrawFail => Failed,
    },
    SourceDone => {
        TargetDepositOk => Committed,
        TargetDepositFail => Compensating,
        TargetDepositUnknown => SourceDone [loop],  // Stay, Infinite Retry
    },
    Compensating => {
        RefundOk => RolledBack,
        RefundFail => Compensating [loop],  // Stay, Infinite Retry
    },
    // Terminal States
    Committed,
    Failed,
    RolledBack,
}
}

Note

The DSL above is used for compile-time validation of state transition validity. Actual runtime state is stored in PostgreSQL and updated via CAS.

2.0.1 Core State Flow (Top Level)

                               ┌─────────────────────────────────────────────────────────┐
                               │              INTERNAL TRANSFER FSM                       │
                               └─────────────────────────────────────────────────────────┘

    ┌─────────────────────────────── Happy Path ────────────────────────────────────────────┐
    │                                                                                       │
    │    ┌─────────┐                    ┌─────────────┐                    ┌───────────────┐  │
    │    │  INIT   │   Source Deduct ✓  │ SOURCE_DONE │   Target Credit ✓  │               │  │
    │    │(Request)│ ─────────────────▶ │ (In-Flight) │ ─────────────────▶ │   COMMITTED   │  │
    │    └─────────┘                    └─────────────┘                    │               │  │
    │         │                               │                            └───────────────┘  │
    │         │                               │                                   ✅          │
    └─────────│───────────────────────────────│───────────────────────────────────────────────┘
              │                               │
              │                               │
              │                               ▼
              │                     ╔══════════════════════════════════════════════════╗
              │                     ║  🔒 ATOMIC COMMIT                               ║
              │                     ║                                                  ║
              │                     ║  IF AND ONLY IF:                                 ║
              │                     ║    FROM.withdraw = SUCCESS  ✓                   ║
              │                     ║    TO.deposit    = SUCCESS  ✓                   ║
              │                     ║                                                  ║
              │                     ║  EXECUTE: CAS(SOURCE_DONE → COMMITTED)           ║
              │                     ║  Must be atomic and non-interruptible.           ║
              │                     ╚══════════════════════════════════════════════════╝
              │                               │
              │ Source Deduction Fail         │ Target Credit Fail (EXPLICIT_FAIL)
              ▼                               ▼
        ┌──────────┐                   ┌──────────────┐
        │  FAILED  │                   │ COMPENSATING │◀───────────┐
        │ (Source) │                   │  (Refunding) │            │ Refund Fail (Infinite Retry)
        └──────────┘                   └──────────────┘────────────┘
             ❌                               │ Refund Success
                                              ▼
                                       ┌─────────────┐
                                       │ ROLLED_BACK │
                                       │ (Restored)  │
                                       └─────────────┘
                                             ↩️

    ╔════════════════════════════════════════════════════════════════════════════════════════╗
    ║  ⚠️ Target Unknown (TIMEOUT/UNKNOWN) → Stay SOURCE_DONE, Infinite Retry, NEVER rollback. ║
    ╚════════════════════════════════════════════════════════════════════════════════════════╝

Core State Description:

Important

SOURCE_DONE is the most critical state - funds have left the source account but have not yet reached the target. At this point, the state MUST NOT be lost; it must eventually reach COMMITTED or ROLLED_BACK.

2.1 States (Exhaustive)

2.2 State Transition Rules (Exhaustive)

┌───────────────────────────────────────────────────────────────────────────────┐
│                         CANONICAL STATE TRANSITIONS                           │
├───────────────────────────────────────────────────────────────────────────────┤
│                                                                               │
│  INIT ──────[CAS OK]───────► SOURCE_PENDING                                   │
│    │                              │                                           │
│    │                              ├──[Adapter OK]────► SOURCE_DONE            │
│    │                              │                         │                 │
│    │                              └──[Adapter FAIL]──► FAILED (Terminal)      │
│    │                                                        │                 │
│    │                                                        │                 │
│    │                              SOURCE_DONE ──[CAS OK]──► TARGET_PENDING    │
│    │                                                             │            │
│    │                        ┌────────────────────────────────────┤            │
│    │                        │                                    │            │
│    │            [Adapter OK]│                       [Adapter FAIL]            │
│    │                        │                                    │            │
│    │                        ▼                                    ▼            │
│    │                   COMMITTED                     ┌───────────────────┐    │
│    │                   (Terminal)                    │ SOURCE REVERSIBLE?│    │
│    │                                                 └─────────┬─────────┘    │
│    │                                                   YES     │     NO       │
│    │                                                   ▼       │     ▼        │
│    │                                           COMPENSATING    │  INFINITE    │
│    │                                                 │         │   RETRY      │
│    │                                    [Refund OK]  │         │ (Stay in     │
│    │                                         ▼       │         │  TARGET_     │
│    │                                    ROLLED_BACK  │         │  PENDING)    │
│    │                                    (Terminal)   │         │              │
│    │                                                 │         │              │
│    └─────────────────────────────────────────────────┴─────────┴──────────────┘

2.3 Reversibility Rule (CRITICAL)

Core Principle: Only when an Adapter returns an explicitly defined failure can we safely rollback.

Caution

Only EXPLICIT_FAIL allows safe rollback. Any unknown state (Timeout, Pending, Network Error) means funds are In-Flight. We cannot know whether the counterparty has processed the request. Rash rollback will cause Double Spend or Fund Loss. Only safe actions: Infinite Retry or Manual Intervention.

3. Transfer Scenarios (Step-by-Step)

3.1 Scenario A: Funding → Spot (Deposit to Trading)

Happy Path:

Failure Path (Target Fails):

3.2 Scenario B: Spot → Funding (Withdraw from Trading)

Happy Path:

Failure Path (Target Fails):

Warning

Only enter COMPENSATING when Target returns EXPLICIT_FAIL. If Timeout or Unknown, funds are In-Flight. Must Infinite Retry or Manual Intervention.

4. Failure Mode and Effects Analysis (FMEA)

4.1 Phase 1 Failures (Source Operation)

4.2 Phase 2 Failures (Target Operation)

4.3 Compensation Failures

5. Idempotency Requirements (MANDATORY)

5.1 Why Idempotency?

Retries are the foundation of crash recovery. Without idempotency, a retry will cause double execution (double deduction, double credit).

5.2 Implementation (Funding Adapter)

Requirement: Given the same req_id, calling withdraw() or deposit() multiple times MUST have the same effect as calling it once.

Mechanism:

1. transfers_tb has UNIQUE(req_id).
2. Atomic Transaction:

   BEGIN;
   -- Check if already processed
   SELECT state FROM transfers_tb WHERE req_id = $1;
   IF state >= expected_post_state THEN
       RETURN 'AlreadyProcessed';
   END IF;
   
   -- Perform balance update
   UPDATE balances_tb SET amount = amount - $2 WHERE user_id = $3 AND asset_id = $4 AND amount >= $2;
   IF NOT FOUND THEN
       RETURN 'InsufficientBalance';
   END IF;
   
   -- Update state
   UPDATE transfers_tb SET state = $new_state, updated_at = NOW() WHERE req_id = $1;
   COMMIT;
   RETURN 'Success';
   

5.3 Implementation (Trading Adapter)

Requirement: Same as above. UBSCore MUST reject duplicate req_id.

Mechanism:

1. InternalOrder includes req_id field (or cid).
2. UBSCore maintains a ProcessedTransferSet (HashSet in RAM, rebuilt from WAL on restart).
3. On receiving Transfer Order:

   IF req_id IN ProcessedTransferSet THEN
       RETURN 'AlreadyProcessed' (Success, no-op)
   ELSE
       ProcessTransfer()
       ProcessedTransferSet.insert(req_id)
       WriteWAL(TransferEvent)
       RETURN 'Success'
   END IF
   

6. Recovery Worker (Zombie Handler)

6.1 Purpose

On Coordinator startup (or periodically), scan for “stuck” transfers and resume them.

6.2 Query

SELECT * FROM transfers_tb 
WHERE state IN (0, 10, 20, 30, -20) -- INIT, SOURCE_PENDING, SOURCE_DONE, TARGET_PENDING, COMPENSATING
  AND updated_at < NOW() - INTERVAL '1 minute'; -- Stale threshold

6.3 Recovery Logic

7. Data Model

7.1 Table: transfers_tb

CREATE TABLE transfers_tb (
    transfer_id   BIGSERIAL PRIMARY KEY,
    req_id        VARCHAR(26) UNIQUE NOT NULL,  -- Server-generated Unique ID (ULID)
    cid           VARCHAR(64) UNIQUE,           -- Client Idempotency Key (Optional)
    user_id       BIGINT NOT NULL,
    asset_id      INTEGER NOT NULL,
    amount        DECIMAL(30, 8) NOT NULL,
    transfer_type SMALLINT NOT NULL,            -- 1 = Funding->Spot, 2 = Spot->Funding
    source_type   SMALLINT NOT NULL,            -- 1 = Funding, 2 = Trading
    state         SMALLINT NOT NULL DEFAULT 0,  -- FSM State ID
    error_message TEXT,                         -- Last error (for debugging)
    retry_count   INTEGER NOT NULL DEFAULT 0,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_transfers_state ON transfers_tb(state) WHERE state NOT IN (40, -10, -30);

7.2 Invariant Check

Run periodically to detect data corruption:

-- Sum of Funding + Trading + In-Flight should be constant per user per asset
-- In-Flight = SUM(amount) WHERE state IN (SOURCE_DONE, TARGET_PENDING, COMPENSATING)

8. API Contract

8.1 Endpoint: POST /api/v1/internal_transfer

Request:

{
  "from": "SPOT",       // Source account type
  "to": "FUNDING",     // Target account type
  "asset": "USDT",
  "amount": "100.00"
}

Account Type Enum (AccountType):

Response:

{
  "transfer_id": 12345,
  "req_id": "01JFVQ2X8Z0Y1M3N4P5R6S7T8U",  // Server-generated (ULID)
  "from": "SPOT",
  "to": "FUNDING",
  "state": "COMMITTED",  // or "PENDING" if async
  "message": "Transfer successful"
}

8.2 Query Endpoint: GET /api/v1/internal_transfer/:req_id

Response:

{
  "transfer_id": 12345,
  "req_id": "sr-1734912345678901234",
  "from": "SPOT",
  "to": "FUNDING",
  "asset": "USDT",
  "amount": "100.00",
  "state": "COMMITTED",
  "created_at": "2024-12-23T14:00:00Z",
  "updated_at": "2024-12-23T14:00:01Z"
}

Important

req_id is SERVER-GENERATED, not client. If client needs idempotency, use optional cid (client_order_id) field. Server will check for duplicates and return existing result.

Error Codes:

9. Implementation Pseudocode (Critical State Checks)

9.1 API Layer

function handle_transfer_request(request, auth_context):
    // ========== Defense-in-Depth Layer 1: API Layer ==========
    
    // 1. Identity Authentication
    if !auth_context.is_valid():
        return Error(UNAUTHORIZED)
    
    // 2. User ID Consistency (Prevent cross-user attacks)
    if request.user_id != auth_context.user_id:
        return Error(FORBIDDEN, "User ID mismatch")
    
    // 3. Account Type Check
    if request.from == request.to:
        return Error(SAME_ACCOUNT)
    
    if request.from NOT IN [FUNDING, SPOT]:
        return Error(INVALID_ACCOUNT_TYPE)
    
    if request.to NOT IN [FUNDING, SPOT]:
        return Error(INVALID_ACCOUNT_TYPE)
    
    // 4. Amount Check
    if request.amount <= 0:
        return Error(INVALID_AMOUNT)
    
    if decimal_places(request.amount) > asset.precision:
        return Error(PRECISION_OVERFLOW)
    
    // 5. Idempotency Check
    if request.cid:
        existing = db.find_by_cid(request.cid)
        if existing:
            return Success(existing)  // Return existing result
    
    // 6. Asset Check
    asset = db.get_asset(request.asset_id)
    if !asset or asset.status != ACTIVE:
        return Error(INVALID_ASSET)
    
    // 7. Call Coordinator
    result = coordinator.create_and_execute(request)
    return result

9.2 Coordinator Layer

function create_and_execute(request):
    // ========== Defense-in-Depth Layer 2: Coordinator ==========
    
    // Re-verify (Prevent internal calls bypassing API)
    ASSERT request.from != request.to
    ASSERT request.amount > 0
    ASSERT request.user_id > 0
    
    // Generate unique ID
    req_id = ulid.new()
    
    // Create transfer record (State = INIT)
    transfer = TransferRecord {
        req_id: req_id,
        user_id: request.user_id,
        from: request.from,
        to: request.to,
        asset_id: request.asset_id,
        amount: request.amount,
        state: INIT,
        created_at: now()
    }
    
    db.insert(transfer)
    log.info("Transfer created", req_id)
    
    // Execute FSM
    return execute_fsm(req_id)

function execute_fsm(req_id):
    loop:
        transfer = db.get(req_id)
        
        if transfer.state.is_terminal():
            return transfer
        
        new_state = step(transfer)
        
        if new_state == transfer.state:
            // No progress, wait for retry
            sleep(RETRY_INTERVAL)
            continue
    
function step(transfer):
    match transfer.state:
        INIT:
            return step_init(transfer)
        SOURCE_PENDING:
            return step_source_pending(transfer)
        SOURCE_DONE:
            return step_source_done(transfer)
        TARGET_PENDING:
            return step_target_pending(transfer)
        COMPENSATING:
            return step_compensating(transfer)
        _:
            return transfer.state  // Terminal, no processing
    
function step_init(transfer):
    // CAS: Persist state BEFORE calling adapter (Persist-Before-Call)
    success = db.cas_update(
        req_id = transfer.req_id,
        old_state = INIT,
        new_state = SOURCE_PENDING
    )
    
    if !success:
        return db.get(transfer.req_id).state
    
    // Get source adapter
    source_adapter = get_adapter(transfer.from)
    
    // ========== Defense-in-Depth Layer 3: Adapter ==========
    result = source_adapter.withdraw(
        req_id = transfer.req_id,
        user_id = transfer.user_id,
        asset_id = transfer.asset_id,
        amount = transfer.amount
    )
    
    match result:
        SUCCESS:
            db.cas_update(transfer.req_id, SOURCE_PENDING, SOURCE_DONE)
            return SOURCE_DONE
        
        EXPLICIT_FAIL(reason):
            db.update_with_error(transfer.req_id, SOURCE_PENDING, FAILED, reason)
            return FAILED
        
        TIMEOUT | PENDING | NETWORK_ERROR | UNKNOWN:
            log.warn("Source withdraw unknown state", transfer.req_id)
            return SOURCE_PENDING

function step_source_done(transfer):
    // ========== Enter SOURCE_DONE: Funds In-Flight, must reach terminal state ==========
    
    // CAS update to TARGET_PENDING
    success = db.cas_update(transfer.req_id, SOURCE_DONE, TARGET_PENDING)
    if !success:
        return db.get(transfer.req_id).state
    
    // Get target adapter
    target_adapter = get_adapter(transfer.to)
    
    // ========== Defense-in-Depth Layer 4: Target Adapter ==========
    result = target_adapter.deposit(
        req_id = transfer.req_id,
        user_id = transfer.user_id,
        asset_id = transfer.asset_id,
        amount = transfer.amount
    )
    
    match result:
        SUCCESS:
            // ╔════════════════════════════════════════════════════════════════╗
            // ║  🔒 ATOMIC COMMIT - CRITICAL STEP!                             ║
            // ║                                                                ║
            // ║  At this point:                                                ║
            // ║    FROM.withdraw = SUCCESS ✓ (already confirmed)               ║
            // ║    TO.deposit    = SUCCESS ✓ (just confirmed)                  ║
            // ║                                                                ║
            // ║  Execute Atomic CAS Commit:                                    ║
            // ║    CAS(TARGET_PENDING → COMMITTED)                            ║
            // ║                                                                ║
            // ║  Once this CAS succeeds, the transfer is irreversible!         ║
            // ╚════════════════════════════════════════════════════════════════╝
            
            commit_success = db.cas_update(transfer.req_id, TARGET_PENDING, COMMITTED)
            
            if !commit_success:
                return db.get(transfer.req_id).state
            
            log.info("🔒 ATOMIC COMMIT SUCCESS", transfer.req_id)
            return COMMITTED
        
        EXPLICIT_FAIL(reason):
            db.update_with_error(transfer.req_id, TARGET_PENDING, COMPENSATING, reason)
            return COMPENSATING
        
        TIMEOUT | PENDING | NETWORK_ERROR | UNKNOWN:
            // ========== CRITICAL: Unknown state, MUST NOT compensate! ==========
            log.critical("Target deposit unknown state - INFINITE RETRY", transfer.req_id)
            alert_ops("Transfer stuck in TARGET_PENDING", transfer.req_id)
            return TARGET_PENDING  // Stay and retry

function step_compensating(transfer):
    source_adapter = get_adapter(transfer.from)
    
    result = source_adapter.refund(
        req_id = transfer.req_id,
        user_id = transfer.user_id,
        asset_id = transfer.asset_id,
        amount = transfer.amount
    )
    
    match result:
        SUCCESS:
            db.cas_update(transfer.req_id, COMPENSATING, ROLLED_BACK)
            log.info("Transfer rolled back", transfer.req_id)
            return ROLLED_BACK
        
        _:
            log.critical("Refund failed - MUST RETRY", transfer.req_id)
            return COMPENSATING

9.3 Adapter Layer (Example: Funding Adapter)

function withdraw(req_id, user_id, asset_id, amount):
    // ========== Defense-in-Depth Layer 3: Adapter Internal Verification ==========
    
    // Re-verify parameters (Do not trust caller)
    ASSERT amount > 0
    ASSERT user_id > 0
    ASSERT asset_id > 0
    
    // Idempotency Check
    existing = db.find_transfer_operation(req_id, "WITHDRAW")
    if existing:
        return existing.result
    
    // Begin transaction
    tx = db.begin_transaction()
    try:
        // SELECT FOR UPDATE
        account = tx.select_for_update(
            "SELECT * FROM balances_tb WHERE user_id = ? AND asset_id = ? AND account_type = 'FUNDING'"
        )
        
        if !account:
            tx.rollback()
            return EXPLICIT_FAIL("SOURCE_ACCOUNT_NOT_FOUND")
        
        if account.status == FROZEN:
            tx.rollback()
            return EXPLICIT_FAIL("ACCOUNT_FROZEN")
        
        if account.available < amount:
            tx.rollback()
            return EXPLICIT_FAIL("INSUFFICIENT_BALANCE")
        
        // Execute deduction
        tx.update("UPDATE balances_tb SET available = available - ? WHERE id = ?", amount, account.id)
        
        // Record operation for idempotency
        tx.insert("INSERT INTO transfer_operations (req_id, op_type, result) VALUES (?, 'WITHDRAW', 'SUCCESS')")
        
        tx.commit()
        return SUCCESS
        
    catch Exception as e:
        tx.rollback()
        log.error("Withdraw failed", req_id, e)
        return UNKNOWN  // Uncertainty requires retry

10. Acceptance Test Plan (Security Critical)

Caution

ALL tests below must pass before going production. Any failure indicates potential fund theft, loss, or creation from thin air.

10.1 Fund Conservation Tests

10.2 External Attack Tests

10.3 Balance & Status Tests

10.4 FSM State Transition Tests

10.5 Crash Recovery Tests

10.6 Concurrency & Race Tests

10.7 Idempotency Tests

10.8 Fund Anomaly Tests (Most Critical)

10.9 Monitoring & Alerting Tests

🇨🇳 中文

📦 代码变更: 查看 Diff

1. 问题陈述

1.1 系统拓扑

1.2 核心约束

这两个系统 无法共享事务。没有 XA/2PC 数据库协议。 因此：我们必须使用外部 FSM 协调器构建自己的两阶段提交。

1.5 安全前置检查 (MANDATORY)

Caution

纵深防御 (Defense-in-Depth) 以下所有检查必须在 每一个独立模块 中执行，不仅仅是 API 层。

• API 层: 第一道防线，拒绝明显非法请求
• Coordinator: 再次验证，防止内部调用绕过 API
• Adapters: 最终防线，每个适配器必须独立验证参数
• UBSCore: 内存操作前最后一次检查

安全 > 性能。重复检查的开销可以接受，安全漏洞不可接受。

1.5.1 身份与授权检查

1.5.2 账户类型检查

1.5.3 金额检查

1.5.4 资产检查

1.5.5 账户状态检查

账户初始化规则（概述）

Note

• 各账户类型的具体初始化行为和业务规则，请参见各账户类型的专用文档。
• 每个账户都有自己的状态定义（如是否允许划转），当前不详细定义。
• 默认状态：账户初始化时，默认允许划转。

账户状态检查表

1.5.6 频率限制 (Rate Limiting) - [P2 未来优化]

Note

此部分为 V2 优化项，V1 可不实现。

1.5.7 幂等性检查

1.5.8 检查顺序 (推荐)

1. 身份认证 (JWT 有效?)
2. 授权检查 (user_id 匹配?)
3. 请求格式 (from/to/amount 有效?)
4. 账户类型 (from != to, 类型支持?)
5. 资产检查 (存在? 启用? 可转账?)
6. 金额检查 (范围? 精度? 溢出?)
7. 频率限制 (超限?)
8. 幂等性 (重复?)
9. 余额检查 (充足?) ← 最后检查，避免无谓查询

2. FSM 设计 (状态机)

2.0 库选择: rust-fsm

使用 rust-fsm 库，提供：

• ✅ 编译时验证 - 非法状态转换在编译时报错
• ✅ 声明式 DSL - 清晰定义状态和转换
• ✅ 类型安全 - 防止遗漏分支

Cargo.toml:

[dependencies]
rust-fsm = "0.7"

DSL 定义:

#![allow(unused)]
fn main() {
use rust_fsm::*;

state_machine! {
    derive(Debug, Clone, Copy, PartialEq, Eq)
    
    TransferFsm(Init)  // 初始状态
    
    // 状态定义
    Init => {
        SourceWithdrawOk => SourceDone,
        SourceWithdrawFail => Failed,
    },
    SourceDone => {
        TargetDepositOk => Committed,
        TargetDepositFail => Compensating,
        TargetDepositUnknown => SourceDone [loop],  // 保持，无限重试
    },
    Compensating => {
        RefundOk => RolledBack,
        RefundFail => Compensating [loop],  // 保持，无限重试
    },
    // 终态
    Committed,
    Failed,
    RolledBack,
}
}

Note

上述 DSL 用于编译时验证状态转换的合法性。 实际运行时状态存储在 PostgreSQL，使用 CAS 更新。

2.0.1 核心状态流程图 (Top Level)

                              ┌─────────────────────────────────────────────────────────┐
                              │              INTERNAL TRANSFER FSM                       │
                              └─────────────────────────────────────────────────────────┘

   ┌─────────────────────────────── 正常路径 (Happy Path) ──────────────────────────────────┐
   │                                                                                        │
   │   ┌─────────┐                    ┌─────────────┐                    ┌───────────────┐  │
   │   │  INIT   │   源扣减成功 ✓     │ SOURCE_DONE │   目标入账成功 ✓   │               │  │
   │   │(用户请求)│ ─────────────────▶ │ (资金在途)  │ ─────────────────▶ │   COMMITTED   │  │
   │   └─────────┘                    └─────────────┘                    │               │  │
   │        │                               │                            └───────────────┘  │
   │        │                               │                                   ✅          │
   └────────│───────────────────────────────│───────────────────────────────────────────────┘
            │                               │
            │                               │
            │                               ▼
            │                     ╔══════════════════════════════════════════════════╗
            │                     ║  🔒 ATOMIC COMMIT (原子提交)                     ║
            │                     ║                                                  ║
            │                     ║  当且仅当:                                       ║
            │                     ║    FROM.withdraw = SUCCESS  ✓                   ║
            │                     ║    TO.deposit    = SUCCESS  ✓                   ║
            │                     ║                                                  ║
            │                     ║  执行: CAS(SOURCE_DONE → COMMITTED)             ║
            │                     ║  此操作必须原子，不可中断                         ║
            │                     ╚══════════════════════════════════════════════════╝
            │                               │
            │ 源扣减失败                     │ 目标入账失败 (明确 EXPLICIT_FAIL)
            ▼                               ▼
      ┌──────────┐                   ┌──────────────┐
      │  FAILED  │                   │ COMPENSATING │◀───────────┐
      │ (源失败)  │                   │  (退款中)    │            │ 退款失败 (无限重试)
      └──────────┘                   └──────────────┘────────────┘
           ❌                               │ 退款成功
                                            ▼
                                     ┌─────────────┐
                                     │ ROLLED_BACK │
                                     │  (已回滚)    │
                                     └─────────────┘
                                           ↩️

   ╔════════════════════════════════════════════════════════════════════════════════════════╗
   ║  ⚠️ 目标入账状态未知 (TIMEOUT/UNKNOWN) → 保持 SOURCE_DONE，无限重试，绝不进入 COMPENSATING║
   ╚════════════════════════════════════════════════════════════════════════════════════════╝

核心状态说明:

Important

SOURCE_DONE 是最关键的状态 - 资金已离开源账户但尚未到达目标。 此时绝不能丢失状态，必须确保最终到达 COMMITTED 或 ROLLED_BACK。

2.1 状态 (穷举)

2.2 状态转换规则 (穷举)

┌───────────────────────────────────────────────────────────────────────────────┐
│                              规范状态转换                                       │
├───────────────────────────────────────────────────────────────────────────────┤
│                                                                               │
│  INIT ──────[CAS成功]───────► SOURCE_PENDING                                  │
│    │                              │                                           │
│    │                              ├──[适配器OK]────► SOURCE_DONE              │
│    │                              │                         │                 │
│    │                              └──[适配器FAIL]──► FAILED (终态)            │
│    │                                                        │                 │
│    │                                                        │                 │
│    │                              SOURCE_DONE ──[CAS成功]──► TARGET_PENDING   │
│    │                                                             │            │
│    │                        ┌────────────────────────────────────┤            │
│    │                        │                                    │            │
│    │            [适配器OK]  │                       [适配器FAIL]              │
│    │                        │                                    │            │
│    │                        ▼                                    ▼            │
│    │                   COMMITTED                     ┌───────────────────┐    │
│    │                   (终态)                        │   源可逆?          │    │
│    │                                                 └─────────┬─────────┘    │
│    │                                                   是      │     否       │
│    │                                                   ▼       │     ▼        │
│    │                                           COMPENSATING    │  无限重试    │
│    │                                                 │         │ (保持在      │
│    │                                    [退款OK]     │         │  TARGET_     │
│    │                                         ▼       │         │  PENDING)    │
│    │                                    ROLLED_BACK  │         │              │
│    │                                    (终态)       │         │              │
│    │                                                 │         │              │
│    └─────────────────────────────────────────────────┴─────────┴──────────────┘

2.3 可逆性规则 (关键)

核心原则: 只有当适配器返回 明确定义的失败 时，才能安全撤销。

Caution

只有 EXPLICIT_FAIL 可以安全撤销。 任何状态未知的情况（超时、Pending、网络错误），资金都处于 In-Flight 中。 我们无法知道对方是否已处理。贸然撤销将导致 双花 或 资金丢失。 唯一安全操作：无限重试 或 人工介入。

3. 转账场景 (逐步)

3.1 场景 A: 资金 → 交易 (充值到交易账户)

正常路径:

失败路径 (目标失败):

3.2 场景 B: 交易 → 资金 (从交易账户提现)

正常路径:

失败路径 (目标失败):

Warning

只有当目标返回 EXPLICIT_FAIL 时才能进入 COMPENSATING。 如果是超时或未知状态，资金处于 In-Flight，必须无限重试或人工介入。

4. 失效模式与影响分析 (FMEA)

4.1 阶段1失败 (源操作)

4.2 阶段2失败 (目标操作)

4.3 补偿失败

5. 幂等性要求 (强制)

5.1 为什么需要幂等性?

重试是崩溃恢复的基础。没有幂等性，重试将导致 双重执行（双重扣减、双重入账）。

5.2 实现 (资金适配器)

要求: 给定相同的 req_id，多次调用 withdraw() 或 deposit() 必须与调用一次效果相同。

机制:

1. transfers_tb 有 UNIQUE(req_id)。
2. 原子事务:

   BEGIN;
   -- 检查是否已处理
   SELECT state FROM transfers_tb WHERE req_id = $1;
   IF state >= expected_post_state THEN
       RETURN 'AlreadyProcessed';
   END IF;
   
   -- 执行余额更新
   UPDATE balances_tb SET amount = amount - $2 WHERE user_id = $3 AND asset_id = $4 AND amount >= $2;
   IF NOT FOUND THEN
       RETURN 'InsufficientBalance';
   END IF;
   
   -- 更新状态
   UPDATE transfers_tb SET state = $new_state, updated_at = NOW() WHERE req_id = $1;
   COMMIT;
   RETURN 'Success';
   

5.3 实现 (交易适配器)

要求: 同上。UBSCore 必须拒绝重复的 req_id。

机制:

1. InternalOrder 包含 req_id 字段（或 cid）。
2. UBSCore 维护一个 ProcessedTransferSet（RAM中的HashSet，重启时从WAL重建）。
3. 收到转账订单时:

   IF req_id IN ProcessedTransferSet THEN
       RETURN 'AlreadyProcessed' (成功，无操作)
   ELSE
       ProcessTransfer()
       ProcessedTransferSet.insert(req_id)
       WriteWAL(TransferEvent)
       RETURN 'Success'
   END IF
   

6. 恢复工作器 (僵尸处理器)

6.1 目的

在协调器启动时（或定期），扫描“卡住“的转账并恢复它们。

6.2 查询

SELECT * FROM transfers_tb 
WHERE state IN (0, 10, 20, 30, -20) -- INIT, SOURCE_PENDING, SOURCE_DONE, TARGET_PENDING, COMPENSATING
  AND updated_at < NOW() - INTERVAL '1 minute'; -- 过期阈值

6.3 恢复逻辑

7. 数据模型

7.1 表: transfers_tb

CREATE TABLE transfers_tb (
    transfer_id   BIGSERIAL PRIMARY KEY,
    req_id        VARCHAR(26) UNIQUE NOT NULL,  -- 服务端生成的唯一 ID (ULID)
    cid           VARCHAR(64) UNIQUE,           -- 客户端幂等键 (可选)
    user_id       BIGINT NOT NULL,
    asset_id      INTEGER NOT NULL,
    amount        DECIMAL(30, 8) NOT NULL,
    transfer_type SMALLINT NOT NULL,            -- 1 = 资金->交易, 2 = 交易->资金
    source_type   SMALLINT NOT NULL,            -- 1 = 资金, 2 = 交易
    state         SMALLINT NOT NULL DEFAULT 0,  -- FSM 状态 ID
    error_message TEXT,                         -- 最后错误（用于调试）
    retry_count   INTEGER NOT NULL DEFAULT 0,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_transfers_state ON transfers_tb(state) WHERE state NOT IN (40, -10, -30);

7.2 不变量检查

定期运行以检测数据损坏:

-- 每个用户每个资产的 资金 + 交易 + 在途 之和应该是常数
-- 在途 = SUM(amount) WHERE state IN (SOURCE_DONE, TARGET_PENDING, COMPENSATING)

8. API 契约

8.1 端点: POST /api/v1/internal_transfer

请求:

{
  "from": "SPOT",       // 源账户类型
  "to": "FUNDING",     // 目标账户类型
  "asset": "USDT",
  "amount": "100.00"
}

账户类型枚举 (AccountType):

响应:

{
  "transfer_id": 12345,
  "req_id": "01JFVQ2X8Z0Y1M3N4P5R6S7T8U",  // 服务端生成 (ULID)
  "from": "SPOT",
  "to": "FUNDING",
  "state": "COMMITTED",  // 或 "PENDING" 如果异步
  "message": "转账成功"
}

8.2 查询端点: GET /api/v1/internal_transfer/:req_id

响应:

{
  "transfer_id": 12345,
  "req_id": "sr-1734912345678901234",
  "from": "SPOT",
  "to": "FUNDING",
  "asset": "USDT",
  "amount": "100.00",
  "state": "COMMITTED",
  "created_at": "2024-12-23T14:00:00Z",
  "updated_at": "2024-12-23T14:00:01Z"
}

Important

req_id 由服务端生成，不是客户端。 客户端如果需要幂等性，应使用 cid (client_order_id) 字段（可选），服务端会检查重复并返回已有结果。

错误码:

9. 实现伪代码 (关键状态检查)

9.1 API 层

function handle_transfer_request(request, auth_context):
    // ========== 纵深防御 Layer 1: API 层 ==========
    
    // 1. 身份认证
    if !auth_context.is_valid():
        return Error(UNAUTHORIZED)
    
    // 2. 用户 ID 一致性（防止跨用户攻击）
    if request.user_id != auth_context.user_id:
        return Error(FORBIDDEN, "User ID mismatch")
    
    // 3. 账户类型检查
    if request.from == request.to:
        return Error(SAME_ACCOUNT)
    
    if request.from NOT IN [FUNDING, SPOT]:
        return Error(INVALID_ACCOUNT_TYPE)
    
    if request.to NOT IN [FUNDING, SPOT]:
        return Error(INVALID_ACCOUNT_TYPE)
    
    // 4. 金额检查
    if request.amount <= 0:
        return Error(INVALID_AMOUNT)
    
    if decimal_places(request.amount) > asset.precision:
        return Error(PRECISION_OVERFLOW)
    
    // 5. 幂等性检查
    if request.cid:
        existing = db.find_by_cid(request.cid)
        if existing:
            return Success(existing)  // 返回已存在的结果
    
    // 6. 资产检查
    asset = db.get_asset(request.asset_id)
    if !asset or asset.status != ACTIVE:
        return Error(INVALID_ASSET)
    
    // 7. 调用 Coordinator
    result = coordinator.create_and_execute(request)
    return result

9.2 Coordinator 层

function create_and_execute(request):
    // ========== 纵深防御 Layer 2: Coordinator ==========
    
    // 再次验证（防止内部调用绕过 API）
    ASSERT request.from != request.to
    ASSERT request.amount > 0
    ASSERT request.user_id > 0
    
    // 生成唯一 ID
    req_id = ulid.new()
    
    // 创建转账记录 (State = INIT)
    transfer = TransferRecord {
        req_id: req_id,
        user_id: request.user_id,
        from: request.from,
        to: request.to,
        asset_id: request.asset_id,
        amount: request.amount,
        state: INIT,
        created_at: now()
    }
    
    db.insert(transfer)
    log.info("Transfer created", req_id)
    
    // 执行 FSM
    return execute_fsm(req_id)

function execute_fsm(req_id):
    loop:
        transfer = db.get(req_id)
        
        if transfer.state.is_terminal():
            return transfer
        
        new_state = step(transfer)
        
        if new_state == transfer.state:
            // 未进展，等待重试
            sleep(RETRY_INTERVAL)
            continue
    
function step(transfer):
    match transfer.state:
        INIT:
            return step_init(transfer)
        SOURCE_PENDING:
            return step_source_pending(transfer)
        SOURCE_DONE:
            return step_source_done(transfer)
        TARGET_PENDING:
            return step_target_pending(transfer)
        COMPENSATING:
            return step_compensating(transfer)
        _:
            return transfer.state  // 终态，不处理

function step_init(transfer):
    // CAS: 先更新状态，再调用适配器（Persist-Before-Call）
    success = db.cas_update(
        req_id = transfer.req_id,
        old_state = INIT,
        new_state = SOURCE_PENDING
    )
    
    if !success:
        // 并发冲突，重新读取
        return db.get(transfer.req_id).state
    
    // 获取源适配器
    source_adapter = get_adapter(transfer.from)
    
    // ========== 纵深防御 Layer 3: Adapter ==========
    result = source_adapter.withdraw(
        req_id = transfer.req_id,
        user_id = transfer.user_id,
        asset_id = transfer.asset_id,
        amount = transfer.amount
    )
    
    match result:
        SUCCESS:
            db.cas_update(transfer.req_id, SOURCE_PENDING, SOURCE_DONE)
            return SOURCE_DONE
        
        EXPLICIT_FAIL(reason):
            // 明确失败，可以安全终止
            db.update_with_error(transfer.req_id, SOURCE_PENDING, FAILED, reason)
            return FAILED
        
        TIMEOUT | PENDING | NETWORK_ERROR | UNKNOWN:
            // 状态未知，保持 SOURCE_PENDING，等待重试
            log.warn("Source withdraw unknown state", transfer.req_id)
            return SOURCE_PENDING

function step_source_done(transfer):
    // ========== 进入 SOURCE_DONE: 资金已在途，必须确保最终到达终态 ==========
    
    // CAS 更新到 TARGET_PENDING
    success = db.cas_update(transfer.req_id, SOURCE_DONE, TARGET_PENDING)
    if !success:
        return db.get(transfer.req_id).state
    
    // 获取目标适配器
    target_adapter = get_adapter(transfer.to)
    
    // ========== 纵深防御 Layer 4: Target Adapter ==========
    result = target_adapter.deposit(
        req_id = transfer.req_id,
        user_id = transfer.user_id,
        asset_id = transfer.asset_id,
        amount = transfer.amount
    )
    
    match result:
        SUCCESS:
            // ╔════════════════════════════════════════════════════════════════╗
            // ║  🔒 ATOMIC COMMIT - 最关键的一步！                             ║
            // ║                                                                ║
            // ║  此时:                                                         ║
            // ║    FROM.withdraw = SUCCESS ✓ (已确认)                         ║
            // ║    TO.deposit    = SUCCESS ✓ (刚确认)                         ║
            // ║                                                                ║
            // ║  执行原子 CAS 提交:                                            ║
            // ║    CAS(TARGET_PENDING → COMMITTED)                            ║
            // ║                                                                ║
            // ║  此 CAS 是最终确认，一旦成功，转账不可逆转！                    ║
            // ╚════════════════════════════════════════════════════════════════╝
            
            commit_success = db.cas_update(transfer.req_id, TARGET_PENDING, COMMITTED)
            
            if !commit_success:
                // 极少发生：另一个 Worker 已经提交，返回当前状态
                return db.get(transfer.req_id).state
            
            log.info("🔒 ATOMIC COMMIT SUCCESS", transfer.req_id)
            return COMMITTED
        
        EXPLICIT_FAIL(reason):
            // 明确失败，可以进入补偿
            db.update_with_error(transfer.req_id, TARGET_PENDING, COMPENSATING, reason)
            return COMPENSATING
        
        TIMEOUT | PENDING | NETWORK_ERROR | UNKNOWN:
            // ========== 关键：状态未知，不能补偿！==========
            log.critical("Target deposit unknown state - INFINITE RETRY", transfer.req_id)
            alert_ops("Transfer stuck in TARGET_PENDING", transfer.req_id)
            return TARGET_PENDING  // 保持状态，等待重试


function step_compensating(transfer):
    source_adapter = get_adapter(transfer.from)
    
    result = source_adapter.refund(
        req_id = transfer.req_id,
        user_id = transfer.user_id,
        asset_id = transfer.asset_id,
        amount = transfer.amount
    )
    
    match result:
        SUCCESS:
            db.cas_update(transfer.req_id, COMPENSATING, ROLLED_BACK)
            log.info("Transfer rolled back", transfer.req_id)
            return ROLLED_BACK
        
        _:
            // 退款失败，必须无限重试
            log.critical("Refund failed - MUST RETRY", transfer.req_id)
            return COMPENSATING

9.3 Adapter 层 (示例: Funding Adapter)

function withdraw(req_id, user_id, asset_id, amount):
    // ========== 纵深防御 Layer 3: Adapter 内部检查 ==========
    
    // 再次验证参数（不信任调用者）
    ASSERT amount > 0
    ASSERT user_id > 0
    ASSERT asset_id > 0
    
    // 幂等性检查
    existing = db.find_transfer_operation(req_id, "WITHDRAW")
    if existing:
        return existing.result  // 返回已处理的结果
    
    // 开始事务
    tx = db.begin_transaction()
    try:
        // 获取账户并锁定
        account = tx.select_for_update(
            "SELECT * FROM balances_tb WHERE user_id = ? AND asset_id = ? AND account_type = 'FUNDING'"
        )
        
        if !account:
            tx.rollback()
            return EXPLICIT_FAIL("SOURCE_ACCOUNT_NOT_FOUND")
        
        if account.status == FROZEN:
            tx.rollback()
            return EXPLICIT_FAIL("ACCOUNT_FROZEN")
        
        if account.available < amount:
            tx.rollback()
            return EXPLICIT_FAIL("INSUFFICIENT_BALANCE")
        
        // 执行扣减
        tx.update("UPDATE balances_tb SET available = available - ? WHERE id = ?", amount, account.id)
        
        // 记录操作（用于幂等性）
        tx.insert("INSERT INTO transfer_operations (req_id, op_type, result) VALUES (?, 'WITHDRAW', 'SUCCESS')")
        
        tx.commit()
        return SUCCESS
        
    catch Exception as e:
        tx.rollback()
        log.error("Withdraw failed", req_id, e)
        return UNKNOWN  // 不确定是否执行，必须重试

10. 验收测试计划 (安全关键)

Caution

以下测试必须全部通过才能上线。 任何失败都可能导致资金被盗、消失或无中生有。

10.1 资金守恒测试

10.2 外部攻击测试

10.3 余额不足测试

10.4 FSM 状态转换测试

10.5 崩溃恢复测试

10.6 并发/竞态测试

10.7 幂等性测试

10.8 资金异常测试 (最关键)

10.9 监控告警测试

📋 Implementation & Verification | 实现与验证

本章的完整实现细节、API 说明、E2E 测试脚本和验证结果请参阅:

For complete implementation details, API documentation, E2E test scripts, and verification results:

👉 Phase 0x0B-a: Implementation & Testing Guide

包含 / Includes:

• 架构实现与核心模块 (Architecture & Core Modules)
• 新增 API 端点 (New API Endpoints)
• 可复用 E2E 测试脚本 (Reusable E2E Test Script)
• 数据库验证方法 (Database Verification)
• 已修复 Bug 清单 (Fixed Bugs)