fix(syb): recover bounded today pagination overlaps (#235)

2026-09-07 14:25:47 +08:00
parent bf58ad0005
commit ff2d0ca0e5
4 changed files with 144 additions and 8 deletions
@@ -0,0 +1,101 @@
+package sybimport
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+)
+
+func freezePaginationToday(t *testing.T) {
+	t.Helper()
+	previous := syncNow
+	syncNow = func() time.Time { return time.Date(2026, 8, 29, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*60*60)) }
+	t.Cleanup(func() { syncNow = previous })
+}
+
+func TestTodayCrossPageOverlapRestartsWithIndependentIDs(t *testing.T) {
+	freezePaginationToday(t)
+	f := &fakeSYB{perDay: map[string]int{"2026-08-29": 4}}
+	f.pageIDs = func(_ string, start, call int) []int64 {
+		if call == 2 {
+			return []int64{1001, 1002}
+		}
+		if call > 2 {
+			return []int64{int64(2000 + start), int64(2001 + start)}
+		}
+		return nil
+	}
+	rows, err := loadDailyListWithRecovery(context.Background(), newSyncClient(t, f), "2026-08-29", 2, 4, 100)
+	if err != nil || len(rows) != 4 || f.listTotalCalls != 2 {
+		t.Fatalf("rows=%d totals=%d err=%v", len(rows), f.listTotalCalls, err)
+	}
+	if got := strings.Join(f.listCalls, ","); got != "2026-08-29:0,2026-08-29:2,2026-08-29:0,2026-08-29:2" {
+		t.Fatalf("did not restart at first page: %s", got)
+	}
+	for index, row := range rows {
+		if row.ID != int64(2000+index) {
+			t.Fatal("rows leaked from abandoned attempt")
+		}
+	}
+}
+
+func TestTodayPersistentOverlapPreservesYesterdayButDoesNotImportToday(t *testing.T) {
+	freezePaginationToday(t)
+	f := &fakeSYB{perDay: map[string]int{"2026-08-28": 2, "2026-08-29": 4}}
+	f.pageIDs = func(date string, start, _ int) []int64 {
+		if date == "2026-08-29" && start == 2 {
+			return []int64{1001, 1002}
+		}
+		return nil
+	}
+	db := newSyncTestDB(t)
+	report, err := Sync(context.Background(), db, newSyncClient(t, f), SyncConfig{PageSize: 2, MaxMatches: 100}, "2026-08-28", "2026-08-29")
+	if err == nil {
+		t.Fatal("overlap was treated as successful sync")
+	}
+	for _, token := range []string{"连续 3 次", "跨页重复", "firstPage=1", "firstRow=2", "page=2", "row=1", "start=2", "pageSize=2", "expectedTotal=4", "unique=2"} {
+		if !strings.Contains(err.Error(), token) {
+			t.Fatalf("missing %s in %v", token, err)
+		}
+	}
+	if strings.Contains(err.Error(), "1001") || strings.Contains(err.Error(), "已保存") {
+		t.Fatalf("unsafe diagnosis/degraded save: %v", err)
+	}
+	if len(f.listCalls) != 7 || f.detailCalls != 1 || report.OrderCount != 2 {
+		t.Fatalf("unexpected retry/import boundary: pages=%d details=%d orders=%d", len(f.listCalls), f.detailCalls, report.OrderCount)
+	}
+	var count int64
+	if e := db.Table("syb_product").Count(&count).Error; e != nil || count != 2 {
+		t.Fatalf("yesterday not preserved: count=%d err=%v", count, e)
+	}
+}
+
+func TestPaginationHardErrorsDoNotUseOverlapRecovery(t *testing.T) {
+	freezePaginationToday(t)
+	for _, tc := range []struct {
+		name, date, message string
+		page                int
+		ids                 []int64
+		calls               int
+	}{
+		{"same page", "2026-08-29", "同页重复", 0, []int64{1000, 1000}, 1},
+		{"mixed overlap and same page", "2026-08-29", "同页重复", 2, []int64{1001, 1001}, 2},
+		{"overlap and invalid ID", "2026-08-29", "非法 id", 2, []int64{1001, 0}, 2},
+		{"historical overlap", "2026-08-28", "跨页重复", 2, []int64{1001, 1002}, 2},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f := &fakeSYB{perDay: map[string]int{tc.date: 4}}
+			f.pageIDs = func(_ string, start, _ int) []int64 {
+				if start == tc.page {
+					return tc.ids
+				}
+				return nil
+			}
+			rows, err := loadDailyListWithRecovery(context.Background(), newSyncClient(t, f), tc.date, 2, 4, 100)
+			if rows != nil || err == nil || !strings.Contains(err.Error(), tc.message) || len(f.listCalls) != tc.calls || f.listTotalCalls != 0 {
+				t.Fatalf("unexpected recovery: rows=%d pages=%d totals=%d err=%v", len(rows), len(f.listCalls), f.listTotalCalls, err)
+			}
+		})
+	}
+}
@@ -358,7 +358,7 @@ func loadDailyListWithRecovery(ctx context.Context, client *sybclient.Client, da
 		}
 		var drift *snapshotDriftError
 		if !errors.As(err, &drift) {
-			return nil, err
+			return nil, fmt.Errorf("今天第 %d/%d 次拉取失败: %w", attempt, maxTodaySnapshotAttempts, err)
 		}
 		last = drift
 	}
