OPC/goauto
2
0
Fork 0
Code Issues 192 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(server): converge purchase retry startup (#159)

Browse Source
This commit is contained in:
QiuSW
2026-08-29 17:30:45 +08:00
parent 77a2b2506e
commit 8afbdbc8ff
3 changed files with 81 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/goauto/purchase/lifecycle.go
+35 -4
View File
@@ -137,6 +137,7 @@ func (s *Service) Start(ctx context.Context, taskID uint64, req ActionRequest, t
return TaskPayload{}, fail(CodeInvalidRequest, "requestId 无效")
}
var out TaskPayload
var committedFailure error
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
d, err := device.NewService(tx).Authenticate(ctx, token)
if err != nil {
@@ -188,12 +189,34 @@ func (s *Service) Start(ctx context.Context, taskID uint64, req ActionRequest, t
if t.SpecSource == "unresolved" {
phase = models.PurchaseAttemptPhaseSpecProbe
}
h := sha256.Sum256([]byte(t.RuleSnapshot))
ruleSnapshotHash := purchaseRuleSnapshotHash(t.RuleSnapshot)
now := s.Now()
var a models.PurchaseTaskAttempt
if e = tx.Where("task_id = ? AND status = ?", t.ID, models.PurchaseAttemptStatusPending).Order("attempt_number DESC, id DESC").First(&a).Error; e == nil {
if a.DeviceID == nil || *a.DeviceID != d.ID || a.RuleSnapshotHash != hex.EncodeToString(h[:]) {
return fail(CodeStateConflict, "待执行 attempt 与当前任务不一致")
if a.DeviceID == nil || *a.DeviceID != d.ID || a.RuleSnapshotHash != ruleSnapshotHash {
failureCode := CodeStateConflict
message := "采购规则快照校验失败，请重新重试任务"
a.Status = models.PurchaseAttemptStatusFailed
a.ErrorCode = &failureCode
a.ErrorMessage = &message
a.FinishedAt = &now
if e = tx.Omit("Task").Save(&a).Error; e != nil {
return internal(e)
}
if e = t.SetStatus(models.PurchaseTaskStatusFailed); e != nil {
return internal(e)
}
t.ErrorCode = &failureCode
t.ErrorMessage = &message
t.LeaseExpiresAt = nil
t.ClaimRequestID = nil
t.StatusVersion++
t.StatusChangedAt = now
if e = tx.Save(&t).Error; e != nil {
return conflictOrInternal(e)
}
committedFailure = fail(CodeStateConflict, message)
return nil
}
a.Status = models.PurchaseAttemptStatusRunning
a.StartRequestID = &req.RequestID
@@ -206,7 +229,7 @@ func (s *Service) Start(ctx context.Context, taskID uint64, req ActionRequest, t
if e = tx.Model(&models.PurchaseTaskAttempt{}).Where("task_id = ?", t.ID).Count(&count).Error; e != nil {
return internal(e)
}
a = models.PurchaseTaskAttempt{TaskID: t.ID, AttemptID: uuid.NewString(), AttemptNumber: int(count) + 1, Phase: phase, Status: models.PurchaseAttemptStatusRunning, DeviceID: &d.ID, RuleSnapshotHash: hex.EncodeToString(h[:]), SpecDecisionSnapshot: t.SpecDecisionSnapshot, StartRequestID: &req.RequestID, StartedAt: &now}
a = models.PurchaseTaskAttempt{TaskID: t.ID, AttemptID: uuid.NewString(), AttemptNumber: int(count) + 1, Phase: phase, Status: models.PurchaseAttemptStatusRunning, DeviceID: &d.ID, RuleSnapshotHash: ruleSnapshotHash, SpecDecisionSnapshot: t.SpecDecisionSnapshot, StartRequestID: &req.RequestID, StartedAt: &now}
if e = tx.Omit("Task").Create(&a).Error; e != nil {
return internal(e)
}
@@ -226,9 +249,17 @@ func (s *Service) Start(ctx context.Context, taskID uint64, req ActionRequest, t
out = *p
return e
})
if err == nil && committedFailure != nil {
return out, committedFailure
}
return out, err
}
func purchaseRuleSnapshotHash(snapshot string) string {
digest := sha256.Sum256([]byte(snapshot))
return hex.EncodeToString(digest[:])
}
func (s *Service) MarkOrderSubmitStarted(ctx context.Context, taskID uint64, req ActionRequest, token string) (TaskPayload, error) {
return s.withRunning(ctx, taskID, token, func(tx *gorm.DB, t *models.PurchaseTask, a *models.PurchaseTaskAttempt, d models.AgentDevice) (TaskPayload, error) {
if t.ExecutionMode != models.PurchaseExecutionModeLive {
server/app/goauto/purchase/reset.go
+7 -4
View File
@@ -2,8 +2,6 @@ package purchase
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -136,11 +134,16 @@ func (s *Service) reset(ctx context.Context, taskID uint64, req PurchaseResetReq
if task.SpecSource == "unresolved" {
phase = models.PurchaseAttemptPhaseSpecProbe
}
digest := sha256.Sum256(rawRule)
// MySQL normalizes values written to a JSON column. Reload the task before
// hashing so the pending attempt uses the exact representation Start will
// read later, rather than the pre-persistence DefaultLiveRule bytes.
if err := tx.First(&task, task.ID).Error; err != nil {
return internal(err)
}
attempt := models.PurchaseTaskAttempt{
TaskID: task.ID, AttemptID: attemptID, AttemptNumber: int(attemptCount) + 1,
Phase: phase, Status: models.PurchaseAttemptStatusPending, DeviceID: &deviceRecord.ID,
RuleSnapshotHash: hex.EncodeToString(digest[:]), SpecDecisionSnapshot: task.SpecDecisionSnapshot,
RuleSnapshotHash: purchaseRuleSnapshotHash(task.RuleSnapshot), SpecDecisionSnapshot: task.SpecDecisionSnapshot,
}
if err := tx.Omit("Task").Create(&attempt).Error; err != nil {
return conflictOrInternal(err)
server/app/goauto/purchase/reset_test.go
+39 -4
View File
@@ -2,8 +2,6 @@ package purchase
import (
"context"
"crypto/sha256"
"encoding/hex"
"testing"
"go-admin/app/goauto/device"
@@ -76,8 +74,7 @@ func TestPurchaseResetReusesTaskRefreshesRuleAndIsIdempotent(t *testing.T) {
if err := db.Where("task_id = ?", failed.ID).Find(&attempts).Error; err != nil || len(attempts) != 1 || attempts[0].Status != models.PurchaseAttemptStatusPending {
t.Fatalf("pending reset attempt mismatch: attempts=%+v err=%v", attempts, err)
}
digest := sha256.Sum256(purchasecontract.DefaultLiveRule())
if attempts[0].RuleSnapshotHash != hex.EncodeToString(digest[:]) {
if attempts[0].RuleSnapshotHash != purchaseRuleSnapshotHash(reset.RuleSnapshot) {
t.Fatalf("attempt rule hash mismatch: %s", attempts[0].RuleSnapshotHash)
}
var taskCount int64
@@ -94,6 +91,44 @@ func TestPurchaseResetReusesTaskRefreshesRuleAndIsIdempotent(t *testing.T) {
}
}
func TestPurchaseStartPersistsPendingAttemptSnapshotMismatchAsFailure(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)
service := testService(db)
failed := failedLiveTask(t, db, service, f)
if _, err := service.ResetForDevice(context.Background(), failed.ID, PurchaseResetRequest{RequestID: uuid.NewString()}, f.token); err != nil {
t.Fatal(err)
}
if err := db.Model(&models.PurchaseTaskAttempt{}).Where("task_id = ?", failed.ID).Update("rule_snapshot_hash", "invalid").Error; err != nil {
t.Fatal(err)
}
if _, err := service.Claim(context.Background(), failed.ID, ActionRequest{RequestID: uuid.NewString()}, f.token); err != nil {
t.Fatal(err)
}
if _, err := service.Start(context.Background(), failed.ID, ActionRequest{RequestID: uuid.NewString()}, f.token); code(err) != CodeStateConflict {
t.Fatalf("snapshot mismatch error = %v", err)
}
var task models.PurchaseTask
if err := db.First(&task, failed.ID).Error; err != nil {
t.Fatal(err)
}
if task.Status != models.PurchaseTaskStatusFailed || task.ErrorCode == nil || *task.ErrorCode != CodeStateConflict ||
task.LeaseExpiresAt != nil || task.ClaimRequestID != nil || task.DeviceRunSlot != nil || task.AccountRunSlot != nil {
t.Fatalf("snapshot mismatch task did not converge: %+v", task)
}
var attempt models.PurchaseTaskAttempt
if err := db.Where("task_id = ?", failed.ID).First(&attempt).Error; err != nil {
t.Fatal(err)
}
if attempt.Status != models.PurchaseAttemptStatusFailed || attempt.ErrorCode == nil || *attempt.ErrorCode != CodeStateConflict || attempt.FinishedAt == nil {
t.Fatalf("snapshot mismatch attempt did not converge: %+v", attempt)
}
if next, err := service.Next(context.Background(), f.token); err != nil || next != nil {
t.Fatalf("failed mismatch task remained dispatchable: next=%+v err=%v", next, err)
}
}
func TestPurchaseResetPendingAttemptIsUsedByStartAndCanRetryAgain(t *testing.T) {
db := testDB(t)
f := seed(t, db, liveCaps(), true)