fix(library): 修正启动收敛误判并提供精确核对
build / build (push) Failing after 1m20s

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 项通过。
This commit is contained in:
姚章浩
2026-11-06 10:51:21 +08:00
parent 146ac832a2
commit 2eff9447e1
12 changed files with 610 additions and 12 deletions
+23
View File
@@ -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
+35 -2
View File
@@ -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,
@@ -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)
}
}
@@ -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)
}
}
@@ -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)
}
}
+75 -5
View File
@@ -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, &currentSnapshot, &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, 作业, 快照) 精确定位一条版本记录。
@@ -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)
}
}
}
@@ -0,0 +1 @@
DROP TABLE IF EXISTS library_job_repairs;
@@ -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);
@@ -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'") {
+26
View File
@@ -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],
+20 -2
View File
@@ -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({
<StageBar view={view} />
{sealJob &&
canReconcileCapture &&
(sealJob.state === 'pending' || sealJob.state === 'running') &&
(sealJob.state === 'pending' || sealJob.state === 'running' || legacyOrphan) &&
stage('export')?.state === 'succeeded' ? (
<div className="rel-between rel-progress-foot" aria-label="核对封存结果">
<span>队列已完成、导出阶段已成功,但作业终态尚未回写。</span>
@@ -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 是本次预备份。