From dc26321babb107b2d0bb2fc6ba0e2fdce71301da Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 26 Sep 2026 00:25:01 +0000 Subject: [PATCH] fix(docgov): join ambient uow in persistApproveAndFAV to avoid row-lock deadlock docflow Commands.run takes SELECT ... FOR UPDATE on document_records inside its unit of work before calling ApproverApprove; persistApproveAndFAV then opened a second pool connection and its UPDATE blocked on that row lock until statement timeout (SQLSTATE 57014), failing every final approval. Join the ambient uow scope (doc sync + FAV staging on the caller tx, commit owned by the uow); only open an own tx when no scope is present. Regression test reproduces the production shape (FOR UPDATE + FAV wiring inside Within). --- internal/docgov/engine_fav.go | 23 +++++ internal/docgov/engine_read_snapshot_test.go | 92 +++++++++++++++++++- 2 files changed, 113 insertions(+), 2 deletions(-) diff --git a/internal/docgov/engine_fav.go b/internal/docgov/engine_fav.go index f76dac4a..7629d5d1 100644 --- a/internal/docgov/engine_fav.go +++ b/internal/docgov/engine_fav.go @@ -6,6 +6,8 @@ import ( "fmt" "github.com/jackc/pgx/v5" + + "chat-agent-system/internal/app/uow" ) func (e *GovernanceEngine) freezeReviewSnapshotLocked(ctx context.Context, doc *DocumentRecord) error { @@ -48,6 +50,27 @@ func (e *GovernanceEngine) persistApproveAndFAV(ctx context.Context, doc *Docume if e.pool == nil { return e.syncDocAndAuditToDB(ctx, doc, audit) } + // Join the ambient unit of work when present. Opening a second + // connection here deadlocks against the caller's FOR UPDATE row lock on + // document_records (docflow Commands.run) — the UPDATE below blocks until + // statement timeout (SQLSTATE 57014). When a scope exists the uow owns + // the transaction boundary and commit; FAV rows join the same tx. + if scope, ok := uow.ScopeFrom(ctx); ok { + tx, err := uow.AsPG(scope) + if err != nil { + return err + } + if err := syncDocAndAuditTX(ctx, tx, doc, audit); err != nil { + return err + } + if e.OnFinalArtifactTX != nil { + md, _ := e.LedgerMarkdown(ctx, doc.ReviewSnapshotID) + if err := e.OnFinalArtifactTX(ctx, tx, doc, []byte(md), doc.ReviewSnapshotID); err != nil { + return err + } + } + return nil + } if e.OnFinalArtifactTX == nil { return e.syncDocAndAuditToDB(ctx, doc, audit) } diff --git a/internal/docgov/engine_read_snapshot_test.go b/internal/docgov/engine_read_snapshot_test.go index d29dbb3f..ad390f79 100644 --- a/internal/docgov/engine_read_snapshot_test.go +++ b/internal/docgov/engine_read_snapshot_test.go @@ -6,9 +6,15 @@ import ( "testing" "time" + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/artifactvault" "chat-agent-system/internal/docgov" + + "github.com/jackc/pgx/v5" ) +func artifactvaultIngest() *artifactvault.Ingest { return &artifactvault.Ingest{Env: "test"} } + // Regression (2026-09-26): loadDocumentFromDBLocked did not select // review_snapshot_id, so any doc loaded from PG after a restart (or cache // miss) lost the frozen review snapshot and ApproverApprove failed with @@ -47,13 +53,13 @@ func TestDocGov_LoadDocumentCarriesReviewSnapshotID(t *testing.T) { err = pool.QueryRow(ctx, ` WITH blob AS ( INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size, created_at) - VALUES ('snap_load_regression_sha', '# regression', 12, now()) + VALUES ($2, '# regression', 12, now()) RETURNING content_sha256 ) INSERT INTO document_snapshots (doc_id, content_sha256, actor_user_id, actor_snapshot, source_kind, created_at) SELECT $1::text, blob.content_sha256, 2, '"张工"'::jsonb, 'human', now() FROM blob RETURNING id - `, testDocID).Scan(&want) + `, testDocID, fmt.Sprintf("snap_load_sha_%d", time.Now().UnixNano())).Scan(&want) if err != nil || want <= 0 { t.Fatalf("seed snapshot: id=%d err=%v", want, err) } @@ -72,3 +78,85 @@ func TestDocGov_LoadDocumentCarriesReviewSnapshotID(t *testing.T) { } _ = doc } + +// Full approve chain against an isolated PG with production-shaped FAV +// wiring: surfaces the true persistApproveAndFAV failure that the HTTP +// envelope redacts. +func TestDocGov_ApproveChainWithFAVWiring(t *testing.T) { + pool := openIsolatedPool(t) + defer pool.Close() + projectID := seedTestProject(t, pool) + + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + + ing := artifactvaultIngest() + engine := docgov.NewGovernanceEngine(pool) + engine.OnFinalArtifactTX = func(c context.Context, tx pgx.Tx, d *docgov.DocumentRecord, md []byte, snapID int64) error { + return ing.ApprovedTX(c, tx, d.ProjectID, d.DocID, snapID, d.ApprovedAt, md) + } + + testDocID := fmt.Sprintf("doc_test_fav_%d", time.Now().UnixNano()) + if _, err := engine.CreateOrGetDocument(ctx, testDocID, projectID, "feasibility_reg", + "FAV 接线批准链", "V0.5_feasibility", "data/dev/workspaces/test_fav", + 2, "张工", 3, "李主管", 1, "Seed Admin"); err != nil { + t.Fatalf("create: %v", err) + } + if _, err := engine.RecordDocumentViewed(ctx, testDocID, 2, docgov.RoleAuthor, "张工"); err != nil { + t.Fatalf("claim: %v", err) + } + if err := engine.MarkDocumentContentUpdated(ctx, testDocID, "fav_sha256", false); err != nil { + t.Fatalf("mark: %v", err) + } + if _, err := engine.SubmitForReview(ctx, testDocID, 2, docgov.RoleAuthor, "张工"); err != nil { + t.Fatalf("submit: %v", err) + } + if _, err := engine.ReviewApprove(ctx, testDocID, 3, docgov.RoleReviewer, "李主管", ""); err != nil { + t.Fatalf("review approve: %v", err) + } + // Freeze a real snapshot like the sealed ledger does in production. + var snapID int64 + err := pool.QueryRow(ctx, ` + WITH blob AS ( + INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size, created_at) + VALUES ($2, '# fav chain', 12, now()) + RETURNING content_sha256 + ) + INSERT INTO document_snapshots (doc_id, content_sha256, actor_user_id, actor_snapshot, source_kind, created_at) + SELECT $1::text, blob.content_sha256, 2, '"张工"'::jsonb, 'human', now() FROM blob + RETURNING id + `, testDocID, fmt.Sprintf("fav_chain_sha_%d", time.Now().UnixNano())).Scan(&snapID) + if err != nil || snapID <= 0 { + t.Fatalf("seed snapshot: %d %v", snapID, err) + } + if _, err := pool.Exec(ctx, `UPDATE document_records SET review_snapshot_id=$1, head_snapshot_id=$1 WHERE doc_id=$2`, snapID, testDocID); err != nil { + t.Fatalf("freeze: %v", err) + } + // Reload through a fresh engine so the approve path loads snapshot ids from PG. + engine = docgov.NewGovernanceEngine(pool) + engine.OnFinalArtifactTX = func(c context.Context, tx pgx.Tx, d *docgov.DocumentRecord, md []byte, snap int64) error { + return ing.ApprovedTX(c, tx, d.ProjectID, d.DocID, snap, d.ApprovedAt, md) + } + // Production shape: docflow Commands.run wraps the approve in a unit of + // work and takes SELECT ... FOR UPDATE on document_records first. The old + // persistApproveAndFAV opened a second connection and deadlocked on that + // row lock until statement timeout. + unit := uow.NewPG(pool) + err = unit.Within(ctx, func(scope uow.TxScope) error { + if _, err := (docgov.Repository{}).GetForUpdate(uow.WithScope(ctx, scope), scope, testDocID); err != nil { + return err + } + _, err := engine.ApproverApprove(uow.WithScope(ctx, scope), testDocID, 1, docgov.RoleApprover, "Seed Admin", "ok") + return err + }) + if err != nil { + t.Fatalf("approver approve inside uow (deadlock regression): %v", err) + } + got, err := engine.GetDocument(ctx, testDocID) + if err != nil { + t.Fatal(err) + } + if got.CurrentState != docgov.StateApproved { + t.Fatalf("state = %s, want approved", got.CurrentState) + } +}