feat: add durable business task outbox example and fault recovery

2026-09-17 23:13:48 +08:00
parent 2b8213a4f7
commit 522fe2ef3e
27 changed files with 2195 additions and 7 deletions
--- a/examples/task-outbox/internal/outbox/store.go
+++ b/examples/task-outbox/internal/outbox/store.go
@@ -0,0 +1,321 @@
+// Package outbox 演示接入方本地业务事务、稳定提交身份及有限发送恢复。
+package outbox
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	_ "embed"
+	"encoding/json"
+	"errors"
+	"regexp"
+	"strings"
+	"time"
+	"unicode"
+	"unicode/utf8"
+
+	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
+	"github.com/go-sql-driver/mysql"
+)
+
+//go:embed schema.sql
+var schema string
+
+var (
+	ErrInvalid           = errors.New("outbox scope, operation or request is invalid")
+	ErrConflict          = errors.New("outbox operation already has different content; keep the original identity")
+	ErrStorage           = errors.New("outbox storage failed; reconcile the original operation")
+	ErrCommitUnconfirmed = errors.New("outbox commit is unconfirmed; reconcile the original operation")
+	ErrIdle              = errors.New("no due outbox record")
+	ErrNotFound          = errors.New("outbox operation not found in this scope")
+	ErrLeaseLost         = errors.New("outbox lease was replaced; do not overwrite its result")
+	idPattern            = regexp.MustCompile(`^[a-f0-9]{32}$`)
+)
+
+const maxSends = 100
+const recordColumns = `namespace_id,queue_id,operation_id,idempotency_key,request_json,request_digest,state,send_count,lease_epoch,wake_at,send_deadline_at,last_code,task_id,receipt_json,created_at,updated_at`
+
+type Store struct {
+	db               *sql.DB
+	namespace, queue string
+}
+
+// Open 仅连接业务方数据库，不创建表，不输出原始DSN或数据库错误。
+func Open(ctx context.Context, dsn string) (*sql.DB, error) {
+	cfg, err := mysql.ParseDSN(dsn)
+	if err != nil || cfg.DBName == "" {
+		return nil, ErrInvalid
+	}
+	cfg.ParseTime = true
+	cfg.Loc = time.UTC
+	cfg.MultiStatements = false
+	cfg.AllowAllFiles = false
+	cfg.Timeout, cfg.ReadTimeout, cfg.WriteTimeout = 5*time.Second, 8*time.Second, 8*time.Second
+	if cfg.Params == nil {
+		cfg.Params = map[string]string{}
+	}
+	cfg.Params["time_zone"] = "'+00:00'"
+	connector, err := mysql.NewConnector(cfg)
+	if err != nil {
+		return nil, ErrStorage
+	}
+	db := sql.OpenDB(connector)
+	db.SetMaxOpenConns(8)
+	db.SetMaxIdleConns(4)
+	db.SetConnMaxLifetime(5 * time.Minute)
+	if err = db.PingContext(ctx); err != nil {
+		db.Close()
+		return nil, ErrStorage
+	}
+	return db, nil
+}
+
+// Init 仅用于专用空库，已有表会报错；生产接入使用业务方自己的迁移。
+func Init(ctx context.Context, db *sql.DB) error {
+	for _, statement := range strings.Split(schema, ";") {
+		if strings.TrimSpace(statement) == "" {
+			continue
+		}
+		if _, err := db.ExecContext(ctx, statement); err != nil {
+			return ErrStorage
+		}
+	}
+	return nil
+}
+func New(db *sql.DB, namespace, queue string) (*Store, error) {
+	if db == nil || !idPattern.MatchString(namespace) || !idPattern.MatchString(queue) {
+		return nil, ErrInvalid
+	}
+	return &Store{db: db, namespace: namespace, queue: queue}, nil
+}
+
+// document 冻结当前v1默认值及UTC表示；Outbox按文档字节去重，不冒充平台语义摘要。
+func (s *Store) document(in orbit.SubmitTaskRequest) ([]byte, error) {
+	// encoding/json会把非法UTF-8替换成U+FFFD；在冻结摘要前拒绝，不能静默改变业务标识。
+	for _, value := range []string{in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, in.BusinessKey, in.ConcurrencyKey} {
+		if !utf8.ValidString(value) {
+			return nil, ErrInvalid
+		}
+	}
+	if in.QueueID != s.queue || in.Handler == "" || in.HandlerVersion == "" || in.PayloadVersion == "" || len(in.Payload) > 256<<10 || !utf8.Valid(in.Payload) || !json.Valid(in.Payload) || bytes.Equal(bytes.TrimSpace(in.Payload), []byte("null")) {
+		return nil, ErrInvalid
+	}
+	if in.TaskTimeoutMS == 0 {
+		in.TaskTimeoutMS = 86400000
+	}
+	if in.RecoverySafety == "" {
+		in.RecoverySafety = orbit.RecoveryReconcileOnly
+	}
+	if in.Retry.MaxAttempts == 0 {
+		in.Retry.MaxAttempts = 1
+	}
+	if in.Retry.AttemptTimeoutMS == 0 {
+		in.Retry.AttemptTimeoutMS = 30000
+	}
+	if in.Retry.MaxAttempts > 1 {
+		if in.Retry.InitialDelayMS == 0 {
+			in.Retry.InitialDelayMS = 1000
+		}
+		if in.Retry.MaxDelayMS == 0 {
+			in.Retry.MaxDelayMS = 30000
+		}
+	}
+	if in.AvailableAt != nil {
+		at := in.AvailableAt.UTC()
+		in.AvailableAt = &at
+	}
+	if in.StartDeadlineAt != nil {
+		at := in.StartDeadlineAt.UTC()
+		in.StartDeadlineAt = &at
+	}
+	raw, err := json.Marshal(in)
+	if err != nil || len(raw) > 512<<10 {
+		return nil, ErrInvalid
+	}
+	return raw, nil
+}
+
+// Enqueue 原子创建业务记录及Outbox。duplicate只表示该本地事务已经提交，不表示平台受理。
+// 调用方在第一次调用前持久化operationID；COMMIT回执未知时仍用原ID、原内容再次调用。
+func (s *Store) Enqueue(parent context.Context, operationID, sourceRef string, in orbit.SubmitTaskRequest) (duplicate bool, err error) {
+	if !idPattern.MatchString(operationID) || !utf8.ValidString(sourceRef) || strings.TrimSpace(sourceRef) == "" || utf8.RuneCountInString(sourceRef) > 512 || strings.ContainsFunc(sourceRef, unicode.IsControl) {
+		return false, ErrInvalid
+	}
+	raw, err := s.document(in)
+	if err != nil {
+		return false, err
+	}
+	digest := sha256.Sum256(raw)
+	ctx, stop := context.WithTimeout(parent, 5*time.Second)
+	defer stop()
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return false, ErrStorage
+	}
+	defer tx.Rollback()
+	_, err = tx.ExecContext(ctx, `INSERT INTO sender_outbox(namespace_id,queue_id,operation_id,idempotency_key,request_json,request_digest,state,send_count,lease_epoch,wake_at,send_deadline_at,last_code,created_at,updated_at) VALUES(?,?,?,?,?,?,'PENDING',0,0,UTC_TIMESTAMP(6),TIMESTAMPADD(HOUR,24,UTC_TIMESTAMP(6)),'',UTC_TIMESTAMP(6),UTC_TIMESTAMP(6))`, s.namespace, s.queue, operationID, "outbox-task-"+operationID, raw, digest[:])
+	if err != nil {
+		var mysqlErr *mysql.MySQLError
+		if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1062 {
+			return false, ErrStorage
+		}
+		if tx.Rollback() != nil {
+			return false, ErrStorage
+		}
+		var saved []byte
+		var queue, source string
+		err = s.db.QueryRowContext(ctx, `SELECT o.request_digest,o.queue_id,j.source_ref FROM sender_outbox o JOIN sender_jobs j ON j.namespace_id=o.namespace_id AND j.operation_id=o.operation_id WHERE o.namespace_id=? AND o.operation_id=?`, s.namespace, operationID).Scan(&saved, &queue, &source)
+		if err != nil {
+			return false, ErrStorage
+		}
+		if queue != s.queue || source != sourceRef || !bytes.Equal(saved, digest[:]) {
+			return false, ErrConflict
+		}
+		return true, nil
+	}
+	// 锁序统一为Outbox→业务行，不能在业务行持锁时反向调用发送确认。
+	if _, err = tx.ExecContext(ctx, `INSERT INTO sender_jobs(namespace_id,operation_id,source_ref,created_at,updated_at) VALUES(?,?,?,UTC_TIMESTAMP(6),UTC_TIMESTAMP(6))`, s.namespace, operationID, sourceRef); err != nil {
+		return false, ErrStorage
+	}
+	if tx.Commit() != nil {
+		return false, ErrCommitUnconfirmed
+	}
+	return false, nil
+}
+
+type scanner interface{ Scan(...any) error }
+
+func scanRecord(row scanner, r *Record, extra ...any) error {
+	args := []any{&r.NamespaceID, &r.QueueID, &r.OperationID, &r.IdempotencyKey, &r.RequestJSON, &r.RequestDigest, &r.State, &r.SendCount, &r.LeaseEpoch, &r.WakeAt, &r.SendDeadlineAt, &r.LastCode, &r.TaskID, &r.ReceiptJSON, &r.CreatedAt, &r.UpdatedAt}
+	return row.Scan(append(args, extra...)...)
+}
+func (s *Store) Status(parent context.Context, id string) (Record, error) {
+	if !idPattern.MatchString(id) {
+		return Record{}, ErrInvalid
+	}
+	ctx, stop := context.WithTimeout(parent, 5*time.Second)
+	defer stop()
+	var r Record
+	err := scanRecord(s.db.QueryRowContext(ctx, `SELECT `+recordColumns+` FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=?`, s.namespace, s.queue, id), &r)
+	if errors.Is(err, sql.ErrNoRows) {
+		return Record{}, ErrNotFound
+	}
+	if err != nil {
+		return Record{}, ErrStorage
+	}
+	return r, nil
+}
+
+func (s *Store) claim(parent context.Context) (Record, error) {
+	ctx, stop := context.WithTimeout(parent, 5*time.Second)
+	defer stop()
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return Record{}, ErrStorage
+	}
+	defer tx.Rollback()
+	var r Record
+	var now time.Time
+	err = scanRecord(tx.QueryRowContext(ctx, `SELECT `+recordColumns+`,UTC_TIMESTAMP(6) FROM sender_outbox FORCE INDEX(ix_sender_due) WHERE namespace_id=? AND queue_id=? AND wake_at<=UTC_TIMESTAMP(6) ORDER BY wake_at,operation_id LIMIT 1 FOR UPDATE SKIP LOCKED`, s.namespace, s.queue), &r, &now)
+	if errors.Is(err, sql.ErrNoRows) {
+		return Record{}, ErrIdle
+	}
+	if err != nil {
+		return Record{}, ErrStorage
+	}
+	if r.State != "PENDING" && r.State != "SENDING" {
+		return Record{}, ErrStorage
+	}
+	if !now.Before(r.SendDeadlineAt) || r.SendCount >= maxSends {
+		r.State = "REVIEW"
+		r.WakeAt = nil
+		r.LastCode = "SEND_BUDGET_EXHAUSTED"
+		if !now.Before(r.SendDeadlineAt) {
+			r.LastCode = "SEND_DEADLINE"
+		}
+	} else {
+		r.State = "SENDING"
+		r.SendCount++
+		r.LeaseEpoch++
+		wake := now.Add(30 * time.Second)
+		if wake.After(r.SendDeadlineAt) {
+			wake = r.SendDeadlineAt
+		}
+		r.WakeAt = &wake
+	}
+	_, err = tx.ExecContext(ctx, `UPDATE sender_outbox SET state=?,send_count=?,lease_epoch=?,wake_at=?,last_code=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, r.State, r.SendCount, r.LeaseEpoch, r.WakeAt, r.LastCode, s.namespace, r.OperationID)
+	if err != nil {
+		return Record{}, ErrStorage
+	}
+	if tx.Commit() != nil {
+		return Record{}, ErrCommitUnconfirmed
+	}
+	return r, nil
+}
+
+// finish 只接受最近代次；过期但尚未被替代的成功回执可落库，重复HTTP只复用原键。
+func (s *Store) finish(parent context.Context, claim Record, state, code string, delay time.Duration, receipt *orbit.SubmitTaskResult) error {
+	ctx, stop := context.WithTimeout(parent, 5*time.Second)
+	defer stop()
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return ErrStorage
+	}
+	defer tx.Rollback()
+	var current Record
+	var now time.Time
+	err = scanRecord(tx.QueryRowContext(ctx, `SELECT `+recordColumns+`,UTC_TIMESTAMP(6) FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=? FOR UPDATE`, s.namespace, s.queue, claim.OperationID), &current, &now)
+	if err != nil {
+		return ErrStorage
+	}
+	if current.State != "SENDING" || current.LeaseEpoch != claim.LeaseEpoch {
+		return ErrLeaseLost
+	}
+	var wake *time.Time
+	var task *string
+	var raw []byte
+	if state == "ACCEPTED" {
+		if receipt == nil || !validReceipt(current, *receipt) {
+			return ErrInvalid
+		}
+		raw, err = json.Marshal(receipt)
+		if err != nil || len(raw) > 1<<20 {
+			return ErrInvalid
+		}
+		task = &receipt.Task.ID
+		var existing *string
+		if err = tx.QueryRowContext(ctx, `SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=? FOR UPDATE`, s.namespace, current.OperationID).Scan(&existing); err != nil {
+			return ErrStorage
+		}
+		if existing != nil && *existing != *task {
+			return ErrConflict
+		}
+		if _, err = tx.ExecContext(ctx, `UPDATE sender_jobs SET task_id=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, *task, s.namespace, current.OperationID); err != nil {
+			return ErrStorage
+		}
+	} else if state == "PENDING" {
+		if delay < time.Second || delay > time.Hour {
+			return ErrInvalid
+		}
+		if !now.Before(current.SendDeadlineAt) || current.SendCount >= maxSends {
+			state = "REVIEW"
+			code = "SEND_BUDGET_EXHAUSTED"
+		} else {
+			next := now.Add(delay)
+			if next.After(current.SendDeadlineAt) {
+				next = current.SendDeadlineAt
+			}
+			wake = &next
+		}
+	} else if state != "REVIEW" {
+		return ErrInvalid
+	}
+	if _, err = tx.ExecContext(ctx, `UPDATE sender_outbox SET state=?,last_code=?,wake_at=?,task_id=?,receipt_json=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, state, code, wake, task, raw, s.namespace, current.OperationID); err != nil {
+		return ErrStorage
+	}
+	if tx.Commit() != nil {
+		return ErrCommitUnconfirmed
+	}
+	return nil
+}