312 lines
11 KiB
Go
312 lines
11 KiB
Go
// Package outbox 将待发任务加入调用方的 MySQL 事务，并通过有限租约发送器恢复提交。
|
||
package outbox
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
_ "embed"
|
||
"encoding/json"
|
||
"errors"
|
||
"regexp"
|
||
"strings"
|
||
"time"
|
||
"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")
|
||
ErrAcceptance = errors.New("outbox acceptance hook failed; confirmation rolled back")
|
||
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`
|
||
|
||
// Store 固定一个业务数据库及目标范围，不拥有连接池的关闭权。
|
||
type Store struct {
|
||
db *sql.DB
|
||
namespace, queue string
|
||
onAccepted AcceptedHook
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// SchemaSQL 返回首版无外键表结构，供业务迁移工具管理；不会自动执行。
|
||
func SchemaSQL() string { return schema }
|
||
|
||
// 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
|
||
}
|
||
|
||
// New 构造可并发复用的适配器，不执行建表或网络请求。
|
||
func New(db *sql.DB, namespace, queue string, options Options) (*Store, error) {
|
||
if db == nil || !idPattern.MatchString(namespace) || !idPattern.MatchString(queue) {
|
||
return nil, ErrInvalid
|
||
}
|
||
return &Store{db: db, namespace: namespace, queue: queue, onAccepted: options.OnAccepted}, 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
|
||
}
|
||
|
||
// EnqueueTx 只写入调用方事务，不开启/提交/回滚事务，不进行 HTTP 请求。
|
||
// tx 必须来自 New 使用的同一业务数据库。失败时调用方必须回滚整个事务。
|
||
// duplicate 只表示事务内已有同身份同内容记录，调用方仍必须确认 Commit 成功。
|
||
func (s *Store) EnqueueTx(ctx context.Context, tx *sql.Tx, operationID string, in orbit.SubmitTaskRequest) (duplicate bool, err error) {
|
||
if tx == nil || !idPattern.MatchString(operationID) {
|
||
return false, ErrInvalid
|
||
}
|
||
raw, err := s.document(in)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
digest := sha256.Sum256(raw)
|
||
_, 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 {
|
||
return false, nil
|
||
}
|
||
var duplicateKey *mysql.MySQLError
|
||
if !errors.As(err, &duplicateKey) || duplicateKey.Number != 1062 {
|
||
return false, ErrStorage
|
||
}
|
||
var saved []byte
|
||
var queue, key string
|
||
// 共享锁当前读：核对 RR 旧快照外的竞争提交，且不把重复 INSERT 的共享锁升级为排它锁。
|
||
err = tx.QueryRowContext(ctx, `SELECT request_digest,queue_id,idempotency_key FROM sender_outbox WHERE namespace_id=? AND operation_id=? FOR SHARE`, s.namespace, operationID).Scan(&saved, &queue, &key)
|
||
if err != nil {
|
||
return false, ErrStorage
|
||
}
|
||
if queue != s.queue || key != "outbox-task-"+operationID || !bytes.Equal(saved, digest[:]) {
|
||
return false, ErrConflict
|
||
}
|
||
return true, 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...)...)
|
||
}
|
||
|
||
// Status 读取固定命名空间和队列内的本地发送记录，不查询平台执行结果。
|
||
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), ¤t, &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
|
||
// 持锁顺序：Outbox → 业务行。回调只能做短 SQL，不得提交、回滚或执行网络副作用。
|
||
if s.onAccepted != nil {
|
||
if err = s.onAccepted(ctx, tx, current, *receipt); err != nil {
|
||
return ErrAcceptance
|
||
}
|
||
}
|
||
} 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
|
||
}
|