feat: transaction implement
This commit is contained in:
@@ -0,0 +1,169 @@
|
||||
package transaction
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
)
|
||||
|
||||
// InspectionStatus 表示外部系统中某一步的实际结果。
|
||||
type InspectionStatus string
|
||||
|
||||
const (
|
||||
InspectionApplied InspectionStatus = "APPLIED"
|
||||
InspectionNotApplied InspectionStatus = "NOT_APPLIED"
|
||||
InspectionUnknown InspectionStatus = "UNKNOWN"
|
||||
)
|
||||
|
||||
// Inspection 是执行器通过 inspect、摘要、健康检查等方式得到的实际状态。
|
||||
type Inspection struct {
|
||||
Status InspectionStatus
|
||||
Result json.RawMessage
|
||||
}
|
||||
|
||||
// Operation 是一个可核对实际结果的外部副作用。
|
||||
// Apply 返回成功只代表调用完成；最终成功必须由 Inspect 确认。
|
||||
type Operation interface {
|
||||
Apply(context.Context) error
|
||||
Inspect(context.Context) (Inspection, error)
|
||||
}
|
||||
|
||||
// UncertainStepError 表示当前无法确认外部副作用是否已经发生。
|
||||
// 这种错误必须保留 INTENT_RECORDED，等待恢复流程再次核对。
|
||||
type UncertainStepError struct {
|
||||
TransactionID string
|
||||
StepKey string
|
||||
Cause error
|
||||
}
|
||||
|
||||
func (e *UncertainStepError) Error() string {
|
||||
return fmt.Sprintf("external step result is uncertain: transaction=%s step=%s: %v", e.TransactionID, e.StepKey, e.Cause)
|
||||
}
|
||||
|
||||
func (e *UncertainStepError) Unwrap() error {
|
||||
return e.Cause
|
||||
}
|
||||
|
||||
// Coordinator 串行化单机更新，并实现“先记录意图、再执行、最后 inspect”的步骤协议。
|
||||
type Coordinator struct {
|
||||
store *Store
|
||||
logger *slog.Logger
|
||||
permit chan struct{}
|
||||
}
|
||||
|
||||
func NewCoordinator(store *Store, logger *slog.Logger) (*Coordinator, error) {
|
||||
if store == nil {
|
||||
return nil, errors.New("transaction store is required")
|
||||
}
|
||||
if logger == nil {
|
||||
logger = slog.Default()
|
||||
}
|
||||
permit := make(chan struct{}, 1)
|
||||
permit <- struct{}{}
|
||||
return &Coordinator{store: store, logger: logger, permit: permit}, nil
|
||||
}
|
||||
|
||||
// RunExclusive 在一个进程内只允许一个完整更新流程进入执行区。
|
||||
func (c *Coordinator) RunExclusive(ctx context.Context, run func(context.Context) error) error {
|
||||
if run == nil {
|
||||
return errors.New("exclusive update function is required")
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-c.permit:
|
||||
}
|
||||
defer func() { c.permit <- struct{}{} }()
|
||||
return run(ctx)
|
||||
}
|
||||
|
||||
// ExecuteStep 执行或恢复一个外部步骤。
|
||||
// 相同 step key 再次调用时先核对现场，禁止直接重复 Apply。
|
||||
func (c *Coordinator) ExecuteStep(ctx context.Context, transactionID string, intent StepIntent, operation Operation) (Step, error) {
|
||||
if operation == nil {
|
||||
return Step{}, errors.New("external operation is required")
|
||||
}
|
||||
step, created, err := c.store.RecordStepIntent(ctx, transactionID, intent)
|
||||
if err != nil {
|
||||
return Step{}, err
|
||||
}
|
||||
if step.Status == StepSucceeded {
|
||||
return step, nil
|
||||
}
|
||||
if step.Status == StepFailed {
|
||||
return step, fmt.Errorf("external step already failed: %s", step.Error)
|
||||
}
|
||||
|
||||
if !created {
|
||||
inspection, err := operation.Inspect(ctx)
|
||||
if err != nil {
|
||||
return step, c.uncertain(ctx, transactionID, intent.Key, err)
|
||||
}
|
||||
switch inspection.Status {
|
||||
case InspectionApplied:
|
||||
return c.completeApplied(ctx, transactionID, intent.Key, inspection)
|
||||
case InspectionUnknown:
|
||||
return step, c.uncertain(ctx, transactionID, intent.Key, errors.New("inspect returned UNKNOWN"))
|
||||
case InspectionNotApplied:
|
||||
// 现场明确未发生副作用后，才允许恢复流程重新 Apply。
|
||||
default:
|
||||
return step, c.uncertain(ctx, transactionID, intent.Key, fmt.Errorf("inspect returned invalid status %q", inspection.Status))
|
||||
}
|
||||
}
|
||||
|
||||
c.logger.InfoContext(ctx, "external step apply started",
|
||||
"transaction_id", transactionID,
|
||||
"step_key", intent.Key,
|
||||
"step_name", intent.Name,
|
||||
)
|
||||
applyErr := operation.Apply(ctx)
|
||||
inspection, inspectErr := operation.Inspect(ctx)
|
||||
if inspectErr != nil {
|
||||
return step, c.uncertain(ctx, transactionID, intent.Key, errors.Join(applyErr, inspectErr))
|
||||
}
|
||||
switch inspection.Status {
|
||||
case InspectionApplied:
|
||||
completed, err := c.completeApplied(ctx, transactionID, intent.Key, inspection)
|
||||
if err == nil {
|
||||
c.logger.InfoContext(ctx, "external step applied",
|
||||
"transaction_id", transactionID,
|
||||
"step_key", intent.Key,
|
||||
)
|
||||
}
|
||||
return completed, err
|
||||
case InspectionUnknown:
|
||||
return step, c.uncertain(ctx, transactionID, intent.Key, errors.Join(applyErr, errors.New("inspect returned UNKNOWN")))
|
||||
case InspectionNotApplied:
|
||||
failure := applyErr
|
||||
if failure == nil {
|
||||
failure = errors.New("operation completed without reaching the expected external state")
|
||||
}
|
||||
completed, err := c.store.CompleteStep(ctx, transactionID, intent.Key, StepFailed, inspection.Result, failure.Error())
|
||||
if err != nil {
|
||||
return Step{}, errors.Join(failure, err)
|
||||
}
|
||||
c.logger.ErrorContext(ctx, "external step failed",
|
||||
"transaction_id", transactionID,
|
||||
"step_key", intent.Key,
|
||||
"error", failure,
|
||||
)
|
||||
return completed, failure
|
||||
default:
|
||||
return step, c.uncertain(ctx, transactionID, intent.Key, fmt.Errorf("inspect returned invalid status %q", inspection.Status))
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Coordinator) completeApplied(ctx context.Context, transactionID, stepKey string, inspection Inspection) (Step, error) {
|
||||
return c.store.CompleteStep(ctx, transactionID, stepKey, StepSucceeded, inspection.Result, "")
|
||||
}
|
||||
|
||||
func (c *Coordinator) uncertain(ctx context.Context, transactionID, stepKey string, cause error) error {
|
||||
c.logger.WarnContext(ctx, "external step result is uncertain",
|
||||
"transaction_id", transactionID,
|
||||
"step_key", stepKey,
|
||||
"error", cause,
|
||||
)
|
||||
return &UncertainStepError{TransactionID: transactionID, StepKey: stepKey, Cause: cause}
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
package transaction
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestCoordinatorRecoversIntentWithoutRepeatingAppliedOperation(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t)
|
||||
record := createTestTransaction(t, store, "recover-applied")
|
||||
intent := StepIntent{Key: "start-green", Name: "start green", Intent: json.RawMessage(`{"slot":"green"}`)}
|
||||
if _, _, err := store.RecordStepIntent(ctx, record.ID, intent); err != nil {
|
||||
t.Fatalf("record crash-window intent: %v", err)
|
||||
}
|
||||
operation := &fakeOperation{applied: true, result: json.RawMessage(`{"running":true}`)}
|
||||
coordinator := newTestCoordinator(t, store)
|
||||
step, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
||||
if err != nil {
|
||||
t.Fatalf("recover applied operation: %v", err)
|
||||
}
|
||||
if step.Status != StepSucceeded || operation.applyCalls.Load() != 0 || operation.inspectCalls.Load() != 1 {
|
||||
t.Fatalf("unexpected recovery result: step=%+v apply=%d inspect=%d", step, operation.applyCalls.Load(), operation.inspectCalls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoordinatorRecoversNotAppliedIntentThenExecutesOnce(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t)
|
||||
record := createTestTransaction(t, store, "recover-not-applied")
|
||||
intent := StepIntent{Key: "write-files", Name: "write files", Intent: json.RawMessage(`{}`)}
|
||||
if _, _, err := store.RecordStepIntent(ctx, record.ID, intent); err != nil {
|
||||
t.Fatalf("record crash-window intent: %v", err)
|
||||
}
|
||||
operation := &fakeOperation{result: json.RawMessage(`{"written":true}`)}
|
||||
coordinator := newTestCoordinator(t, store)
|
||||
step, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
||||
if err != nil {
|
||||
t.Fatalf("recover not-applied operation: %v", err)
|
||||
}
|
||||
if step.Status != StepSucceeded || operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 2 {
|
||||
t.Fatalf("unexpected recovery result: step=%+v apply=%d inspect=%d", step, operation.applyCalls.Load(), operation.inspectCalls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoordinatorLeavesIntentPendingWhenInspectionIsUnknown(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t)
|
||||
record := createTestTransaction(t, store, "recover-unknown")
|
||||
intent := StepIntent{Key: "reload-gateway", Name: "reload gateway", Intent: json.RawMessage(`{}`)}
|
||||
operation := &fakeOperation{inspectErr: errors.New("gateway unavailable")}
|
||||
coordinator := newTestCoordinator(t, store)
|
||||
step, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
||||
var uncertain *UncertainStepError
|
||||
if !errors.As(err, &uncertain) {
|
||||
t.Fatalf("expected uncertain step error, got step=%+v err=%v", step, err)
|
||||
}
|
||||
pending, err := store.PendingSteps(ctx, record.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("read pending steps: %v", err)
|
||||
}
|
||||
if len(pending) != 1 || pending[0].Status != StepIntentRecorded {
|
||||
t.Fatalf("uncertain step did not remain pending: %+v", pending)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoordinatorExclusiveExecutionHonorsContext(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := openTestStore(t)
|
||||
coordinator := newTestCoordinator(t, store)
|
||||
firstEntered := make(chan struct{})
|
||||
releaseFirst := make(chan struct{})
|
||||
firstDone := make(chan error, 1)
|
||||
go func() {
|
||||
firstDone <- coordinator.RunExclusive(context.Background(), func(context.Context) error {
|
||||
close(firstEntered)
|
||||
<-releaseFirst
|
||||
return nil
|
||||
})
|
||||
}()
|
||||
<-firstEntered
|
||||
cancelled, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
err := coordinator.RunExclusive(cancelled, func(context.Context) error {
|
||||
t.Fatal("cancelled update entered exclusive section")
|
||||
return nil
|
||||
})
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("expected context cancellation, got %v", err)
|
||||
}
|
||||
close(releaseFirst)
|
||||
if err := <-firstDone; err != nil {
|
||||
t.Fatalf("first exclusive update failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
type fakeOperation struct {
|
||||
applied bool
|
||||
result json.RawMessage
|
||||
applyErr error
|
||||
inspectErr error
|
||||
applyCalls atomic.Int32
|
||||
inspectCalls atomic.Int32
|
||||
}
|
||||
|
||||
func (o *fakeOperation) Apply(context.Context) error {
|
||||
o.applyCalls.Add(1)
|
||||
if o.applyErr == nil {
|
||||
o.applied = true
|
||||
}
|
||||
return o.applyErr
|
||||
}
|
||||
|
||||
func (o *fakeOperation) Inspect(context.Context) (Inspection, error) {
|
||||
o.inspectCalls.Add(1)
|
||||
if o.inspectErr != nil {
|
||||
return Inspection{}, o.inspectErr
|
||||
}
|
||||
if o.applied {
|
||||
return Inspection{Status: InspectionApplied, Result: o.result}, nil
|
||||
}
|
||||
return Inspection{Status: InspectionNotApplied}, nil
|
||||
}
|
||||
|
||||
func createTestTransaction(t *testing.T, store *Store, suffix string) Transaction {
|
||||
t.Helper()
|
||||
record, _, err := store.CreateTransaction(context.Background(), CreateRequest{
|
||||
ID: "transaction-" + suffix,
|
||||
IdempotencyKey: "request-" + suffix,
|
||||
Source: "test",
|
||||
Service: "backend",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create test transaction: %v", err)
|
||||
}
|
||||
return record
|
||||
}
|
||||
|
||||
func newTestCoordinator(t *testing.T, store *Store) *Coordinator {
|
||||
t.Helper()
|
||||
coordinator, err := NewCoordinator(store, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
if err != nil {
|
||||
t.Fatalf("create coordinator: %v", err)
|
||||
}
|
||||
return coordinator
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package transaction
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrNotFound = errors.New("transaction record not found")
|
||||
ErrActiveExists = errors.New("an unfinished transaction already exists")
|
||||
ErrStepConflict = errors.New("step key already refers to different intent")
|
||||
ErrStepNotPending = errors.New("step is not waiting for an execution result")
|
||||
)
|
||||
|
||||
// ActiveTransactionError 告知调用方当前阻塞新请求的事务。
|
||||
type ActiveTransactionError struct {
|
||||
TransactionID string
|
||||
}
|
||||
|
||||
func (e *ActiveTransactionError) Error() string {
|
||||
return fmt.Sprintf("%s: %s", ErrActiveExists, e.TransactionID)
|
||||
}
|
||||
|
||||
func (e *ActiveTransactionError) Unwrap() error {
|
||||
return ErrActiveExists
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package transaction
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Transaction 是一次更新请求的服务端持久化快照。
|
||||
type Transaction struct {
|
||||
ID string
|
||||
IdempotencyKey string
|
||||
Source string
|
||||
Service string
|
||||
Request json.RawMessage
|
||||
State State
|
||||
Version int64
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
// CreateRequest 包含创建事务所需的不可变请求信息。
|
||||
type CreateRequest struct {
|
||||
ID string
|
||||
IdempotencyKey string
|
||||
Source string
|
||||
Service string
|
||||
Request json.RawMessage
|
||||
}
|
||||
|
||||
// StepStatus 是外部步骤的持久化执行状态。
|
||||
type StepStatus string
|
||||
|
||||
const (
|
||||
StepIntentRecorded StepStatus = "INTENT_RECORDED"
|
||||
StepSucceeded StepStatus = "SUCCEEDED"
|
||||
StepFailed StepStatus = "FAILED"
|
||||
)
|
||||
|
||||
// Step 记录一次外部副作用的意图和最终核对结果。
|
||||
type Step struct {
|
||||
TransactionID string
|
||||
Key string
|
||||
Name string
|
||||
Status StepStatus
|
||||
Intent json.RawMessage
|
||||
Result json.RawMessage
|
||||
Error string
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
// StepIntent 是执行外部操作前必须先持久化的内容。
|
||||
type StepIntent struct {
|
||||
Key string
|
||||
Name string
|
||||
Intent json.RawMessage
|
||||
}
|
||||
|
||||
// Event 是供查询和 WSS 断联恢复使用的顺序事件。
|
||||
type Event struct {
|
||||
Sequence int64
|
||||
TransactionID string
|
||||
StepKey string
|
||||
Kind string
|
||||
FromState State
|
||||
ToState State
|
||||
Message string
|
||||
CreatedAt time.Time
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package transaction
|
||||
|
||||
import "fmt"
|
||||
|
||||
// State 是服务端更新事务的持久化状态。
|
||||
type State string
|
||||
|
||||
const (
|
||||
StateCreated State = "CREATED"
|
||||
StateValidating State = "VALIDATING"
|
||||
StatePrepared State = "PREPARED"
|
||||
StateStarting State = "STARTING"
|
||||
StateSwitching State = "SWITCHING"
|
||||
StateVerifying State = "VERIFYING"
|
||||
StateDraining State = "DRAINING"
|
||||
StateCommitted State = "COMMITTED"
|
||||
StateRollingBack State = "ROLLING_BACK"
|
||||
StateRolledBack State = "ROLLED_BACK"
|
||||
StateFailed State = "FAILED"
|
||||
)
|
||||
|
||||
var forwardTransitions = map[State]State{
|
||||
StateCreated: StateValidating,
|
||||
StateValidating: StatePrepared,
|
||||
StatePrepared: StateStarting,
|
||||
StateStarting: StateSwitching,
|
||||
StateSwitching: StateVerifying,
|
||||
StateVerifying: StateDraining,
|
||||
StateDraining: StateCommitted,
|
||||
}
|
||||
|
||||
// Valid 报告状态是否属于当前状态机协议。
|
||||
func (s State) Valid() bool {
|
||||
switch s {
|
||||
case StateCreated,
|
||||
StateValidating,
|
||||
StatePrepared,
|
||||
StateStarting,
|
||||
StateSwitching,
|
||||
StateVerifying,
|
||||
StateDraining,
|
||||
StateCommitted,
|
||||
StateRollingBack,
|
||||
StateRolledBack,
|
||||
StateFailed:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Terminal 报告事务是否已经不可再推进。
|
||||
func (s State) Terminal() bool {
|
||||
return s == StateCommitted || s == StateRolledBack || s == StateFailed
|
||||
}
|
||||
|
||||
// CanTransitionTo 校验正常推进、回滚和不可恢复失败三类转换。
|
||||
func (s State) CanTransitionTo(next State) bool {
|
||||
if !s.Valid() || !next.Valid() || s.Terminal() {
|
||||
return false
|
||||
}
|
||||
if s == StateRollingBack {
|
||||
return next == StateRolledBack || next == StateFailed
|
||||
}
|
||||
if next == StateRollingBack || next == StateFailed {
|
||||
return true
|
||||
}
|
||||
return forwardTransitions[s] == next
|
||||
}
|
||||
|
||||
// TransitionError 表示状态机拒绝了一次转换。
|
||||
type TransitionError struct {
|
||||
From State
|
||||
To State
|
||||
}
|
||||
|
||||
func (e *TransitionError) Error() string {
|
||||
return fmt.Sprintf("transaction state transition is not allowed: %s -> %s", e.From, e.To)
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package transaction
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestStateTransitions(t *testing.T) {
|
||||
t.Parallel()
|
||||
forward := []State{
|
||||
StateCreated,
|
||||
StateValidating,
|
||||
StatePrepared,
|
||||
StateStarting,
|
||||
StateSwitching,
|
||||
StateVerifying,
|
||||
StateDraining,
|
||||
StateCommitted,
|
||||
}
|
||||
for index := 0; index < len(forward)-1; index++ {
|
||||
if !forward[index].CanTransitionTo(forward[index+1]) {
|
||||
t.Fatalf("expected transition %s -> %s", forward[index], forward[index+1])
|
||||
}
|
||||
}
|
||||
if StateCreated.CanTransitionTo(StatePrepared) {
|
||||
t.Fatal("state machine accepted a skipped forward state")
|
||||
}
|
||||
if !StateSwitching.CanTransitionTo(StateRollingBack) {
|
||||
t.Fatal("state machine rejected rollback")
|
||||
}
|
||||
if !StateRollingBack.CanTransitionTo(StateRolledBack) {
|
||||
t.Fatal("state machine rejected rollback completion")
|
||||
}
|
||||
if StateCommitted.CanTransitionTo(StateRollingBack) {
|
||||
t.Fatal("terminal state accepted another transition")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,704 @@
|
||||
package transaction
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/ncruces/go-sqlite3/driver"
|
||||
)
|
||||
|
||||
const schemaVersion = 1
|
||||
|
||||
const schemaV1 = `
|
||||
CREATE TABLE transactions (
|
||||
id TEXT PRIMARY KEY,
|
||||
idempotency_key TEXT NOT NULL UNIQUE,
|
||||
source TEXT NOT NULL,
|
||||
service TEXT NOT NULL,
|
||||
request_json TEXT NOT NULL,
|
||||
state TEXT NOT NULL CHECK (state IN (
|
||||
'CREATED', 'VALIDATING', 'PREPARED', 'STARTING', 'SWITCHING',
|
||||
'VERIFYING', 'DRAINING', 'COMMITTED', 'ROLLING_BACK',
|
||||
'ROLLED_BACK', 'FAILED'
|
||||
)),
|
||||
version INTEGER NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
) STRICT;
|
||||
|
||||
CREATE UNIQUE INDEX one_unfinished_transaction
|
||||
ON transactions ((1))
|
||||
WHERE state NOT IN ('COMMITTED', 'ROLLED_BACK', 'FAILED');
|
||||
|
||||
CREATE TABLE transaction_steps (
|
||||
transaction_id TEXT NOT NULL REFERENCES transactions(id) ON DELETE CASCADE,
|
||||
step_key TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
status TEXT NOT NULL CHECK (status IN ('INTENT_RECORDED', 'SUCCEEDED', 'FAILED')),
|
||||
intent_json TEXT NOT NULL,
|
||||
result_json TEXT,
|
||||
error_message TEXT NOT NULL DEFAULT '',
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (transaction_id, step_key)
|
||||
) STRICT;
|
||||
|
||||
CREATE TABLE transaction_events (
|
||||
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
transaction_id TEXT NOT NULL REFERENCES transactions(id) ON DELETE CASCADE,
|
||||
step_key TEXT NOT NULL DEFAULT '',
|
||||
kind TEXT NOT NULL,
|
||||
from_state TEXT NOT NULL DEFAULT '',
|
||||
to_state TEXT NOT NULL DEFAULT '',
|
||||
message TEXT NOT NULL DEFAULT '',
|
||||
created_at TEXT NOT NULL
|
||||
) STRICT;
|
||||
|
||||
CREATE INDEX transaction_events_by_transaction
|
||||
ON transaction_events (transaction_id, sequence);
|
||||
|
||||
PRAGMA user_version = 1;
|
||||
`
|
||||
|
||||
// Store 是服务端 SQLite 事务记录。一个进程只应创建一个 Store。
|
||||
type Store struct {
|
||||
db *sql.DB
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
// OpenStore 打开本地 SQLite，并强制校验第一版持久化参数。
|
||||
func OpenStore(ctx context.Context, path string) (*Store, error) {
|
||||
if path == "" {
|
||||
return nil, errors.New("sqlite path is required")
|
||||
}
|
||||
absPath, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("resolve sqlite path: %w", err)
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(absPath), 0o750); err != nil {
|
||||
return nil, fmt.Errorf("create sqlite directory: %w", err)
|
||||
}
|
||||
|
||||
dsn := (&url.URL{
|
||||
Scheme: "file",
|
||||
Path: absPath,
|
||||
RawQuery: url.Values{"_txlock": {"immediate"}}.Encode(),
|
||||
}).String()
|
||||
db, err := driver.Open(dsn)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open sqlite driver: %w", err)
|
||||
}
|
||||
// 单连接是服务端事务串行化的一部分，也保证连接级 PRAGMA 始终生效。
|
||||
db.SetMaxOpenConns(1)
|
||||
db.SetMaxIdleConns(1)
|
||||
|
||||
closeOnError := func(cause error) (*Store, error) {
|
||||
_ = db.Close()
|
||||
return nil, cause
|
||||
}
|
||||
if err := db.PingContext(ctx); err != nil {
|
||||
return closeOnError(fmt.Errorf("ping sqlite: %w", err))
|
||||
}
|
||||
if err := configureSQLite(ctx, db); err != nil {
|
||||
return closeOnError(err)
|
||||
}
|
||||
if err := migrate(ctx, db); err != nil {
|
||||
return closeOnError(err)
|
||||
}
|
||||
if err := os.Chmod(absPath, 0o600); err != nil {
|
||||
return closeOnError(fmt.Errorf("set sqlite permissions: %w", err))
|
||||
}
|
||||
return &Store{db: db, now: time.Now}, nil
|
||||
}
|
||||
|
||||
func configureSQLite(ctx context.Context, db *sql.DB) error {
|
||||
var journalMode string
|
||||
if err := db.QueryRowContext(ctx, "PRAGMA journal_mode = DELETE").Scan(&journalMode); err != nil {
|
||||
return fmt.Errorf("set sqlite journal mode: %w", err)
|
||||
}
|
||||
if journalMode != "delete" {
|
||||
return fmt.Errorf("sqlite journal mode mismatch: got %q, want %q", journalMode, "delete")
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, "PRAGMA synchronous = EXTRA"); err != nil {
|
||||
return fmt.Errorf("set sqlite synchronous: %w", err)
|
||||
}
|
||||
var synchronous int
|
||||
if err := db.QueryRowContext(ctx, "PRAGMA synchronous").Scan(&synchronous); err != nil {
|
||||
return fmt.Errorf("read sqlite synchronous: %w", err)
|
||||
}
|
||||
if synchronous != 3 {
|
||||
return fmt.Errorf("sqlite synchronous mismatch: got %d, want 3 (EXTRA)", synchronous)
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil {
|
||||
return fmt.Errorf("enable sqlite foreign keys: %w", err)
|
||||
}
|
||||
var foreignKeys int
|
||||
if err := db.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil {
|
||||
return fmt.Errorf("read sqlite foreign_keys: %w", err)
|
||||
}
|
||||
if foreignKeys != 1 {
|
||||
return errors.New("sqlite foreign_keys is not enabled")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func migrate(ctx context.Context, db *sql.DB) error {
|
||||
var version int
|
||||
if err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil {
|
||||
return fmt.Errorf("read sqlite schema version: %w", err)
|
||||
}
|
||||
if version > schemaVersion {
|
||||
return fmt.Errorf("sqlite schema version %d is newer than supported version %d", version, schemaVersion)
|
||||
}
|
||||
if version == schemaVersion {
|
||||
return nil
|
||||
}
|
||||
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin sqlite migration: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if _, err := tx.ExecContext(ctx, schemaV1); err != nil {
|
||||
return fmt.Errorf("apply sqlite schema version 1: %w", err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("commit sqlite migration: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close 关闭服务端 SQLite。
|
||||
func (s *Store) Close() error {
|
||||
return s.db.Close()
|
||||
}
|
||||
|
||||
// CreateTransaction 原子处理幂等重试和单活动事务约束。created=false 表示返回已有幂等事务。
|
||||
func (s *Store) CreateTransaction(ctx context.Context, request CreateRequest) (record Transaction, created bool, err error) {
|
||||
if err := validateCreateRequest(&request); err != nil {
|
||||
return Transaction{}, false, err
|
||||
}
|
||||
if request.ID == "" {
|
||||
request.ID = rand.Text()
|
||||
}
|
||||
now := s.now().UTC()
|
||||
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
|
||||
if err != nil {
|
||||
return Transaction{}, false, fmt.Errorf("begin create transaction: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
existing, err := getTransactionByIdempotencyKey(ctx, tx, request.IdempotencyKey)
|
||||
if err == nil {
|
||||
return existing, false, nil
|
||||
}
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
return Transaction{}, false, err
|
||||
}
|
||||
active, err := getActiveTransaction(ctx, tx)
|
||||
if err == nil {
|
||||
return Transaction{}, false, &ActiveTransactionError{TransactionID: active.ID}
|
||||
}
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
return Transaction{}, false, err
|
||||
}
|
||||
|
||||
requestJSON := string(request.Request)
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
INSERT INTO transactions (
|
||||
id, idempotency_key, source, service, request_json,
|
||||
state, version, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)`,
|
||||
request.ID,
|
||||
request.IdempotencyKey,
|
||||
request.Source,
|
||||
request.Service,
|
||||
requestJSON,
|
||||
StateCreated,
|
||||
formatTime(now),
|
||||
formatTime(now),
|
||||
)
|
||||
if err != nil {
|
||||
return Transaction{}, false, fmt.Errorf("insert transaction: %w", err)
|
||||
}
|
||||
if err := insertEvent(ctx, tx, request.ID, "", "TRANSACTION_CREATED", "", StateCreated, "", now); err != nil {
|
||||
return Transaction{}, false, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return Transaction{}, false, fmt.Errorf("commit create transaction: %w", err)
|
||||
}
|
||||
return Transaction{
|
||||
ID: request.ID,
|
||||
IdempotencyKey: request.IdempotencyKey,
|
||||
Source: request.Source,
|
||||
Service: request.Service,
|
||||
Request: cloneJSON(request.Request),
|
||||
State: StateCreated,
|
||||
Version: 1,
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}, true, nil
|
||||
}
|
||||
|
||||
func validateCreateRequest(request *CreateRequest) error {
|
||||
if strings.TrimSpace(request.IdempotencyKey) == "" {
|
||||
return errors.New("idempotency key is required")
|
||||
}
|
||||
if strings.TrimSpace(request.Source) == "" {
|
||||
return errors.New("transaction source is required")
|
||||
}
|
||||
if strings.TrimSpace(request.Service) == "" {
|
||||
return errors.New("transaction service is required")
|
||||
}
|
||||
if len(request.Request) == 0 {
|
||||
request.Request = json.RawMessage(`{}`)
|
||||
}
|
||||
if !json.Valid(request.Request) {
|
||||
return errors.New("transaction request is not valid JSON")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Transaction 返回指定事务的最新持久化快照。
|
||||
func (s *Store) Transaction(ctx context.Context, id string) (Transaction, error) {
|
||||
return getTransactionByID(ctx, s.db, id)
|
||||
}
|
||||
|
||||
// ActiveTransaction 返回当前唯一未结束事务。
|
||||
func (s *Store) ActiveTransaction(ctx context.Context) (Transaction, error) {
|
||||
return getActiveTransaction(ctx, s.db)
|
||||
}
|
||||
|
||||
// Transition 校验并原子提交状态变化及其恢复事件。
|
||||
func (s *Store) Transition(ctx context.Context, id string, next State, message string) (Transaction, error) {
|
||||
if !next.Valid() {
|
||||
return Transaction{}, fmt.Errorf("unknown transaction state: %q", next)
|
||||
}
|
||||
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
|
||||
if err != nil {
|
||||
return Transaction{}, fmt.Errorf("begin state transition: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
record, err := getTransactionByID(ctx, tx, id)
|
||||
if err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
if record.State == next {
|
||||
return record, nil
|
||||
}
|
||||
if !record.State.CanTransitionTo(next) {
|
||||
return Transaction{}, &TransitionError{From: record.State, To: next}
|
||||
}
|
||||
now := s.now().UTC()
|
||||
result, err := tx.ExecContext(ctx, `
|
||||
UPDATE transactions
|
||||
SET state = ?, version = version + 1, updated_at = ?
|
||||
WHERE id = ? AND version = ?`,
|
||||
next, formatTime(now), id, record.Version,
|
||||
)
|
||||
if err != nil {
|
||||
return Transaction{}, fmt.Errorf("update transaction state: %w", err)
|
||||
}
|
||||
rows, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
return Transaction{}, fmt.Errorf("read updated transaction rows: %w", err)
|
||||
}
|
||||
if rows != 1 {
|
||||
return Transaction{}, errors.New("transaction changed concurrently")
|
||||
}
|
||||
if err := insertEvent(ctx, tx, id, "", "TRANSACTION_STATE_CHANGED", record.State, next, message, now); err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return Transaction{}, fmt.Errorf("commit state transition: %w", err)
|
||||
}
|
||||
record.State = next
|
||||
record.Version++
|
||||
record.UpdatedAt = now
|
||||
return record, nil
|
||||
}
|
||||
|
||||
// RecordStepIntent 先于外部副作用持久化步骤意图。created=false 表示相同意图已经存在。
|
||||
func (s *Store) RecordStepIntent(ctx context.Context, transactionID string, intent StepIntent) (record Step, created bool, err error) {
|
||||
if strings.TrimSpace(intent.Key) == "" {
|
||||
return Step{}, false, errors.New("step key is required")
|
||||
}
|
||||
if strings.TrimSpace(intent.Name) == "" {
|
||||
return Step{}, false, errors.New("step name is required")
|
||||
}
|
||||
if len(intent.Intent) == 0 {
|
||||
intent.Intent = json.RawMessage(`{}`)
|
||||
}
|
||||
if !json.Valid(intent.Intent) {
|
||||
return Step{}, false, errors.New("step intent is not valid JSON")
|
||||
}
|
||||
|
||||
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
|
||||
if err != nil {
|
||||
return Step{}, false, fmt.Errorf("begin record step intent: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
transactionRecord, err := getTransactionByID(ctx, tx, transactionID)
|
||||
if err != nil {
|
||||
return Step{}, false, err
|
||||
}
|
||||
if transactionRecord.State.Terminal() {
|
||||
return Step{}, false, fmt.Errorf("cannot record a step for terminal transaction %s", transactionID)
|
||||
}
|
||||
existing, err := getStep(ctx, tx, transactionID, intent.Key)
|
||||
if err == nil {
|
||||
if existing.Name != intent.Name || !bytes.Equal(existing.Intent, intent.Intent) {
|
||||
return Step{}, false, ErrStepConflict
|
||||
}
|
||||
return existing, false, nil
|
||||
}
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
return Step{}, false, err
|
||||
}
|
||||
|
||||
now := s.now().UTC()
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
INSERT INTO transaction_steps (
|
||||
transaction_id, step_key, name, status, intent_json,
|
||||
result_json, error_message, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, NULL, '', ?, ?)`,
|
||||
transactionID,
|
||||
intent.Key,
|
||||
intent.Name,
|
||||
StepIntentRecorded,
|
||||
string(intent.Intent),
|
||||
formatTime(now),
|
||||
formatTime(now),
|
||||
)
|
||||
if err != nil {
|
||||
return Step{}, false, fmt.Errorf("insert step intent: %w", err)
|
||||
}
|
||||
if err := insertEvent(ctx, tx, transactionID, intent.Key, "STEP_INTENT_RECORDED", "", "", intent.Name, now); err != nil {
|
||||
return Step{}, false, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return Step{}, false, fmt.Errorf("commit step intent: %w", err)
|
||||
}
|
||||
return Step{
|
||||
TransactionID: transactionID,
|
||||
Key: intent.Key,
|
||||
Name: intent.Name,
|
||||
Status: StepIntentRecorded,
|
||||
Intent: cloneJSON(intent.Intent),
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}, true, nil
|
||||
}
|
||||
|
||||
// CompleteStep 原子记录外部状态核对后的最终结果。
|
||||
func (s *Store) CompleteStep(ctx context.Context, transactionID, stepKey string, status StepStatus, result json.RawMessage, errorMessage string) (Step, error) {
|
||||
if status != StepSucceeded && status != StepFailed {
|
||||
return Step{}, fmt.Errorf("invalid final step status: %q", status)
|
||||
}
|
||||
if len(result) > 0 && !json.Valid(result) {
|
||||
return Step{}, errors.New("step result is not valid JSON")
|
||||
}
|
||||
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
|
||||
if err != nil {
|
||||
return Step{}, fmt.Errorf("begin complete step: %w", err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
record, err := getStep(ctx, tx, transactionID, stepKey)
|
||||
if err != nil {
|
||||
return Step{}, err
|
||||
}
|
||||
if record.Status == status {
|
||||
if !bytes.Equal(record.Result, result) || record.Error != errorMessage {
|
||||
return Step{}, ErrStepConflict
|
||||
}
|
||||
return record, nil
|
||||
}
|
||||
if record.Status != StepIntentRecorded {
|
||||
return Step{}, ErrStepNotPending
|
||||
}
|
||||
now := s.now().UTC()
|
||||
var resultValue any
|
||||
if len(result) > 0 {
|
||||
resultValue = string(result)
|
||||
}
|
||||
updateResult, err := tx.ExecContext(ctx, `
|
||||
UPDATE transaction_steps
|
||||
SET status = ?, result_json = ?, error_message = ?, updated_at = ?
|
||||
WHERE transaction_id = ? AND step_key = ? AND status = ?`,
|
||||
status,
|
||||
resultValue,
|
||||
errorMessage,
|
||||
formatTime(now),
|
||||
transactionID,
|
||||
stepKey,
|
||||
StepIntentRecorded,
|
||||
)
|
||||
if err != nil {
|
||||
return Step{}, fmt.Errorf("update step result: %w", err)
|
||||
}
|
||||
updatedRows, err := updateResult.RowsAffected()
|
||||
if err != nil {
|
||||
return Step{}, fmt.Errorf("read updated step rows: %w", err)
|
||||
}
|
||||
if updatedRows != 1 {
|
||||
return Step{}, errors.New("step changed concurrently")
|
||||
}
|
||||
eventKind := "STEP_SUCCEEDED"
|
||||
if status == StepFailed {
|
||||
eventKind = "STEP_FAILED"
|
||||
}
|
||||
if err := insertEvent(ctx, tx, transactionID, stepKey, eventKind, "", "", errorMessage, now); err != nil {
|
||||
return Step{}, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return Step{}, fmt.Errorf("commit step result: %w", err)
|
||||
}
|
||||
record.Status = status
|
||||
record.Result = cloneJSON(result)
|
||||
record.Error = errorMessage
|
||||
record.UpdatedAt = now
|
||||
return record, nil
|
||||
}
|
||||
|
||||
// PendingSteps 返回重启后必须先核对实际外部状态的步骤。
|
||||
func (s *Store) PendingSteps(ctx context.Context, transactionID string) ([]Step, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT transaction_id, step_key, name, status, intent_json,
|
||||
result_json, error_message, created_at, updated_at
|
||||
FROM transaction_steps
|
||||
WHERE transaction_id = ? AND status = ?
|
||||
ORDER BY created_at, step_key`, transactionID, StepIntentRecorded)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query pending steps: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var records []Step
|
||||
for rows.Next() {
|
||||
record, err := scanStep(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
records = append(records, record)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("iterate pending steps: %w", err)
|
||||
}
|
||||
return records, nil
|
||||
}
|
||||
|
||||
// EventsAfter 返回指定顺序位置之后的事务事件。
|
||||
func (s *Store) EventsAfter(ctx context.Context, transactionID string, afterSequence int64, limit int) ([]Event, error) {
|
||||
if afterSequence < 0 {
|
||||
return nil, errors.New("event sequence must not be negative")
|
||||
}
|
||||
if limit < 1 || limit > 1000 {
|
||||
return nil, errors.New("event limit must be between 1 and 1000")
|
||||
}
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT sequence, transaction_id, step_key, kind,
|
||||
from_state, to_state, message, created_at
|
||||
FROM transaction_events
|
||||
WHERE transaction_id = ? AND sequence > ?
|
||||
ORDER BY sequence
|
||||
LIMIT ?`, transactionID, afterSequence, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query transaction events: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var events []Event
|
||||
for rows.Next() {
|
||||
var event Event
|
||||
var fromState, toState, createdAt string
|
||||
if err := rows.Scan(
|
||||
&event.Sequence,
|
||||
&event.TransactionID,
|
||||
&event.StepKey,
|
||||
&event.Kind,
|
||||
&fromState,
|
||||
&toState,
|
||||
&event.Message,
|
||||
&createdAt,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan transaction event: %w", err)
|
||||
}
|
||||
event.FromState = State(fromState)
|
||||
event.ToState = State(toState)
|
||||
event.CreatedAt, err = parseTime(createdAt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
events = append(events, event)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("iterate transaction events: %w", err)
|
||||
}
|
||||
return events, nil
|
||||
}
|
||||
|
||||
type queryRower interface {
|
||||
QueryRowContext(context.Context, string, ...any) *sql.Row
|
||||
}
|
||||
|
||||
type rowScanner interface {
|
||||
Scan(...any) error
|
||||
}
|
||||
|
||||
func getTransactionByID(ctx context.Context, query queryRower, id string) (Transaction, error) {
|
||||
return scanTransaction(query.QueryRowContext(ctx, `
|
||||
SELECT id, idempotency_key, source, service, request_json,
|
||||
state, version, created_at, updated_at
|
||||
FROM transactions
|
||||
WHERE id = ?`, id))
|
||||
}
|
||||
|
||||
func getTransactionByIdempotencyKey(ctx context.Context, query queryRower, key string) (Transaction, error) {
|
||||
return scanTransaction(query.QueryRowContext(ctx, `
|
||||
SELECT id, idempotency_key, source, service, request_json,
|
||||
state, version, created_at, updated_at
|
||||
FROM transactions
|
||||
WHERE idempotency_key = ?`, key))
|
||||
}
|
||||
|
||||
func getActiveTransaction(ctx context.Context, query queryRower) (Transaction, error) {
|
||||
return scanTransaction(query.QueryRowContext(ctx, `
|
||||
SELECT id, idempotency_key, source, service, request_json,
|
||||
state, version, created_at, updated_at
|
||||
FROM transactions
|
||||
WHERE state NOT IN (?, ?, ?)
|
||||
LIMIT 1`, StateCommitted, StateRolledBack, StateFailed))
|
||||
}
|
||||
|
||||
func scanTransaction(row rowScanner) (Transaction, error) {
|
||||
var record Transaction
|
||||
var requestJSON, state, createdAt, updatedAt string
|
||||
if err := row.Scan(
|
||||
&record.ID,
|
||||
&record.IdempotencyKey,
|
||||
&record.Source,
|
||||
&record.Service,
|
||||
&requestJSON,
|
||||
&state,
|
||||
&record.Version,
|
||||
&createdAt,
|
||||
&updatedAt,
|
||||
); err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return Transaction{}, ErrNotFound
|
||||
}
|
||||
return Transaction{}, fmt.Errorf("scan transaction: %w", err)
|
||||
}
|
||||
record.Request = json.RawMessage(requestJSON)
|
||||
record.State = State(state)
|
||||
var err error
|
||||
record.CreatedAt, err = parseTime(createdAt)
|
||||
if err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
record.UpdatedAt, err = parseTime(updatedAt)
|
||||
if err != nil {
|
||||
return Transaction{}, err
|
||||
}
|
||||
return record, nil
|
||||
}
|
||||
|
||||
func getStep(ctx context.Context, query queryRower, transactionID, stepKey string) (Step, error) {
|
||||
return scanStep(query.QueryRowContext(ctx, `
|
||||
SELECT transaction_id, step_key, name, status, intent_json,
|
||||
result_json, error_message, created_at, updated_at
|
||||
FROM transaction_steps
|
||||
WHERE transaction_id = ? AND step_key = ?`, transactionID, stepKey))
|
||||
}
|
||||
|
||||
func scanStep(row rowScanner) (Step, error) {
|
||||
var record Step
|
||||
var status, intentJSON, createdAt, updatedAt string
|
||||
var resultJSON sql.NullString
|
||||
if err := row.Scan(
|
||||
&record.TransactionID,
|
||||
&record.Key,
|
||||
&record.Name,
|
||||
&status,
|
||||
&intentJSON,
|
||||
&resultJSON,
|
||||
&record.Error,
|
||||
&createdAt,
|
||||
&updatedAt,
|
||||
); err != nil {
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return Step{}, ErrNotFound
|
||||
}
|
||||
return Step{}, fmt.Errorf("scan transaction step: %w", err)
|
||||
}
|
||||
record.Status = StepStatus(status)
|
||||
record.Intent = json.RawMessage(intentJSON)
|
||||
if resultJSON.Valid {
|
||||
record.Result = json.RawMessage(resultJSON.String)
|
||||
}
|
||||
var err error
|
||||
record.CreatedAt, err = parseTime(createdAt)
|
||||
if err != nil {
|
||||
return Step{}, err
|
||||
}
|
||||
record.UpdatedAt, err = parseTime(updatedAt)
|
||||
if err != nil {
|
||||
return Step{}, err
|
||||
}
|
||||
return record, nil
|
||||
}
|
||||
|
||||
func insertEvent(
|
||||
ctx context.Context,
|
||||
tx *sql.Tx,
|
||||
transactionID string,
|
||||
stepKey string,
|
||||
kind string,
|
||||
fromState State,
|
||||
toState State,
|
||||
message string,
|
||||
createdAt time.Time,
|
||||
) error {
|
||||
_, err := tx.ExecContext(ctx, `
|
||||
INSERT INTO transaction_events (
|
||||
transaction_id, step_key, kind, from_state, to_state, message, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
||||
transactionID,
|
||||
stepKey,
|
||||
kind,
|
||||
fromState,
|
||||
toState,
|
||||
message,
|
||||
formatTime(createdAt),
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("insert transaction event: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func formatTime(value time.Time) string {
|
||||
return value.UTC().Format(time.RFC3339Nano)
|
||||
}
|
||||
|
||||
func parseTime(value string) (time.Time, error) {
|
||||
parsed, err := time.Parse(time.RFC3339Nano, value)
|
||||
if err != nil {
|
||||
return time.Time{}, fmt.Errorf("parse persisted time %q: %w", value, err)
|
||||
}
|
||||
return parsed, nil
|
||||
}
|
||||
|
||||
func cloneJSON(value json.RawMessage) json.RawMessage {
|
||||
if len(value) == 0 {
|
||||
return nil
|
||||
}
|
||||
return bytes.Clone(value)
|
||||
}
|
||||
@@ -0,0 +1,220 @@
|
||||
package transaction
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t)
|
||||
request := CreateRequest{
|
||||
ID: "transaction-1",
|
||||
IdempotencyKey: "request-1",
|
||||
Source: "test",
|
||||
Service: "backend",
|
||||
Request: json.RawMessage(`{"service":"backend"}`),
|
||||
}
|
||||
created, isNew, err := store.CreateTransaction(ctx, request)
|
||||
if err != nil {
|
||||
t.Fatalf("create transaction: %v", err)
|
||||
}
|
||||
if !isNew || created.State != StateCreated {
|
||||
t.Fatalf("unexpected created transaction: %+v new=%v", created, isNew)
|
||||
}
|
||||
|
||||
retried, isNew, err := store.CreateTransaction(ctx, request)
|
||||
if err != nil {
|
||||
t.Fatalf("retry transaction: %v", err)
|
||||
}
|
||||
if isNew || retried.ID != created.ID {
|
||||
t.Fatalf("idempotent retry created another transaction: %+v", retried)
|
||||
}
|
||||
|
||||
_, _, err = store.CreateTransaction(ctx, CreateRequest{
|
||||
ID: "transaction-2",
|
||||
IdempotencyKey: "request-2",
|
||||
Source: "test",
|
||||
Service: "frontend",
|
||||
})
|
||||
var activeErr *ActiveTransactionError
|
||||
if !errors.As(err, &activeErr) || activeErr.TransactionID != created.ID {
|
||||
t.Fatalf("expected active transaction error, got %v", err)
|
||||
}
|
||||
|
||||
if _, err := store.Transition(ctx, created.ID, StateFailed, "test terminal state"); err != nil {
|
||||
t.Fatalf("finish first transaction: %v", err)
|
||||
}
|
||||
second, isNew, err := store.CreateTransaction(ctx, CreateRequest{
|
||||
ID: "transaction-2",
|
||||
IdempotencyKey: "request-2",
|
||||
Source: "test",
|
||||
Service: "frontend",
|
||||
})
|
||||
if err != nil || !isNew || second.ID != "transaction-2" {
|
||||
t.Fatalf("create transaction after terminal state: record=%+v new=%v err=%v", second, isNew, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStoreTransitionStepAndEventPersistence(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
databasePath := filepath.Join(t.TempDir(), "transaction.db")
|
||||
store, err := OpenStore(ctx, databasePath)
|
||||
if err != nil {
|
||||
t.Fatalf("open store: %v", err)
|
||||
}
|
||||
fixedTime := time.Date(2026, time.August, 15, 10, 0, 0, 0, time.UTC)
|
||||
store.now = func() time.Time { return fixedTime }
|
||||
record, _, err := store.CreateTransaction(ctx, CreateRequest{
|
||||
ID: "transaction-persisted",
|
||||
IdempotencyKey: "request-persisted",
|
||||
Source: "test",
|
||||
Service: "all",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create transaction: %v", err)
|
||||
}
|
||||
if _, err := store.Transition(ctx, record.ID, StateValidating, "validation started"); err != nil {
|
||||
t.Fatalf("transition transaction: %v", err)
|
||||
}
|
||||
if _, err := store.Transition(ctx, record.ID, StatePrepared, "validation completed"); err != nil {
|
||||
t.Fatalf("transition transaction: %v", err)
|
||||
}
|
||||
step, isNew, err := store.RecordStepIntent(ctx, record.ID, StepIntent{
|
||||
Key: "prepare-files",
|
||||
Name: "prepare immutable files",
|
||||
Intent: json.RawMessage(`{"sha256":"abc"}`),
|
||||
})
|
||||
if err != nil || !isNew || step.Status != StepIntentRecorded {
|
||||
t.Fatalf("record step intent: step=%+v new=%v err=%v", step, isNew, err)
|
||||
}
|
||||
if err := store.Close(); err != nil {
|
||||
t.Fatalf("close store: %v", err)
|
||||
}
|
||||
|
||||
reopened, err := OpenStore(ctx, databasePath)
|
||||
if err != nil {
|
||||
t.Fatalf("reopen store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = reopened.Close() })
|
||||
persisted, err := reopened.Transaction(ctx, record.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("read persisted transaction: %v", err)
|
||||
}
|
||||
if persisted.State != StatePrepared || persisted.Version != 3 {
|
||||
t.Fatalf("unexpected persisted transaction: %+v", persisted)
|
||||
}
|
||||
pending, err := reopened.PendingSteps(ctx, record.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("read pending steps: %v", err)
|
||||
}
|
||||
if len(pending) != 1 || pending[0].Key != step.Key {
|
||||
t.Fatalf("unexpected pending steps: %+v", pending)
|
||||
}
|
||||
completed, err := reopened.CompleteStep(ctx, record.ID, step.Key, StepSucceeded, json.RawMessage(`{"installed":true}`), "")
|
||||
if err != nil || completed.Status != StepSucceeded {
|
||||
t.Fatalf("complete step: step=%+v err=%v", completed, err)
|
||||
}
|
||||
events, err := reopened.EventsAfter(ctx, record.ID, 0, 100)
|
||||
if err != nil {
|
||||
t.Fatalf("read events: %v", err)
|
||||
}
|
||||
if len(events) != 5 {
|
||||
t.Fatalf("unexpected event count: got %d events=%+v", len(events), events)
|
||||
}
|
||||
for index := 1; index < len(events); index++ {
|
||||
if events[index].Sequence <= events[index-1].Sequence {
|
||||
t.Fatalf("events are not ordered: %+v", events)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStoreRejectsInvalidTransitionAndConflictingStepIntent(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t)
|
||||
record, _, err := store.CreateTransaction(ctx, CreateRequest{
|
||||
ID: "transaction-conflict",
|
||||
IdempotencyKey: "request-conflict",
|
||||
Source: "test",
|
||||
Service: "backend",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create transaction: %v", err)
|
||||
}
|
||||
_, err = store.Transition(ctx, record.ID, StatePrepared, "skip validation")
|
||||
var transitionErr *TransitionError
|
||||
if !errors.As(err, &transitionErr) {
|
||||
t.Fatalf("expected transition error, got %v", err)
|
||||
}
|
||||
intent := StepIntent{Key: "same-key", Name: "first", Intent: json.RawMessage(`{"value":1}`)}
|
||||
if _, _, err := store.RecordStepIntent(ctx, record.ID, intent); err != nil {
|
||||
t.Fatalf("record first step intent: %v", err)
|
||||
}
|
||||
_, _, err = store.RecordStepIntent(ctx, record.ID, StepIntent{
|
||||
Key: intent.Key,
|
||||
Name: "different",
|
||||
Intent: intent.Intent,
|
||||
})
|
||||
if !errors.Is(err, ErrStepConflict) {
|
||||
t.Fatalf("expected step conflict, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStoreSerializesConcurrentCreates(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t)
|
||||
const workers = 12
|
||||
var wait sync.WaitGroup
|
||||
wait.Add(workers)
|
||||
results := make(chan error, workers)
|
||||
for index := 0; index < workers; index++ {
|
||||
go func(index int) {
|
||||
defer wait.Done()
|
||||
_, _, err := store.CreateTransaction(ctx, CreateRequest{
|
||||
IdempotencyKey: "concurrent-" + string(rune('A'+index)),
|
||||
Source: "test",
|
||||
Service: "backend",
|
||||
})
|
||||
results <- err
|
||||
}(index)
|
||||
}
|
||||
wait.Wait()
|
||||
close(results)
|
||||
var created, rejected int
|
||||
for err := range results {
|
||||
switch {
|
||||
case err == nil:
|
||||
created++
|
||||
case errors.Is(err, ErrActiveExists):
|
||||
rejected++
|
||||
default:
|
||||
t.Fatalf("unexpected create error: %v", err)
|
||||
}
|
||||
}
|
||||
if created != 1 || rejected != workers-1 {
|
||||
t.Fatalf("unexpected concurrent result: created=%d rejected=%d", created, rejected)
|
||||
}
|
||||
}
|
||||
|
||||
func openTestStore(t *testing.T) *Store {
|
||||
t.Helper()
|
||||
store, err := OpenStore(context.Background(), filepath.Join(t.TempDir(), "transaction.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("open test store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := store.Close(); err != nil {
|
||||
t.Errorf("close test store: %v", err)
|
||||
}
|
||||
})
|
||||
return store
|
||||
}
|
||||
Reference in New Issue
Block a user