merge: #241 服务端按地址后缀批量回填订单号与下单时间

合并 feat/241-order-backfill-endpoint(290a17e、72b8b5d)。

代码自动合并无冲突;与今天 #302(PaymentPageObservedAt)、#303(原地重试)
同文件的改动经 purchase / purchasecontract / task 测试验证无语义冲突。

docs 三个镜像冲突取主线版本:主线镜像于 2026-09-11 同步 Wiki,已包含
#241/#242 说明及其后 #254、#271 内容;分支侧为 2026-09-08 旧快照。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NTDbDcwbDw1TSAcE6wfh2F
This commit is contained in:
QiuSW
2026-09-17 15:49:30 +08:00
co-authored by Claude Opus 5
13 changed files with 915 additions and 5 deletions
+9 -1
View File
@@ -206,7 +206,15 @@ func (task *PurchaseTask) BeforeCreate(_ *gorm.DB) error {
return task.syncPurchaseGuardSlots()
}
func (task *PurchaseTask) BeforeSave(_ *gorm.DB) error { return task.syncPurchaseGuardSlots() }
func (task *PurchaseTask) BeforeSave(tx *gorm.DB) error {
if err := task.syncPurchaseGuardSlots(); err != nil {
return err
}
if task.PDDOrderNo != nil && *task.PDDOrderNo != "" {
return CheckPurchaseOrderNumber(tx, task.ID, *task.PDDOrderNo)
}
return nil
}
func (task *PurchaseTask) SetStatus(status string) error {
task.Status = status
@@ -0,0 +1,41 @@
package models
import (
"errors"
"fmt"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var ErrPurchaseOrderNumberUsed = errors.New("purchase order number belongs to another task")
type PurchaseOrderNumberUsedError struct {
TaskID uint64
}
func (e *PurchaseOrderNumberUsedError) Error() string {
return fmt.Sprintf("订单号已属于任务 CG-%d", e.TaskID)
}
func (e *PurchaseOrderNumberUsedError) Unwrap() error { return ErrPurchaseOrderNumberUsed }
// CheckPurchaseOrderNumber must run inside the caller's write transaction.
// The existing singleton setting row serializes order assignments across
// processes, including an absent order number, without relying on gap locks or
// a new schema constraint. Locking reads see the latest committed assignment.
// A missing singleton fails closed. Deadlocks roll back the losing transaction.
func CheckPurchaseOrderNumber(tx *gorm.DB, taskID uint64, orderNo string) error {
var setting PurchaseRuleSetting
if err := tx.Session(&gorm.Session{NewDB: true}).Clauses(clause.Locking{Strength: "UPDATE"}).First(&setting, 1).Error; err != nil {
return err
}
var others []PurchaseTask
if err := tx.Session(&gorm.Session{NewDB: true}).Select("id").Clauses(clause.Locking{Strength: "UPDATE"}).Where("pdd_order_no = ? AND id <> ?", orderNo, taskID).Find(&others).Error; err != nil {
return err
}
if len(others) != 0 {
return &PurchaseOrderNumberUsedError{TaskID: others[0].ID}
}
return nil
}
+1 -1
View File
@@ -542,7 +542,7 @@ func writeError(c *gin.Context, err error) {
status = http.StatusForbidden
case CodeTaskNotFound:
status = http.StatusNotFound
case CodeStateConflict, CodeCapabilityMismatch, CodeDeviceBusy, CodeTaskClaimed, CodeLeaseExpired, CodeMappingRequired, CodeResultConflict, CodeRePurchaseRequired:
case CodeStateConflict, CodeCapabilityMismatch, CodeDeviceBusy, CodeTaskClaimed, CodeLeaseExpired, CodeMappingRequired, CodeResultConflict, CodeRePurchaseRequired, CodeOrderNumberUsed:
status = http.StatusConflict
}
c.JSON(status, gin.H{"code": code, "message": msg, "retryable": retryable})
+25 -1
View File
@@ -6,6 +6,7 @@ import (
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"unicode"
@@ -379,6 +380,22 @@ func (s *Service) SubmitResult(ctx context.Context, taskID uint64, req ResultReq
t.PDDOrderNo = &req.PDDOrderNo
t.OrderSubmittedAt = req.OrderSubmittedAt
t.ActualUnitPriceCent = req.ActualUnitPriceCent
// Keep the assignment lock until commit. A conflicting observation
// after the irreversible boundary must reach manual resolution,
// not roll back the result or claim another task's order number.
if e := models.CheckPurchaseOrderNumber(tx, t.ID, req.PDDOrderNo); e != nil {
var conflict *models.PurchaseOrderNumberUsedError
if !errors.As(e, &conflict) {
return TaskPayload{}, conflictOrInternal(e)
}
next = models.PurchaseTaskStatusOrderResultUnknown
t.PDDOrderNo = nil
failureCode := CodeOrderNumberUsed
message := fmt.Sprintf("读到订单号 %s,但该号已属于任务 %s", req.PDDOrderNo, taskNumber(conflict.TaskID))
t.ErrorCode, t.ErrorMessage = &failureCode, &message
a.Status = models.PurchaseAttemptStatusFailed
a.ErrorCode, a.ErrorMessage = &failureCode, &message
}
case "order_result_unknown":
if t.ExecutionMode != models.PurchaseExecutionModeLive || t.Status != models.PurchaseTaskStatusOrderSubmitStarted {
return TaskPayload{}, fail(CodeStateConflict, "当前任务不能标记订单结果未知")
@@ -600,7 +617,10 @@ func (s *Service) applySpecDecision(ctx context.Context, taskID uint64, req Spec
}
t.StatusVersion++
t.StatusChangedAt = s.Now()
return tx.Save(&t).Error
if e := tx.Save(&t).Error; e != nil {
return conflictOrInternal(e)
}
return nil
})
return t, replayed, err
}
@@ -803,6 +823,10 @@ func purchaseNotFound(err error) error {
return internal(err)
}
func conflictOrInternal(err error) error {
var conflict *models.PurchaseOrderNumberUsedError
if errors.As(err, &conflict) {
return fail(CodeOrderNumberUsed, conflict.Error())
}
if isDuplicate(err) {
return fail(CodeDeviceBusy, "设备或拼多多账号已有运行任务")
}
+8 -2
View File
@@ -80,7 +80,10 @@ func (s *Service) SelectWriteback(ctx context.Context, id uint64, req ManualRequ
}
out.WritebackStatus = models.PurchaseWritebackStatusPending
out.WritebackSelectRequestID = &req.RequestID
return tx.Save(&out).Error
if e := tx.Save(&out).Error; e != nil {
return conflictOrInternal(e)
}
return nil
})
return out, replayed, err
}
@@ -166,7 +169,10 @@ func (s *Service) manual(ctx context.Context, id uint64, req ManualRequest, appl
return internal(e)
}
}
return tx.Save(&out).Error
if e := tx.Save(&out).Error; e != nil {
return conflictOrInternal(e)
}
return nil
})
return out, replayed, err
}
@@ -0,0 +1,202 @@
package purchase
import (
"context"
"errors"
"strings"
"time"
"unicode/utf8"
"go-admin/app/goauto/device"
"go-admin/app/goauto/models"
"go-admin/app/goauto/purchasecontract"
"github.com/google/uuid"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"gorm.io/gorm/logger"
)
const (
MaxOrderBackfillItems = 50
CodeBackfillSuffix = "PURCHASE_BACKFILL_SUFFIX_INVALID"
CodeBackfillDevice = "PURCHASE_BACKFILL_DEVICE_MISMATCH"
CodeBackfillOrderConflict = "PURCHASE_BACKFILL_ORDER_CONFLICT"
CodeBackfillBatchConflict = "PURCHASE_BACKFILL_BATCH_CONFLICT"
CodeBackfillOrderUsed = "PURCHASE_BACKFILL_ORDER_ALREADY_USED"
)
type OrderBackfillRequest struct {
RequestID string `json:"requestId"`
Items []OrderBackfillItem `json:"items"`
}
type OrderBackfillItem struct {
AddressSuffix string `json:"addressSuffix"`
PDDOrderNo string `json:"pddOrderNo"`
// A string keeps an invalid page timestamp local to this item.
OrderSubmittedAt *string `json:"orderSubmittedAt,omitempty"`
}
type OrderBackfillResult struct {
Index int `json:"index"`
TaskID uint64 `json:"taskId,omitempty"`
Result string `json:"result"`
Code string `json:"code"`
Status string `json:"status,omitempty"`
StatusVersion uint64 `json:"statusVersion,omitempty"`
PDDOrderNo *string `json:"pddOrderNo,omitempty"`
OrderSubmittedAt *time.Time `json:"orderSubmittedAt,omitempty"`
TimeSource string `json:"timeSource,omitempty"`
Retryable bool `json:"retryable"`
}
type OrderBackfillResponse struct {
RequestID string `json:"requestId"`
Items []OrderBackfillResult `json:"items"`
}
func (s *Service) BackfillOrders(ctx context.Context, req OrderBackfillRequest, token string) (OrderBackfillResponse, error) {
out := OrderBackfillResponse{RequestID: req.RequestID}
d, err := device.NewService(s.DB).Authenticate(ctx, token)
if err != nil {
return out, err
}
if _, err := uuid.Parse(req.RequestID); err != nil || len(req.Items) == 0 || len(req.Items) > MaxOrderBackfillItems {
return out, fail(CodeInvalidRequest, "requestId 必须为 UUID,items 必须包含 1 到 50 条")
}
ids := make([]uint64, len(req.Items))
orders := make(map[uint64]string)
conflicts := make(map[uint64]bool)
for i, item := range req.Items {
id, err := purchasecontract.ParseAddressSuffix(item.AddressSuffix)
if err != nil {
continue
}
ids[i] = id
if previous, ok := orders[id]; ok && previous != item.PDDOrderNo {
conflicts[id] = true
}
orders[id] = item.PDDOrderNo
}
out.Items = make([]OrderBackfillResult, len(req.Items))
for i, item := range req.Items {
r := OrderBackfillResult{Index: i, TaskID: ids[i], Result: "failed"}
if ids[i] == 0 {
r.Code = CodeBackfillSuffix
} else {
r = s.backfillOrder(ctx, d.ID, ids[i], req.RequestID, item, conflicts[ids[i]])
r.Index = i
}
out.Items[i] = r
}
return out, nil
}
func (s *Service) backfillOrder(ctx context.Context, deviceID, taskID uint64, requestID string, item OrderBackfillItem, batchConflict bool) OrderBackfillResult {
r := OrderBackfillResult{TaskID: taskID, Result: "failed"}
var task models.PurchaseTask
// SQL errors must not print bound order numbers or the task's address snapshot.
db := s.DB.Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent)}).WithContext(ctx)
err := db.Transaction(func(tx *gorm.DB) error {
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&task, taskID).Error; err != nil {
return purchaseNotFound(err)
}
if task.DeviceID == nil || *task.DeviceID != deviceID {
return fail(CodeBackfillDevice, "任务不属于当前设备")
}
if batchConflict {
return fail(CodeBackfillBatchConflict, "同批任务有不同订单号")
}
if task.ExecutionMode != models.PurchaseExecutionModeLive || (task.Status != models.PurchaseTaskStatusOrderResultUnknown && task.Status != models.PurchaseTaskStatusOrderCreated) {
return fail(CodeStateConflict, "当前任务不允许回填")
}
if item.PDDOrderNo == "" || strings.TrimSpace(item.PDDOrderNo) != item.PDDOrderNo || utf8.RuneCountInString(item.PDDOrderNo) > 100 || strings.ContainsAny(item.PDDOrderNo, "\r\n\t") {
return fail(CodeInvalidRequest, "订单号无效")
}
if task.PDDOrderNo != nil && *task.PDDOrderNo != "" && *task.PDDOrderNo != item.PDDOrderNo {
return fail(CodeBackfillOrderConflict, "已有不同订单号")
}
// The shared model guard also protects manual resolution and late results.
if err := models.CheckPurchaseOrderNumber(tx, taskID, item.PDDOrderNo); err != nil {
return err
}
if task.Status == models.PurchaseTaskStatusOrderCreated {
if task.PDDOrderNo == nil || *task.PDDOrderNo != item.PDDOrderNo {
return fail(CodeStateConflict, "已创建订单缺少匹配订单号")
}
r.Result, r.Code = "already_backfilled", "ALREADY_BACKFILLED"
return nil
}
var submitted time.Time
source := "page"
if item.OrderSubmittedAt != nil {
var err error
submitted, err = time.Parse(time.RFC3339Nano, *item.OrderSubmittedAt)
if err != nil || submitted.IsZero() || submitted.Year() < 1000 || submitted.Year() > 9999 {
return fail(CodeOrderTimeInvalid, "下单时间必须为 RFC3339")
}
} else {
if task.IrreversibleAt == nil || task.IrreversibleAt.IsZero() {
return fail(CodeOrderTimeMissing, "下单时间和不可逆时间均缺失")
}
submitted, source = *task.IrreversibleAt, "irreversible_at"
}
submitted = submitted.UTC()
task.PDDOrderNo, task.OrderSubmittedAt = &item.PDDOrderNo, &submitted
if err := task.SetStatus(models.PurchaseTaskStatusOrderCreated); err != nil {
return internal(err)
}
task.StatusVersion++
task.StatusChangedAt = s.Now()
task.ErrorCode, task.ErrorMessage = nil, nil
task.LeaseExpiresAt = nil
// Reuse the existing resolution request slot. Scope a batch UUID to a
// task, and retain provenance without a schema change or replay cache.
marker := "backfill:" + source + ":" + uuid.NewSHA1(uuid.NameSpaceOID, []byte(requestID+":"+item.AddressSuffix)).String()
task.UnknownResolveRequestID = &marker
if err := tx.Save(&task).Error; err != nil {
return err
}
r.Result, r.Code = "backfilled", "BACKFILLED"
return nil
})
if err != nil {
r.Result, r.Code = "failed", CodeInternal
r.Retryable = true
var se *ServiceError
if errors.As(err, &se) {
r.Code, r.Retryable = se.Code, se.Retryable
}
if errors.Is(err, models.ErrPurchaseOrderNumberUsed) {
r.Code, r.Retryable = CodeBackfillOrderUsed, false
}
if r.Code == CodeBackfillBatchConflict || r.Code == CodeBackfillOrderConflict || r.Code == CodeBackfillOrderUsed {
r.Result = "conflict"
}
}
// Return only this device's committed facts, including on a rejected item.
// Never return in-memory changes from a rolled back transaction.
saved := task
readable := err == nil
if !readable {
saved = models.PurchaseTask{}
readable = db.Where("id = ? AND device_id = ?", taskID, deviceID).First(&saved).Error == nil
}
if readable {
r.Status, r.StatusVersion = saved.Status, saved.StatusVersion
r.PDDOrderNo, r.OrderSubmittedAt = saved.PDDOrderNo, saved.OrderSubmittedAt
if saved.OrderSubmittedAt != nil {
r.TimeSource = "existing_unknown"
if saved.UnknownResolveRequestID != nil {
if strings.HasPrefix(*saved.UnknownResolveRequestID, "backfill:page:") {
r.TimeSource = "page"
}
if strings.HasPrefix(*saved.UnknownResolveRequestID, "backfill:irreversible_at:") {
r.TimeSource = "irreversible_at"
}
}
}
}
return r
}
@@ -0,0 +1,25 @@
package purchase
import (
"net/http"
"github.com/gin-gonic/gin"
)
func (h Handler) BackfillOrders(c *gin.Context) {
var req OrderBackfillRequest
if !decode(c, &req) {
return
}
s, ok := h.service(c)
if !ok {
return
}
out, err := s.BackfillOrders(c.Request.Context(), req, bearer(c.GetHeader("Authorization")))
if err != nil {
writeError(c, err)
return
}
c.Header("Cache-Control", "no-store")
c.JSON(http.StatusOK, gin.H{"data": out})
}
@@ -0,0 +1,498 @@
package purchase
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"sync"
"testing"
"time"
"go-admin/app/goauto/device"
"go-admin/app/goauto/models"
"go-admin/app/goauto/purchasecontract"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"gorm.io/gorm"
)
func backfillTask(t *testing.T, db *gorm.DB, f fixture, status string) models.PurchaseTask {
t.Helper()
now := testService(db).Now()
task := models.PurchaseTask{TaskType: models.PurchaseTaskTypeStock, ExecutionMode: models.PurchaseExecutionModeLive,
Status: status, DeviceID: &f.device.ID, PDDProductID: f.pdd.ID, Quantity: 1, Currency: "CNY",
CreateRequestID: uuid.NewString(), RuleSnapshot: string(purchasecontract.DefaultLiveRule()),
SpecDecisionSnapshot: `{}`, RequiredCapabilitiesJSON: `[]`, IrreversibleAt: &now,
ErrorCode: strptr("ORIGINAL_ERROR"), ErrorMessage: strptr("original failure")}
if err := db.Create(&task).Error; err != nil {
t.Fatal(err)
}
return task
}
func strptr(s string) *string { return &s }
func backfillItem(id uint64, order string) OrderBackfillItem {
return OrderBackfillItem{AddressSuffix: purchasecontract.AddressSuffix(id), PDDOrderNo: order}
}
func runBackfill(t *testing.T, s *Service, token, requestID string, items ...OrderBackfillItem) []OrderBackfillResult {
t.Helper()
out, err := s.BackfillOrders(context.Background(), OrderBackfillRequest{RequestID: requestID, Items: items}, token)
if err != nil {
t.Fatal(err)
}
if len(out.Items) != len(items) || out.RequestID != requestID {
t.Fatalf("bad envelope: %+v", out)
}
return out.Items
}
func loadBackfillTask(t *testing.T, db *gorm.DB, id uint64) models.PurchaseTask {
t.Helper()
var task models.PurchaseTask
if err := db.First(&task, id).Error; err != nil {
t.Fatal(err)
}
return task
}
func TestOrderBackfillMixedBatchAndReplay(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
a := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
a.TaskType, a.SYBProductID = models.PurchaseTaskTypeSYBOrder, &f.syb.ID
if err := db.Save(&a).Error; err != nil {
t.Fatal(err)
}
b := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
c := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
if err := db.Model(&c).Update("irreversible_at", nil).Error; err != nil {
t.Fatal(err)
}
page := backfillItem(b.ID, "ORDER-B")
page.OrderSubmittedAt = strptr("2026-09-08T20:30:00+08:00")
rid := uuid.NewString()
items := []OrderBackfillItem{backfillItem(a.ID, "ORDER-A"), {AddressSuffix: "_cg0", PDDOrderNo: "bad"}, page, backfillItem(c.ID, "ORDER-C"), backfillItem(99999, "missing")}
results := runBackfill(t, s, f.token, rid, items...)
want := []string{"BACKFILLED", CodeBackfillSuffix, "BACKFILLED", CodeOrderTimeMissing, CodeTaskNotFound}
for i, r := range results {
if r.Code != want[i] || r.Index != i {
t.Fatalf("item %d: %+v", i, r)
}
}
if results[0].TimeSource != "irreversible_at" || !results[0].OrderSubmittedAt.Equal(*a.IrreversibleAt) {
t.Fatalf("fallback: %+v", results[0])
}
if results[2].TimeSource != "page" || results[2].OrderSubmittedAt.Format(time.RFC3339) != "2026-09-08T12:30:00Z" {
t.Fatalf("page: %+v", results[2])
}
saved := loadBackfillTask(t, db, a.ID)
if saved.StatusVersion != a.StatusVersion+1 || saved.ErrorCode != nil || saved.ErrorMessage != nil || saved.DeviceRunSlot != nil || saved.AccountRunSlot != nil || saved.ActiveSlot == nil || saved.Status != models.PurchaseTaskStatusOrderCreated {
t.Fatalf("state metadata: %+v", saved)
}
if saved.PaymentReviewStatus != a.PaymentReviewStatus || saved.LogisticsStatus != a.LogisticsStatus || saved.WritebackStatus != a.WritebackStatus || saved.RuleSnapshot != a.RuleSnapshot {
t.Fatal("unrelated business facts changed")
}
for _, replayID := range []string{rid, uuid.NewString()} {
item := items[0]
item.OrderSubmittedAt = strptr("2026-09-09T00:00:00Z")
r := runBackfill(t, s, f.token, replayID, item)[0]
if r.Result != "already_backfilled" || r.TimeSource != "irreversible_at" {
t.Fatalf("replay: %+v", r)
}
if got := loadBackfillTask(t, db, a.ID); !reflect.DeepEqual(saved, got) {
t.Fatal("replay changed persisted task")
}
}
if got := loadBackfillTask(t, db, c.ID); got.PDDOrderNo != nil || got.StatusVersion != c.StatusVersion {
t.Fatal("missing time wrote data")
}
}
func TestOrderBackfillRejectsOwnershipStatesAndInvalidTime(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
for _, status := range []string{models.PurchaseTaskStatusPending, models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted, models.PurchaseTaskStatusSpecProbePending, models.PurchaseTaskStatusFailed, models.PurchaseTaskStatusCancelled, models.PurchaseTaskStatusRehearsalCompleted} {
task := backfillTask(t, db, f, status)
before := loadBackfillTask(t, db, task.ID)
r := runBackfill(t, s, f.token, uuid.NewString(), backfillItem(task.ID, "ORDER"))[0]
if r.Code != CodeStateConflict {
t.Fatalf("%s: %+v", status, r)
}
if got := loadBackfillTask(t, db, task.ID); !reflect.DeepEqual(got, before) {
t.Fatal("rejection wrote data")
}
// Release the fixture's device slot before testing the next running state.
if err := task.SetStatus(models.PurchaseTaskStatusCancelled); err != nil {
t.Fatal(err)
}
if err := db.Save(&task).Error; err != nil {
t.Fatal(err)
}
}
task := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
if err := db.Model(&task).Update("device_id", nil).Error; err != nil {
t.Fatal(err)
}
r := runBackfill(t, s, f.token, uuid.NewString(), backfillItem(task.ID, "ORDER"))[0]
if r.Code != CodeBackfillDevice || r.Status != "" || r.PDDOrderNo != nil {
t.Fatalf("ownership leaked: %+v", r)
}
other, err := device.NewService(db).Register(context.Background(), device.RegisterRequest{RequestID: uuid.NewString(), InstallID: uuid.NewString(), Name: "Other", Manufacturer: "Test", Model: "Test", AndroidVersion: "15", AgentVersion: "1", PDDVersion: "7", Capabilities: liveCaps()}, "")
if err != nil {
t.Fatal(err)
}
if err := db.Model(&task).Update("device_id", other.DeviceID).Error; err != nil {
t.Fatal(err)
}
if r := runBackfill(t, s, f.token, uuid.NewString(), backfillItem(task.ID, "ORDER"))[0]; r.Code != CodeBackfillDevice {
t.Fatalf("cross device: %+v", r)
}
if err := db.Model(&task).Updates(map[string]any{"device_id": f.device.ID, "execution_mode": models.PurchaseExecutionModeRehearsal}).Error; err != nil {
t.Fatal(err)
}
if r := runBackfill(t, s, f.token, uuid.NewString(), backfillItem(task.ID, "ORDER"))[0]; r.Code != CodeStateConflict {
t.Fatalf("rehearsal: %+v", r)
}
if err := db.Model(&task).Update("execution_mode", models.PurchaseExecutionModeLive).Error; err != nil {
t.Fatal(err)
}
for _, raw := range []string{"", "2026-09-08 12:00:00", "0001-01-01T00:00:00Z", "garbage"} {
item := backfillItem(task.ID, "ORDER")
item.OrderSubmittedAt = &raw
if r := runBackfill(t, s, f.token, uuid.NewString(), item)[0]; r.Code != CodeOrderTimeInvalid {
t.Fatalf("invalid time: %+v", r)
}
}
}
func TestOrderBackfillConflictsNeverOverwrite(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
a := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
b := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
rid := uuid.NewString()
r := runBackfill(t, s, f.token, rid, backfillItem(a.ID, "A"), backfillItem(a.ID, "B"), backfillItem(b.ID, "B"))
if r[0].Code != CodeBackfillBatchConflict || r[1].Code != CodeBackfillBatchConflict || r[2].Code != "BACKFILLED" {
t.Fatalf("batch: %+v", r)
}
r = runBackfill(t, s, f.token, rid, backfillItem(a.ID, "B"), backfillItem(b.ID, "C"))
if r[0].Code != CodeBackfillOrderUsed || r[1].Code != CodeBackfillOrderConflict {
t.Fatalf("changed requestId payload bypassed checks: %+v", r)
}
if got := loadBackfillTask(t, db, b.ID); *got.PDDOrderNo != "B" || got.StatusVersion != b.StatusVersion+1 {
t.Fatal("conflict overwrote")
}
if got := loadBackfillTask(t, db, a.ID); got.PDDOrderNo != nil {
t.Fatal("conflict wrote data")
}
// Even an unknown task with an existing conflicting value must preserve it.
a.PDDOrderNo = strptr("OLD")
if err := db.Save(&a).Error; err != nil {
t.Fatal(err)
}
if r := runBackfill(t, s, f.token, uuid.NewString(), backfillItem(a.ID, "NEW"))[0]; r.Code != CodeBackfillOrderConflict {
t.Fatalf("unknown existing: %+v", r)
}
}
func TestOrderBackfillConcurrentResolveUnknown(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
// SQLite serializes transactions through one connection. These concurrent
// service calls verify both winner orders; they do not certify MySQL locks.
sqlDB, _ := db.DB()
sqlDB.SetMaxOpenConns(1)
for i := 0; i < 12; i++ {
task := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
start := make(chan struct{})
var wg sync.WaitGroup
wg.Add(2)
var out OrderBackfillResponse
var backErr, manualErr error
go func() {
defer wg.Done()
<-start
out, backErr = s.BackfillOrders(context.Background(), OrderBackfillRequest{RequestID: uuid.NewString(), Items: []OrderBackfillItem{backfillItem(task.ID, "BACK-"+purchasecontract.AddressSuffix(task.ID))}}, f.token)
}()
go func() {
defer wg.Done()
<-start
_, _, manualErr = s.ResolveUnknown(context.Background(), task.ID, ManualRequest{RequestID: uuid.NewString(), OperatorID: 1, Status: models.PurchaseTaskStatusOrderCreated, PDDOrderNo: "MANUAL-" + purchasecontract.AddressSuffix(task.ID), OrderSubmittedAt: task.IrreversibleAt})
}()
close(start)
wg.Wait()
if backErr != nil {
t.Fatal(backErr)
}
got := loadBackfillTask(t, db, task.ID)
if got.StatusVersion != task.StatusVersion+1 || got.Status != models.PurchaseTaskStatusOrderCreated {
t.Fatal("competing writes changed version twice")
}
if manualErr == nil {
if out.Items[0].Code != CodeBackfillOrderConflict || !strings.HasPrefix(*got.PDDOrderNo, "MANUAL-") {
t.Fatalf("manual winner: %+v", out)
}
} else if code(manualErr) != CodeStateConflict || out.Items[0].Code != "BACKFILLED" || !strings.HasPrefix(*got.PDDOrderNo, "BACK-") {
t.Fatalf("backfill winner: %+v %v", out, manualErr)
}
}
}
func TestOrderBackfillConcurrentLateResultAndOtherTask(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
sqlDB, _ := db.DB()
sqlDB.SetMaxOpenConns(1)
a := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
attempt := models.PurchaseTaskAttempt{TaskID: a.ID, AttemptID: uuid.NewString(), AttemptNumber: 1, Phase: models.PurchaseAttemptPhasePurchase, Status: models.PurchaseAttemptStatusFailed, DeviceID: &f.device.ID, RuleSnapshotHash: purchaseRuleSnapshotHash(a.RuleSnapshot), SpecDecisionSnapshot: `{}`}
if err := db.Omit("Task").Create(&attempt).Error; err != nil {
t.Fatal(err)
}
if err := db.First(&attempt, attempt.ID).Error; err != nil {
t.Fatal(err)
}
start := make(chan struct{})
var wg sync.WaitGroup
wg.Add(2)
var out OrderBackfillResponse
var backErr, lateErr error
go func() {
defer wg.Done()
<-start
out, backErr = s.BackfillOrders(context.Background(), OrderBackfillRequest{RequestID: uuid.NewString(), Items: []OrderBackfillItem{backfillItem(a.ID, "BACK")}}, f.token)
}()
go func() {
defer wg.Done()
<-start
_, lateErr = s.SubmitResult(context.Background(), a.ID, ResultRequest{RequestID: uuid.NewString(), TaskAttemptID: attempt.AttemptID, ResultType: "order_created", PDDOrderNo: "LATE", OrderSubmittedAt: a.IrreversibleAt}, f.token)
}()
close(start)
wg.Wait()
if backErr != nil || out.Items[0].Code != "BACKFILLED" || code(lateErr) != CodeStateConflict {
t.Fatalf("late race: %+v %v %v", out, backErr, lateErr)
}
var savedAttempt models.PurchaseTaskAttempt
if err := db.First(&savedAttempt, attempt.ID).Error; err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(savedAttempt, attempt) {
t.Fatal("backfill rewrote attempt")
}
b := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
_, _, err := s.ResolveUnknown(context.Background(), b.ID, ManualRequest{RequestID: uuid.NewString(), OperatorID: 1, Status: models.PurchaseTaskStatusOrderCreated, PDDOrderNo: "BACK", OrderSubmittedAt: b.IrreversibleAt})
if err == nil {
t.Fatal("manual path assigned another task's order")
}
if got := loadBackfillTask(t, db, b.ID); got.PDDOrderNo != nil || got.StatusVersion != b.StatusVersion {
t.Fatal("other task changed on conflict")
}
}
func TestOrderBackfillHTTPBoundary(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
task := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
gin.SetMode(gin.TestMode)
r := gin.New()
r.POST("/order-backfill", device.RequireAgentHTTPS(false, false), (Handler{DB: db}).BackfillOrders)
body, _ := json.Marshal(OrderBackfillRequest{RequestID: uuid.NewString(), Items: []OrderBackfillItem{backfillItem(task.ID, "HTTP")}})
for _, test := range []struct {
body, token string
status int
}{
{string(body), "", http.StatusUnauthorized},
{`{"requestId":"bad","items":[]}`, f.token, http.StatusUnprocessableEntity},
{`{"requestId":"x","address":"forbidden"}`, f.token, http.StatusUnprocessableEntity},
{string(body), f.token, http.StatusOK},
} {
req := httptest.NewRequest(http.MethodPost, "/order-backfill", strings.NewReader(test.body))
req.Header.Set("Authorization", "Bearer "+test.token)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != test.status {
t.Fatalf("HTTP %d: %s", w.Code, w.Body.String())
}
}
_, err := testService(db).BackfillOrders(context.Background(), OrderBackfillRequest{RequestID: uuid.NewString(), Items: make([]OrderBackfillItem, 51)}, f.token)
if code(err) != CodeInvalidRequest {
t.Fatalf("batch limit: %v", err)
}
}
func TestOrderBackfillConcurrentSameOrderDifferentTasks(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
sqlDB, _ := db.DB()
sqlDB.SetMaxOpenConns(1)
a := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
b := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
start := make(chan struct{})
results := make(chan OrderBackfillResponse, 2)
errors := make(chan error, 2)
for _, id := range []uint64{a.ID, b.ID} {
go func(id uint64) {
<-start
out, err := s.BackfillOrders(context.Background(), OrderBackfillRequest{RequestID: uuid.NewString(), Items: []OrderBackfillItem{backfillItem(id, "SAME")}}, f.token)
results <- out
errors <- err
}(id)
}
close(start)
codes := make(map[string]int)
for i := 0; i < 2; i++ {
out := <-results
if err := <-errors; err != nil {
t.Fatal(err)
}
codes[out.Items[0].Code]++
}
if codes["BACKFILLED"] != 1 || codes[CodeBackfillOrderUsed] != 1 {
t.Fatalf("concurrent assignments: %+v", codes)
}
var count int64
if err := db.Model(&models.PurchaseTask{}).Where("pdd_order_no = ?", "SAME").Count(&count).Error; err != nil || count != 1 {
t.Fatalf("duplicate order: %d %v", count, err)
}
}
func TestOrderBackfillRejectsLateAssignmentOfSameOrder(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
a := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
if r := runBackfill(t, s, f.token, uuid.NewString(), backfillItem(a.ID, "SHARED"))[0]; r.Code != "BACKFILLED" {
t.Fatal(r)
}
b := backfillTask(t, db, f, models.PurchaseTaskStatusOrderSubmitStarted)
lease := s.Now().Add(time.Minute)
b.LeaseExpiresAt = &lease
if err := db.Save(&b).Error; err != nil {
t.Fatal(err)
}
attempt := models.PurchaseTaskAttempt{TaskID: b.ID, AttemptID: uuid.NewString(), AttemptNumber: 1, Phase: models.PurchaseAttemptPhasePurchase, Status: models.PurchaseAttemptStatusRunning, DeviceID: &f.device.ID, RuleSnapshotHash: purchaseRuleSnapshotHash(b.RuleSnapshot), SpecDecisionSnapshot: `{}`}
if err := db.Omit("Task").Create(&attempt).Error; err != nil {
t.Fatal(err)
}
req := ResultRequest{RequestID: uuid.NewString(), TaskAttemptID: attempt.AttemptID, ResultType: "order_created", PDDOrderNo: "SHARED", OrderSubmittedAt: b.IrreversibleAt}
out, err := s.SubmitResult(context.Background(), b.ID, req, f.token)
if err != nil || out.Status != models.PurchaseTaskStatusOrderResultUnknown {
t.Fatalf("conflicting result must commit as unknown: %+v %v", out, err)
}
got := loadBackfillTask(t, db, b.ID)
if got.Status != models.PurchaseTaskStatusOrderResultUnknown || got.StatusVersion != b.StatusVersion+1 || got.PDDOrderNo != nil {
t.Fatalf("duplicate assignment was not safely downgraded: %+v", got)
}
wantMessage := "读到订单号 SHARED,但该号已属于任务 " + taskNumber(a.ID)
if got.ErrorCode == nil || *got.ErrorCode != CodeOrderNumberUsed || got.ErrorMessage == nil || *got.ErrorMessage != wantMessage {
t.Fatalf("conflict evidence missing: %+v", got)
}
if got.OrderSubmittedAt == nil || !got.OrderSubmittedAt.Equal(*req.OrderSubmittedAt) || got.IrreversibleAt == nil || got.LeaseExpiresAt != nil || got.DeviceRunSlot != nil || got.AccountRunSlot != nil {
t.Fatalf("boundary evidence or released lease missing: %+v", got)
}
var saved models.PurchaseTaskAttempt
if err := db.First(&saved, attempt.ID).Error; err != nil {
t.Fatal(err)
}
if saved.Status != models.PurchaseAttemptStatusFailed || saved.ResultRequestID == nil || *saved.ResultRequestID != req.RequestID || saved.ResultHash == nil || saved.ResultType == nil || *saved.ResultType != "order_created" || saved.FinishedAt == nil || saved.ErrorCode == nil || *saved.ErrorCode != CodeOrderNumberUsed || saved.ErrorMessage == nil || *saved.ErrorMessage != wantMessage {
t.Fatalf("attempt result and conflict evidence missing: %+v", saved)
}
out, err = s.SubmitResult(context.Background(), b.ID, req, f.token)
if err != nil || !out.Replayed || out.Status != models.PurchaseTaskStatusOrderResultUnknown || loadBackfillTask(t, db, b.ID).StatusVersion != got.StatusVersion {
t.Fatalf("unknown result replay failed: %+v %v", out, err)
}
owner := loadBackfillTask(t, db, a.ID)
if owner.PDDOrderNo == nil || *owner.PDDOrderNo != "SHARED" || owner.Status != models.PurchaseTaskStatusOrderCreated {
t.Fatalf("existing owner changed: %+v", owner)
}
var count int64
if err := db.Model(&models.PurchaseTask{}).Where("pdd_order_no = ?", "SHARED").Count(&count).Error; err != nil || count != 1 {
t.Fatalf("duplicate order: %d %v", count, err)
}
resolved, _, err := s.ResolveUnknown(context.Background(), b.ID, ManualRequest{RequestID: uuid.NewString(), OperatorID: 1, Status: models.PurchaseTaskStatusOrderCreated, PDDOrderNo: "CORRECTED", OrderSubmittedAt: req.OrderSubmittedAt})
if err != nil || resolved.Status != models.PurchaseTaskStatusOrderCreated {
t.Fatalf("manual resolution unavailable: %+v %v", resolved, err)
}
}
func TestOrderBackfillHTTPTransportPolicy(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, allow := range []string{"false", "true"} {
t.Setenv("GOAUTO_ALLOW_INSECURE_AGENT_HTTP", allow)
r := gin.New()
r.POST("/order-backfill", device.RequireAgentHTTPS(true, false), (Handler{}).BackfillOrders)
w := httptest.NewRecorder()
r.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/order-backfill", strings.NewReader(`{}`)))
if allow == "false" && w.Code != http.StatusUpgradeRequired {
t.Fatalf("HTTPS bypass: %d", w.Code)
}
if allow == "true" && w.Code == http.StatusUpgradeRequired {
t.Fatal("HTTP compatibility broken")
}
}
}
func TestOrderBackfillMultiConnectionResolveRace(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
sqlDB.SetMaxOpenConns(4)
task := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
start := make(chan struct{})
var wg sync.WaitGroup
wg.Add(2)
var back OrderBackfillResponse
var backErr, manualErr error
req := OrderBackfillRequest{RequestID: uuid.NewString(), Items: []OrderBackfillItem{backfillItem(task.ID, "BACK")}}
go func() { defer wg.Done(); <-start; back, backErr = s.BackfillOrders(context.Background(), req, f.token) }()
go func() {
defer wg.Done()
<-start
_, _, manualErr = s.ResolveUnknown(context.Background(), task.ID, ManualRequest{RequestID: uuid.NewString(), OperatorID: 1, Status: models.PurchaseTaskStatusOrderCreated, PDDOrderNo: "MANUAL", OrderSubmittedAt: task.IrreversibleAt})
}()
close(start)
wg.Wait()
// SQLite returns table-lock errors rather than waiting on FOR UPDATE.
// Only that documented DB contention or a domain conflict is acceptable;
// after the competing calls finish, replay must converge without overwrite.
if backErr != nil && !strings.Contains(backErr.Error(), "locked") {
t.Fatal(backErr)
}
if manualErr != nil && code(manualErr) != CodeStateConflict && !strings.Contains(manualErr.Error(), "locked") {
t.Fatal(manualErr)
}
if backErr == nil && back.Items[0].Code != "BACKFILLED" && back.Items[0].Code != CodeBackfillOrderConflict && !(back.Items[0].Code == CodeInternal && back.Items[0].Retryable) {
t.Fatalf("unexpected race result: %+v", back)
}
before := loadBackfillTask(t, db, task.ID)
replay := runBackfill(t, s, f.token, req.RequestID, req.Items...)[0]
after := loadBackfillTask(t, db, task.ID)
if before.PDDOrderNo != nil && !reflect.DeepEqual(before, after) {
t.Fatal("replay overwrote the concurrent winner")
}
if after.StatusVersion != task.StatusVersion+1 || after.Status != models.PurchaseTaskStatusOrderCreated {
t.Fatal("race did not converge to a single transition")
}
if manualErr == nil {
if *after.PDDOrderNo != "MANUAL" || replay.Code != CodeBackfillOrderConflict {
t.Fatal("manual winner overwritten")
}
} else if *after.PDDOrderNo != "BACK" || (replay.Code != "BACKFILLED" && replay.Code != "ALREADY_BACKFILLED") {
t.Fatalf("backfill did not converge: %+v", replay)
}
}
@@ -0,0 +1,74 @@
package purchase
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"go-admin/app/goauto/models"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"gorm.io/gorm"
)
func TestOrderNumberConflictBusinessErrors(t *testing.T) {
for _, path := range []string{"resolve_unknown", "cancel", "lifecycle"} {
t.Run(path, func(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
s := testService(db)
owner := backfillTask(t, db, f, models.PurchaseTaskStatusOrderResultUnknown)
if r := runBackfill(t, s, f.token, uuid.NewString(), backfillItem(owner.ID, "SHARED"))[0]; r.Code != "BACKFILLED" {
t.Fatal(r)
}
status := models.PurchaseTaskStatusOrderResultUnknown
if path == "lifecycle" {
status = models.PurchaseTaskStatusRunning
}
task := backfillTask(t, db, f, status)
if path != "resolve_unknown" {
// Model legacy duplicate data predating the global save guard.
if err := db.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).Where("id = ?", task.ID).Updates(map[string]any{"pdd_order_no": "SHARED", "lease_expires_at": s.Now().Add(time.Minute)}).Error; err != nil {
t.Fatal(err)
}
}
req := ManualRequest{RequestID: uuid.NewString(), OperatorID: 1, Status: models.PurchaseTaskStatusOrderCreated, PDDOrderNo: "SHARED", OrderSubmittedAt: task.IrreversibleAt, Reason: "人工取消"}
var err error
switch path {
case "resolve_unknown":
_, _, err = s.ResolveUnknown(context.Background(), task.ID, req)
case "cancel":
_, _, err = s.Cancel(context.Background(), task.ID, req)
case "lifecycle":
attempt := models.PurchaseTaskAttempt{TaskID: task.ID, AttemptID: uuid.NewString(), AttemptNumber: 1, Phase: models.PurchaseAttemptPhasePurchase, Status: models.PurchaseAttemptStatusRunning, DeviceID: &f.device.ID, RuleSnapshotHash: purchaseRuleSnapshotHash(task.RuleSnapshot), SpecDecisionSnapshot: `{}`}
if e := db.Omit("Task").Create(&attempt).Error; e != nil {
t.Fatal(e)
}
_, err = s.MarkOrderSubmitStarted(context.Background(), task.ID, ActionRequest{RequestID: req.RequestID}, f.token)
}
want := "订单号已属于任务 " + taskNumber(owner.ID)
if code(err) != CodeOrderNumberUsed || err.Error() != want {
t.Fatalf("unmapped conflict: %v", err)
}
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
writeError(c, err)
var body struct {
Code string `json:"code"`
Message string `json:"message"`
Retryable bool `json:"retryable"`
}
if e := json.Unmarshal(w.Body.Bytes(), &body); e != nil || w.Code != http.StatusConflict || body.Code != CodeOrderNumberUsed || body.Message != want || body.Retryable {
t.Fatalf("unexpected HTTP error: %d %s (%v)", w.Code, w.Body.String(), e)
}
got := loadBackfillTask(t, db, task.ID)
if got.Status != status || got.StatusVersion != task.StatusVersion || got.UnknownResolveRequestID != nil || got.CancelRequestID != nil || got.OrderSubmitRequestID != nil {
t.Fatalf("rejected mutation persisted: %+v", got)
}
})
}
}
+1
View File
@@ -18,6 +18,7 @@ func InitRouter(engine *gin.Engine, auth *jwt.GinJWTMiddleware) {
agent := engine.Group("/api/agent/v1/purchase-tasks").Use(device.RequireAgentHTTPS(config.ApplicationConfig.Mode == "prod", trust))
agent.GET("", h.AgentHistory)
agent.GET("/next", h.Next)
agent.POST("/order-backfill", h.BackfillOrders)
agent.GET("/:taskId", h.AgentHistoryDetail)
agent.POST("/:taskId/retry", h.AgentRetry)
agent.POST("/:taskId/reset", h.AgentReset)
+1
View File
@@ -22,6 +22,7 @@ const (
CodeRetryStale = "PURCHASE_RETRY_STALE"
CodeSpecReprobeRejected = "PURCHASE_SPEC_REPROBE_REJECTED"
CodeOrderResultUnknown = "PURCHASE_ORDER_RESULT_UNKNOWN"
CodeOrderNumberUsed = "PURCHASE_ORDER_NUMBER_ALREADY_USED"
CodeOrderEmptyTimeout = "PURCHASE_ORDER_EMPTY_TIMEOUT"
CodeOrderChooserBack = "PURCHASE_ORDER_CHOOSER_BACK_FAILED"
CodeOrderWechatRestore = "PURCHASE_ORDER_WECHAT_RESTORE_FAILED"
@@ -0,0 +1,17 @@
package purchasecontract
import "testing"
func TestParseAddressSuffix(t *testing.T) {
for _, id := range []uint64{7, 72, ^uint64(0)} {
got, err := ParseAddressSuffix(AddressSuffix(id))
if err != nil || got != id {
t.Fatalf("id=%d got=%d err=%v", id, got, err)
}
}
for _, raw := range []string{"", "_cg", "_cg0", "_cg00", "_cg07", "_cg+7", "_cg-7", "_cg18446744073709551616", "_CG7", "_cg7x", "_cg7_cg72", "address_cg7", " _cg7", "_cg7 ", "_cg7", "_cg7\n"} {
if id, err := ParseAddressSuffix(raw); err == nil || id != 0 {
t.Errorf("accepted %q: %d", raw, id)
}
}
}
@@ -9,6 +9,7 @@ import (
"math"
"regexp"
"sort"
"strconv"
"strings"
"unicode/utf8"
)
@@ -360,6 +361,18 @@ func RequiredCapabilities(rule RuleSnapshot) []string {
func AddressSuffix(taskID uint64) string { return fmt.Sprintf("_cg%d", taskID) }
// ParseAddressSuffix accepts only the exact canonical suffix, never an address.
func ParseAddressSuffix(suffix string) (uint64, error) {
if !strings.HasPrefix(suffix, "_cg") {
return 0, errors.New("invalid address suffix")
}
id, err := strconv.ParseUint(strings.TrimPrefix(suffix, "_cg"), 10, 64)
if err != nil || id == 0 || AddressSuffix(id) != suffix {
return 0, errors.New("invalid address suffix")
}
return id, nil
}
func ensureEOF(decoder *json.Decoder) error {
var extra any
if err := decoder.Decode(&extra); err != io.EOF {