diff --git a/server/app/goauto/purchase/lifecycle.go b/server/app/goauto/purchase/lifecycle.go index f1274c2..8a9e54a 100644 --- a/server/app/goauto/purchase/lifecycle.go +++ b/server/app/goauto/purchase/lifecycle.go @@ -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 { diff --git a/server/app/goauto/purchase/reset.go b/server/app/goauto/purchase/reset.go index 0eb1714..9a2916d 100644 --- a/server/app/goauto/purchase/reset.go +++ b/server/app/goauto/purchase/reset.go @@ -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) diff --git a/server/app/goauto/purchase/reset_test.go b/server/app/goauto/purchase/reset_test.go index 1ffddfc..02f4cdd 100644 --- a/server/app/goauto/purchase/reset_test.go +++ b/server/app/goauto/purchase/reset_test.go @@ -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)