只改写入侧会让"行内容摘要"对不上:校验拿变换前的归档行与目标里的变换后行比较, 真实同步在 verify 阶段报"生产 resources 内容与快照不一致"并回滚。 把行变换抽成 rowTransform,写入与校验共用同一份。 试过:make verify 15 项全绿;真实同步复现该失败后修复(下一轮重跑验证)。
This commit is contained in:
@@ -321,25 +321,8 @@ func (t *ProductionLibraryTarget) insertRows(ctx context.Context, tx pgx.Tx, arc
|
||||
continue
|
||||
}
|
||||
if table == "resources" {
|
||||
// 资源定位符里的桶必须换成本次写入的目标桶:字节就写在那里。
|
||||
raw, ok := row["physical_path"]
|
||||
if !ok {
|
||||
return errors.New("资源行缺少定位符")
|
||||
}
|
||||
var locator string
|
||||
if err := json.Unmarshal(raw, &locator); err != nil {
|
||||
return errors.New("资源定位符不可解析")
|
||||
}
|
||||
if t.config.ObjectsBucket != "" {
|
||||
rewritten, ok := rewriteLocatorBucket(locator, t.config.ObjectsBucket)
|
||||
if !ok {
|
||||
return fmt.Errorf("资源定位符无法改写到目标桶:%s", locator)
|
||||
}
|
||||
encoded, err := json.Marshal(rewritten)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
row["physical_path"] = encoded
|
||||
if err := t.applyResourceBucket(row); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if columns == nil {
|
||||
@@ -388,6 +371,41 @@ type resourceLocation struct {
|
||||
Provider string
|
||||
}
|
||||
|
||||
// rowTransform 是"写入时对行做过的变换"的统一入口:校验要用同一份,否则摘要比的是
|
||||
// 变换前的行、目标里是变换后的行,永远不一致。
|
||||
func (t *ProductionLibraryTarget) rowTransform(table string) func(map[string]json.RawMessage) error {
|
||||
if table == "resources" {
|
||||
return t.applyResourceBucket
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// applyResourceBucket 把资源行定位符里的桶换成本次写入的目标桶。
|
||||
// 写入与校验必须用同一个变换:只改写入侧会让"行内容摘要"对不上(真跑踩过)。
|
||||
func (t *ProductionLibraryTarget) applyResourceBucket(row map[string]json.RawMessage) error {
|
||||
if t.config.ObjectsBucket == "" {
|
||||
return nil
|
||||
}
|
||||
raw, ok := row["physical_path"]
|
||||
if !ok {
|
||||
return errors.New("资源行缺少定位符")
|
||||
}
|
||||
var locator string
|
||||
if err := json.Unmarshal(raw, &locator); err != nil {
|
||||
return errors.New("资源定位符不可解析")
|
||||
}
|
||||
rewritten, ok := rewriteLocatorBucket(locator, t.config.ObjectsBucket)
|
||||
if !ok {
|
||||
return fmt.Errorf("资源定位符无法改写到目标桶:%s", locator)
|
||||
}
|
||||
encoded, err := json.Marshal(rewritten)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
row["physical_path"] = encoded
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *ProductionLibraryTarget) importResources(ctx context.Context, archive *librarysnapshot.Archive, manifest librarysnapshot.Manifest) error {
|
||||
paths, err := t.resourcePaths(ctx, resourcePartNames(manifest))
|
||||
if err != nil {
|
||||
@@ -542,7 +560,7 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
|
||||
wantCount, wantDigest, err := archivedTableState(archive, part)
|
||||
if errors.Is(err, errMissingIdentity) {
|
||||
// 个别表(例如日志表)没有代理主键:身份集合无从比,但行数与逐行内容摘要仍然必须一致。
|
||||
wantContentCount, wantContentDigest, columns, contentErr := archivedTableContent(archive, part)
|
||||
wantContentCount, wantContentDigest, columns, contentErr := archivedTableContent(archive, part, t.rowTransform(part.Name))
|
||||
if contentErr != nil {
|
||||
return contentErr
|
||||
}
|
||||
@@ -573,7 +591,7 @@ func (t *ProductionLibraryTarget) Verify(ctx context.Context, expect libraryrele
|
||||
return fmt.Errorf("生产 %s 与快照不一致:行数 %d/%d", part.Name, count, wantCount)
|
||||
}
|
||||
// 行数/身份一致还要比内容:同 ID 改正文、改 Wiki、改向量必须被发现。
|
||||
wantContentCount, wantContentDigest, columns, err := archivedTableContent(archive, part)
|
||||
wantContentCount, wantContentDigest, columns, err := archivedTableContent(archive, part, t.rowTransform(part.Name))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -98,7 +98,8 @@ func (t *ProductionLibraryTarget) productionTableContent(ctx context.Context, ta
|
||||
}
|
||||
|
||||
// archivedTableContent 读取归档侧规范内容摘要,并返回归档里出现过的列集合。
|
||||
func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot.Part) (int64, string, map[string]bool, error) {
|
||||
func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot.Part,
|
||||
transform func(map[string]json.RawMessage) error) (int64, string, map[string]bool, error) {
|
||||
reader, err := archive.OpenPart(part)
|
||||
if err != nil {
|
||||
return 0, "", nil, err
|
||||
@@ -116,6 +117,12 @@ func archivedTableContent(archive *librarysnapshot.Archive, part librarysnapshot
|
||||
if err := decoder.Decode(&raw); err != nil {
|
||||
return 0, "", nil, err
|
||||
}
|
||||
if transform != nil {
|
||||
// 写入时对行做过的变换(例如资源定位符换桶)校验时必须同样应用。
|
||||
if err := transform(raw); err != nil {
|
||||
return 0, "", nil, err
|
||||
}
|
||||
}
|
||||
for key := range raw {
|
||||
columns[key] = true
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user