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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NTDbDcwbDw1TSAcE6wfh2F
2026-09-23 10:48:38 +08:00
parent 0c5e5559d5
commit 131163c8c4
2 changed files with 318 additions and 1 deletions
@@ -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 "<nil>" by
+// fmt.Sprint on a nil/zero value) or it was an explicit empty string.
+func isEmptyExternal(v string) bool { return v == "" || v == "<nil>" }
+
+// 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)
@@ -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)
+	}
+}