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).
2026-09-26 00:25:01 +00:00
parent 043c66ccfb
commit dc26321bab
2 changed files with 113 additions and 2 deletions
@@ -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)
 	}
@@ -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)
+	}
+}