From 131163c8c446b02c412200ffa8788da1ea69679c Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Wed, 23 Sep 2026 10:48:38 +0800 Subject: [PATCH] fix(yeeke): reclaim stale sync lease and stabilize item keys (#336) Two reviewer-identified defects in the yeeke return sync: - acquire() wrote LeaseExpiresAt but nothing ever read it back, so a crash/restart mid-run left a permanent active_slot=1 row blocking every future sync. acquire() now runs a conditional takeover UPDATE first (status=running AND lease_expires_at <= now -> failed, active_slot cleared, error_message recorded), following the lease-with-expiry- takeover idiom in order_writeback_worker.go. The takeover UPDATE is a single statement so it is atomic per-row, and the ux_yeeke_sync_run_active_slot unique index arbitrates a concurrent takeover race the same way it already arbitrates two brand-new runs. - itemKey() always appended the positional index, so a package whose items come back in a different order on a later sync got new keys and duplicate rows. The index fallback is now used only when i.ID, i.ItemID and i.VariationID are all empty. Added tests: TestStaleLeaseIsTakenOverOnNextAcquire, TestValidLeaseIsNotTakenOver, TestConcurrentTakeoverExactlyOneWins, TestItemKeyStableAcrossReorder, TestItemKeyIndexFallbackForItemsLackingAllIDs, TestItemKeyDistinctVariationsOfSameItemID. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NTDbDcwbDw1TSAcE6wfh2F --- server/app/goauto/yeeke/sync.go | 51 +++- server/app/goauto/yeeke/sync_paging_test.go | 268 ++++++++++++++++++++ 2 files changed, 318 insertions(+), 1 deletion(-) diff --git a/server/app/goauto/yeeke/sync.go b/server/app/goauto/yeeke/sync.go index b8837c2..d69b5e4 100644 --- a/server/app/goauto/yeeke/sync.go +++ b/server/app/goauto/yeeke/sync.go @@ -61,10 +61,59 @@ func packageKey(p yeekeclient.ReturnPackage) string { } return p.Ordersn + "/" + p.TrackingNo + "/" + external(p.ShopID) + "/" + p.CreateTime.String() } + +// itemKey builds the stable per-item identity used to upsert +// models.YeekeReturnItem without duplicating rows across syncs. It prefers +// the yeeke-issued identifiers (i.ID, then i.ItemID/i.VariationID) because +// those stay the same regardless of the order the API returns items in +// within a package; the positional index n is only used as a last resort +// when none of those identifiers are present, since in that case the index +// is the sole thing distinguishing items of the same package (#336). func itemKey(p yeekeclient.ReturnPackage, i yeekeclient.ReturnItem, n int) string { - return packageKey(p) + "/" + external(i.ID) + "/" + external(i.ItemID) + "/" + external(i.VariationID) + "/" + strconv.Itoa(n) + id, itemID, variationID := external(i.ID), external(i.ItemID), external(i.VariationID) + if isEmptyExternal(id) && isEmptyExternal(itemID) && isEmptyExternal(variationID) { + return packageKey(p) + "/" + id + "/" + itemID + "/" + variationID + "/" + strconv.Itoa(n) + } + return packageKey(p) + "/" + id + "/" + itemID + "/" + variationID } + +// isEmptyExternal reports whether external() produced a value that carries +// no real identity: either the field was unset (formatted as "" by +// fmt.Sprint on a nil/zero value) or it was an explicit empty string. +func isEmptyExternal(v string) bool { return v == "" || v == "" } + +// takeoverStaleLease reclaims a run whose lease has expired, e.g. because the +// process crashed or was restarted mid-sync. It matches the +// lease-with-expiry-takeover idiom used by +// app/goauto/purchase/order_writeback_worker.go: a single conditional UPDATE +// guarded by "status = running AND lease_expires_at <= now" flips the stale +// row to a terminal status and frees active_slot in one statement, so it is +// atomic without a separate row lock. The stale row is never deleted — it is +// left in place with status "failed" and an error_message explaining why, so +// history stays auditable. If two callers race this same UPDATE, only the +// first to reach the database actually changes any row; the second's WHERE +// clause no longer matches (status is no longer "running") and it affects +// zero rows, which is a harmless no-op. Whichever caller then wins the +// subsequent Create (see acquire) is arbitrated by the ux_yeeke_sync_run_active_slot +// unique index, exactly as it already is for two brand-new concurrent runs. +func (s *Service) takeoverStaleLease(ctx context.Context) error { + now := time.Now().UTC() + return s.db.WithContext(ctx).Model(&models.YeekeSyncRun{}). + Where("status = ? AND active_slot = ? AND lease_expires_at IS NOT NULL AND lease_expires_at <= ?", "running", 1, now). + Updates(map[string]any{ + "status": "failed", + "active_slot": nil, + "lease_owner": "", + "error_message": "lease expired: run interrupted, likely a process restart mid-sync (stale lease takeover)", + "lease_expires_at": nil, + "finished_at": now, + }).Error +} + func (s *Service) acquire(ctx context.Context, trigger string) (*models.YeekeSyncRun, error) { + if e := s.takeoverStaleLease(ctx); e != nil { + return nil, e + } now := time.Now().UTC() owner := fmt.Sprintf("%d", now.UnixNano()) slot := uint8(1) diff --git a/server/app/goauto/yeeke/sync_paging_test.go b/server/app/goauto/yeeke/sync_paging_test.go index 6d5f295..8b6bac1 100644 --- a/server/app/goauto/yeeke/sync_paging_test.go +++ b/server/app/goauto/yeeke/sync_paging_test.go @@ -460,3 +460,271 @@ func setTestYeekeBaseURL(t *testing.T, url string) func() { config.ExtConfig.Yeeke.BaseURL = url return func() { config.ExtConfig.Yeeke = before } } + +// packageWithItems builds a raw list-page record for one package carrying an +// arbitrary, caller-ordered set of items, so tests can reorder items between +// two syncs of the same package. +func packageWithItems(pkgID string, items ...map[string]any) string { + b, _ := json.Marshal(map[string]any{ + "id": pkgID, "ordersn": "o-" + pkgID, "trackingNo": "t-" + pkgID, "status": 1, + "items": items, + }) + return string(b) +} + +func item(id, itemID, variationID string) map[string]any { + return map[string]any{"id": id, "itemId": itemID, "variationId": variationID, "itemName": "n", "variationName": "v", "variationQuantityPurchased": 1} +} + +// itemNoIDs builds an item carrying no yeeke-issued identifiers at all +// (id/itemId/variationId all empty), the case itemKey's positional-index +// fallback exists for. +func itemNoIDs() map[string]any { + return map[string]any{"id": "", "itemId": "", "variationId": "", "itemName": "n", "variationName": "v", "variationQuantityPurchased": 1} +} + +// --- Defect 1 (#336): stale lease takeover ------------------------------- + +// TestStaleLeaseIsTakenOverOnNextAcquire simulates a crash: a "running" row +// is left behind with an active_slot and a lease that has already expired +// (as if the process died mid-sync, long before the lease's normal +// duration). The very next Sync call — scheduled or manual — must reclaim +// the slot rather than being permanently blocked, and the abandoned row must +// end up in a clear terminal state (not silently deleted) recording why. +func TestStaleLeaseIsTakenOverOnNextAcquire(t *testing.T) { + db := testDB(t) + slot := uint8(1) + past := time.Now().UTC().Add(-time.Hour) + stale := models.YeekeSyncRun{ + Status: "running", Trigger: "scheduled", StartedAt: past.Add(-time.Minute), + ActiveSlot: &slot, LeaseOwner: "dead-process", LeaseExpiresAt: &past, + } + if e := db.Create(&stale).Error; e != nil { + t.Fatal(e) + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, page(nil, 0, 0)) + })) + defer srv.Close() + c, _ := yeekeclient.New(srv.URL) + s := NewService(db, c, Config{PageSize: 10}) + + rep, err := s.Sync(context.Background(), "manual") + if err != nil { + t.Fatalf("resume after a stale lease must succeed, got err=%v", err) + } + if rep.Status != "succeeded" { + t.Fatalf("rep=%+v", rep) + } + + var reclaimed models.YeekeSyncRun + if e := db.First(&reclaimed, stale.ID).Error; e != nil { + t.Fatal(e) + } + if reclaimed.Status != "failed" { + t.Fatalf("stale run status=%q, want a terminal status (not silently left running or deleted)", reclaimed.Status) + } + if reclaimed.ErrorMessage == "" { + t.Fatal("stale run must record why it was taken over") + } + if reclaimed.ActiveSlot != nil { + t.Fatal("stale run must release active_slot on takeover") + } + + // The new run's own row clears active_slot on completion just like any + // other successful run (see run()'s defer), so what proves the takeover + // happened is that a second, distinct run row now exists alongside the + // reclaimed stale one. + var totalRuns int64 + db.Model(&models.YeekeSyncRun{}).Count(&totalRuns) + if totalRuns != 2 { + t.Fatalf("expected the stale row plus exactly one new run after takeover, got %d run rows", totalRuns) + } + if rep.RunID == stale.ID { + t.Fatal("the new run must not reuse the stale run's row") + } +} + +// TestActiveSlotLeaseRejectsConcurrentRuns above must still pass unmodified: +// a lease that has NOT expired must keep blocking a second run. This test +// pins that same guarantee at the acquire() level directly. +func TestValidLeaseIsNotTakenOver(t *testing.T) { + db := testDB(t) + slot := uint8(1) + future := time.Now().UTC().Add(time.Hour) + holding := models.YeekeSyncRun{ + Status: "running", Trigger: "manual", StartedAt: time.Now().UTC(), + ActiveSlot: &slot, LeaseOwner: "still-alive", LeaseExpiresAt: &future, + } + if e := db.Create(&holding).Error; e != nil { + t.Fatal(e) + } + c, _ := yeekeclient.New("http://unused.invalid") + s := NewService(db, c, Config{PageSize: 10}) + if _, e := s.acquire(context.Background(), "scheduled"); e == nil { + t.Fatal("a still-valid lease must not be taken over or bypassed") + } + var row models.YeekeSyncRun + if e := db.First(&row, holding.ID).Error; e != nil { + t.Fatal(e) + } + if row.Status != "running" || row.ActiveSlot == nil { + t.Fatalf("holder must be untouched: %+v", row) + } +} + +// TestConcurrentTakeoverExactlyOneWins races two acquire() calls against the +// same stale, expired-lease row. Both attempt the takeover UPDATE and then a +// Create; the takeover UPDATE is idempotent (the loser affects zero rows +// since the row's status is no longer "running" by the time it runs), and +// the ux_yeeke_sync_run_active_slot unique index arbitrates the Create race +// the same way it already does for two brand-new concurrent runs. Exactly +// one goroutine must come away holding the slot. +func TestConcurrentTakeoverExactlyOneWins(t *testing.T) { + db := testDB(t) + slot := uint8(1) + past := time.Now().UTC().Add(-time.Hour) + stale := models.YeekeSyncRun{ + Status: "running", Trigger: "scheduled", StartedAt: past.Add(-time.Minute), + ActiveSlot: &slot, LeaseOwner: "dead-process", LeaseExpiresAt: &past, + } + if e := db.Create(&stale).Error; e != nil { + t.Fatal(e) + } + // SQLite only allows one writer at a time; serialize connections through + // the Go pool (same pattern as app/goauto/purchase/order_backfill_test.go + // and friends) so the race is decided by acquire()'s own logic rather + // than by spurious "database is locked" errors. + if sqlDB, e := db.DB(); e == nil { + sqlDB.SetMaxOpenConns(1) + } + c, _ := yeekeclient.New("http://unused.invalid") + s := NewService(db, c, Config{PageSize: 10}) + + const n = 8 + var wg sync.WaitGroup + oks := make([]bool, n) + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, e := s.acquire(context.Background(), "manual") + oks[i] = e == nil + }(i) + } + wg.Wait() + + winners := 0 + for _, ok := range oks { + if ok { + winners++ + } + } + if winners != 1 { + t.Fatalf("winners=%d, want exactly 1 (active_slot must arbitrate concurrent takeover attempts)", winners) + } + var holders int64 + db.Model(&models.YeekeSyncRun{}).Where("active_slot = ?", 1).Count(&holders) + if holders != 1 { + t.Fatalf("holders=%d, want exactly 1 row holding active_slot after the race", holders) + } +} + +// --- Defect 2 (#336): itemKey must not depend on item order -------------- + +// TestItemKeyStableAcrossReorder syncs the same package twice with its two +// items in reversed order the second time. Reordering must not create new +// rows: each item's identity must key off its own IDs, not its position. +func TestItemKeyStableAcrossReorder(t *testing.T) { + db := testDB(t) + var call int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + n := atomic.AddInt32(&call, 1) + if n == 1 { + fmt.Fprint(w, page([]string{packageWithItems("p1", item("i1", "item-a", "var-a"), item("i2", "item-b", "var-b"))}, 1, 1)) + return + } + if n == 2 { + // Same package, items reordered. + fmt.Fprint(w, page([]string{packageWithItems("p1", item("i2", "item-b", "var-b"), item("i1", "item-a", "var-a"))}, 1, 1)) + return + } + fmt.Fprint(w, page(nil, 0, 0)) + })) + defer srv.Close() + c, _ := yeekeclient.New(srv.URL) + s := NewService(db, c, Config{PageSize: 10}) + + if _, err := s.Sync(context.Background(), "manual"); err != nil { + t.Fatal(err) + } + if _, err := s.Sync(context.Background(), "manual"); err != nil { + t.Fatal(err) + } + + var pkg models.YeekeReturnPackage + if e := db.Where("external_id = ?", "p1").First(&pkg).Error; e != nil { + t.Fatal(e) + } + var n int64 + db.Model(&models.YeekeReturnItem{}).Where("package_id = ?", pkg.ID).Count(&n) + if n != 2 { + t.Fatalf("items=%d, want 2 (reordering the same items must not duplicate rows)", n) + } +} + +// TestItemKeyIndexFallbackForItemsLackingAllIDs covers a package whose items +// carry no yeeke-issued identifiers at all: the positional index is the only +// thing that can distinguish them, so the fallback must still apply and keep +// them as separate rows. +func TestItemKeyIndexFallbackForItemsLackingAllIDs(t *testing.T) { + db := testDB(t) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, page([]string{packageWithItems("p1", itemNoIDs(), itemNoIDs())}, 1, 1)) + })) + defer srv.Close() + c, _ := yeekeclient.New(srv.URL) + s := NewService(db, c, Config{PageSize: 10}) + if _, err := s.Sync(context.Background(), "manual"); err != nil { + t.Fatal(err) + } + var pkg models.YeekeReturnPackage + if e := db.Where("external_id = ?", "p1").First(&pkg).Error; e != nil { + t.Fatal(e) + } + var n int64 + db.Model(&models.YeekeReturnItem{}).Where("package_id = ?", pkg.ID).Count(&n) + if n != 2 { + t.Fatalf("items=%d, want 2 (items lacking all IDs must still be distinguished by position)", n) + } +} + +// TestItemKeyDistinctVariationsOfSameItemID pins existing behavior: two +// items sharing the same itemID but different variationIDs are, and must +// remain, two distinct rows. +func TestItemKeyDistinctVariationsOfSameItemID(t *testing.T) { + db := testDB(t) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, page([]string{packageWithItems("p1", item("i1", "item-a", "var-1"), item("i2", "item-a", "var-2"))}, 1, 1)) + })) + defer srv.Close() + c, _ := yeekeclient.New(srv.URL) + s := NewService(db, c, Config{PageSize: 10}) + if _, err := s.Sync(context.Background(), "manual"); err != nil { + t.Fatal(err) + } + var pkg models.YeekeReturnPackage + if e := db.Where("external_id = ?", "p1").First(&pkg).Error; e != nil { + t.Fatal(e) + } + var n int64 + db.Model(&models.YeekeReturnItem{}).Where("package_id = ?", pkg.ID).Count(&n) + if n != 2 { + t.Fatalf("items=%d, want 2 (same itemID with different variationID must stay distinct)", n) + } +}