test(r39): 重放前后逐对象实取字节摘要比对
build / build (push) Successful in 1m55s

原重放段只比 Listing 的 key@versionId#etag#size,同一版本号下的字节
被改写它看不见,因此"重放后字节不变"缺实测证据。

改为重放前后各调一次 r38ObjectVersionDigests:逐 key+versionId 真实
GET 并本地算 SHA256,落完整版本清单(含版本、ETag、size、摘要),
再逐项比较;GET 失败、清单缺条目、版本增减或摘要不同都判失败。
治理 PG/ES/ledger/receipt 零新增断言保持原样。

离线负控 TestR39ReplayObjectByteControl:固定"旧判据对字节改写报相同"
的盲区,并要求新判据在同一元数据上报 bytes changed、正例不误报、
空清单/缺摘要必须红;用变异(去掉摘要比较)确认该负控会红后恢复。

台账:04-baseline-and-auth.json 分开记录授权 201 与基线 202,
不再让后一次状态码覆盖前一次。
This commit is contained in:
姚章浩
2026-11-02 10:14:00 +08:00
parent d2ff089b95
commit f1cddfb601
3 changed files with 176 additions and 8 deletions
@@ -19,6 +19,7 @@ import (
"time"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/jackc/pgx/v5/pgxpool"
"knowledge-governance/internal/governance"
@@ -528,6 +529,82 @@ func r39CheckPublishedObjects(t *testing.T, ctx context.Context, fx *writeSetFix
"versions_after": len(digests), "version_digests": digests}
}
// r39ObjectVersionReadback 逐 key+versionId 实取 GET 并本地算 SHA256,
// 返回完整版本清单(含 key/versionId/ETag/size/SHA256)与同一键的摘要映射。
// GET 失败、字节数与清单不符、清单里有条目没取到摘要都直接失败(非零):
// ETag/size 只是旁证,不能代替字节。
func r39ObjectVersionReadback(t *testing.T, ctx context.Context, fx *writeSetFixture,
inventory *s3.ListObjectVersionsOutput) ([]map[string]any, map[string]string) {
t.Helper()
digests := r38ObjectVersionDigests(t, ctx, fx, inventory)
manifest := make([]map[string]any, 0, len(inventory.Versions))
for _, version := range inventory.Versions {
key, id := aws.ToString(version.Key), aws.ToString(version.VersionId)
digest, ok := digests[key+"@"+id]
if !ok || digest == "" {
t.Fatalf("version %s@%s has no measured SHA256", key, id)
}
manifest = append(manifest, map[string]any{"key": key, "version_id": id,
"etag": aws.ToString(version.ETag), "size": aws.ToInt64(version.Size), "sha256": digest})
}
sort.Slice(manifest, func(i, j int) bool {
return fmt.Sprint(manifest[i]["key"], "@", manifest[i]["version_id"]) <
fmt.Sprint(manifest[j]["key"], "@", manifest[j]["version_id"])
})
return manifest, digests
}
// r39CompareObjectVersionBytes 逐 key+versionId 比较重放前后的完整清单和实取 SHA256。
// 版本增减/消失、摘要缺失、字节不同都返回问题;空清单也算问题——
// 否则"一个对象都没读到"会被当成"完全相同"。
func r39CompareObjectVersionBytes(label string, before, after []map[string]any,
beforeDigests, afterDigests map[string]string) []string {
problems := []string{}
if len(before) == 0 || len(after) == 0 {
return []string{fmt.Sprintf("%s object version manifest is empty: %d -> %d", label, len(before), len(after))}
}
if len(before) != len(after) {
problems = append(problems, fmt.Sprintf("%s object versions %d -> %d", label, len(before), len(after)))
}
ids := func(manifest []map[string]any) []string {
out := make([]string, 0, len(manifest))
for _, row := range manifest {
out = append(out, fmt.Sprint(row["key"], "@", row["version_id"]))
}
sort.Strings(out)
return out
}
beforeIDs, afterIDs := ids(before), ids(after)
beforeSet := make(map[string]bool, len(beforeIDs))
afterSet := make(map[string]bool, len(afterIDs))
for _, id := range beforeIDs {
beforeSet[id] = true
}
for _, id := range afterIDs {
afterSet[id] = true
}
for _, id := range beforeIDs {
if !afterSet[id] {
problems = append(problems, fmt.Sprintf("%s object version disappeared: %s", label, id))
continue
}
if beforeDigests[id] == "" || afterDigests[id] == "" {
problems = append(problems, fmt.Sprintf("%s object version %s has no measured SHA256", label, id))
continue
}
if beforeDigests[id] != afterDigests[id] {
problems = append(problems, fmt.Sprintf("%s object version %s bytes changed: %s -> %s",
label, id, beforeDigests[id], afterDigests[id]))
}
}
for _, id := range afterIDs {
if !beforeSet[id] {
problems = append(problems, fmt.Sprintf("%s object version appeared: %s", label, id))
}
}
return problems
}
func r39Flatten(values map[string][]string) []string {
tables := make([]string, 0, len(values))
for table := range values {
@@ -0,0 +1,75 @@
package v071
// r39 重放字节判据的离线负控:原判据只比 Listing 里的 key@versionId#etag#size,
// 同一个 versionId 的字节被改写它完全看不见。这里先把那个盲区固定下来,
// 再要求新判据逐 key+versionId 比较实取 SHA256,缺失/空清单也必须报问题。
import (
"crypto/sha256"
"encoding/hex"
"reflect"
"strings"
"testing"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/aws-sdk-go-v2/service/s3/types"
)
func r39ControlVersion() types.ObjectVersion {
return types.ObjectVersion{
Key: aws.String("kb-source/docs/1"), VersionId: aws.String("v-before"),
ETag: aws.String("\"etag-same\""), Size: aws.Int64(4)}
}
func r39ControlDigest(body string) string {
sum := sha256.Sum256([]byte(body))
return hex.EncodeToString(sum[:])
}
func r39ControlManifest(digest string) ([]map[string]any, map[string]string) {
manifest := []map[string]any{{"key": "kb-source/docs/1", "version_id": "v-before",
"etag": "\"etag-same\"", "size": int64(4), "sha256": digest}}
return manifest, map[string]string{"kb-source/docs/1@v-before": digest}
}
func TestR39ReplayObjectByteControl(t *testing.T) {
// 1) 旧判据的盲区:Listing 只有 key/versionId/etag/size,没有字节。
// 同一 versionId 的字节被改写,旧判据仍然报"完全相同"。
beforeListing := &s3.ListObjectVersionsOutput{Versions: []types.ObjectVersion{r39ControlVersion()}}
afterListing := &s3.ListObjectVersionsOutput{Versions: []types.ObjectVersion{r39ControlVersion()}}
if !reflect.DeepEqual(objectVersionIDs(beforeListing), objectVersionIDs(afterListing)) {
t.Fatalf("control setup is wrong: the legacy version-ID predicate is expected to see identical metadata")
}
beforeManifest, beforeDigests := r39ControlManifest(r39ControlDigest("abcd"))
afterManifest, afterDigests := r39ControlManifest(r39ControlDigest("abce"))
if beforeDigests["kb-source/docs/1@v-before"] == afterDigests["kb-source/docs/1@v-before"] {
t.Fatalf("control setup is wrong: the two bodies must hash differently")
}
// 2) 新判据必须在"版本元数据完全相同 + 字节不同"的输入上报出差异。
problems := r39CompareObjectVersionBytes("replay", beforeManifest, afterManifest, beforeDigests, afterDigests)
if len(problems) == 0 {
t.Fatalf("replay byte comparison missed a body change that kept key/versionId/etag/size identical")
}
if !strings.Contains(strings.Join(problems, "\n"), "bytes changed") {
t.Fatalf("byte change must be reported as bytes changed, got: %v", problems)
}
// 3) 正例对照:字节相同不得报问题。
if same := r39CompareObjectVersionBytes("replay", beforeManifest, beforeManifest, beforeDigests, beforeDigests); len(same) != 0 {
t.Fatalf("identical manifests must not report problems: %v", same)
}
// 4) 版本消失、空清单、摘要缺失都必须非零。
if got := r39CompareObjectVersionBytes("replay", beforeManifest, nil, beforeDigests, map[string]string{}); len(got) == 0 {
t.Fatalf("a disappeared/empty version manifest must be reported")
}
if got := r39CompareObjectVersionBytes("replay", nil, nil, nil, nil); len(got) == 0 {
t.Fatalf("an empty version manifest must be reported")
}
noDigest := map[string]string{}
if got := r39CompareObjectVersionBytes("replay", beforeManifest, beforeManifest, beforeDigests, noDigest); len(got) == 0 {
t.Fatalf("a version without a measured SHA256 must be reported")
}
}
@@ -740,20 +740,22 @@ UPDATE knowledge_bases SET embedding_model_id='r38-model',vector_store_id='r38-v
// 发布授权:管理员通过真实接口签发,绑定目标快照与参数摘要。
code, authBody := api.do(t, http.MethodPost, "/api/v1/library/authorizations",
map[string]any{"scope": "internal", "snapshot_id": source, "action": "sync", "expires_in_minutes": 120})
require("authorization-issued", "sync 授权必须由真实接口签发", evidence("04-baseline-and-auth.json"), code == http.StatusCreated)
save("04-baseline-and-auth.json", map[string]any{"authorization_http": code, "authorization": authBody})
authCode := code
require("authorization-issued", "sync 授权必须由真实接口签发", evidence("04-baseline-and-auth.json"), authCode == http.StatusCreated)
save("04-baseline-and-auth.json", map[string]any{"authorization_http": authCode, "authorization": authBody})
// 目标基线:目标非空且没有生效版本 → 只能走正式基线捕获。
code, baseBody := api.do(t, http.MethodPost, "/api/v1/library/baselines", map[string]any{"scope": "internal"})
require("baseline-accepted", "目标基线的唯一正式入口必须接受本次请求", evidence("04-baseline-and-auth.json"), code == http.StatusAccepted)
baselineCode := code
require("baseline-accepted", "目标基线的唯一正式入口必须接受本次请求", evidence("04-baseline-and-auth.json"), baselineCode == http.StatusAccepted)
baseJob, _ := baseBody["job"].(map[string]any)
baselineJobID, _ := baseJob["id"].(string)
if baselineJobID == "" {
t.Fatalf("baseline response carries no job id: %v", baseBody)
}
baseline := r39WaitLibraryJob(t, store, baselineJobID, governance.LibraryJobSucceeded, 420)
save("04-baseline-and-auth.json", map[string]any{"authorization_http": code, "authorization": authBody,
"baseline": map[string]any{"http": http.StatusAccepted, "job": baselineJobID, "terminal": baseline}})
save("04-baseline-and-auth.json", map[string]any{"authorization_http": authCode, "authorization": authBody,
"baseline": map[string]any{"http": baselineCode, "job": baselineJobID, "terminal": baseline}})
require("baseline-succeeded", "基线必须由 release worker 真实捕获+校验+登记后成功",
evidence("04-baseline-and-auth.json"), baseline.State == governance.LibraryJobSucceeded)
baselineVersion, hasBaseline, err := store.LibraryCurrentVersion(ctx, "internal")
@@ -961,7 +963,8 @@ UPDATE knowledge_bases SET embedding_model_id='r38-model',vector_store_id='r38-v
beforeReplayGovernance := r39DumpTables(t, gov, r39GovernanceTables)
beforeReplayPG := universeDump(t, fx.pool)
beforeReplayIndex := recoveryIndexReadback(t, fx)
beforeReplayObjects := objectVersionIDs(objectInventoryR38(t, ctx, fx))
beforeReplayInventory := objectInventoryR38(t, ctx, fx)
beforeReplayObjects, beforeReplayDigests := r39ObjectVersionReadback(t, ctx, fx, beforeReplayInventory)
saveIn("08-replay", "before.json", map[string]any{"governance": beforeReplayGovernance, "pg": beforeReplayPG,
"index": beforeReplayIndex, "object_versions": beforeReplayObjects})
replayCode, replayBody := api.do(t, http.MethodPost, "/api/v1/library/syncs", json.RawMessage(rawRequest))
@@ -972,7 +975,15 @@ UPDATE knowledge_bases SET embedding_model_id='r38-model',vector_store_id='r38-v
afterReplayGovernance := r39DumpTables(t, gov, r39GovernanceTables)
afterReplayPG := universeDump(t, fx.pool)
afterReplayIndex := recoveryIndexReadback(t, fx)
afterReplayObjects := objectVersionIDs(objectInventoryR38(t, ctx, fx))
afterReplayInventory := objectInventoryR38(t, ctx, fx)
afterReplayObjects, afterReplayDigests := r39ObjectVersionReadback(t, ctx, fx, afterReplayInventory)
objectByteProblems := r39CompareObjectVersionBytes("replay", beforeReplayObjects, afterReplayObjects,
beforeReplayDigests, afterReplayDigests)
saveIn("08-replay", "object-bytes.json", map[string]any{
"before_count": len(beforeReplayObjects), "after_count": len(afterReplayObjects),
"before_versions": beforeReplayObjects, "after_versions": afterReplayObjects,
"match": len(objectByteProblems) == 0, "problems": objectByteProblems,
"method": "每个 key+versionId 单独 GET,本地 SHA256;ETag/size 只作旁证"})
problems := r39DiffDump(beforeReplayGovernance, afterReplayGovernance)
problems = append(problems, r39DiffStrings("pg", r39Flatten(beforeReplayPG), r39Flatten(afterReplayPG))...)
if !reflect.DeepEqual(beforeReplayIndex, afterReplayIndex) {
@@ -981,10 +992,13 @@ UPDATE knowledge_bases SET embedding_model_id='r38-model',vector_store_id='r38-v
if !reflect.DeepEqual(beforeReplayObjects, afterReplayObjects) {
problems = append(problems, fmt.Sprintf("object versions changed on replay: %d -> %d", len(beforeReplayObjects), len(afterReplayObjects)))
}
problems = append(problems, objectByteProblems...)
replayRecord := map[string]any{"request_body": json.RawMessage(rawRequest), "status": replayCode,
"response": replayBody, "reused": replayBody["reused"], "job_id": replayed["id"], "operation_id": replayed["operation_id"],
"original_job_id": jobID, "original_operation_id": operationID,
"governance_diff": problems, "new_execution_evidence": map[string]any{
"governance_diff": problems, "object_versions_before": len(beforeReplayObjects),
"object_versions_after": len(afterReplayObjects), "object_byte_problems": objectByteProblems,
"new_execution_evidence": map[string]any{
"jobs": len(afterReplayGovernance["jobs"]) - len(beforeReplayGovernance["jobs"]),
"library_jobs": len(afterReplayGovernance["library_jobs"]) - len(beforeReplayGovernance["library_jobs"]),
"library_attempts": len(afterReplayGovernance["library_attempts"]) - len(beforeReplayGovernance["library_attempts"]),
@@ -994,6 +1008,8 @@ UPDATE knowledge_bases SET embedding_model_id='r38-model',vector_store_id='r38-v
save("08-replay-http.json", replayRecord)
saveIn("08-replay", "after.json", map[string]any{"governance": afterReplayGovernance, "pg": afterReplayPG,
"index": afterReplayIndex, "object_versions": afterReplayObjects})
require("replay-object-version-bytes", "重放前后每个 key+versionId 实取 GET 的完整版本清单与 SHA256 必须完全相同",
evidence("08-replay/object-bytes.json"), len(objectByteProblems) == 0)
require("replay-zero-side-effects", "重放不得新增执行/写入:治理行、目标内容、索引、对象版本必须逐字节不变",
evidence("08-replay-http.json"), len(problems) == 0)
afterOrderPG := universeDump(t, fx.pool)