diff --git a/internal/docgov/engine.go b/internal/docgov/engine.go index 7e066959..f0220df7 100644 --- a/internal/docgov/engine.go +++ b/internal/docgov/engine.go @@ -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 { diff --git a/internal/docgov/lease.go b/internal/docgov/lease.go index 73f017a1..c7e31233 100644 --- a/internal/docgov/lease.go +++ b/internal/docgov/lease.go @@ -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) + } } }