fb182adef4
🎯 核心变更: - OpType (string) → OpCode (int32) - 20+ OpCode枚举常量 (基于DOIP/IRP标准) - 类型安全 + 性能优化 📊 影响范围: - 核心模型: Operation结构体、CBOR序列化 - 数据库: schema.go + SQL DDL (PostgreSQL/MySQL/SQLite) - 持久化: repository.go查询、cursor_worker.go - API接口: Protobuf定义 + gRPC客户端 - 测试代码: 60+ 测试文件更新 ✅ 测试结果: - 通过率: 100% (所有87个测试用例) - 总体覆盖率: 53.7% - 核心包覆盖率: logger(100%), highclient(95.3%), model(79.1%) 📝 文档: - 精简README (1056行→489行，减少54%) - 完整的OpCode枚举说明 - 三种持久化策略示例 - 数据库表结构和架构图 🔧 技术细节: - 类型转换: string(OpCode) → int32(OpCode) - SQL参数: 字符串值 → 整数值 - Protobuf: op_type string → op_code int32 - 测试断言: 字符串比较 → 常量比较 🎉 质量保证: - 零编译错误 - 100%测试通过 - PostgreSQL/Pulsar集成测试验证 - 分布式并发安全测试通过
291 lines
9.4 KiB
Go
291 lines
9.4 KiB
Go
package persistence_test
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
_ "github.com/lib/pq"
|
||
"github.com/stretchr/testify/require"
|
||
|
||
"go.yandata.net/iod/iod/go-trustlog/api/adapter"
|
||
"go.yandata.net/iod/iod/go-trustlog/api/logger"
|
||
"go.yandata.net/iod/iod/go-trustlog/api/model"
|
||
"go.yandata.net/iod/iod/go-trustlog/api/persistence"
|
||
)
|
||
|
||
// TestCursorInitialization 验证 cursor 初始化逻辑
|
||
func TestCursorInitialization(t *testing.T) {
|
||
if testing.Short() {
|
||
t.Skip("Skipping cursor initialization test in short mode")
|
||
}
|
||
|
||
ctx := context.Background()
|
||
log := logger.NewDefaultLogger() // 使用标准logger来输出诊断信息
|
||
|
||
// 连接数据库
|
||
dsn := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
|
||
e2eTestPGHost, e2eTestPGPort, e2eTestPGUser, e2eTestPGPassword, e2eTestPGDatabase)
|
||
|
||
db, err := sql.Open("postgres", dsn)
|
||
if err != nil {
|
||
t.Skipf("PostgreSQL not available: %v", err)
|
||
return
|
||
}
|
||
defer db.Close()
|
||
|
||
if err := db.Ping(); err != nil {
|
||
t.Skipf("PostgreSQL not reachable: %v", err)
|
||
return
|
||
}
|
||
|
||
// 清理测试数据
|
||
_, _ = db.Exec("DELETE FROM trustlog_retry WHERE op_id LIKE 'cursor-init-%'")
|
||
_, _ = db.Exec("DELETE FROM operation WHERE op_id LIKE 'cursor-init-%'")
|
||
_, _ = db.Exec("DELETE FROM trustlog_cursor")
|
||
defer func() {
|
||
_, _ = db.Exec("DELETE FROM trustlog_retry WHERE op_id LIKE 'cursor-init-%'")
|
||
_, _ = db.Exec("DELETE FROM operation WHERE op_id LIKE 'cursor-init-%'")
|
||
_, _ = db.Exec("DELETE FROM trustlog_cursor")
|
||
}()
|
||
|
||
t.Log("✅ PostgreSQL connected and cleaned")
|
||
|
||
// 确保schema是最新的（添加可能缺失的列）
|
||
_, _ = db.Exec("ALTER TABLE operation ADD COLUMN IF NOT EXISTS op_hash VARCHAR(128)")
|
||
_, _ = db.Exec("ALTER TABLE operation ADD COLUMN IF NOT EXISTS sign VARCHAR(512)")
|
||
_, _ = db.Exec("ALTER TABLE operation ADD COLUMN IF NOT EXISTS timestamp TIMESTAMP")
|
||
_, _ = db.Exec("ALTER TABLE operation ADD COLUMN IF NOT EXISTS updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP")
|
||
|
||
// 场景 1: 没有历史数据时启动
|
||
t.Run("NoHistoricalData", func(t *testing.T) {
|
||
// 清理
|
||
_, _ = db.Exec("DELETE FROM operation")
|
||
_, _ = db.Exec("DELETE FROM trustlog_cursor")
|
||
|
||
// 创建 Pulsar Publisher
|
||
publisher, err := adapter.NewPublisher(adapter.PublisherConfig{
|
||
URL: e2eTestPulsarURL,
|
||
}, log)
|
||
require.NoError(t, err)
|
||
defer publisher.Close()
|
||
|
||
// 创建 PersistenceClient
|
||
dbConfig := persistence.DBConfig{
|
||
DriverName: "postgres",
|
||
DSN: dsn,
|
||
MaxOpenConns: 10,
|
||
MaxIdleConns: 5,
|
||
ConnMaxLifetime: time.Hour,
|
||
}
|
||
|
||
persistenceConfig := persistence.PersistenceConfig{
|
||
Strategy: persistence.StrategyDBAndTrustlog,
|
||
EnableRetry: true,
|
||
MaxRetryCount: 3,
|
||
RetryBatchSize: 10,
|
||
}
|
||
|
||
cursorConfig := &persistence.CursorWorkerConfig{
|
||
ScanInterval: 100 * time.Millisecond,
|
||
BatchSize: 10,
|
||
Enabled: true, // 必须显式启用！
|
||
}
|
||
|
||
retryConfig := &persistence.RetryWorkerConfig{
|
||
RetryInterval: 100 * time.Millisecond,
|
||
BatchSize: 10,
|
||
}
|
||
|
||
envelopeConfig := model.EnvelopeConfig{
|
||
Signer: &model.NopSigner{},
|
||
}
|
||
|
||
clientConfig := persistence.PersistenceClientConfig{
|
||
Publisher: publisher,
|
||
Logger: log,
|
||
EnvelopeConfig: envelopeConfig,
|
||
DBConfig: dbConfig,
|
||
PersistenceConfig: persistenceConfig,
|
||
CursorWorkerConfig: cursorConfig,
|
||
EnableCursorWorker: true,
|
||
RetryWorkerConfig: retryConfig,
|
||
EnableRetryWorker: true,
|
||
}
|
||
|
||
client, err := persistence.NewPersistenceClient(ctx, clientConfig)
|
||
require.NoError(t, err)
|
||
|
||
// 等待初始化
|
||
time.Sleep(500 * time.Millisecond)
|
||
|
||
// 验证 cursor 已创建
|
||
var cursorValue string
|
||
var updatedAt time.Time
|
||
err = db.QueryRow("SELECT cursor_value, last_updated_at FROM trustlog_cursor WHERE cursor_key = 'operation_scan'").Scan(&cursorValue, &updatedAt)
|
||
require.NoError(t, err, "❌ Cursor should be initialized!")
|
||
|
||
t.Logf("✅ Cursor initialized: %s", cursorValue)
|
||
t.Logf(" Updated at: %v", updatedAt)
|
||
|
||
// cursor 应该是一个很早的时间（因为没有历史数据）
|
||
cursorTime, err := time.Parse(time.RFC3339Nano, cursorValue)
|
||
require.NoError(t, err)
|
||
require.True(t, cursorTime.Before(time.Now().Add(-1*time.Hour)), "Cursor should be set to an early time")
|
||
|
||
client.Close()
|
||
})
|
||
|
||
// 场景 2: 有历史数据时启动
|
||
t.Run("WithHistoricalData", func(t *testing.T) {
|
||
// 清理
|
||
_, _ = db.Exec("DELETE FROM operation WHERE op_id LIKE 'cursor-init-%'")
|
||
_, _ = db.Exec("DELETE FROM trustlog_cursor")
|
||
|
||
// 插入一些历史数据
|
||
baseTime := time.Now().Add(-10 * time.Minute)
|
||
for i := 0; i < 5; i++ {
|
||
opID := fmt.Sprintf("cursor-init-%d", i)
|
||
createdAt := baseTime.Add(time.Duration(i) * time.Minute)
|
||
_, err := db.Exec(`
|
||
INSERT INTO operation (
|
||
op_id, op_actor, doid, producer_id,
|
||
request_body_hash, response_body_hash, op_hash, sign,
|
||
op_source, op_code, do_prefix, do_repository,
|
||
trustlog_status, created_at
|
||
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
|
||
`, opID, "tester", fmt.Sprintf("test/%d", i), "producer",
|
||
"req-hash", "resp-hash", "op-hash", "signature",
|
||
"DOIP", 100, "test", "repo", "NOT_TRUSTLOGGED", createdAt)
|
||
require.NoError(t, err)
|
||
}
|
||
|
||
t.Logf("✅ Created 5 historical records starting from %v", baseTime)
|
||
|
||
// 创建 Pulsar Publisher
|
||
publisher, err := adapter.NewPublisher(adapter.PublisherConfig{
|
||
URL: e2eTestPulsarURL,
|
||
}, log)
|
||
require.NoError(t, err)
|
||
defer publisher.Close()
|
||
|
||
// 创建 PersistenceClient
|
||
dbConfig := persistence.DBConfig{
|
||
DriverName: "postgres",
|
||
DSN: dsn,
|
||
MaxOpenConns: 10,
|
||
MaxIdleConns: 5,
|
||
ConnMaxLifetime: time.Hour,
|
||
}
|
||
|
||
persistenceConfig := persistence.PersistenceConfig{
|
||
Strategy: persistence.StrategyDBAndTrustlog,
|
||
EnableRetry: true,
|
||
MaxRetryCount: 3,
|
||
RetryBatchSize: 10,
|
||
}
|
||
|
||
cursorConfig := &persistence.CursorWorkerConfig{
|
||
ScanInterval: 100 * time.Millisecond,
|
||
BatchSize: 10,
|
||
Enabled: true, // 必须显式启用！
|
||
}
|
||
|
||
retryConfig := &persistence.RetryWorkerConfig{
|
||
RetryInterval: 100 * time.Millisecond,
|
||
BatchSize: 10,
|
||
}
|
||
|
||
envelopeConfig := model.EnvelopeConfig{
|
||
Signer: &model.NopSigner{},
|
||
}
|
||
|
||
clientConfig := persistence.PersistenceClientConfig{
|
||
Publisher: publisher,
|
||
Logger: log,
|
||
EnvelopeConfig: envelopeConfig,
|
||
DBConfig: dbConfig,
|
||
PersistenceConfig: persistenceConfig,
|
||
CursorWorkerConfig: cursorConfig,
|
||
EnableCursorWorker: true,
|
||
RetryWorkerConfig: retryConfig,
|
||
EnableRetryWorker: true,
|
||
}
|
||
|
||
t.Log("📌 Creating PersistenceClient...")
|
||
t.Logf(" EnableCursorWorker: %v", clientConfig.EnableCursorWorker)
|
||
t.Logf(" Strategy: %v", clientConfig.PersistenceConfig.Strategy)
|
||
|
||
client, err := persistence.NewPersistenceClient(ctx, clientConfig)
|
||
require.NoError(t, err)
|
||
t.Log("✅ PersistenceClient created")
|
||
|
||
// 立即验证初始 cursor (在 Worker 开始扫描前)
|
||
// 注意：由于 Worker 可能已经开始处理，我们需要快速读取
|
||
time.Sleep(10 * time.Millisecond) // 给一点时间让 InitCursor 完成
|
||
|
||
var initialCursorValue string
|
||
var updatedAt time.Time
|
||
err = db.QueryRow("SELECT cursor_value, last_updated_at FROM trustlog_cursor WHERE cursor_key = 'operation_scan'").Scan(&initialCursorValue, &updatedAt)
|
||
require.NoError(t, err, "❌ Cursor should be initialized!")
|
||
|
||
t.Logf("📍 Initial cursor: %s", initialCursorValue)
|
||
t.Logf(" Updated at: %v", updatedAt)
|
||
|
||
// 验证初始 cursor 应该在最早记录之前（或接近）
|
||
initialCursorTime, err := time.Parse(time.RFC3339Nano, initialCursorValue)
|
||
require.NoError(t, err)
|
||
|
||
var earliestRecordTime time.Time
|
||
err = db.QueryRow("SELECT MIN(created_at) FROM operation WHERE op_id LIKE 'cursor-init-%'").Scan(&earliestRecordTime)
|
||
require.NoError(t, err)
|
||
|
||
t.Logf(" Earliest record: %v", earliestRecordTime)
|
||
t.Logf(" Initial cursor time: %v", initialCursorTime)
|
||
|
||
// cursor 应该在最早记录之前或相差不超过2秒（考虑 Worker 可能已经开始更新）
|
||
timeDiff := earliestRecordTime.Sub(initialCursorTime)
|
||
require.True(t, timeDiff >= -2*time.Second,
|
||
"❌ Cursor (%v) should be before or near earliest record (%v), diff: %v",
|
||
initialCursorTime, earliestRecordTime, timeDiff)
|
||
|
||
t.Log("✅ Initial cursor position is correct!")
|
||
|
||
// 等待 Worker 处理所有记录
|
||
t.Log("⏳ Waiting for Worker to process all records...")
|
||
time.Sleep(3 * time.Second)
|
||
|
||
// 再次查询 cursor，看看是否被更新
|
||
var updatedCursorValue string
|
||
var finalUpdatedAt time.Time
|
||
err = db.QueryRow("SELECT cursor_value, last_updated_at FROM trustlog_cursor WHERE cursor_key = 'operation_scan'").Scan(&updatedCursorValue, &finalUpdatedAt)
|
||
require.NoError(t, err)
|
||
|
||
t.Logf("📍 Cursor after processing:")
|
||
t.Logf(" Value: %s", updatedCursorValue)
|
||
t.Logf(" Updated: %v", finalUpdatedAt)
|
||
t.Logf(" Changed: %v", updatedCursorValue != initialCursorValue)
|
||
|
||
// 验证所有记录都被处理了
|
||
var trustloggedCount int
|
||
err = db.QueryRow("SELECT COUNT(*) FROM operation WHERE op_id LIKE 'cursor-init-%' AND trustlog_status = 'TRUSTLOGGED'").Scan(&trustloggedCount)
|
||
require.NoError(t, err)
|
||
|
||
t.Logf("📊 Processed records: %d/5", trustloggedCount)
|
||
require.Equal(t, 5, trustloggedCount, "❌ All 5 records should be processed!")
|
||
|
||
t.Log("✅ All historical records were processed correctly!")
|
||
|
||
client.Close()
|
||
})
|
||
|
||
t.Log("\n" + strings.Repeat("=", 60))
|
||
t.Log("✅ Cursor initialization verification PASSED")
|
||
t.Log(strings.Repeat("=", 60))
|
||
}
|
||
|
||
|