From 2eff9447e1a6f345c4fd79d67e702f0d517cfd3f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A7=9A=E7=AB=A0=E6=B5=A9?= Date: Fri, 6 Nov 2026 10:51:21 +0800 Subject: [PATCH] =?UTF-8?q?fix(library):=20=E4=BF=AE=E6=AD=A3=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=E6=94=B6=E6=95=9B=E8=AF=AF=E5=88=A4=E5=B9=B6=E6=8F=90?= =?UTF-8?q?=E4=BE=9B=E7=B2=BE=E7=A1=AE=E6=A0=B8=E5=AF=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ReconcileLibraryJobs 不再把 completed capture queue 当成不存在，保留待核对 pending 封存作业；failed/cancelled/missing 仍按真实队列状态处置。管理员核对 入口只允许精确的启动误判形态，并要求 completed queue、export succeeded、 唯一 test_sealed 版本、来源身份与完整归档复核。新增 library_job_repairs 不可变审计表，在同一事务内 CAS 保存原 state/stage/error/recovery/finished_at 及本次 actor/原因/证据后修复；真实失败/取消不开放。 验证：真实隔离 PostgreSQL 启动收敛 RED/GREEN、启动→handler 纵向 pending 与 系统误判 failed、审计保留/幂等、负例；真实 Go 全量与 make verify 17 项通过。 --- CHANGELOG.md | 23 ++ internal/httpapi/library_capture_reconcile.go | 37 +++- .../httpapi/library_capture_reconcile_test.go | 88 ++++++++ ...rary_capture_reconcile_vertical_pg_test.go | 204 ++++++++++++++++++ .../library_capture_reconcile_pg_test.go | 36 +++- internal/platform/postgres/library_jobs.go | 80 ++++++- .../library_jobs_startup_reconcile_pg_test.go | 79 +++++++ .../000080_library_job_repair_audit.down.sql | 1 + .../000080_library_job_repair_audit.up.sql | 19 ++ internal/platform/postgres/migrations_test.go | 7 +- web/src/ReleaseTargets.test.tsx | 26 +++ web/src/ReleaseTargets.tsx | 22 +- 12 files changed, 610 insertions(+), 12 deletions(-) create mode 100644 internal/httpapi/library_capture_reconcile_vertical_pg_test.go create mode 100644 internal/platform/postgres/library_jobs_startup_reconcile_pg_test.go create mode 100644 internal/platform/postgres/migrations/000080_library_job_repair_audit.down.sql create mode 100644 internal/platform/postgres/migrations/000080_library_job_repair_audit.up.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index e79a661..8d557d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,28 @@ # Changelog +## 开发快照 0.1.366（启动收敛误判修复与误判核对审计） + +### Fixed + +- `ReconcileLibraryJobs` 启动收敛不再把仍为 pending/running、但 capture queue 已 + `completed` 的封存业务作业误判为“队列不存在”；completed 队列保留待核对状态， + failed/cancelled/missing 仍按真实队列状态处置。 +- 管理员核对入口增加唯一允许的“系统误判”失败形态：原 state=failed、stage 为空、 + snapshot 为空且 error/recovery 精确匹配历史启动误判；仍须 completed queue、 + export succeeded、唯一 test_sealed 版本、来源身份和归档字节复核。真实失败/取消/ + 其他 error 一律拒绝。 +- 新增 `library_job_repairs` 不可变审计表（migration 80）：在同一事务内 CAS 比对原 + state/stage/error/recovery/finished_at，保存原值、actor、原因和精确证据后再修复原行； + 审计写入失败则回滚，不静默覆盖。 +- UI 仅在该精确误判形态显示“核对封存结果”，真实失败没有入口；成功后刷新原作业 + 和快照状态。 + +### Tests + +- 真实隔离 PostgreSQL：启动收敛 pending+completed 保留 RED/GREEN；真实 handler + 纵向启动→核对 pending 与系统误判 failed；审计原字段与重复幂等；错误 failed/取消/ + 缺失/错证据拒绝。 + ## 开发快照 0.1.365（封存终态回写与精确核对） ### Fixed diff --git a/internal/httpapi/library_capture_reconcile.go b/internal/httpapi/library_capture_reconcile.go index b387033..7686527 100644 --- a/internal/httpapi/library_capture_reconcile.go +++ b/internal/httpapi/library_capture_reconcile.go @@ -24,8 +24,14 @@ type libraryCaptureReconciler interface { LibraryTestSealedVersionForJob(ctx context.Context, scope, jobID string) (governance.LibraryVersion, bool, error) EnsureLibraryOperation(ctx context.Context, op governance.LibraryOperation) (governance.LibraryOperation, bool, error) ReconcileLibraryCaptureTerminal(ctx context.Context, jobID, scope, queueJobID, snapshotID string, now time.Time) (bool, error) + ReconcileMisclassifiedLibraryCapture(ctx context.Context, jobID, scope, queueJobID, snapshotID, expectedStage, expectedError, expectedRecovery string, expectedFinished *time.Time, auditID, actor, reason string, evidence map[string]any, now time.Time) (bool, error) } +const ( + legacyOrphanError = "队列中已无对应作业，业务作业未收到结果" + legacyOrphanRecovery = "封存作业由 agent 恢复来源；确认来源 healthy 后可重发" +) + type libraryCaptureQueueReader interface { QueueJob(ctx context.Context, id string) (jobs.Job, bool, error) } @@ -61,8 +67,11 @@ func libraryCaptureReconcile(w http.ResponseWriter, r *http.Request, config Conf conflict(w, "只有封存作业可以核对封存结果") return } - if job.State != governance.LibraryJobPending && job.State != governance.LibraryJobRunning && job.State != governance.LibraryJobSucceeded { - conflict(w, "作业已是失败/取消终态，不能改写成成功") + systemMisclassified := job.State == governance.LibraryJobFailed && job.Stage == "" && + job.Error == legacyOrphanError && job.Recovery == legacyOrphanRecovery && job.SnapshotID == "" + if job.State != governance.LibraryJobPending && job.State != governance.LibraryJobRunning && + job.State != governance.LibraryJobSucceeded && !systemMisclassified { + conflict(w, "作业是真实失败/取消终态，不能改写成成功") return } if job.QueueJobID == "" { @@ -176,6 +185,30 @@ func libraryCaptureReconcile(w http.ResponseWriter, r *http.Request, config Conf writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build reconcile audit digest"}) return } + if systemMisclassified { + auditID := "librepair-" + digest[:16] + updated, repairErr := store.ReconcileMisclassifiedLibraryCapture(r.Context(), job.ID, job.Scope, + job.QueueJobID, version.SnapshotID, job.Stage, legacyOrphanError, legacyOrphanRecovery, + job.FinishedAt, auditID, actor, "completed capture queue + export succeeded + exact version + archive verified; legacy orphan misclassification", map[string]any{ + "original_state": job.State, "original_error": job.Error, "original_recovery": job.Recovery, + "original_finished_at": job.FinishedAt, "queue_job_id": queue.ID, + "snapshot_id": version.SnapshotID, "version_id": version.ID, "archive_verified": true, + }, now) + if repairErr != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot persist misclassification repair"}) + return + } + if !updated { + conflict(w, "原误判行状态/错误/完成时间已变化，拒绝覆盖") + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "repaired": true, "already_succeeded": false, "misclassification_repaired": true, + "job_id": job.ID, "snapshot_id": version.SnapshotID, "queue_job_id": queue.ID, + "stage": "export", "version_id": version.ID, "audit_repair_id": auditID, + }) + return + } opID := "libop-capture-reconcile-" + digest[:16] op, _, err := store.EnsureLibraryOperation(r.Context(), governance.LibraryOperation{ ID: opID, Scope: job.Scope, Kind: governance.LibraryJobRecover, diff --git a/internal/httpapi/library_capture_reconcile_test.go b/internal/httpapi/library_capture_reconcile_test.go index a5dfe6a..7ccd1be 100644 --- a/internal/httpapi/library_capture_reconcile_test.go +++ b/internal/httpapi/library_capture_reconcile_test.go @@ -99,6 +99,30 @@ func (f *fakeLibraryStore) ReconcileLibraryCaptureTerminal(_ context.Context, jo } return false, nil } + +func (f *fakeLibraryStore) ReconcileMisclassifiedLibraryCapture(_ context.Context, jobID, scope, queueJobID, snapshotID, expectedStage, expectedError, expectedRecovery string, expectedFinished *time.Time, _ string, _ string, _ string, _ map[string]any, _ time.Time) (bool, error) { + if f.reconcileDenied { + return false, nil + } + for index := range f.jobs { + j := &f.jobs[index] + if j.ID != jobID || j.Scope != scope || j.Kind != governance.LibraryJobSeal || j.QueueJobID != queueJobID || + j.State != governance.LibraryJobFailed || j.Stage != expectedStage || j.Error != expectedError || + j.Recovery != expectedRecovery || j.SnapshotID != "" || (j.FinishedAt == nil) != (expectedFinished == nil) || + (j.FinishedAt != nil && expectedFinished != nil && !j.FinishedAt.Equal(*expectedFinished)) { + return false, nil + } + j.State = governance.LibraryJobSucceeded + j.SnapshotID = snapshotID + j.Stage = "export" + f.lastState = governance.LibraryJobSucceeded + return true, nil + } + return false, nil +} + +var _ libraryCaptureReconciler = (*fakeLibraryStore)(nil) + func runReconcile(t *testing.T, config Config, jobID string) *httptest.ResponseRecorder { t.Helper() req := httptest.NewRequest(http.MethodPost, "/api/v1/library/jobs/"+jobID+"/reconcile-capture", nil) @@ -271,3 +295,67 @@ func TestLibraryCaptureReconcileConditionalWriteRejectsConcurrentState(t *testin t.Fatalf("code=%d advanced=%q body=%s", rec.Code, store.lastState, rec.Body.String()) } } + +func TestLibraryCaptureReconcileRepairsExactLegacyMisclassification(t *testing.T) { + store, config, job, _ := reconcileFixture(t) + finished := time.Now().UTC() + job.State = governance.LibraryJobFailed + job.Stage = "" + job.Error = legacyOrphanError + job.Recovery = legacyOrphanRecovery + job.FinishedAt = &finished + store.jobs[0] = job + rec := runReconcile(t, config, job.ID) + if rec.Code != http.StatusOK { + t.Fatalf("code=%d body=%s", rec.Code, rec.Body.String()) + } + var body map[string]any + _ = json.Unmarshal(rec.Body.Bytes(), &body) + if body["misclassification_repaired"] != true || store.lastState != governance.LibraryJobSucceeded { + t.Fatalf("body=%v advanced=%q", body, store.lastState) + } +} + +func TestLibraryCaptureReconcileRejectsOtherFailedError(t *testing.T) { + store, config, job, _ := reconcileFixture(t) + finished := time.Now().UTC() + job.State = governance.LibraryJobFailed + job.Stage = "" + job.Error = "capture export failed" + job.Recovery = legacyOrphanRecovery + job.FinishedAt = &finished + store.jobs[0] = job + rec := runReconcile(t, config, job.ID) + if rec.Code != http.StatusConflict || store.lastState != "" { + t.Fatalf("code=%d advanced=%q body=%s", rec.Code, store.lastState, rec.Body.String()) + } +} + +func TestLibraryCaptureReconcileRejectsRealFailureAndCancel(t *testing.T) { + for _, state := range []governance.LibraryJobState{governance.LibraryJobFailed, governance.LibraryJobCancelled} { + store, config, job, _ := reconcileFixture(t) + job.State = state + job.Error = "real failure" + store.jobs[0] = job + rec := runReconcile(t, config, job.ID) + if rec.Code != http.StatusConflict || store.lastState != "" { + t.Fatalf("state=%s code=%d advanced=%q", state, rec.Code, store.lastState) + } + } +} + +func TestLibraryCaptureReconcileMisclassificationCASRejectsChange(t *testing.T) { + store, config, job, _ := reconcileFixture(t) + finished := time.Now().UTC() + job.State = governance.LibraryJobFailed + job.Stage = "" + job.Error = legacyOrphanError + job.Recovery = legacyOrphanRecovery + job.FinishedAt = &finished + store.jobs[0] = job + store.reconcileDenied = true + rec := runReconcile(t, config, job.ID) + if rec.Code != http.StatusConflict || store.lastState != "" { + t.Fatalf("code=%d advanced=%q", rec.Code, store.lastState) + } +} diff --git a/internal/httpapi/library_capture_reconcile_vertical_pg_test.go b/internal/httpapi/library_capture_reconcile_vertical_pg_test.go new file mode 100644 index 0000000..848f4d8 --- /dev/null +++ b/internal/httpapi/library_capture_reconcile_vertical_pg_test.go @@ -0,0 +1,204 @@ +package httpapi + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "knowledge-governance/internal/governance" + "knowledge-governance/internal/librarysnapshot" + postgresstore "knowledge-governance/internal/platform/postgres" +) + +// Vertical PostgreSQL path: the real worker startup reconciliation must leave +// a pending seal with a completed capture queue intact, then the actual HTTP +// handler must verify archive/version/stage and repair it. This is not a fake +// store or a direct helper test. +func verticalReconcileFixture(t *testing.T) (*postgresstore.Store, Config, string, *pgxpool.Pool) { + t.Helper() + dsn := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("requires explicitly isolated kg_library_test_ PostgreSQL database") + } + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil || !strings.HasPrefix(cfg.ConnConfig.Database, "kg_library_test_") || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" { + t.Fatal("isolated database required") + } + cfg.MaxConns = 1 + cfg.ConnConfig.RuntimeParams["search_path"] = "pg_temp" + ctx := context.Background() + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + if _, err := pool.Exec(ctx, ` +CREATE TEMP TABLE library_jobs( + id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL, + state varchar(16) NOT NULL, stage varchar(32) NOT NULL DEFAULT '', + idempotency_key varchar(255) NOT NULL DEFAULT '', request_digest varchar(64) NOT NULL DEFAULT '', + queue_job_id varchar(64) NOT NULL DEFAULT '', snapshot_id varchar(64) NOT NULL DEFAULT '', + target_snapshot_id varchar(64) NOT NULL DEFAULT '', actor varchar(128) NOT NULL DEFAULT '', + error text NOT NULL DEFAULT '', recovery varchar(255) NOT NULL DEFAULT '', + operation_id varchar(64) NOT NULL DEFAULT '', attempt_epoch integer NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL, updated_at timestamptz NOT NULL, finished_at timestamptz, + target_id text NOT NULL DEFAULT ''); +CREATE TEMP TABLE library_job_stages( + job_id varchar(64) NOT NULL, stage varchar(32) NOT NULL, state varchar(16) NOT NULL, + total bigint, completed bigint, bytes_total bigint, bytes_done bigint, + detail text NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, PRIMARY KEY(job_id,stage)); +CREATE TEMP TABLE library_versions( + id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(24) NOT NULL, + snapshot_id varchar(64) NOT NULL DEFAULT '', version_label varchar(64) NOT NULL DEFAULT '', + source_job_id varchar(64) NOT NULL DEFAULT '', operation_id varchar(64) NOT NULL DEFAULT '', + attempt_epoch integer NOT NULL DEFAULT 0, evidence jsonb NOT NULL DEFAULT '{}'::jsonb, + created_by varchar(128) NOT NULL DEFAULT '', created_at timestamptz NOT NULL, superseded_at timestamptz); +CREATE TEMP TABLE library_job_repairs( + id varchar(64) PRIMARY KEY, job_id varchar(64) NOT NULL, scope varchar(16) NOT NULL, + kind varchar(16) NOT NULL, queue_job_id varchar(64) NOT NULL, snapshot_id varchar(64) NOT NULL, + original_state varchar(16) NOT NULL, original_stage varchar(32) NOT NULL DEFAULT '', + original_error text NOT NULL DEFAULT '', original_recovery varchar(255) NOT NULL DEFAULT '', + original_finished_at timestamptz, reason varchar(255) NOT NULL, actor varchar(128) NOT NULL DEFAULT '', + evidence jsonb NOT NULL DEFAULT '{}'::jsonb, created_at timestamptz NOT NULL); +CREATE TEMP TABLE library_operations( + id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL, + target_snapshot_id varchar(64) NOT NULL DEFAULT '', request_digest varchar(64) NOT NULL, + operation_key varchar(255) NOT NULL, actor varchar(128) NOT NULL DEFAULT '', + created_at timestamptz NOT NULL, target_id text NOT NULL DEFAULT '', + first_backup_snapshot_id varchar(64), first_backup_bytes bigint, first_backup_at timestamptz, + writeset_digest varchar(64), writeset_rows bigint, target_fingerprint varchar(128), anchors_at timestamptz); +CREATE UNIQUE INDEX library_operations_key_test ON library_operations(scope,kind,operation_key); +CREATE TEMP TABLE jobs( + id varchar(64) PRIMARY KEY, type varchar(64) NOT NULL, payload jsonb NOT NULL, + state varchar(16) NOT NULL, priority integer NOT NULL DEFAULT 0, available_at timestamptz NOT NULL, + lease_owner varchar(128), lease_until timestamptz, attempt integer NOT NULL DEFAULT 0, + max_attempts integer NOT NULL DEFAULT 1, last_error text, idempotency_key varchar(255) NOT NULL, + created_at timestamptz NOT NULL, updated_at timestamptz NOT NULL);`); err != nil { + t.Fatal(err) + } + root := libraryFixtureRoot(t) + snapshotID := sealSharedVersions(t, root, 1, 4)[0] + archive, err := librarysnapshot.Open(root, 256*1024*1024) + if err != nil { + t.Fatal(err) + } + manifest, err := archive.Load(snapshotID) + archive.Close() + if err != nil { + t.Fatal(err) + } + config := Config{LibraryArchiveRoot: root, + LibrarySourceProject: "fixture-project", LibrarySourceApp: "fixture-app", + LibrarySourceTenantID: manifest.TenantID, LibrarySourceInternalKB: manifest.KnowledgeBaseID} + source := configuredLibrarySource(config, "internal") + sourceID, err := source.ID() + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + payload, _ := json.Marshal(map[string]string{"scope": "internal", "capture_job": "vertical"}) + if _, err := pool.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,available_at,idempotency_key,created_at,updated_at) + VALUES('q-vertical','library.capture',$1,'completed',$2,'library.capture:vertical',$2,$2)`, payload, now); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `INSERT INTO library_versions(id,scope,kind,snapshot_id,source_job_id,evidence,created_at) + VALUES('ver-vertical','internal','test_sealed',$1,'libseal-vertical',$2,$3)`, snapshotID, + map[string]any{"source_id": sourceID, "capture_job": "vertical"}, now); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `INSERT INTO library_job_stages(job_id,stage,state,detail,updated_at) + VALUES('libseal-vertical','export','succeeded','已封存快照', $1)`, now); err != nil { + t.Fatal(err) + } + store := postgresstore.NewStore(pool) + config.Store = store + if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,idempotency_key,request_digest,queue_job_id,created_at,updated_at) + VALUES('libseal-vertical','internal','seal','pending','k','d','q-vertical',$1,$1)`, now); err != nil { + t.Fatal(err) + } + return store, config, snapshotID, pool +} + +func TestStartupReconcileThenHandlerRepairsPendingCapture(t *testing.T) { + store, config, _, _ := verticalReconcileFixture(t) + ctx := context.Background() + now := time.Now().UTC() + if n, err := store.ReconcileLibraryJobs(ctx, now, []string{"seal"}); err != nil || n != 0 { + t.Fatalf("startup reconcile n=%d err=%v", n, err) + } + job, err := store.LibraryJob(ctx, "libseal-vertical") + if err != nil || job.State != governance.LibraryJobPending { + t.Fatalf("startup changed job: %+v err=%v", job, err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/library/jobs/libseal-vertical/reconcile-capture", nil) + req.SetPathValue("id", "libseal-vertical") + rec := httptest.NewRecorder() + libraryCaptureReconcile(rec, req, config) + if rec.Code != http.StatusOK { + t.Fatalf("handler code=%d body=%s", rec.Code, rec.Body.String()) + } + job, err = store.LibraryJob(ctx, "libseal-vertical") + if err != nil || job.State != governance.LibraryJobSucceeded || job.SnapshotID == "" { + t.Fatalf("handler did not repair: %+v err=%v", job, err) + } +} + +func TestStartupReconcilePreservesFailedCapturesForExactReconcile(t *testing.T) { + store, config, snapshotID, pool := verticalReconcileFixture(t) + ctx := context.Background() + now := time.Now().UTC() + source := configuredLibrarySource(config, "internal") + sourceID, err := source.ID() + if err != nil { + t.Fatal(err) + } + payload, _ := json.Marshal(map[string]string{"scope": "internal", "capture_job": "vertical-failed"}) + if _, err := pool.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,available_at,idempotency_key,created_at,updated_at) + VALUES('q-failed','library.capture',$1,'completed',$2,'library.capture:vertical-failed',$2,$2)`, payload, now); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `INSERT INTO library_versions(id,scope,kind,snapshot_id,source_job_id,evidence,created_at) + VALUES('ver-failed','internal','test_sealed',$1,'libseal-vertical-failed',$2,$3)`, + snapshotID, map[string]any{"source_id": sourceID, "capture_job": "vertical-failed"}, now); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `INSERT INTO library_job_stages(job_id,stage,state,detail,updated_at) + VALUES('libseal-vertical-failed','export','succeeded','已封存快照', $1)`, now); err != nil { + t.Fatal(err) + } + finished := now.Truncate(time.Microsecond) + if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,idempotency_key,request_digest,queue_job_id,error,recovery,created_at,updated_at,finished_at) + VALUES('libseal-vertical-failed','internal','seal','failed','kf','df','q-failed',$1,$2,$3,$3,$3)`, + "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", finished); err != nil { + t.Fatal(err) + } + if n, err := store.ReconcileLibraryJobs(ctx, now, []string{"seal"}); err != nil || n != 0 { + t.Fatalf("startup reconcile n=%d err=%v", n, err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/library/jobs/libseal-vertical-failed/reconcile-capture", nil) + req.SetPathValue("id", "libseal-vertical-failed") + rec := httptest.NewRecorder() + libraryCaptureReconcile(rec, req, config) + if rec.Code != http.StatusOK { + t.Fatalf("handler code=%d body=%s", rec.Code, rec.Body.String()) + } + job, err := store.LibraryJob(ctx, "libseal-vertical-failed") + if err != nil || job.State != governance.LibraryJobSucceeded || job.SnapshotID != snapshotID { + t.Fatalf("failed misclassification not repaired: %+v err=%v", job, err) + } + var originalState, originalError string + if err := pool.QueryRow(ctx, `SELECT original_state, original_error FROM library_job_repairs WHERE job_id='libseal-vertical-failed'`). + Scan(&originalState, &originalError); err != nil { + t.Fatal(err) + } + if originalState != "failed" || originalError == "" { + t.Fatalf("audit missing original: state=%q error=%q", originalState, originalError) + } +} diff --git a/internal/platform/postgres/library_capture_reconcile_pg_test.go b/internal/platform/postgres/library_capture_reconcile_pg_test.go index fcbbefd..6c87ec5 100644 --- a/internal/platform/postgres/library_capture_reconcile_pg_test.go +++ b/internal/platform/postgres/library_capture_reconcile_pg_test.go @@ -35,10 +35,17 @@ CREATE TEMP TABLE library_jobs( id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL, state varchar(16) NOT NULL, queue_job_id varchar(64) NOT NULL DEFAULT '', stage varchar(32) NOT NULL DEFAULT '', snapshot_id varchar(64) NOT NULL DEFAULT '', - error text NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, finished_at timestamptz); + error text NOT NULL DEFAULT '', recovery varchar(255) NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, finished_at timestamptz); CREATE TEMP TABLE library_job_stages( job_id varchar(64) NOT NULL, stage varchar(32) NOT NULL, state varchar(16) NOT NULL, detail text NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, PRIMARY KEY(job_id,stage)); +CREATE TEMP TABLE library_job_repairs( + id varchar(64) PRIMARY KEY, job_id varchar(64) NOT NULL, scope varchar(16) NOT NULL, + kind varchar(16) NOT NULL, queue_job_id varchar(64) NOT NULL, snapshot_id varchar(64) NOT NULL, + original_state varchar(16) NOT NULL, original_stage varchar(32) NOT NULL DEFAULT '', + original_error text NOT NULL DEFAULT '', original_recovery varchar(255) NOT NULL DEFAULT '', + original_finished_at timestamptz, reason varchar(255) NOT NULL, actor varchar(128) NOT NULL DEFAULT '', + evidence jsonb NOT NULL DEFAULT '{}'::jsonb, created_at timestamptz NOT NULL); CREATE TEMP TABLE library_versions( id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(24) NOT NULL, snapshot_id varchar(64) NOT NULL DEFAULT '', version_label varchar(64) NOT NULL DEFAULT '', @@ -94,4 +101,31 @@ INSERT INTO library_versions(id,scope,kind,snapshot_id,source_job_id,created_at) if updated, err := store.ReconcileLibraryCaptureTerminal(ctx, "libseal-other", "internal", "libcap-3", "snap-3", now); err != nil || updated { t.Fatalf("different existing snapshot must not be overwritten: updated=%v err=%v", updated, err) } + + legacyFinished := now.Add(-time.Minute).Truncate(time.Microsecond) + if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,queue_job_id,error,recovery,updated_at,finished_at) + VALUES('libseal-misclassified','internal','seal','failed','libcap-4',$1,$2,$3,$3)`, + "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", legacyFinished); err != nil { + t.Fatal(err) + } + repaired, err := store.ReconcileMisclassifiedLibraryCapture(ctx, "libseal-misclassified", "internal", "libcap-4", "snap-4", + "", "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", &legacyFinished, + "librepair-test", "tester", "test repair", map[string]any{"archive_verified": true}, now) + if err != nil || !repaired { + t.Fatalf("misclassification repair=%v err=%v", repaired, err) + } + var originalState, originalError string + var originalFinished *time.Time + if err := pool.QueryRow(ctx, `SELECT original_state, original_error, original_finished_at FROM library_job_repairs WHERE id='librepair-test'`). + Scan(&originalState, &originalError, &originalFinished); err != nil { + t.Fatal(err) + } + if originalState != "failed" || originalError == "" || originalFinished == nil { + t.Fatalf("audit lost original fields: state=%q error=%q finished=%v", originalState, originalError, originalFinished) + } + if repaired, err := store.ReconcileMisclassifiedLibraryCapture(ctx, "libseal-misclassified", "internal", "libcap-4", "snap-4", + "", "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", &legacyFinished, + "librepair-test-2", "tester", "test repair", map[string]any{}, now); err != nil || repaired { + t.Fatalf("second repair must be false: repaired=%v err=%v", repaired, err) + } } diff --git a/internal/platform/postgres/library_jobs.go b/internal/platform/postgres/library_jobs.go index 320c56d..fc61cb2 100644 --- a/internal/platform/postgres/library_jobs.go +++ b/internal/platform/postgres/library_jobs.go @@ -346,6 +346,73 @@ WHERE id=$1 AND scope=$2 AND kind='seal' AND queue_job_id=$3 return tag.RowsAffected() == 1, nil } +// ReconcileMisclassifiedLibraryCapture repairs ONLY the known startup +// misclassification: a failed seal whose queue actually completed, with no +// stage/snapshot and the exact legacy error/recovery. The original fields are +// copied to the immutable audit row and the job update is CAS-bound to them; +// any other failed/cancelled row is rejected. +func (s *Store) ReconcileMisclassifiedLibraryCapture(ctx context.Context, jobID, scope, queueJobID, snapshotID, + expectedStage, expectedError, expectedRecovery string, expectedFinished *time.Time, + auditID, actor, reason string, evidence map[string]any, now time.Time) (bool, error) { + if jobID == "" || scope == "" || queueJobID == "" || snapshotID == "" || auditID == "" { + return false, errors.New("misclassification repair requires job/scope/queue/snapshot/audit") + } + rawEvidence, err := json.Marshal(evidence) + if err != nil { + return false, fmt.Errorf("encode repair evidence: %w", err) + } + tx, err := s.db.Begin(ctx) + if err != nil { + return false, fmt.Errorf("begin repair: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + var state, stage, errText, recovery, currentSnapshot string + var finished *time.Time + if err := tx.QueryRow(ctx, ` +SELECT state, stage, error, recovery, snapshot_id, finished_at +FROM library_jobs WHERE id=$1 AND scope=$2 AND kind='seal' AND queue_job_id=$3 +FOR UPDATE`, jobID, scope, queueJobID).Scan(&state, &stage, &errText, &recovery, ¤tSnapshot, &finished); err != nil { + return false, fmt.Errorf("read repair target: %w", err) + } + if state != "failed" || stage != expectedStage || errText != expectedError || + recovery != expectedRecovery || currentSnapshot != "" || + (finished == nil) != (expectedFinished == nil) || + (finished != nil && expectedFinished != nil && !finished.Equal(*expectedFinished)) { + return false, nil + } + tag, err := tx.Exec(ctx, ` +UPDATE library_jobs +SET state='succeeded', stage='export', error='', recovery='', snapshot_id=$3, + updated_at=$4, finished_at=$4 +WHERE id=$1 AND scope=$2 AND kind='seal' AND queue_job_id=$5 + AND state='failed' AND stage=$6 AND error=$7 AND recovery=$8 AND snapshot_id='' + AND finished_at IS NOT DISTINCT FROM $9`, + jobID, scope, snapshotID, now.UTC(), queueJobID, expectedStage, expectedError, expectedRecovery, expectedFinished) + if err != nil { + return false, fmt.Errorf("repair misclassified seal: %w", err) + } + if tag.RowsAffected() != 1 { + return false, nil + } + tag, err = tx.Exec(ctx, ` +INSERT INTO library_job_repairs + (id, job_id, scope, kind, queue_job_id, snapshot_id, original_state, original_stage, + original_error, original_recovery, original_finished_at, reason, actor, evidence, created_at) +VALUES ($1,$2,$3,'seal',$4,$5,'failed',$6,$7,$8,$9,$10,$11,$12,$13) +ON CONFLICT (id) DO NOTHING`, auditID, jobID, scope, queueJobID, snapshotID, + expectedStage, expectedError, expectedRecovery, expectedFinished, reason, actor, rawEvidence, now.UTC()) + if err != nil { + return false, fmt.Errorf("write repair audit: %w", err) + } + if tag.RowsAffected() != 1 { + return false, errors.New("repair audit already exists") + } + if err := tx.Commit(ctx); err != nil { + return false, fmt.Errorf("commit repair: %w", err) + } + return true, nil +} + // RecordLibraryVersion 写入一条**不可变版本事件**，并在需要时推进 current 指针。 // // 与旧实现的区别（F02/F09 的根因）： @@ -724,7 +791,7 @@ func (s *Store) ReconcileLibraryJobs(ctx context.Context, now time.Time, kinds [ if len(kinds) == 0 { return 0, errors.New("reconcile library jobs requires the kinds this executor owns") } - if _, err := s.db.Exec(ctx, ` + tag, err := s.db.Exec(ctx, ` UPDATE library_jobs AS lj SET state='failed', error='队列中已无对应作业，业务作业未收到结果', recovery=CASE WHEN lj.kind IN ('sync','restore','recover','baseline') @@ -737,12 +804,15 @@ WHERE lj.state IN ('pending','running') AND lj.queue_job_id <> '' AND NOT EXISTS ( SELECT 1 FROM jobs AS j - WHERE j.id = lj.queue_job_id AND j.state IN ('pending','running','retry') - )`, now.UTC(), kinds, s.libraryTargetID); err != nil { + WHERE j.id = lj.queue_job_id + AND j.state IN ('pending','running','retry','completed') + )`, now.UTC(), kinds, s.libraryTargetID) + if err != nil { return 0, fmt.Errorf("reconcile library jobs: %w", err) } + first := int(tag.RowsAffected()) // 建了业务作业但队列投递没完成（queue_job_id 为空）超过 5 分钟，也是孤儿。 - tag, err := s.db.Exec(ctx, ` + tag, err = s.db.Exec(ctx, ` UPDATE library_jobs SET state='failed', error='队列投递未完成，作业从未开始', recovery='无需恢复来源；可直接重发', updated_at=$1, finished_at=$1 WHERE state IN ('pending','running') AND queue_job_id='' AND kind = ANY($3) AND (kind='seal' OR target_id=$4) AND created_at < $2`, @@ -750,7 +820,7 @@ WHERE state IN ('pending','running') AND queue_job_id='' AND kind = ANY($3) AND if err != nil { return 0, fmt.Errorf("reconcile unattached library jobs: %w", err) } - return int(tag.RowsAffected()), nil + return first + int(tag.RowsAffected()), nil } // LibraryVersionByRef 按 (scope, kind, 作业, 快照) 精确定位一条版本记录。 diff --git a/internal/platform/postgres/library_jobs_startup_reconcile_pg_test.go b/internal/platform/postgres/library_jobs_startup_reconcile_pg_test.go new file mode 100644 index 0000000..3574176 --- /dev/null +++ b/internal/platform/postgres/library_jobs_startup_reconcile_pg_test.go @@ -0,0 +1,79 @@ +package postgres + +import ( + "context" + "os" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// The worker startup reconciliation must not turn a completed capture queue +// into the legacy "queue does not exist" failure. This is a real PostgreSQL +// fixture with the actual varchar shapes; the RED case is the old SQL which +// only accepted pending/running/retry. +func TestStartupReconcileKeepsCompletedCaptureJob(t *testing.T) { + dsn := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("requires explicitly isolated kg_library_test_ PostgreSQL database") + } + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil || !strings.HasPrefix(cfg.ConnConfig.Database, "kg_library_test_") || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" { + t.Fatal("isolated database required") + } + cfg.MaxConns = 1 + cfg.ConnConfig.RuntimeParams["search_path"] = "pg_temp" + ctx := context.Background() + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + now := time.Now().UTC() + if _, err := pool.Exec(ctx, ` +CREATE TEMP TABLE library_jobs( + id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL, + state varchar(16) NOT NULL, stage varchar(32) NOT NULL DEFAULT '', + queue_job_id varchar(64) NOT NULL DEFAULT '', target_id text NOT NULL DEFAULT '', + error text NOT NULL DEFAULT '', recovery varchar(255) NOT NULL DEFAULT '', + snapshot_id varchar(64) NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, finished_at timestamptz, created_at timestamptz NOT NULL DEFAULT now()); +CREATE TEMP TABLE jobs(id varchar(64) PRIMARY KEY, type varchar(64) NOT NULL, state varchar(16) NOT NULL);`); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,queue_job_id,updated_at) + VALUES('libseal-completed','internal','seal','pending','q-completed',$1), + ('libseal-failed','internal','seal','pending','q-failed',$1), + ('libseal-missing','internal','seal','pending','q-missing',$1)`, now); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, `INSERT INTO jobs(id,type,state) VALUES('q-completed','library.capture','completed'), + ('q-failed','library.capture','failed')`); err != nil { + t.Fatal(err) + } + store := NewStore(pool) + n, err := store.ReconcileLibraryJobs(ctx, now, []string{"seal"}) + if err != nil { + t.Fatal(err) + } + var state, errText string + if err := pool.QueryRow(ctx, `SELECT state, error FROM library_jobs WHERE id='libseal-completed'`).Scan(&state, &errText); err != nil { + t.Fatal(err) + } + if state != "pending" || errText != "" { + t.Fatalf("completed queue was misclassified: state=%q error=%q", state, errText) + } + if n != 2 { + t.Fatalf("reconciled=%d, want only failed/missing queue jobs (2)", n) + } + for _, id := range []string{"libseal-failed", "libseal-missing"} { + var got string + if err := pool.QueryRow(ctx, `SELECT state FROM library_jobs WHERE id=$1`, id).Scan(&got); err != nil { + t.Fatal(err) + } + if got != "failed" { + t.Fatalf("%s state=%q, want failed", id, got) + } + } +} diff --git a/internal/platform/postgres/migrations/000080_library_job_repair_audit.down.sql b/internal/platform/postgres/migrations/000080_library_job_repair_audit.down.sql new file mode 100644 index 0000000..1ff3720 --- /dev/null +++ b/internal/platform/postgres/migrations/000080_library_job_repair_audit.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS library_job_repairs; diff --git a/internal/platform/postgres/migrations/000080_library_job_repair_audit.up.sql b/internal/platform/postgres/migrations/000080_library_job_repair_audit.up.sql new file mode 100644 index 0000000..6362738 --- /dev/null +++ b/internal/platform/postgres/migrations/000080_library_job_repair_audit.up.sql @@ -0,0 +1,19 @@ +-- Immutable audit for the only permitted automatic/system-misclassification repair. +CREATE TABLE library_job_repairs ( + id VARCHAR(64) PRIMARY KEY, + job_id VARCHAR(64) NOT NULL, + scope VARCHAR(16) NOT NULL, + kind VARCHAR(16) NOT NULL, + queue_job_id VARCHAR(64) NOT NULL, + snapshot_id VARCHAR(64) NOT NULL, + original_state VARCHAR(16) NOT NULL, + original_stage VARCHAR(32) NOT NULL DEFAULT '', + original_error TEXT NOT NULL DEFAULT '', + original_recovery VARCHAR(255) NOT NULL DEFAULT '', + original_finished_at TIMESTAMPTZ, + reason VARCHAR(255) NOT NULL, + actor VARCHAR(128) NOT NULL DEFAULT '', + evidence JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL +); +CREATE UNIQUE INDEX library_job_repairs_identity ON library_job_repairs(job_id, snapshot_id); diff --git a/internal/platform/postgres/migrations_test.go b/internal/platform/postgres/migrations_test.go index 402bde7..e3e11ec 100644 --- a/internal/platform/postgres/migrations_test.go +++ b/internal/platform/postgres/migrations_test.go @@ -11,11 +11,14 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) { if err != nil { t.Fatalf("loadMigrations() error = %v", err) } - if len(migrations) != 79 { - t.Fatalf("migration count = %d, want 79", len(migrations)) + if len(migrations) != 80 { + t.Fatalf("migration count = %d, want 80", len(migrations)) } // 79 是恢复验证：verify 成为独立业务作业并与写作业共用互斥， // 证据按 target_id 落库（换目标旧证据自动失效）。 + if repair := migrations[79]; repair.Version != 80 { + t.Fatalf("migration 80 = %#v", repair) + } if verification := migrations[78]; verification.Version != 79 || !strings.Contains(verification.UpSQL, "library_restore_verifications") || !strings.Contains(verification.UpSQL, "'verify'") { diff --git a/web/src/ReleaseTargets.test.tsx b/web/src/ReleaseTargets.test.tsx index ee11eb8..f3d72e0 100644 --- a/web/src/ReleaseTargets.test.tsx +++ b/web/src/ReleaseTargets.test.tsx @@ -744,6 +744,32 @@ describe('ReleaseTargetsView（确认稿布局）', () => { expect(engineer).not.toContain('核对封存结果') }) + it('系统误判的封存作业（已完成队列+成功导出）仍可由管理员核对；真实失败不显示', () => { + const misclassified: LibraryJobRecord = { + ...persistedSeal, + state: 'failed', + stage: '', + snapshot_id: '', + error: '队列中已无对应作业，业务作业未收到结果', + recovery: '封存作业由 agent 恢复来源；确认来源 healthy 后可重发', + stages: [{ ...persistedSeal.stages[0], state: 'succeeded' }], + } + const admin = render({ + canReconcileCapture: true, + overview: { ...overview, job: misclassified, jobs: [misclassified] }, + }) + expect(admin).toContain('核对封存结果') + const realFailure = render({ + canReconcileCapture: true, + overview: { + ...overview, + job: { ...misclassified, error: '真实导出失败' }, + jobs: [{ ...misclassified, error: '真实导出失败' }], + }, + }) + expect(realFailure).not.toContain('核对封存结果') + }) + it('封存作业只显示只读导出步骤，不冒充生产备份/同步/生效或目标维护', () => { const html = render({ jobs: [runningJob], diff --git a/web/src/ReleaseTargets.tsx b/web/src/ReleaseTargets.tsx index 14d338f..beed84a 100644 --- a/web/src/ReleaseTargets.tsx +++ b/web/src/ReleaseTargets.tsx @@ -1107,6 +1107,14 @@ function ProgressBody({ // 不是"只有 sync 才有"。写死 sync 会让恢复作业的分项全显示"未接入"。 const dataJob = job && job.kind !== 'seal' ? job : undefined const running = job?.state === 'running' || job?.state === 'pending' + // Startup reconciliation used to rewrite a completed capture as "队列不存在". + // Only that exact legacy shape gets an admin reconcile action; real failures do not. + const legacyOrphan = + sealJob?.state === 'failed' && + sealJob.stage === '' && + sealJob.snapshot_id === '' && + sealJob.error === '队列中已无对应作业，业务作业未收到结果' && + sealJob.recovery === '封存作业由 agent 恢复来源；确认来源 healthy 后可重发' const stage = (name: string) => (dataJob ?? sealJob)?.stages.find((item) => item.stage === name) // F11: counts must belong to the snapshot this job actually used. A newly @@ -1197,7 +1205,7 @@ function ProgressBody({ {sealJob && canReconcileCapture && - (sealJob.state === 'pending' || sealJob.state === 'running') && + (sealJob.state === 'pending' || sealJob.state === 'running' || legacyOrphan) && stage('export')?.state === 'succeeded' ? (
队列已完成、导出阶段已成功，但作业终态尚未回写。 @@ -1869,7 +1877,17 @@ export function ReleaseTargetsView(props: ViewProps) { const internal = scopeView(page, jobs, 'internal', scopedOverview('internal')) const external = scopeView(page, jobs, 'external', scopedOverview('external')) const view = scope === 'internal' ? internal : external - const active = Boolean(view.activeJob) || view.job?.state === 'running' || view.job?.state === 'pending' + const legacyOrphanJob = Boolean( + view.job && + view.job.kind === 'seal' && + view.job.state === 'failed' && + view.job.stage === '' && + view.job.snapshot_id === '' && + view.job.error === '队列中已无对应作业，业务作业未收到结果' && + view.job.recovery === '封存作业由 agent 恢复来源；确认来源 healthy 后可重发' + ) + const active = + Boolean(view.activeJob) || view.job?.state === 'running' || view.job?.state === 'pending' || legacyOrphanJob // F06: 运行中禁止再次启动，也禁止把主按钮继续显示成可启动的绿色动作。 const runningJob = active && view.job ? view.job : undefined // 三类版本严格分开：target 是测试封存，production 是生产权威版本，prebackup 是本次预备份。