weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(docgov): surface lease/audit persist failures and read back fencing token

Browse Source 
- persistLeaseLocked now returns error; AcquireLease rolls back in-memory
  lease on DB failure so memory and PG stay consistent
- appendAuditLocked uses request ctx and logs audit INSERT failures
  instead of swallowing them with context.Background()
- loadDocumentFromDBLocked now SELECTs lease_fencing_token so fencing
  survives process restart
- SaveDocumentContent/RecordDocumentViewed/RecordDocumentEdited/
  DecideModificationProposal/SetDocumentWorkDir log persist failures
  via slog.Error instead of silent _, _

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-19 04:50:40 +00:00
co-authored by Cursor
parent e8ed33b0aa
commit 11f375f53d
2 changed files with 78 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/docgov/engine.go
+34 -18
View File
@@ -315,11 +315,13 @@ func (e *GovernanceEngine) SetDocumentWorkDir(ctx context.Context, docID string,
doc.UpdatedAt = time.Now()
if e.pool != nil {
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
UPDATE document_records
SET work_dir = $1, updated_at = $2
WHERE doc_id = $3
`, doc.WorkDir, doc.UpdatedAt, docID)
`, doc.WorkDir, doc.UpdatedAt, docID); err != nil {
slog.Error("docgov_workdir_persist_failed", "doc_id", docID, "error", err)
}
}
return nil
@@ -389,7 +391,7 @@ func (e *GovernanceEngine) RecordDocumentViewed(ctx context.Context, docID strin
e.memAudits[docID] = append(e.memAudits[docID], audit)
if e.pool != nil {
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
UPDATE document_records
SET author_claimed = $1, author_claimed_at = $2, author_last_viewed_at = $3,
reviewer_claimed = $4, reviewer_claimed_at = $5,
@@ -399,14 +401,18 @@ func (e *GovernanceEngine) RecordDocumentViewed(ctx context.Context, docID strin
`, doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastViewedAt,
doc.ReviewerClaimed, doc.ReviewerClaimedAt,
doc.ApproverClaimed, doc.ApproverClaimedAt,
doc.UpdatedAt, docID)
doc.UpdatedAt, docID); err != nil {
slog.Error("docgov_viewed_persist_failed", "doc_id", docID, "error", err)
}
detJSON, _ := json.Marshal(audit.Details)
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole),
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt)
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil {
slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err)
}
}
return doc, nil
@@ -467,18 +473,22 @@ func (e *GovernanceEngine) RecordDocumentEdited(ctx context.Context, docID strin
e.memAudits[docID] = append(e.memAudits[docID], audit)
if e.pool != nil {
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
UPDATE document_records
SET author_claimed = $1, author_claimed_at = $2, author_last_edited_at = $3, updated_at = $4
WHERE doc_id = $5
`, doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, docID)
`, doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, docID); err != nil {
slog.Error("docgov_edited_persist_failed", "doc_id", docID, "error", err)
}
detJSON, _ := json.Marshal(audit.Details)
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole),
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt)
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil {
slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err)
}
}
return doc, nil
@@ -647,11 +657,13 @@ func (e *GovernanceEngine) ProposeModification(
}
detJSON, _ := json.Marshal(audit.Details)
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole),
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt)
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil {
slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err)
}
}
return proposal, nil
@@ -740,18 +752,22 @@ func (e *GovernanceEngine) DecideModificationProposal(
e.memAudits[proposal.DocID] = append(e.memAudits[proposal.DocID], audit)
if e.pool != nil {
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
UPDATE document_change_proposals
SET status = $1, author_decision_reason = $2, decided_at = $3
WHERE proposal_id = $4
`, string(proposal.Status), proposal.AuthorDecisionReason, proposal.DecidedAt, proposalID)
`, string(proposal.Status), proposal.AuthorDecisionReason, proposal.DecidedAt, proposalID); err != nil {
slog.Error("docgov_proposal_decide_persist_failed", "proposal_id", proposalID, "error", err)
}
detJSON, _ := json.Marshal(audit.Details)
_, _ = e.pool.Exec(ctx, `
if _, err := e.pool.Exec(ctx, `
INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole),
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt)
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil {
slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err)
}
}
return proposal, nil
@@ -1087,7 +1103,7 @@ func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID s
last_rejected_reason, last_rejected_by_role, last_rejected_by_name, last_rejected_at,
approved_at, config, created_at, updated_at,
COALESCE(lease_holder_type, ''), COALESCE(lease_holder_id, ''), COALESCE(lease_work_dir, ''),
lease_expires_at, lease_heartbeat_at,
lease_expires_at, lease_heartbeat_at, COALESCE(lease_fencing_token, ''),
COALESCE(duty_locked, false), COALESCE(duty_policy, '')
FROM document_records
WHERE doc_id = $1
@@ -1106,7 +1122,7 @@ func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID s
&doc.ReviewerClaimed, &doc.ReviewerClaimedAt, &doc.ApproverClaimed, &doc.ApproverClaimedAt,
&doc.LastRejectedReason, &doc.LastRejectedByRole, &doc.LastRejectedByName, &doc.LastRejectedAt,
&doc.ApprovedAt, &cfgJSON, &doc.CreatedAt, &doc.UpdatedAt,
&leaseType, &leaseID, &leaseWD, &doc.LeaseExpiresAt, &doc.LeaseHeartbeatAt,
&leaseType, &leaseID, &leaseWD, &doc.LeaseExpiresAt, &doc.LeaseHeartbeatAt, &doc.LeaseFencingToken,
&doc.DutyLocked, &doc.DutyPolicy,
)
if err != nil {
internal/docgov/lease.go
+44 -18
View File
@@ -7,6 +7,7 @@ import (
"encoding/json"
"fmt"
"io"
"log/slog"
"path/filepath"
"strings"
"time"
@@ -72,14 +73,16 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq
if doc.HasActiveLease(now) {
sameHolder := doc.LeaseHolderType == req.HolderType && doc.LeaseHolderID == req.HolderID
if !sameHolder && req.HolderType == LeaseHolderSquad && e.shouldReclaimSquadLease(ctx, doc) {
e.appendAuditLocked(doc, "lease_reclaim_orphan", 0, string(req.HolderType), RoleSystem, map[string]any{
e.appendAuditLocked(ctx, doc, "lease_reclaim_orphan", 0, string(req.HolderType), RoleSystem, map[string]any{
"previous_holder_type": doc.LeaseHolderType,
"previous_holder_id": doc.LeaseHolderID,
})
doc.ClearLease()
e.persistLeaseLocked(ctx, doc)
if perr := e.persistLeaseLocked(ctx, doc); perr != nil {
slog.Error("docgov_lease_reclaim_persist_failed", "doc_id", doc.DocID, "error", perr)
}
} else if !sameHolder {
e.appendAuditLocked(doc, "lease_deny", 0, string(req.HolderType), RoleSystem, map[string]any{
e.appendAuditLocked(ctx, doc, "lease_deny", 0, string(req.HolderType), RoleSystem, map[string]any{
"requester_type": req.HolderType,
"requester_id": req.HolderID,
"holder_type": doc.LeaseHolderType,
@@ -98,6 +101,9 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq
}
expires := now.Add(ttl)
// 先持久化再更新内存，失败时不污染内存租约状态
prevType, prevID, prevWD := doc.LeaseHolderType, doc.LeaseHolderID, doc.LeaseWorkDir
prevExpires, prevHeartbeat, prevFencing := doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken
doc.LeaseHolderType = req.HolderType
doc.LeaseHolderID = req.HolderID
if req.WorkDir != "" {
@@ -108,8 +114,12 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq
doc.LeaseFencingToken = newLeaseFencingToken()
doc.UpdatedAt = now
e.persistLeaseLocked(ctx, doc)
e.appendAuditLocked(doc, "lease_acquire", 0, string(req.HolderType), RoleSystem, map[string]any{
if perr := e.persistLeaseLocked(ctx, doc); perr != nil {
doc.LeaseHolderType, doc.LeaseHolderID, doc.LeaseWorkDir = prevType, prevID, prevWD
doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken = prevExpires, prevHeartbeat, prevFencing
return nil, perr
}
e.appendAuditLocked(ctx, doc, "lease_acquire", 0, string(req.HolderType), RoleSystem, map[string]any{
"holder_type": req.HolderType,
"holder_id": req.HolderID,
"work_dir": doc.LeaseWorkDir,
@@ -134,7 +144,9 @@ func (e *GovernanceEngine) ReleaseLease(ctx context.Context, docID, holderType,
if !doc.HasActiveLease(now) {
doc.ClearLease()
doc.UpdatedAt = now
e.persistLeaseLocked(ctx, doc)
if perr := e.persistLeaseLocked(ctx, doc); perr != nil {
slog.Error("docgov_lease_release_persist_failed", "doc_id", doc.DocID, "error", perr)
}
return doc, nil
}
@@ -144,8 +156,10 @@ func (e *GovernanceEngine) ReleaseLease(ctx context.Context, docID, holderType,
doc.ClearLease()
doc.UpdatedAt = now
e.persistLeaseLocked(ctx, doc)
e.appendAuditLocked(doc, "lease_release", 0, holderType, RoleSystem, map[string]any{
if perr := e.persistLeaseLocked(ctx, doc); perr != nil {
slog.Error("docgov_lease_release_persist_failed", "doc_id", doc.DocID, "error", perr)
}
e.appendAuditLocked(ctx, doc, "lease_release", 0, holderType, RoleSystem, map[string]any{
"holder_type": holderType,
"holder_id": holderID,
})
@@ -181,7 +195,9 @@ func (e *GovernanceEngine) HeartbeatLease(ctx context.Context, docID, holderType
doc.LeaseExpiresAt = &expires
doc.LeaseHeartbeatAt = &now
doc.UpdatedAt = now
e.persistLeaseLocked(ctx, doc)
if perr := e.persistLeaseLocked(ctx, doc); perr != nil {
slog.Error("docgov_lease_heartbeat_persist_failed", "doc_id", doc.DocID, "error", perr)
}
return doc, nil
}
@@ -402,16 +418,19 @@ func (e *GovernanceEngine) SaveDocumentContent(ctx context.Context, req ContentS
}
if e.pool != nil {
_, _ = e.pool.Exec(ctx, `
if _, perr := e.pool.Exec(ctx, `
UPDATE document_records
SET content_sha256 = $1, is_template_only = $2, work_dir = $3,
author_claimed = $4, author_claimed_at = $5, author_last_edited_at = $6, updated_at = $7
WHERE doc_id = $8
`, doc.ContentSHA256, doc.IsTemplateOnly, doc.WorkDir,
doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, doc.DocID)
doc.AuthorClaimed, doc.AuthorClaimedAt, doc.AuthorLastEditedAt, doc.UpdatedAt, doc.DocID); perr != nil {
slog.Error("docgov_content_persist_failed", "doc_id", doc.DocID, "error", perr)
return nil, fmt.Errorf("docgov: persist content: %w", perr)
}
}
e.appendAuditLocked(doc, "author_edit_content", req.ActorID, req.ActorName, req.ActorRole, map[string]any{
e.appendAuditLocked(ctx, doc, "author_edit_content", req.ActorID, req.ActorName, req.ActorRole, map[string]any{
"content_sha256": shaStr,
"bytes": len(canon),
})
@@ -453,20 +472,25 @@ func (e *GovernanceEngine) getDocLocked(ctx context.Context, docID string) (*Doc
return doc, nil
}
func (e *GovernanceEngine) persistLeaseLocked(ctx context.Context, doc *DocumentRecord) {
func (e *GovernanceEngine) persistLeaseLocked(ctx context.Context, doc *DocumentRecord) error {
if e.pool == nil {
return
return nil
}
_, _ = e.pool.Exec(ctx, `
_, err := e.pool.Exec(ctx, `
UPDATE document_records
SET lease_holder_type = $1, lease_holder_id = $2, lease_work_dir = $3,
lease_expires_at = $4, lease_heartbeat_at = $5, lease_fencing_token = $6, updated_at = $7
WHERE doc_id = $8
`, string(doc.LeaseHolderType), doc.LeaseHolderID, doc.LeaseWorkDir,
doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken, doc.UpdatedAt, doc.DocID)
if err != nil {
slog.Error("docgov_lease_persist_failed", "doc_id", doc.DocID, "holder", doc.LeaseHolderID, "error", err)
return fmt.Errorf("docgov: persist lease: %w", err)
}
return nil
}
func (e *GovernanceEngine) appendAuditLocked(doc *DocumentRecord, action string, actorID int64, actorName string, actorRole UserRole, details map[string]any) {
func (e *GovernanceEngine) appendAuditLocked(ctx context.Context, doc *DocumentRecord, action string, actorID int64, actorName string, actorRole UserRole, details map[string]any) {
now := time.Now()
audit := &AuditLog{
DocID: doc.DocID,
@@ -482,10 +506,12 @@ func (e *GovernanceEngine) appendAuditLocked(doc *DocumentRecord, action string,
e.memAudits[doc.DocID] = append(e.memAudits[doc.DocID], audit)
if e.pool != nil {
detJSON, _ := json.Marshal(details)
_, _ = e.pool.Exec(context.Background(), `
if _, err := e.pool.Exec(ctx, `
INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole),
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt)
string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt); err != nil {
slog.Error("docgov_audit_persist_failed", "doc_id", audit.DocID, "action", audit.Action, "error", err)
}
}
}