@@ -370,7 +370,8 @@ func loadDailyListWithRecovery(ctx context.Context, client *sybclient.Client, da

 func loadDailyList(ctx context.Context, client *sybclient.Client, date string, pageSize, expectedTotal int) ([]sybclient.StockRow, error) {
 	rows := make([]sybclient.StockRow, 0, expectedTotal)
-	seen := make(map[int64]struct{}, expectedTotal)
+	type position struct{ page, row int }
+	seen := make(map[int64]position, expectedTotal)

 	for start := 0; start < expectedTotal; start += pageSize {
 		pageIndex := start/pageSize + 1
@@ -386,14 +387,27 @@ func loadDailyList(ctx context.Context, client *sybclient.Client, date string, p
 		if pageCount != len(page) {
 			return nil, fmt.Errorf("%s 货运单列表第 %d 页响应条数不自洽：total=%d，list=%d", date, pageIndex, pageCount, len(page))
 		}
-		for _, row := range page {
+		// Validate the whole page first: a same-page duplicate must not be
+		// hidden behind an earlier, recoverable cross-page overlap.
+		pageSeen := make(map[int64]int, len(page))
+		for index, row := range page {
 			if row.ID <= 0 {
 				return nil, fmt.Errorf("%s 货运单列表包含非法 id=%d", date, row.ID)
 			}
-			if _, duplicate := seen[row.ID]; duplicate {
-				return nil, fmt.Errorf("%s 货运单列表重复返回 id=%d", date, row.ID)
+			if firstRow, duplicate := pageSeen[row.ID]; duplicate {
+				return nil, fmt.Errorf("%s 货运单列表同页重复 [firstPage=%d,firstRow=%d,page=%d,row=%d,start=%d,pageSize=%d,expectedTotal=%d,unique=%d]",
+					date, pageIndex, firstRow, pageIndex, index+1, start, pageSize, expectedTotal, len(rows))
 			}
-			seen[row.ID] = struct{}{}
+			pageSeen[row.ID] = index + 1
+		}
+		for index, row := range page {
+			if first, duplicate := seen[row.ID]; duplicate {
+				// No rows/valid flag: overlapping pages are never eligible for
+				// the existing final-attempt degraded save, even after deduping.
+				return nil, &snapshotDriftError{message: fmt.Sprintf("%s 货运单列表跨页重复 [firstPage=%d,firstRow=%d,page=%d,row=%d,start=%d,pageSize=%d,expectedTotal=%d,unique=%d]",
+					date, first.page, first.row, pageIndex, index+1, start, pageSize, expectedTotal, len(rows))}
+			}
+			seen[row.ID] = position{page: pageIndex, row: index + 1}
 			rows = append(rows, row)
 		}
 		if len(page) != expectedPageCount {
@@ -60,6 +60,9 @@ type fakeSYB struct {
 	totalOverride    map[int]int
 	shortPageAtIndex int
 	detailDropID     int64
+	listCalls        []string
+	pageIDs          func(date string, start, call int) []int64
+	detailCalls      int
 	// shopNames 按货运单序号轮换；留空表示全部用「测试店铺」。
 	shopNames []string
 	// detailShopName 非空时，明细响应里的 shopName 用它覆盖，
@@ -96,6 +99,7 @@ func (f *fakeSYB) server(t *testing.T) *httptest.Server {
 		start := int(body["start"].(float64))
 		pageSize := int(body["length"].(float64))
 		total := f.perDay[date]
+		f.listCalls = append(f.listCalls, fmt.Sprintf("%s:%d", date, start))

 		rows := []map[string]any{}
 		for i := start; i < total && len(rows) < pageSize; i++ {
@@ -108,9 +112,18 @@ func (f *fakeSYB) server(t *testing.T) *httptest.Server {
 		if f.shortPageAtIndex > 0 && start/pageSize+1 == f.shortPageAtIndex && len(rows) > 0 {
 			rows = rows[:len(rows)-1]
 		}
+		if f.pageIDs != nil {
+			if ids := f.pageIDs(date, start, len(f.listCalls)); ids != nil {
+				rows = nil
+				for _, id := range ids {
+					rows = append(rows, map[string]any{"id": id, "code": "TEST", "shopName": f.shopFor(0)})
+				}
+			}
+		}
 		writeEnvelope(w, map[string]any{"list": rows, "total": len(rows)})
 	})
 	mux.HandleFunc("/am/stock/detail/listByStock", func(w http.ResponseWriter, r *http.Request) {
+		f.detailCalls++
 		var body struct {
 			IDs []int64 `json:"ids"`
 		}