fix: 恢复并发 SKU 规格唯一键冲突 (#283)
This commit is contained in:
@@ -336,10 +336,23 @@ func findCatalogSKUBySpec(q Execer, goodsID, specKey string) (*catalogSKUStored,
|
|||||||
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE goods_id=? AND spec_key=?`, goodsID, specKey))
|
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE goods_id=? AND spec_key=?`, goodsID, specKey))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func findCatalogSKUBySpecForUpdate(q Execer, goodsID, specKey string) (*catalogSKUStored, error) {
|
||||||
|
// 规格唯一键冲突后必须使用当前读。普通 SELECT 在 MySQL 默认的
|
||||||
|
// REPEATABLE READ 事务中会继续使用第一次查询的快照,看不到竞争方
|
||||||
|
// 刚刚提交的 SKU,导致无法恢复本来可幂等的并发写入。
|
||||||
|
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE goods_id=? AND spec_key=? FOR UPDATE`, goodsID, specKey))
|
||||||
|
}
|
||||||
|
|
||||||
func findCatalogSKUByExternalID(q Execer, skuID string) (*catalogSKUStored, error) {
|
func findCatalogSKUByExternalID(q Execer, skuID string) (*catalogSKUStored, error) {
|
||||||
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE shopee_sku_id=?`, skuID))
|
return scanCatalogSKU(q.QueryRow(`SELECT sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,source_observed_at,field_sources,field_observed_at FROM shopee_skus WHERE shopee_sku_id=?`, skuID))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func isShopeeSKUSpecUniqueConflict(err error) bool {
|
||||||
|
var mysqlErr *mysql.MySQLError
|
||||||
|
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 &&
|
||||||
|
strings.Contains(mysqlErr.Message, "uq_shopee_skus_spec")
|
||||||
|
}
|
||||||
|
|
||||||
// UpsertCatalogShopeeSKU 用真实 ID 或商品规格身份定位同一内部记录,人工行永不覆盖。
|
// UpsertCatalogShopeeSKU 用真实 ID 或商品规格身份定位同一内部记录,人工行永不覆盖。
|
||||||
func UpsertCatalogShopeeSKU(q Execer, in CatalogShopeeSKUInput) (CatalogSKUOutcome, error) {
|
func UpsertCatalogShopeeSKU(q Execer, in CatalogShopeeSKUInput) (CatalogSKUOutcome, error) {
|
||||||
bySpec, err := findCatalogSKUBySpec(q, in.GoodsID, in.SpecKey)
|
bySpec, err := findCatalogSKUBySpec(q, in.GoodsID, in.SpecKey)
|
||||||
@@ -372,10 +385,23 @@ func UpsertCatalogShopeeSKU(q Execer, in CatalogShopeeSKUInput) (CatalogSKUOutco
|
|||||||
sourcesJSON, _ := json.Marshal(fieldSources)
|
sourcesJSON, _ := json.Marshal(fieldSources)
|
||||||
timesJSON, _ := json.Marshal(fieldTimes)
|
timesJSON, _ := json.Marshal(fieldTimes)
|
||||||
_, err = q.Exec(`INSERT INTO shopee_skus(sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,field_sources,field_observed_at,source_observed_at,created_at,updated_at) VALUES(?,NULLIF(?,''),?,?,?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),?,NULLIF(?,''),0,?,?,?,?,?,?)`, in.RecordID, in.ShopeeSKUID, in.GoodsID, in.SpecRaw, in.SpecKey, in.Color, in.Size, in.Advice, parse, in.SKUCode, in.Source, string(sourcesJSON), string(timesJSON), in.ObservedAt, in.Now, in.Now)
|
_, err = q.Exec(`INSERT INTO shopee_skus(sku_id,shopee_sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,sku_code,is_manual,source,field_sources,field_observed_at,source_observed_at,created_at,updated_at) VALUES(?,NULLIF(?,''),?,?,?,NULLIF(?,''),NULLIF(?,''),NULLIF(?,''),?,NULLIF(?,''),0,?,?,?,?,?,?)`, in.RecordID, in.ShopeeSKUID, in.GoodsID, in.SpecRaw, in.SpecKey, in.Color, in.Size, in.Advice, parse, in.SKUCode, in.Source, string(sourcesJSON), string(timesJSON), in.ObservedAt, in.Now, in.Now)
|
||||||
if err != nil {
|
if err == nil {
|
||||||
|
return CatalogSKUCreated, nil
|
||||||
|
}
|
||||||
|
if !isShopeeSKUSpecUniqueConflict(err) {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
return CatalogSKUCreated, nil
|
|
||||||
|
// 目录导入和 SYB 同步可以同时写同一个规格。唯一键已经证明竞争方
|
||||||
|
// 已提交该业务身份;重新读取后沿用下面原有的来源优先级规则,不能
|
||||||
|
// 把它当成同步失败,也不能用 INSERT IGNORE 静默跳过字段补全。
|
||||||
|
current, err = findCatalogSKUBySpecForUpdate(q, in.GoodsID, in.SpecKey)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("规格唯一键冲突后重读商品 %s 的 SKU 失败: %w", in.GoodsID, err)
|
||||||
|
}
|
||||||
|
if current == nil {
|
||||||
|
return "", fmt.Errorf("规格唯一键冲突后未找到商品 %s 的规格 %s", in.GoodsID, in.SpecKey)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if current.ShopeeSKUID.Valid && in.ShopeeSKUID != "" && current.ShopeeSKUID.String != in.ShopeeSKUID {
|
if current.ShopeeSKUID.Valid && in.ShopeeSKUID != "" && current.ShopeeSKUID.String != in.ShopeeSKUID {
|
||||||
return "", fmt.Errorf("商品 %s 的规格 %s 已对应真实蝦皮 SKU %s", in.GoodsID, in.SpecRaw, current.ShopeeSKUID.String)
|
return "", fmt.Errorf("商品 %s 的规格 %s 已对应真实蝦皮 SKU %s", in.GoodsID, in.SpecRaw, current.ShopeeSKUID.String)
|
||||||
|
|||||||
@@ -4,13 +4,96 @@ import (
|
|||||||
"database/sql"
|
"database/sql"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/go-sql-driver/mysql"
|
||||||
_ "modernc.org/sqlite"
|
_ "modernc.org/sqlite"
|
||||||
|
|
||||||
"cmautobuy/admin/model"
|
"cmautobuy/admin/model"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
type skuSpecConflictExecer struct {
|
||||||
|
*sql.DB
|
||||||
|
conflictKey string
|
||||||
|
inserted bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *skuSpecConflictExecer) Exec(query string, args ...any) (sql.Result, error) {
|
||||||
|
if !q.inserted && strings.HasPrefix(strings.TrimSpace(query), "INSERT INTO shopee_skus") {
|
||||||
|
q.inserted = true
|
||||||
|
_, err := q.DB.Exec(`INSERT INTO shopee_skus(
|
||||||
|
sku_id,goods_id,spec_raw,spec_key,color,size,advice,parse_ok,is_manual,
|
||||||
|
source,field_sources,field_observed_at,source_observed_at,created_at,updated_at
|
||||||
|
) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||||||
|
"catalog:existing", "GOODS-1", "白色,M", "白色,M", "目录白", "M", "",
|
||||||
|
1, 0, "xlsx-processed", `{"color":"xlsx-processed","size":"xlsx-processed","advice":"","sku_code":""}`,
|
||||||
|
`{"color":"2026-08-21T00:00:00Z","size":"2026-08-21T00:00:00Z","advice":"","sku_code":""}`,
|
||||||
|
"2026-08-21T00:00:00Z", "2026-08-21T00:00:00Z", "2026-08-21T00:00:00Z")
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return nil, &mysql.MySQLError{Number: 1062, Message: "Duplicate entry for key '" + q.conflictKey + "'"}
|
||||||
|
}
|
||||||
|
return q.DB.Exec(query, args...)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *skuSpecConflictExecer) QueryRow(query string, args ...any) *sql.Row {
|
||||||
|
return q.DB.QueryRow(strings.TrimSuffix(query, " FOR UPDATE"), args...)
|
||||||
|
}
|
||||||
|
|
||||||
|
func newSKUConflictTestDB(t *testing.T, conflictKey string) *skuSpecConflictExecer {
|
||||||
|
t.Helper()
|
||||||
|
db, err := sql.Open("sqlite", ":memory:")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { db.Close() })
|
||||||
|
if _, err := db.Exec(`CREATE TABLE shopee_skus (
|
||||||
|
sku_id TEXT PRIMARY KEY, shopee_sku_id TEXT, goods_id TEXT NOT NULL,
|
||||||
|
spec_raw TEXT NOT NULL, spec_key TEXT NOT NULL, color TEXT, size TEXT,
|
||||||
|
advice TEXT, parse_ok INTEGER NOT NULL, sku_code TEXT, is_manual INTEGER NOT NULL,
|
||||||
|
source TEXT, field_sources TEXT, field_observed_at TEXT, source_observed_at TEXT,
|
||||||
|
created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
|
||||||
|
UNIQUE(goods_id, spec_key)
|
||||||
|
)`); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return &skuSpecConflictExecer{DB: db, conflictKey: conflictKey}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpsertCatalogShopeeSKU_规格唯一键并发冲突后安全重读(t *testing.T) {
|
||||||
|
q := newSKUConflictTestDB(t, "uq_shopee_skus_spec")
|
||||||
|
outcome, err := UpsertCatalogShopeeSKU(q, CatalogShopeeSKUInput{
|
||||||
|
RecordID: "syb:incoming", GoodsID: "GOODS-1", SpecRaw: "白色,M", SpecKey: "白色,M",
|
||||||
|
Color: "顺运宝白", Size: "M", ParseOK: true, Source: "syb",
|
||||||
|
ObservedAt: "2026-08-21T00:01:00Z", Now: "2026-08-21T00:01:00Z", UpdatePolicy: "fill_missing",
|
||||||
|
})
|
||||||
|
if err != nil || outcome != CatalogSKUSkipped {
|
||||||
|
t.Fatalf("并发撞键后应重读并按既有规则跳过:outcome=%q err=%v", outcome, err)
|
||||||
|
}
|
||||||
|
var id, color, source string
|
||||||
|
if err := q.QueryRow(`SELECT sku_id,color,source FROM shopee_skus WHERE goods_id='GOODS-1' AND spec_key='白色,M'`).Scan(&id, &color, &source); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if id != "catalog:existing" || color != "目录白" || source != "xlsx-processed" {
|
||||||
|
t.Fatalf("竞争方目录数据不得被 SYB 覆盖:id=%q color=%q source=%q", id, color, source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpsertCatalogShopeeSKU_非规格唯一键冲突仍失败(t *testing.T) {
|
||||||
|
q := newSKUConflictTestDB(t, "uq_shopee_skus_external")
|
||||||
|
_, err := UpsertCatalogShopeeSKU(q, CatalogShopeeSKUInput{
|
||||||
|
RecordID: "syb:incoming", GoodsID: "GOODS-1", SpecRaw: "白色,M", SpecKey: "白色,M",
|
||||||
|
Color: "顺运宝白", Size: "M", ParseOK: true, Source: "syb",
|
||||||
|
ObservedAt: "2026-08-21T00:01:00Z", Now: "2026-08-21T00:01:00Z", UpdatePolicy: "fill_missing",
|
||||||
|
})
|
||||||
|
var mysqlErr *mysql.MySQLError
|
||||||
|
if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1062 {
|
||||||
|
t.Fatalf("非规格唯一键冲突必须原样失败:%v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestCatalogImportRun_登记查询重放和冲突(t *testing.T) {
|
func TestCatalogImportRun_登记查询重放和冲突(t *testing.T) {
|
||||||
db, err := sql.Open("sqlite", ":memory:")
|
db, err := sql.Open("sqlite", ":memory:")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user