diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a5aaa3..918e937 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,35 @@ # Changelog +## 开发快照(安全门禁 + 崩溃接管 + 五项真实入口验收) + +### Fixed + +- **预备份持久化成为硬门禁**:`BackupObserver.BackupPrepared` 现在返回 error, + 引用/版本登记失败时 `Run` 直接返回失败且**一个字节都不写目标**(此前只写日志继续写)。 +- 预备份同时登记成 `production_prebackup` 版本,失败与成功路径都留证据。 +- 中断接管按**队列租约**判定(原来只看 state=running,worker 刚被杀而租约未过期时会漏掉); + 运行期每 30 秒扫一次,不再只在启动时扫一次。 +- 接管用一次条件更新原子抢占(`running→pending` 只有一个 worker 成功),队列作业 ID/幂等键固定, + 并发接管只会得到同一条队列作业;重复接管时把 attempt 归零,否则 `attempt 0 { message := "生产" + kindLabel(kind) + "被闸口拒绝:" + strings.Join(gateReasons, ";") recorder.Stage(ctx, "prebackup", "failed", message) + // 接管场景下生产可能已经是半写状态:这时"无需恢复"是错的, + // 必须把既有预备份作为恢复基线留住,并保持目标被占用直到处理完。 + recovery := "未对生产执行任何写入,无需恢复" + if resumed { + recovery = "生产可能处于半写状态,必须先按预备份 " + previous.SnapshotID + " 恢复后再重试" + } _ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "prebackup", message, - "未对生产执行任何写入,无需恢复", "", "", time.Now().UTC()) + recovery, previous.SnapshotID, "", time.Now().UTC()) return errors.New(message) } @@ -66,24 +75,68 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a if err != nil || !found { return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, "没有与目标快照匹配的有效发布授权", "prebackup") } - // 上一次生产生效版本登记的实际索引映射:这是本生产索引的基线。 + // 生产索引映射基线:优先当前生效版本,其次任何一次成功预备份(它也记录了生产实际映射)。 + // 只认"生效版本"会在恢复把版本 supersede 掉之后误判"索引被改动"。 knownMapping := "" if versions, versionErr := store.LibraryVersions(ctx, payload.Scope, 50); versionErr == nil { for _, item := range versions { - if item.Kind != governance.LibraryVersionProductionEffective || item.SupersededAt != nil { - continue + if item.Kind == governance.LibraryVersionProductionEffective && item.SupersededAt == nil { + if raw, ok := item.Evidence["index_mapping_sha256"].(string); ok && raw != "" { + knownMapping = raw + break + } } - if raw, ok := item.Evidence["index_mapping_sha256"].(string); ok { - knownMapping = raw + } + if knownMapping == "" { + for _, item := range versions { + if raw, ok := item.Evidence["index_mapping_sha256"].(string); ok && raw != "" { + knownMapping = raw + break + } } - break } } target, backups, wireErr := buildLibraryProductionStack(ctx, config, payload.Scope, payload.TargetSnapshotID, knownMapping) + if wireErr == nil { + if digest, mapErr := target.Search().MappingSHA256(ctx); mapErr == nil { + recorder.mappingDigest = digest + } + } if wireErr != nil { return failLibraryJob(ctx, store, recorder, payload.LibraryJobID, wireErr.Error(), "prebackup") } defer target.Close() + if payload.FinalizeOnly { + // 写入已完成、只是版本登记失败:只做校验 + 登记,绝不重写生产。 + recorder.Stage(ctx, "finalize", "running", "续作收尾:先校验生产与目标快照一致") + if verifyErr := target.Verify(ctx, libraryrelease.Expectation{SnapshotID: payload.TargetSnapshotID}); verifyErr != nil { + message := "生产与目标快照不一致,不能登记为生效版本:" + verifyErr.Error() + recorder.Stage(ctx, "finalize", "failed", message) + _ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "finalize", message, + "生产可能处于半写状态,需按预备份恢复后重试", previous.SnapshotID, payload.TargetSnapshotID, time.Now().UTC()) + return errors.New(message) + } + mappingDigest, _ := target.Search().MappingSHA256(ctx) + if err := store.RecordLibraryVersion(ctx, governance.LibraryVersion{ + ID: governance.LibraryVersionID(governance.LibraryVersionProductionEffective, payload.Scope, payload.TargetSnapshotID), + Scope: payload.Scope, Kind: governance.LibraryVersionProductionEffective, + SnapshotID: payload.TargetSnapshotID, SourceJobID: payload.LibraryJobID, CreatedBy: "worker", + Evidence: map[string]any{"backup_snapshot": previous.SnapshotID, "action": "finalize_only", + "index_mapping_sha256": mappingDigest}, + CreatedAt: time.Now().UTC(), + }); err != nil { + message := "收尾续作仍无法登记生效版本:" + err.Error() + recorder.Stage(ctx, "finalize", "failed", message) + _ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "finalize", message, + "预备份 "+previous.SnapshotID+" 保留,可再次续作", previous.SnapshotID, payload.TargetSnapshotID, time.Now().UTC()) + return errors.New(message) + } + recorder.Stage(ctx, "finalize", "succeeded", "已登记生效版本(仅收尾,未重写生产)") + _ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobSucceeded, "finalize", "", "", + previous.SnapshotID, payload.TargetSnapshotID, time.Now().UTC()) + return nil + } + var backupSource libraryrelease.BackupStore = backups if resumed { backupSource = fixedBackupStore{ref: libraryrelease.BackupRef{ID: "backup-" + previous.SnapshotID, @@ -144,8 +197,12 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a outcome.Error, outcome.Rollback, "", "", finishedAt) return fmt.Errorf("%s(%s)", outcome.Error, outcome.Rollback) default: + recovery := outcome.Recovery + if recovery == "" { + recovery = "未对生产执行任何写入,无需恢复" + } _ = store.AdvanceLibraryJob(ctx, payload.LibraryJobID, governance.LibraryJobFailed, "prebackup", - outcome.Error, "未对生产执行任何写入,无需恢复", "", "", finishedAt) + outcome.Error, recovery, outcome.Backup.SnapshotID, "", finishedAt) return errors.New(outcome.Error) } } @@ -168,9 +225,13 @@ func resumeInterruptedLibraryJobs(ctx context.Context, store *postgresstore.Stor failed++ continue } - if err := store.AdvanceLibraryJob(ctx, job.ID, governance.LibraryJobPending, "prebackup", "", "", - "", "", now); err != nil { - log.Printf("library job %s cannot be marked for resume: %v", job.ID, err) + // 原子抢占:多个 worker 同时接管也只有一个能把 running 改成 pending。 + claimed, claimErr := store.ClaimLibraryJobForResume(ctx, job.ID, now) + if claimErr != nil { + log.Printf("library job %s cannot be claimed for resume: %v", job.ID, claimErr) + continue + } + if !claimed { continue } queueType := librarySyncJobType @@ -179,15 +240,35 @@ func resumeInterruptedLibraryJobs(ctx context.Context, store *postgresstore.Stor } payload, _ := json.Marshal(map[string]string{"library_job_id": job.ID, "scope": job.Scope, "target_snapshot_id": job.TargetSnapshotID}) + // 队列作业 ID 与幂等键都按作业固定:重复接管/重复点击只会得到同一个队列作业。 queueJob, err := store.Enqueue(ctx, jobs.EnqueueRequest{ - ID: governance.QueueJobIDFor(job.ID) + "-r" + now.Format("050405"), Type: queueType, Payload: payload, + ID: governance.QueueJobIDFor(job.ID), Type: queueType, Payload: payload, AvailableAt: now, MaxAttempts: 1, - IdempotencyKey: "library.resume.queue:" + job.ID + ":" + now.Format("150405"), + IdempotencyKey: "library.resume.queue:" + job.ID, }, now) if err != nil || queueJob == nil { + // 队列作业 ID 是确定性的:并发接管时另一个 worker 已经建好了, + // 这里按"已存在"处理并继续接管,不去覆盖别人的队列行。 + if err != nil && strings.Contains(err.Error(), "duplicate key") { + queueID := governance.QueueJobIDFor(job.ID) + if resetErr := store.ResetQueueJobForResume(ctx, queueID, now); resetErr != nil { + log.Printf("library job %s resume reset existing queue failed: %v", job.ID, resetErr) + continue + } + if attachErr := store.AttachLibraryQueueJob(ctx, job.ID, queueID, now); attachErr != nil { + log.Printf("library job %s resume attach existing queue failed: %v", job.ID, attachErr) + } else { + resumed++ + } + continue + } log.Printf("library job %s resume enqueue failed: %v", job.ID, err) continue } + if resetErr := store.ResetQueueJobForResume(ctx, queueJob.ID, now); resetErr != nil { + log.Printf("library job %s resume reset queue failed: %v", job.ID, resetErr) + continue + } if attachErr := store.AttachLibraryQueueJob(ctx, job.ID, queueJob.ID, now); attachErr != nil { log.Printf("library job %s resume attach failed: %v", job.ID, attachErr) continue @@ -205,12 +286,25 @@ func (s fixedBackupStore) Prepare(context.Context, string, libraryrelease.Curren return s.ref, nil } -// BackupPrepared 让 recorder 兼作 BackupObserver,把预备份快照立刻写进作业。 -func (r libraryStageRecorder) BackupPrepared(ctx context.Context, backup libraryrelease.BackupRef) { +// BackupPrepared 把预备份写进作业**并**登记成版本。任何一步失败都返回错误: +// 上游据此在写目标前中止,保证"没有可恢复基线就不写"。 +func (r libraryStageRecorder) BackupPrepared(ctx context.Context, backup libraryrelease.BackupRef) error { + now := time.Now().UTC() if err := r.store.AdvanceLibraryJob(ctx, r.jobID, governance.LibraryJobRunning, "", "", "", - backup.SnapshotID, "", time.Now().UTC()); err != nil { - log.Printf("library job %s cannot persist backup ref: %v", r.jobID, err) + backup.SnapshotID, "", now); err != nil { + return fmt.Errorf("持久化预备份引用失败: %w", err) } + if err := r.store.RecordLibraryVersion(ctx, governance.LibraryVersion{ + ID: governance.LibraryVersionID(governance.LibraryVersionProductionPrebackup, r.scope, backup.SnapshotID), + Scope: r.scope, Kind: governance.LibraryVersionProductionPrebackup, SnapshotID: backup.SnapshotID, + SourceJobID: r.jobID, CreatedBy: "worker", + Evidence: map[string]any{"backup_snapshot": backup.SnapshotID, "backup_bytes": backup.Bytes, + "job": r.jobID, "index_mapping_sha256": r.mappingDigest}, + CreatedAt: now, + }); err != nil { + return fmt.Errorf("登记预备份版本失败: %w", err) + } + return nil } func failLibraryJob(ctx context.Context, store *postgresstore.Store, recorder libraryStageRecorder, jobID, message, stage string) error { diff --git a/cmd/worker/library_wiring.go b/cmd/worker/library_wiring.go index fd370c9..5be0d33 100644 --- a/cmd/worker/library_wiring.go +++ b/cmd/worker/library_wiring.go @@ -85,7 +85,9 @@ func buildLibraryProductionStack(ctx context.Context, config appconfig.Config, s // 它才是这份生产索引的既有事实;快照契约只管"来自哪份数据"。 baseline := knownMapping if baseline == "" { - baseline = contract.IndexMappingSHA256 + // 没有历史基线时(首次接管这套生产索引),把当前实际映射当作基线并随 + // 预备份登记;之后任何改动都会与登记值不符而被拒绝。 + baseline = mapping } if baseline != "" && mapping != baseline { current, currentErr := target.Current(ctx) diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 815da94..a35737c 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -31,6 +31,8 @@ type libraryStageRecorder struct { store *postgresstore.Store jobID string scope string + // mappingDigest 是本次读到的生产索引映射,随预备份一起登记成版本证据。 + mappingDigest string } func (r libraryStageRecorder) Stage(ctx context.Context, stage, state, detail string) { @@ -80,6 +82,24 @@ func run(ctx context.Context, getenv func(string) string) error { } else if resumed > 0 || failed > 0 { log.Printf("library job resume: %d resumed, %d failed without backup", resumed, failed) } + // 运行期也要持续接管:只在启动扫一次会漏掉"worker 在两次重启之间被杀"的情况。 + go func() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + resumed, failed, err := resumeInterruptedLibraryJobs(ctx, store, config) + if err != nil { + log.Printf("library job takeover sweep: %v", err) + } else if resumed > 0 || failed > 0 { + log.Printf("library job takeover sweep: %d resumed, %d failed without backup", resumed, failed) + } + } + } + }() if repaired, reconcileErr := store.ReconcileLibraryJobs(context.Background(), time.Now().UTC()); reconcileErr != nil { log.Printf("library job reconcile failed: %v", reconcileErr) } else if repaired > 0 { diff --git a/internal/httpapi/library_authorization.go b/internal/httpapi/library_authorization.go index 9d3f44d..a030266 100644 --- a/internal/httpapi/library_authorization.go +++ b/internal/httpapi/library_authorization.go @@ -13,8 +13,11 @@ import ( // 生产发布授权:管理员把"目标 + 快照 + 参数摘要 + 有效期"钉死。 // 换快照、换目标或改参数都会得到不同摘要,旧授权自然失效;到期或被撤销也不能用。 type libraryAuthorizationRequest struct { - Scope string `json:"scope"` - SnapshotID string `json:"snapshot_id"` + Scope string `json:"scope"` + SnapshotID string `json:"snapshot_id"` + // Action 决定授权的动作:sync(默认)或 restore。两者摘要不同, + // 同步授权不能当恢复授权用,反之亦然。 + Action string `json:"action"` ExpiresInMinutes int `json:"expires_in_minutes"` } @@ -69,14 +72,28 @@ func libraryAuthorizationCreate(w http.ResponseWriter, r *http.Request, config C writeJSON(w, http.StatusBadRequest, map[string]string{"error": "expires_in_minutes must be between 5 and 1440"}) return } - digest, err := librarySyncParamsDigest(request.Scope, request.SnapshotID) + action := strings.TrimSpace(request.Action) + if action == "" { + action = "sync" + } + var digest string + var err error + switch action { + case "sync": + digest, err = librarySyncParamsDigest(request.Scope, request.SnapshotID) + case "restore": + digest, err = libraryRestoreParamsDigest(request.Scope, request.SnapshotID) + default: + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "action must be sync or restore"}) + return + } if err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest params"}) return } now := time.Now().UTC() item := governance.LibraryReleaseAuthorization{ - ID: "libauth-" + request.Scope + "-" + now.Format("20060102T150405.000000"), + ID: "libauth-" + action + "-" + request.Scope + "-" + now.Format("20060102T150405.000000000"), Scope: request.Scope, SnapshotID: request.SnapshotID, ParamsDigest: digest, diff --git a/internal/httpapi/library_flow.go b/internal/httpapi/library_flow.go index a212454..eb32b37 100644 --- a/internal/httpapi/library_flow.go +++ b/internal/httpapi/library_flow.go @@ -597,7 +597,8 @@ func librarySyncEnqueue(w http.ResponseWriter, r *http.Request, config Config) { TargetSnapshotID: request.Target, Actor: actor, }, governance.SyncStages(), now) if errors.Is(err, governance.ErrLibraryJobActive) { - writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_active", "error": "该目标已有在飞的同步作业"}) + writeJSON(w, http.StatusConflict, map[string]any{"code": "library_sync_active", + "error": "该目标已有在飞的生产写作业(同步与恢复互斥)"}) return } if errors.Is(err, governance.ErrLibraryJobIdempotencyConflict) { diff --git a/internal/httpapi/library_restore.go b/internal/httpapi/library_restore.go index 312e2e2..311f142 100644 --- a/internal/httpapi/library_restore.go +++ b/internal/httpapi/library_restore.go @@ -228,8 +228,13 @@ func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) { writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "resumed_from": sourceID, "reused": true}) return } - payload, _ := json.Marshal(map[string]string{"library_job_id": job.ID, "scope": source.Scope, - "target_snapshot_id": source.TargetSnapshotID}) + // 写入已完成、只差登记版本的情况:续作只做幂等收尾,不重写生产。 + body := map[string]any{"library_job_id": job.ID, "scope": source.Scope, + "target_snapshot_id": source.TargetSnapshotID} + if source.Stage == "finalize" && source.SnapshotID != "" { + body["finalize_only"] = true + } + payload, _ := json.Marshal(body) queueJob, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{ ID: governance.QueueJobIDFor(job.ID), Type: queueType, Payload: payload, AvailableAt: now, MaxAttempts: 1, IdempotencyKey: "library.resume.queue:" + job.ID, diff --git a/internal/libraryrelease/release.go b/internal/libraryrelease/release.go index 3de941d..2f9bcc0 100644 --- a/internal/libraryrelease/release.go +++ b/internal/libraryrelease/release.go @@ -93,7 +93,9 @@ type Recorder interface { // BackupObserver 让调用方在预备份完成时立刻持久化快照身份: // worker 被杀后重启必须能接管这份预备份,而不是重新备份覆盖它。 type BackupObserver interface { - BackupPrepared(ctx context.Context, backup BackupRef) + // 必须返回错误:预备份引用没能持久化时,worker 被杀就没有恢复基线, + // 这种情况下一个字节都不许写目标。调用方据此直接失败。 + BackupPrepared(ctx context.Context, backup BackupRef) error } // FinalizeStages 是"写入完成之后的收尾":先登记权威版本,再把作业标成功。 @@ -117,6 +119,7 @@ type Outcome struct { FailedAt Stage `json:"failed_at,omitempty"` Error string `json:"error,omitempty"` Rollback string `json:"rollback,omitempty"` + Recovery string `json:"recovery,omitempty"` Finished time.Time `json:"finished"` } @@ -178,7 +181,14 @@ func Run(ctx context.Context, request Request, target Target, backups BackupStor } outcome.Backup = backup if observer, ok := recorder.(BackupObserver); ok { - observer.BackupPrepared(ctx, backup) + if err := observer.BackupPrepared(ctx, backup); err != nil { + // 硬门禁:持久化失败 = 没有可恢复基线 = 绝不写目标。 + outcome.State = "failed" + outcome.Error = "预备份已生成但引用无法持久化,未对生产执行任何写入:" + err.Error() + outcome.Recovery = "预备份 " + backup.SnapshotID + " 已落盘,先修复治理库再重试" + recorder.Stage(ctx, "prebackup", "failed", outcome.Error) + return outcome + } } recorder.Stage(ctx, "prebackup", "succeeded", "预备份 "+backup.ID) diff --git a/internal/libraryrelease/release_gate_test.go b/internal/libraryrelease/release_gate_test.go new file mode 100644 index 0000000..4f1232a --- /dev/null +++ b/internal/libraryrelease/release_gate_test.go @@ -0,0 +1,38 @@ +package libraryrelease + +import ( + "context" + "errors" + "testing" + "time" +) + +// failingObserver 模拟"预备份引用无法持久化":真实场景是治理库写入失败。 +type failingObserver struct{ stageLog } + +func (f *failingObserver) BackupPrepared(context.Context, BackupRef) error { + return errors.New("persist backup ref failed") +} + +// 预备份引用持久化失败时必须零写入:没有可恢复基线就不许碰目标。 +func TestRunRefusesToWriteWhenBackupReferenceCannotBePersisted(t *testing.T) { + target := &fakeTarget{} + backups := &fakeBackups{} + recorder := &failingObserver{} + outcome := Run(context.Background(), request(time.Now()), target, backups, recorder) + if outcome.State == "succeeded" { + t.Fatalf("state = %s, must not succeed without a recoverable backup", outcome.State) + } + if len(target.applied) != 0 { + t.Fatalf("applied stages = %v, must be zero when the backup reference is lost", target.applied) + } + if target.restored != 0 { + t.Fatalf("restore must not run when nothing was written (restored=%d)", target.restored) + } + if !recorder.has("prebackup:failed") { + t.Fatalf("stage log = %v, want prebackup:failed", recorder.entries) + } + if outcome.Recovery == "" { + t.Fatal("failed outcome must explain how to recover") + } +} \ No newline at end of file diff --git a/internal/platform/postgres/library_jobs.go b/internal/platform/postgres/library_jobs.go index 1d9dfe5..83c7b92 100644 --- a/internal/platform/postgres/library_jobs.go +++ b/internal/platform/postgres/library_jobs.go @@ -239,7 +239,13 @@ WHERE scope=$1 AND kind='production_effective' AND superseded_at IS NULL`, versi if _, err := tx.Exec(ctx, ` INSERT INTO library_versions (id, scope, kind, snapshot_id, version_label, source_job_id, evidence, created_by, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) -ON CONFLICT (id) DO NOTHING`, +ON CONFLICT (id) DO UPDATE SET + -- 恢复/重放同一个快照时必须把它重新"激活":否则它还是被 supersede 的旧行, + -- 页面会显示"生产当前版本未知",而实际生产已经恢复到这个快照。 + superseded_at = NULL, + source_job_id = EXCLUDED.source_job_id, + evidence = EXCLUDED.evidence, + created_at = EXCLUDED.created_at`, version.ID, version.Scope, string(version.Kind), version.SnapshotID, version.VersionLabel, version.SourceJobID, evidence, version.CreatedBy, version.CreatedAt.UTC()); err != nil { return fmt.Errorf("record library version: %w", err) @@ -250,10 +256,19 @@ ON CONFLICT (id) DO NOTHING`, // InterruptedLibraryJobs 找出"业务作业 running、但队列里已没有活跃作业"的作业: // worker 被杀/重启后它们需要被接管,而不是被当成失败丢掉已完成的预备份。 func (s *Store) InterruptedLibraryJobs(ctx context.Context) ([]governance.LibraryJob, error) { + // lease 过期才算中断:worker 刚被杀但 lease 未到期时,队列里还是 running, + // 直接判"没中断"会让作业永远悬着;所以这里按 lease_until 判断。 rows, err := s.db.Query(ctx, ` -SELECT id FROM library_jobs lj -WHERE lj.state='running' AND lj.kind IN ('sync','restore') - AND NOT EXISTS (SELECT 1 FROM jobs j WHERE j.id=lj.queue_job_id AND j.state IN ('pending','running','retry')) +SELECT lj.id FROM library_jobs lj +LEFT JOIN jobs j ON j.id=lj.queue_job_id +WHERE lj.kind IN ('sync','restore') + AND ( + -- 业务作业在跑、队列已经没了/停了/租约过期 + (lj.state='running' AND (j.id IS NULL OR j.state IN ('failed','completed','cancelled','dead') + OR (j.state IN ('pending','running','retry') AND j.lease_until IS NOT NULL AND j.lease_until < now()))) + -- 业务作业已被前一轮接管置为 pending,但队列作业的租约也没了(worker 又死了) + OR (lj.state='pending' AND j.state='running' AND j.lease_until IS NOT NULL AND j.lease_until < now()) + ) ORDER BY lj.created_at`) if err != nil { return nil, fmt.Errorf("list interrupted library jobs: %w", err) @@ -281,6 +296,28 @@ ORDER BY lj.created_at`) return jobs, nil } +// ResetQueueJobForResume 把中断/失败过的队列作业重新排队:attempt 必须归零, +// 否则 attempt 0 { + return fmt.Errorf("生产索引存在 %d 条孤儿文档(所属文档已不在数据库),首个 %s", len(orphans), orphans[0]) + } if len(snapshot.Documents) != len(archived) || snapshot.DocumentsSHA256 != wantDigest { // 摘要不一致时给出第一处差异文档,运维才能定位是"少了/多了/被改了"。 return fmt.Errorf("生产索引与快照不一致:文档 %d/%d,首个差异 %s",