108 lines
4.1 KiB
Go
108 lines
4.1 KiB
Go
package outbox
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/json"
|
||
"errors"
|
||
"math/rand/v2"
|
||
"time"
|
||
|
||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||
)
|
||
|
||
// SendOnce 有界领取并发送一项；网络交换在数据库事务外，ErrIdle表示本次无到期项。
|
||
// 返回记录只包含本地发送状态；ACCEPTED不是业务执行成功，REVIEW必须人工核查原身份。
|
||
func (s *Store) SendOnce(parent context.Context, api *orbit.Client) (Record, error) {
|
||
if api == nil {
|
||
return Record{}, ErrInvalid
|
||
}
|
||
// 整个网络阶段从领取前即受单调时钟15秒约束，短于30秒租约；暂停进程后不会获得新发送预算。
|
||
work, stop := context.WithTimeout(parent, 15*time.Second)
|
||
defer stop()
|
||
r, err := s.claim(work)
|
||
if err != nil {
|
||
return Record{}, err
|
||
}
|
||
if r.State == "REVIEW" {
|
||
return s.Status(parent, r.OperationID)
|
||
}
|
||
state, code, delay, receipt := s.deliver(work, api, r)
|
||
// 网络超时不能阻止保存未知结果分类，收尾仍服从调用方总上下文且最多5秒。
|
||
if err = s.finish(parent, r, state, code, delay, receipt); err != nil {
|
||
return Record{}, err
|
||
}
|
||
return s.Status(parent, r.OperationID)
|
||
}
|
||
|
||
func (s *Store) deliver(ctx context.Context, api *orbit.Client, r Record) (string, string, time.Duration, *orbit.SubmitTaskResult) {
|
||
digest := sha256.Sum256(r.RequestJSON)
|
||
if !bytes.Equal(digest[:], r.RequestDigest) || r.NamespaceID != s.namespace || r.QueueID != s.queue || r.IdempotencyKey != "outbox-task-"+r.OperationID {
|
||
return "REVIEW", "INVALID_DOCUMENT", 0, nil
|
||
}
|
||
var in orbit.SubmitTaskRequest
|
||
decoder := json.NewDecoder(bytes.NewReader(r.RequestJSON))
|
||
decoder.DisallowUnknownFields()
|
||
if decoder.Decode(&in) != nil || in.QueueID != s.queue {
|
||
return "REVIEW", "INVALID_DOCUMENT", 0, nil
|
||
}
|
||
// 先通过当前凭据读取固定队列；不靠配置声称命名空间正确，不向错误环境盲发命令。
|
||
queue, err := api.GetQueue(ctx, s.queue)
|
||
if err != nil {
|
||
state, code, delay := classify(err, r.SendCount)
|
||
return state, code, delay, nil
|
||
}
|
||
if queue.ID != s.queue || queue.NamespaceID != s.namespace {
|
||
return "REVIEW", "TARGET_MISMATCH", 0, nil
|
||
}
|
||
// 数据库时间决定剩余截止预算；额外查询不授予新租约，换代或临近截止时不再开始HTTP。
|
||
var remaining int64
|
||
err = s.db.QueryRowContext(ctx, `SELECT TIMESTAMPDIFF(MICROSECOND,UTC_TIMESTAMP(6),LEAST(wake_at,send_deadline_at)) FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=? AND state='SENDING' AND lease_epoch=?`, s.namespace, s.queue, r.OperationID, r.LeaseEpoch).Scan(&remaining)
|
||
if err != nil || remaining <= 0 {
|
||
return "PENDING", "UNCONFIRMED", time.Second, nil
|
||
}
|
||
send, stop := context.WithTimeout(ctx, min(10*time.Second, time.Duration(remaining)*time.Microsecond))
|
||
defer stop()
|
||
result, err := api.SubmitTask(send, r.IdempotencyKey, in)
|
||
if err != nil {
|
||
state, code, delay := classify(err, r.SendCount)
|
||
return state, code, delay, nil
|
||
}
|
||
if !validReceipt(r, result) {
|
||
return "REVIEW", "INVALID_RECEIPT", 0, nil
|
||
}
|
||
return "ACCEPTED", "ACCEPTED", 0, &result
|
||
}
|
||
|
||
func validReceipt(r Record, receipt orbit.SubmitTaskResult) bool {
|
||
t := receipt.Task
|
||
return idPattern.MatchString(t.ID) && t.NamespaceID == r.NamespaceID && t.QueueID == r.QueueID && t.Revision > 0
|
||
}
|
||
|
||
func classify(err error, count uint32) (state, code string, delay time.Duration) {
|
||
state, code = "PENDING", "UNCONFIRMED"
|
||
delay = min(time.Minute, time.Second*time.Duration(uint64(1)<<min(max(count, 1)-1, 6))) + time.Duration(rand.IntN(1000))*time.Millisecond
|
||
var api *orbit.APIError
|
||
if errors.As(err, &api) {
|
||
if api.StatusCode == 429 {
|
||
code = "BACKPRESSURE"
|
||
delay = max(delay, api.RetryAfter)
|
||
} else if api.StatusCode < 500 && api.StatusCode != 408 {
|
||
state, code = "REVIEW", "HTTP_REJECTED"
|
||
if api.StatusCode == 401 || api.StatusCode == 403 {
|
||
code = "AUTHORIZATION"
|
||
}
|
||
if api.Code == "IDEMPOTENCY_CONFLICT" {
|
||
code = "IDEMPOTENCY_CONFLICT"
|
||
}
|
||
}
|
||
} else {
|
||
var httpErr *orbit.HTTPError
|
||
if errors.As(err, &httpErr) && httpErr.StatusCode < 500 && httpErr.StatusCode != 408 && httpErr.StatusCode != 429 {
|
||
state, code = "REVIEW", "HTTP_REJECTED"
|
||
}
|
||
}
|
||
return state, code, min(delay, time.Hour)
|
||
}
|