fix(syb): persist validated pages and report partial sync success (#239)
This commit is contained in:
@@ -171,6 +171,9 @@ func runImport(db *gorm.DB, runID uint64, request ImportRequest, settings config
|
||||
status := SyncRunSucceeded
|
||||
if err != nil {
|
||||
status = SyncRunFailed
|
||||
if report.Created+report.Updated > 0 {
|
||||
status = SyncRunPartial
|
||||
}
|
||||
}
|
||||
finishCtx, finishCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer finishCancel()
|
||||
|
||||
@@ -4,44 +4,9 @@ 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)
|
||||
func TestTodayOverlapPreservesCommittedPagesWithoutImportingOverlappingPage(t *testing.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 {
|
||||
@@ -54,7 +19,7 @@ func TestTodayPersistentOverlapPreservesYesterdayButDoesNotImportToday(t *testin
|
||||
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"} {
|
||||
for _, token := range []string{"跨页重复", "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)
|
||||
}
|
||||
@@ -62,7 +27,7 @@ func TestTodayPersistentOverlapPreservesYesterdayButDoesNotImportToday(t *testin
|
||||
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 {
|
||||
if len(f.listCalls) != 3 || f.detailCalls != 2 || report.OrderCount != 4 || report.Created != 2 || report.Updated != 2 {
|
||||
t.Fatalf("unexpected retry/import boundary: pages=%d details=%d orders=%d", len(f.listCalls), f.detailCalls, report.OrderCount)
|
||||
}
|
||||
var count int64
|
||||
@@ -72,7 +37,6 @@ func TestTodayPersistentOverlapPreservesYesterdayButDoesNotImportToday(t *testin
|
||||
}
|
||||
|
||||
func TestPaginationHardErrorsDoNotUseOverlapRecovery(t *testing.T) {
|
||||
freezePaginationToday(t)
|
||||
for _, tc := range []struct {
|
||||
name, date, message string
|
||||
page int
|
||||
@@ -92,7 +56,7 @@ func TestPaginationHardErrorsDoNotUseOverlapRecovery(t *testing.T) {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
rows, err := loadDailyListWithRecovery(context.Background(), newSyncClient(t, f), tc.date, 2, 4, 100)
|
||||
rows, err := loadDailyList(context.Background(), newSyncClient(t, f), tc.date, 2, 4)
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -41,8 +41,7 @@ const (
|
||||
// maxSyncDays bounds one request's window. It is a guard against a typo in
|
||||
// the date range turning into tens of thousands of remote reads before
|
||||
// MaxMatches trips.
|
||||
maxSyncDays = 31
|
||||
maxTodaySnapshotAttempts = 3
|
||||
maxSyncDays = 31
|
||||
)
|
||||
|
||||
// SyncReport summarises one sync run.
|
||||
@@ -82,8 +81,6 @@ type SyncProgress struct {
|
||||
DaysProcessed int
|
||||
}
|
||||
|
||||
var syncNow = time.Now
|
||||
|
||||
type ProgressFunc func(SyncProgress) error
|
||||
|
||||
type snapshotDriftError struct {
|
||||
@@ -94,14 +91,6 @@ type snapshotDriftError struct {
|
||||
|
||||
func (err *snapshotDriftError) Error() string { return err.message }
|
||||
|
||||
func shanghaiToday() string {
|
||||
location, err := time.LoadLocation("Asia/Shanghai")
|
||||
if err != nil {
|
||||
location = time.FixedZone("Asia/Shanghai", 8*60*60)
|
||||
}
|
||||
return syncNow().In(location).Format("2006-01-02")
|
||||
}
|
||||
|
||||
// Sync pulls every shipment order in [dateFrom, dateTo] and folds each detail
|
||||
// line into the SYB/Shopee archive through ApplyDetail.
|
||||
//
|
||||
@@ -109,16 +98,10 @@ func shanghaiToday() string {
|
||||
// a separate concern (see Connect); a session that dies mid-run surfaces as
|
||||
// sybclient.ErrSessionInvalid and is treated like any other mid-run failure.
|
||||
//
|
||||
// `[必须]` Failure stops the run immediately. Rows already written are NOT
|
||||
// rolled back — ApplyDetail is idempotent on (order_code, detail_id), so a
|
||||
// re-run overwrites them rather than duplicating. What must not happen is
|
||||
// reporting a partial run as a complete one, which would let the missing
|
||||
// orders go unnoticed forever.
|
||||
//
|
||||
// `[必须]` Every day is verified for completeness before anything is written:
|
||||
// the per-day total is re-read after paging and must not have drifted. SYB's
|
||||
// list endpoint returns the *page* size in `total` (§4.3, confirmed against
|
||||
// live data), so the paging loop is driven by listTotal, never by list.total.
|
||||
// Each validated page is committed independently. Date-local read failures
|
||||
// preserve committed pages and allow later dates to run. Database, cancellation
|
||||
// and progress persistence failures stop the run. Completeness is verified at
|
||||
// the end of each date; partial results must never be reported as full success.
|
||||
func Sync(ctx context.Context, db *gorm.DB, client *sybclient.Client, cfg SyncConfig, dateFrom, dateTo string) (SyncReport, error) {
|
||||
return SyncWithProgress(ctx, db, client, cfg, dateFrom, dateTo, nil)
|
||||
}
|
||||
@@ -190,106 +173,38 @@ func SyncWithShopSnapshot(ctx context.Context, db *gorm.DB, client *sybclient.Cl
|
||||
plans = append(plans, dayPlan{date: date, total: total})
|
||||
}
|
||||
|
||||
for dayIndex, plan := range plans {
|
||||
if plan.total == 0 {
|
||||
if err := emit(dayIndex + 1); err != nil {
|
||||
return report, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
// `[必须]` Completeness first, filtering second. loadDailyList proves the
|
||||
// day's snapshot is whole; filtering before that would let drift among
|
||||
// other shops' orders hide a hole in the ones we do want
|
||||
// (docs/12-syb-erp-interface.md §8).
|
||||
rows, listErr := loadDailyListWithRecovery(ctx, client, plan.date, pageSize, plan.total, maxMatches)
|
||||
if listErr != nil && len(rows) == 0 {
|
||||
return report, listErr
|
||||
}
|
||||
report.OrderCount += len(rows)
|
||||
|
||||
byID := make(map[int64]sybclient.StockRow, len(rows))
|
||||
ids := make([]int64, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
name := stringField(row.Raw, "shopName")
|
||||
label := name
|
||||
if sybshop.IsBlank(label) {
|
||||
label = "(无店铺名)"
|
||||
}
|
||||
display, ok := allowed[sybshop.Normalize(name)]
|
||||
if ok {
|
||||
label = display
|
||||
}
|
||||
if !ok {
|
||||
entry := report.ShopBreakdown[label]
|
||||
entry.Skipped++
|
||||
report.ShopBreakdown[label] = entry
|
||||
report.ShopSkipped++
|
||||
continue
|
||||
}
|
||||
entry := report.ShopBreakdown[label]
|
||||
entry.Accepted++
|
||||
report.ShopBreakdown[label] = entry
|
||||
report.AcceptedCount++
|
||||
byID[row.ID] = row
|
||||
ids = append(ids, row.ID)
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
if err := emit(dayIndex + 1); err != nil {
|
||||
return report, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
for start := 0; start < len(ids); start += detailBatch {
|
||||
end := start + detailBatch
|
||||
if end > len(ids) {
|
||||
end = len(ids)
|
||||
}
|
||||
batch := ids[start:end]
|
||||
details, err := client.DetailListByStock(ctx, batch)
|
||||
if err != nil {
|
||||
return report, fmt.Errorf("拉取 %s 货运单明细失败(本次同步停止;"+
|
||||
"已写入的数据保留,重跑会按 (order_code, detail_id) 覆盖): %w", plan.date, err)
|
||||
}
|
||||
if err := validateDetailBatch(batch, details); err != nil {
|
||||
return report, fmt.Errorf("%s 货运单明细不完整:%w;本次同步停止", plan.date, err)
|
||||
}
|
||||
for _, detail := range details {
|
||||
// `[必须]` Re-check the shop on the detail response. The list said
|
||||
// this order belongs to an enabled shop; if the detail disagrees,
|
||||
// the two views are inconsistent and importing it would write a
|
||||
// row for a shop nobody enabled.
|
||||
name := stringField(detail.Raw, "shopName")
|
||||
if sybshop.IsBlank(name) {
|
||||
moveAcceptedToSkipped(&report, stringField(byID[detail.ID].Raw, "shopName"), "(无店铺名)", allowed)
|
||||
report.AcceptedCount--
|
||||
report.ShopSkipped++
|
||||
continue
|
||||
}
|
||||
if _, ok := allowed[sybshop.Normalize(name)]; !ok {
|
||||
moveAcceptedToSkipped(&report, stringField(byID[detail.ID].Raw, "shopName"), name, allowed)
|
||||
report.AcceptedCount--
|
||||
report.ShopSkipped++
|
||||
continue
|
||||
}
|
||||
if err := applyStockDetail(ctx, db, byID[detail.ID], detail, &report); err != nil {
|
||||
return report, err
|
||||
}
|
||||
}
|
||||
if err := emit(dayIndex); err != nil {
|
||||
return report, err
|
||||
}
|
||||
}
|
||||
if listErr != nil {
|
||||
return report, listErr
|
||||
}
|
||||
if err := emit(dayIndex + 1); err != nil {
|
||||
daysCompleted := 0
|
||||
var failures []error
|
||||
for _, plan := range plans {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return report, err
|
||||
}
|
||||
_, dayErr := loadDailyList(ctx, client, plan.date, pageSize, plan.total, func(rows []sybclient.StockRow, page int) error {
|
||||
report.OrderCount += len(rows)
|
||||
err := importSyncPage(ctx, db, client, rows, allowed, &report)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s 第 %d 页明细/入库阶段失败: %w", plan.date, page, err)
|
||||
}
|
||||
if err := emit(daysCompleted); err != nil {
|
||||
return &syncFatalError{err}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if dayErr != nil {
|
||||
failures = append(failures, dayErr)
|
||||
var fatal *syncFatalError
|
||||
if ctx.Err() != nil || errors.As(dayErr, &fatal) || errors.Is(dayErr, sybclient.ErrSessionInvalid) {
|
||||
return report, errors.Join(failures...)
|
||||
}
|
||||
continue
|
||||
}
|
||||
daysCompleted++
|
||||
if err := emit(daysCompleted); err != nil {
|
||||
return report, errors.Join(append(failures, err)...)
|
||||
}
|
||||
}
|
||||
|
||||
report.FinishedAt = time.Now().UTC()
|
||||
return report, nil
|
||||
return report, errors.Join(failures...)
|
||||
}
|
||||
|
||||
func shopSnapshotNames(allowed map[string]string) []string {
|
||||
@@ -336,39 +251,7 @@ func moveAcceptedToSkipped(report *SyncReport, listName, detailName string, allo
|
||||
// `[必须]` The loop bound comes from expectedTotal (listTotal), because
|
||||
// list.total is the current page's row count, not the filtered total (§4.3).
|
||||
// Driving the loop with the response's own total would stop after page one.
|
||||
func loadDailyListWithRecovery(ctx context.Context, client *sybclient.Client, date string, pageSize, expectedTotal, maxMatches int) ([]sybclient.StockRow, error) {
|
||||
if date != shanghaiToday() {
|
||||
return loadDailyList(ctx, client, date, pageSize, expectedTotal)
|
||||
}
|
||||
var last *snapshotDriftError
|
||||
for attempt := 1; attempt <= maxTodaySnapshotAttempts; attempt++ {
|
||||
if attempt > 1 {
|
||||
total, err := client.ListTotal(ctx, date, date, pageSize)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("重新查询今天 %s 货运单总数失败: %w", date, err)
|
||||
}
|
||||
if total > maxMatches {
|
||||
return nil, fmt.Errorf("今天 %s 的货运单总数 %d 超过单次同步上限 %d", date, total, maxMatches)
|
||||
}
|
||||
expectedTotal = total
|
||||
}
|
||||
rows, err := loadDailyList(ctx, client, date, pageSize, expectedTotal)
|
||||
if err == nil {
|
||||
return rows, nil
|
||||
}
|
||||
var drift *snapshotDriftError
|
||||
if !errors.As(err, &drift) {
|
||||
return nil, fmt.Errorf("今天第 %d/%d 次拉取失败: %w", attempt, maxTodaySnapshotAttempts, err)
|
||||
}
|
||||
last = drift
|
||||
}
|
||||
if last != nil && last.valid && len(last.rows) > 0 {
|
||||
return last.rows, fmt.Errorf("今天持续变化,已保存本次取得的完整明细,但未形成稳定快照;下次同步继续覆盖: %w", last)
|
||||
}
|
||||
return nil, fmt.Errorf("今天持续变化,连续 %d 次未形成稳定快照,本次同步停止: %w", maxTodaySnapshotAttempts, last)
|
||||
}
|
||||
|
||||
func loadDailyList(ctx context.Context, client *sybclient.Client, date string, pageSize, expectedTotal int) ([]sybclient.StockRow, error) {
|
||||
func loadDailyList(ctx context.Context, client *sybclient.Client, date string, pageSize, expectedTotal int, consume ...func([]sybclient.StockRow, int) error) ([]sybclient.StockRow, error) {
|
||||
rows := make([]sybclient.StockRow, 0, expectedTotal)
|
||||
type position struct{ page, row int }
|
||||
seen := make(map[int64]position, expectedTotal)
|
||||
@@ -413,6 +296,11 @@ func loadDailyList(ctx context.Context, client *sybclient.Client, date string, p
|
||||
if len(page) != expectedPageCount {
|
||||
return nil, &snapshotDriftError{message: fmt.Sprintf("%s 货运单列表第 %d 页不完整且相对初始总数发生变化:预期 %d 行,实际 %d 行", date, pageIndex, expectedPageCount, len(page)), rows: rows, valid: len(page) > expectedPageCount}
|
||||
}
|
||||
if len(consume) > 0 {
|
||||
if err := consume[0](page, pageIndex); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Re-read the total: if it moved while we paged, some order was inserted or
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
package sybimport
|
||||
|
||||
import (
|
||||
"context"
|
||||
"go-admin/app/goauto/sybclient"
|
||||
"go-admin/app/goauto/sybshop"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// Database/progress errors stop the entire run, not merely the current date.
|
||||
type syncFatalError struct{ error }
|
||||
|
||||
func (e *syncFatalError) Unwrap() error { return e.error }
|
||||
|
||||
// Read all remote details before opening a page transaction. Publish counters
|
||||
// only after commit, so a rollback never reports records as saved.
|
||||
func importSyncPage(ctx context.Context, db *gorm.DB, client *sybclient.Client, rows []sybclient.StockRow, allowed map[string]string, report *SyncReport) error {
|
||||
next := *report
|
||||
next.ShopBreakdown = make(map[string]ShopBreakdown, len(report.ShopBreakdown))
|
||||
for k, v := range report.ShopBreakdown {
|
||||
next.ShopBreakdown[k] = v
|
||||
}
|
||||
byID := make(map[int64]sybclient.StockRow)
|
||||
var ids []int64
|
||||
for _, row := range rows {
|
||||
name := stringField(row.Raw, "shopName")
|
||||
label := name
|
||||
if sybshop.IsBlank(label) {
|
||||
label = "(无店铺名)"
|
||||
}
|
||||
display, ok := allowed[sybshop.Normalize(name)]
|
||||
if ok {
|
||||
label = display
|
||||
}
|
||||
entry := next.ShopBreakdown[label]
|
||||
if ok {
|
||||
entry.Accepted++
|
||||
next.AcceptedCount++
|
||||
ids = append(ids, row.ID)
|
||||
byID[row.ID] = row
|
||||
} else {
|
||||
entry.Skipped++
|
||||
next.ShopSkipped++
|
||||
}
|
||||
next.ShopBreakdown[label] = entry
|
||||
}
|
||||
var details []sybclient.StockDetail
|
||||
for start := 0; start < len(ids); start += detailBatch {
|
||||
end := start + detailBatch
|
||||
if end > len(ids) {
|
||||
end = len(ids)
|
||||
}
|
||||
batch, err := client.DetailListByStock(ctx, ids[start:end])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := validateDetailBatch(ids[start:end], batch); err != nil {
|
||||
return err
|
||||
}
|
||||
details = append(details, batch...)
|
||||
}
|
||||
err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
for _, detail := range details {
|
||||
name := stringField(detail.Raw, "shopName")
|
||||
_, ok := allowed[sybshop.Normalize(name)]
|
||||
if sybshop.IsBlank(name) || !ok {
|
||||
moveAcceptedToSkipped(&next, stringField(byID[detail.ID].Raw, "shopName"), name, allowed)
|
||||
next.AcceptedCount--
|
||||
next.ShopSkipped++
|
||||
continue
|
||||
}
|
||||
if err := applyStockDetail(ctx, tx, byID[detail.ID], detail, &next); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return &syncFatalError{err}
|
||||
}
|
||||
*report = next
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
package sybimport
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"go-admin/app/goauto/models"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestPageFailurePreservesEarlierPagesContinuesDatesAndRerunsIdempotently(t *testing.T) {
|
||||
db := newSyncTestDB(t)
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-01": 3, "2026-08-02": 1}, shortPageAtIndex: 2}
|
||||
var last SyncProgress
|
||||
report, err := SyncWithProgress(context.Background(), db, newSyncClient(t, f), SyncConfig{PageSize: 2, MaxMatches: 100}, "2026-08-01", "2026-08-02", func(p SyncProgress) error { last = p; return nil })
|
||||
if err == nil || !strings.Contains(err.Error(), "第 2 页") || report.Created != 2 || report.Updated != 1 || last.DaysProcessed != 1 {
|
||||
t.Fatalf("report=%+v progress=%+v err=%v", report, last, err)
|
||||
}
|
||||
var count int64
|
||||
db.Model(&models.SYBProduct{}).Count(&count)
|
||||
if count != 2 {
|
||||
t.Fatal(count)
|
||||
}
|
||||
f.shortPageAtIndex = 0
|
||||
report, err = Sync(context.Background(), db, newSyncClient(t, f), SyncConfig{PageSize: 2, MaxMatches: 100}, "2026-08-01", "2026-08-02")
|
||||
if err != nil || report.Created != 1 || report.Updated != 3 {
|
||||
t.Fatalf("rerun=%+v err=%v", report, err)
|
||||
}
|
||||
db.Model(&models.SYBProduct{}).Count(&count)
|
||||
if count != 3 {
|
||||
t.Fatal(count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPageTransactionRollbackDoesNotPublishCountersAndStopsDates(t *testing.T) {
|
||||
db := newSyncTestDB(t)
|
||||
if err := db.Exec("CREATE TRIGGER reject_second BEFORE INSERT ON syb_product WHEN NEW.detail_id = 10011 BEGIN SELECT RAISE(ABORT, 'test database failure'); END").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-01": 3, "2026-08-02": 1}}
|
||||
report, err := Sync(context.Background(), db, newSyncClient(t, f), SyncConfig{PageSize: 2, MaxMatches: 100}, "2026-08-01", "2026-08-02")
|
||||
var fatal *syncFatalError
|
||||
if !errors.As(err, &fatal) || report.Created != 0 || report.DetailCount != 0 || len(f.listCalls) != 1 {
|
||||
t.Fatalf("report=%+v calls=%v err=%v", report, f.listCalls, err)
|
||||
}
|
||||
var count int64
|
||||
db.Model(&models.SYBProduct{}).Count(&count)
|
||||
if count != 0 {
|
||||
t.Fatal(count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPageProgressFailureStopsAfterCommittedPage(t *testing.T) {
|
||||
db := newSyncTestDB(t)
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-01": 3, "2026-08-02": 1}}
|
||||
calls := 0
|
||||
report, err := SyncWithProgress(context.Background(), db, newSyncClient(t, f), SyncConfig{PageSize: 2, MaxMatches: 100}, "2026-08-01", "2026-08-02", func(SyncProgress) error {
|
||||
calls++
|
||||
if calls == 2 {
|
||||
return errors.New("progress unavailable")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err == nil || report.Created != 2 || len(f.listCalls) != 1 {
|
||||
t.Fatalf("report=%+v err=%v", report, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPageMissingDetailPreservesPriorPageAndContinuesNextDate(t *testing.T) {
|
||||
db := newSyncTestDB(t)
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-01": 3, "2026-08-02": 1}, detailDropID: 1002}
|
||||
report, err := Sync(context.Background(), db, newSyncClient(t, f), SyncConfig{PageSize: 2, MaxMatches: 100}, "2026-08-01", "2026-08-02")
|
||||
if err == nil || report.Created != 2 || report.Updated != 1 || report.OrderCount != 4 || len(f.listCalls) != 3 {
|
||||
t.Fatalf("report=%+v err=%v", report, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPageCancellationPreservesCommitAndStopsNextPage(t *testing.T) {
|
||||
db := newSyncTestDB(t)
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-01": 3, "2026-08-02": 1}}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
report, err := SyncWithProgress(ctx, db, newSyncClient(t, f), SyncConfig{PageSize: 2, MaxMatches: 100}, "2026-08-01", "2026-08-02", func(p SyncProgress) error {
|
||||
if p.Report.Created > 0 {
|
||||
cancel()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if !errors.Is(err, context.Canceled) || report.Created != 2 || len(f.listCalls) != 1 {
|
||||
t.Fatalf("report=%+v err=%v", report, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPartialFinishPreservesProgressReleasesSlotAndIsFilterable(t *testing.T) {
|
||||
db := newSyncTestDB(t)
|
||||
ctx := context.Background()
|
||||
s := NewSyncRunService(db)
|
||||
run, err := s.Create(ctx, CreateSyncRunInput{DateFrom: "2026-08-01", DateTo: "2026-08-02"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
report := SyncReport{Created: 2, DetailCount: 2, ShopBreakdown: map[string]ShopBreakdown{}}
|
||||
if err := s.UpdateProgress(ctx, run.ID, SyncProgress{Report: report, DaysTotal: 2, DaysProcessed: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.Finish(ctx, run.ID, SyncRunPartial, report, errors.New("第 2 页失败")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
row, err := s.Detail(ctx, run.ID)
|
||||
if err != nil || row.Status != SyncRunPartial || row.ActiveSlot != nil || row.ProgressPercent != 50 || row.Created != 2 {
|
||||
t.Fatalf("row=%+v err=%v", row, err)
|
||||
}
|
||||
result, err := s.List(ctx, SyncRunListRequest{Page: 1, PageSize: 20, Status: SyncRunPartial})
|
||||
if err != nil || result.Total != 1 {
|
||||
t.Fatalf("result=%+v err=%v", result, err)
|
||||
}
|
||||
}
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
const (
|
||||
SyncRunRunning = "running"
|
||||
SyncRunSucceeded = "succeeded"
|
||||
SyncRunPartial = "partial_success"
|
||||
SyncRunFailed = "failed"
|
||||
SyncRunInterrupted = "interrupted"
|
||||
)
|
||||
@@ -75,7 +76,7 @@ func (s *SyncRunService) UpdateProgress(ctx context.Context, id uint64, progress
|
||||
}
|
||||
|
||||
func (s *SyncRunService) Finish(ctx context.Context, id uint64, status string, report SyncReport, runErr error) error {
|
||||
if status != SyncRunSucceeded && status != SyncRunFailed {
|
||||
if status != SyncRunSucceeded && status != SyncRunPartial && status != SyncRunFailed {
|
||||
return fmt.Errorf("invalid terminal sync status %q", status)
|
||||
}
|
||||
payload, err := marshalSyncRunPayload(report.ShopBreakdown, report.ShopFilterSnapshot)
|
||||
|
||||
@@ -20,7 +20,7 @@ func (handler Handler) ListSyncRuns(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
status := strings.TrimSpace(c.Query("status"))
|
||||
if status != "" && status != SyncRunRunning && status != SyncRunSucceeded && status != SyncRunFailed && status != SyncRunInterrupted {
|
||||
if status != "" && status != SyncRunRunning && status != SyncRunSucceeded && status != SyncRunPartial && status != SyncRunFailed && status != SyncRunInterrupted {
|
||||
writeError(c, invalidRequest("status 无效"))
|
||||
return
|
||||
}
|
||||
|
||||
@@ -36,4 +36,12 @@ func TestSyncRunListAndDetailHandlers(t *testing.T) {
|
||||
if detail.Code != http.StatusOK || !strings.Contains(detail.Body.String(), `"operatorName":"管理员"`) {
|
||||
t.Fatalf("详情响应不正确: %d %s", detail.Code, detail.Body.String())
|
||||
}
|
||||
if err := NewSyncRunService(db).Finish(context.Background(), run.ID, SyncRunPartial, SyncReport{Created: 1}, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
partial := httptest.NewRecorder()
|
||||
engine.ServeHTTP(partial, httptest.NewRequest(http.MethodGet, "/sync-runs?status=partial_success", nil))
|
||||
if partial.Code != http.StatusOK || !strings.Contains(partial.Body.String(), `"total":1`) {
|
||||
t.Fatalf("partial filter: %d %s", partial.Code, partial.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go-admin/app/goauto/migrations"
|
||||
"go-admin/app/goauto/models"
|
||||
@@ -276,39 +275,6 @@ func TestSyncStopsWhenTotalDriftsDuringPaging(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestTodaySnapshotDriftRetriesFromFirstPage(t *testing.T) {
|
||||
originalNow := 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 = originalNow })
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-29": 10}, totalOverride: map[int]int{1: 11, 2: 10, 3: 10}}
|
||||
rows, err := loadDailyListWithRecovery(context.Background(), newSyncClient(t, f), "2026-08-29", 10, 10, 1000)
|
||||
if err != nil || len(rows) != 10 || f.listTotalCalls != 3 {
|
||||
t.Fatalf("rows=%d totalCalls=%d err=%v", len(rows), f.listTotalCalls, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTodayThirdDriftReturnsValidLastListForDegradedSave(t *testing.T) {
|
||||
originalNow := 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 = originalNow })
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-29": 10}, totalOverride: map[int]int{1: 11, 2: 10, 3: 11, 4: 10, 5: 11}}
|
||||
rows, err := loadDailyListWithRecovery(context.Background(), newSyncClient(t, f), "2026-08-29", 10, 10, 1000)
|
||||
if err == nil || len(rows) != 10 || !strings.Contains(err.Error(), "已保存本次取得的完整明细") {
|
||||
t.Fatalf("rows=%d err=%v", len(rows), err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHistoricalSnapshotDriftDoesNotRetry(t *testing.T) {
|
||||
originalNow := 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 = originalNow })
|
||||
f := &fakeSYB{perDay: map[string]int{"2026-08-28": 10}, totalOverride: map[int]int{1: 11}}
|
||||
rows, err := loadDailyListWithRecovery(context.Background(), newSyncClient(t, f), "2026-08-28", 10, 10, 1000)
|
||||
if err == nil || rows != nil || f.listTotalCalls != 1 {
|
||||
t.Fatalf("rows=%v totalCalls=%d err=%v", rows, f.listTotalCalls, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 某页行数少于预期同样是不完整快照。
|
||||
func TestSyncStopsOnShortPage(t *testing.T) {
|
||||
db := newSyncTestDB(t)
|
||||
|
||||
@@ -36,8 +36,9 @@
|
||||
<el-drawer v-model="detail.open" title="同步记录详情" size="760px">
|
||||
<div v-loading="detail.loading" class="drawer-body">
|
||||
<template v-if="detail.item">
|
||||
<el-alert v-if="detail.item.status === 'failed' || detail.item.status === 'interrupted'" :title="detail.item.errorMessage || statusMeta(detail.item.status).label" :type="detail.item.status === 'failed' ? 'error' : 'warning'" show-icon :closable="false" class="notice" />
|
||||
<el-alert v-if="detail.item.status === 'failed' || detail.item.status === 'interrupted' || detail.item.status === 'partial_success'" :title="detail.item.errorMessage || statusMeta(detail.item.status).label" :type="detail.item.status === 'failed' ? 'error' : 'warning'" show-icon :closable="false" class="notice" />
|
||||
<el-alert v-else-if="detail.item.status === 'running'" title="任务正在后台运行,关闭本页不会中断导入。" type="info" show-icon :closable="false" class="notice" />
|
||||
<el-alert v-if="detail.item.status === 'partial_success'" title="已保存的数据会保留。请重新同步该日期范围补齐缺失数据;重跑不会重复新增相同明细。" type="warning" show-icon :closable="false" class="notice" />
|
||||
<el-descriptions :column="2" border>
|
||||
<el-descriptions-item label="记录编号">{{ detail.item.id }}</el-descriptions-item>
|
||||
<el-descriptions-item label="状态"><el-tag :type="statusMeta(detail.item.status).type">{{ statusMeta(detail.item.status).label }}</el-tag></el-descriptions-item>
|
||||
@@ -81,7 +82,7 @@ export default {
|
||||
detail: { open: false, loading: false, item: null },
|
||||
statusOptions: [
|
||||
{ label: '执行中', value: 'running' }, { label: '成功', value: 'succeeded' },
|
||||
{ label: '失败', value: 'failed' }, { label: '已中断', value: 'interrupted' }
|
||||
{ label: '部分成功', value: 'partial_success' }, { label: '失败', value: 'failed' }, { label: '已中断', value: 'interrupted' }
|
||||
]
|
||||
}
|
||||
},
|
||||
@@ -92,7 +93,7 @@ export default {
|
||||
created() { this.load().then(() => { const id = Number(this.$route.query.runId); if (id > 0) this.openDetail(id) }) },
|
||||
beforeUnmount() { this.stopPolling() },
|
||||
methods: {
|
||||
statusMeta(status) { return { running: { label: '执行中', type: 'primary' }, succeeded: { label: '成功', type: 'success' }, failed: { label: '失败', type: 'danger' }, interrupted: { label: '已中断', type: 'warning' }}[status] || { label: status || '-', type: 'info' } },
|
||||
statusMeta(status) { return { running: { label: '执行中', type: 'primary' }, succeeded: { label: '成功', type: 'success' }, partial_success: { label: '部分成功', type: 'warning' }, failed: { label: '失败', type: 'danger' }, interrupted: { label: '已中断', type: 'warning' }}[status] || { label: status || '-', type: 'info' } },
|
||||
formatTime(value) { if (!value) return '—'; return new Date(value).toLocaleString('zh-CN', { hour12: false }) },
|
||||
async load() {
|
||||
this.loading = true; this.loadError = ''
|
||||
|
||||
@@ -6,6 +6,34 @@ async function authenticate(context: any) {
|
||||
|
||||
const syncMenu = [{ path: '/syb-sync-runs', component: 'Layout', visible: '0', menuName: 'SybSync', title: 'SYB 同步', children: [{ path: 'index', component: '/goauto/syb-sync-runs/index', visible: '0', menuName: 'GoAutoSybSyncRuns', title: 'SYB 同步记录' }] }];
|
||||
|
||||
test('部分成功可筛选并查看已保存数量和补齐提示', async ({ page, context }) => {
|
||||
await authenticate(context);
|
||||
const item = { id: 239, status: 'partial_success', dateFrom: '2026-08-01', dateTo: '2026-08-02', daysProcessed: 1, daysTotal: 2, progressPercent: 50, created: 2, updated: 1, orderCount: 3, detailCount: 3, acceptedCount: 3, shopSkipped: 0, errorMessage: '2026-08-01 第 2 页读取失败', shopBreakdown: [] };
|
||||
let selected = '';
|
||||
await page.route('**/api/**', route => {
|
||||
const url = new URL(route.request().url());
|
||||
if (url.pathname.startsWith('/src/api/')) return route.continue();
|
||||
if (url.pathname.endsWith('/getinfo')) return route.fulfill({ json: { code: 200, data: { roles: ['purchaser'], name: '测试用户', avatar: '', introduction: '', permissions: [] } } });
|
||||
if (url.pathname.endsWith('/sync-runs/239')) return route.fulfill({ json: { code: 200, data: { item } } });
|
||||
if (url.pathname.endsWith('/sync-runs')) { selected = url.searchParams.get('status') || ''; return route.fulfill({ json: { code: 200, data: { items: [item], total: 1 } } }); }
|
||||
return route.fulfill({ json: { code: 200, data: url.pathname.endsWith('/menurole') ? syncMenu : [] } });
|
||||
});
|
||||
await page.goto('/#/syb-sync-runs/index');
|
||||
await expect(page.locator('.el-table').getByText('部分成功')).toBeVisible();
|
||||
await page.locator('.search-form .el-select').click();
|
||||
await page.getByRole('option', { name: '部分成功' }).click();
|
||||
await page.getByRole('button', { name: '查询', exact: true }).click();
|
||||
await expect.poll(() => selected).toBe('partial_success');
|
||||
await page.getByRole('button', { name: '详情', exact: true }).click();
|
||||
await expect(page.getByText(item.errorMessage)).toBeVisible();
|
||||
await expect(page.getByText('已保存的数据会保留。', { exact: false })).toBeVisible();
|
||||
await expect(page.locator('.el-drawer').getByText('2 / 1', { exact: true })).toBeVisible();
|
||||
await expect.poll(async () => {
|
||||
const box = await page.locator('.el-drawer').boundingBox();
|
||||
return box ? Math.round(box.x + box.width) : 0;
|
||||
}).toBe(page.viewportSize()!.width);
|
||||
});
|
||||
|
||||
test('同步记录展示失败原因和按店铺统计', async ({ page, context }) => {
|
||||
await authenticate(context);
|
||||
await page.route('**/api/**', async route => {
|
||||
|
||||
Reference in New Issue
Block a user