From a90cce73b905116bfd10eb4bde957baf60bb45b8 Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 28 Sep 2026 08:52:44 +0000 Subject: [PATCH] =?UTF-8?q?fix(docgov):=20=E4=BF=9D=E6=8A=A4=E5=85=B1?= =?UTF-8?q?=E4=BA=AB=E5=86=85=E5=AD=98=20map=EF=BC=8C=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E8=AF=BB=E5=86=99=E6=8A=8A=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E6=89=93=E5=B4=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit per-doc 锁只串行同一文档，不同 docID 仍裸读写 memDocs 等 map。GetDocument 与另一文档写入撞上时运行时 fatal，代理把全部 API 变成 500。 --- internal/docgov/duty_project.go | 17 ++-- internal/docgov/engine.go | 6 +- internal/docgov/engine_claim.go | 16 ++- internal/docgov/engine_document.go | 8 +- internal/docgov/engine_proposal.go | 50 +++------- internal/docgov/engine_read.go | 22 ++--- internal/docgov/engine_review_flow.go | 12 +-- internal/docgov/lease.go | 19 ++-- internal/docgov/lock_split.go | 135 +++++++++++++++++++++++++- internal/docgov/map_race_test.go | 49 ++++++++++ internal/docgov/materialize.go | 22 ++--- internal/docgov/project_docs.go | 6 +- 12 files changed, 254 insertions(+), 108 deletions(-) create mode 100644 internal/docgov/map_race_test.go diff --git a/internal/docgov/duty_project.go b/internal/docgov/duty_project.go index 66e1d471..c7f172a6 100644 --- a/internal/docgov/duty_project.go +++ b/internal/docgov/duty_project.go @@ -35,11 +35,12 @@ func (e *GovernanceEngine) FindByProjectTaskKey(ctx context.Context, projectID i } } - for _, doc := range e.memDocs { - if doc != nil && doc.ProjectID == projectID && doc.TaskKey == taskKey { - cp := *doc - return &cp, nil - } + doc := e.memDocsFind(func(doc *DocumentRecord) bool { + return doc.ProjectID == projectID && doc.TaskKey == taskKey + }) + if doc != nil { + cp := *doc + return &cp, nil } return nil, ErrDocumentNotFound } @@ -94,9 +95,9 @@ func (e *GovernanceEngine) ProjectDocTaskDuties( } } - for _, doc := range e.memDocs { + e.forEachMemDoc(func(doc *DocumentRecord) { if doc == nil || doc.ProjectID != projectID || doc.TaskKey != taskKey { - continue + return } doc.AuthorID = authorID doc.AuthorName = authorName @@ -106,6 +107,6 @@ func (e *GovernanceEngine) ProjectDocTaskDuties( doc.ApproverName = approverName doc.DutyPolicy = strings.TrimSpace(dutyPolicy) doc.UpdatedAt = now - } + }) return nil } diff --git a/internal/docgov/engine.go b/internal/docgov/engine.go index 489476f0..0ec1d892 100644 --- a/internal/docgov/engine.go +++ b/internal/docgov/engine.go @@ -13,7 +13,11 @@ type GovernanceEngine struct { pool *pgxpool.Pool sm *StateMachine - mu sync.RWMutex + mu sync.RWMutex + // memMu 只保护下面四张共享 map 的结构。per-doc 锁串行同一文档的状态机， + // 不同 docID 仍会并发读写这些 map；T20 拆锁后漏了这层，会导致进程崩溃。 + // 锁序：docLocks → e.mu → memMu。持有 memMu 时不得再获取 docLocks 或 e.mu。 + memMu sync.RWMutex memDocs map[string]*DocumentRecord memComms map[string][]*DocumentComment memProps map[string]*ChangeProposal diff --git a/internal/docgov/engine_claim.go b/internal/docgov/engine_claim.go index 6caf5a03..0e0ded48 100644 --- a/internal/docgov/engine_claim.go +++ b/internal/docgov/engine_claim.go @@ -80,7 +80,7 @@ func (e *GovernanceEngine) RecordDocumentViewed(ctx context.Context, docID strin }, CreatedAt: now, } - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if e.pool != nil { if _, err := e.pool.Exec(ctx, ` @@ -158,7 +158,7 @@ func (e *GovernanceEngine) RecordDocumentEdited(ctx context.Context, docID strin Details: map[string]any{"timestamp": now}, CreatedAt: now, } - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if e.pool != nil { if _, err := e.pool.Exec(ctx, ` @@ -235,13 +235,9 @@ func (e *GovernanceEngine) CheckModificationPermission(ctx context.Context, docI } // 3. 已认领：检查是否有待编制人处理的挂起提案 - var pendingProp *ChangeProposal - for _, p := range e.memProps { - if p.DocID == docID && p.Status == ProposalPendingAuthorApproval { - pendingProp = p - break - } - } + pendingProp := e.memPropsFind(func(p *ChangeProposal) bool { + return p.DocID == docID && p.Status == ProposalPendingAuthorApproval + }) if pendingProp == nil && e.pool != nil { var p ChangeProposal var stat string @@ -254,7 +250,7 @@ func (e *GovernanceEngine) CheckModificationPermission(ctx context.Context, docI if err == nil { p.Status = ProposalStatus(stat) pendingProp = &p - e.memProps[p.ProposalID] = &p + e.memPropsPut(p.ProposalID, &p) } } diff --git a/internal/docgov/engine_document.go b/internal/docgov/engine_document.go index dc85a273..9f1a0ee8 100644 --- a/internal/docgov/engine_document.go +++ b/internal/docgov/engine_document.go @@ -28,7 +28,7 @@ func (e *GovernanceEngine) CreateOrGetDocument( defer g.Release() now := time.Now() - if doc, exists := e.memDocs[docID]; exists { + if doc, exists := e.memDocsGet(docID); exists { return doc, nil } @@ -123,7 +123,7 @@ func (e *GovernanceEngine) CreateOrGetDocument( doc.ApproverClaimed = appClaim } - e.memDocs[docID] = doc + e.memDocsPut(docID, doc) return doc, nil } @@ -197,7 +197,7 @@ func (e *GovernanceEngine) MarkDocumentContentUpdated(ctx context.Context, docID } func (e *GovernanceEngine) ensureDocumentLocked(ctx context.Context, docID string) (*DocumentRecord, error) { - if doc, exists := e.memDocs[docID]; exists { + if doc, exists := e.memDocsGet(docID); exists { return doc, nil } if e.pool != nil { @@ -214,7 +214,7 @@ func (e *GovernanceEngine) SetDocumentWorkDir(ctx context.Context, docID string, g := e.acquireDocLock(docID) defer g.Release() - doc, exists := e.memDocs[docID] + doc, exists := e.memDocsGet(docID) if !exists { return ErrDocumentNotFound } diff --git a/internal/docgov/engine_proposal.go b/internal/docgov/engine_proposal.go index b122bbab..9eb1b5b3 100644 --- a/internal/docgov/engine_proposal.go +++ b/internal/docgov/engine_proposal.go @@ -41,7 +41,7 @@ func (e *GovernanceEngine) ProposeModification( CreatedAt: now, } - e.memProps[proposalID] = proposal + e.memPropsPut(proposalID, proposal) // 记录审计 audit := &AuditLog{ @@ -60,7 +60,7 @@ func (e *GovernanceEngine) ProposeModification( }, CreatedAt: now, } - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if e.pool != nil { query := ` @@ -103,7 +103,7 @@ func (e *GovernanceEngine) DecideModificationProposal( g := e.acquireDocLock("*proposal:" + proposalID) defer g.Release() - proposal, exists := e.memProps[proposalID] + proposal, exists := e.memPropsGet(proposalID) if !exists && e.pool != nil { var p ChangeProposal var stat string @@ -115,7 +115,7 @@ func (e *GovernanceEngine) DecideModificationProposal( if err == nil { p.Status = ProposalStatus(stat) proposal = &p - e.memProps[proposalID] = &p + e.memPropsPut(proposalID, &p) exists = true } } @@ -126,7 +126,7 @@ func (e *GovernanceEngine) DecideModificationProposal( return nil, ErrProposalAlreadyDecided } - doc, docExists := e.memDocs[proposal.DocID] + doc, docExists := e.memDocsGet(proposal.DocID) if !docExists && e.pool != nil { var err error doc, err = e.loadDocumentFromDBLocked(ctx, proposal.DocID) @@ -172,7 +172,7 @@ func (e *GovernanceEngine) DecideModificationProposal( }, CreatedAt: now, } - e.memAudits[proposal.DocID] = append(e.memAudits[proposal.DocID], audit) + e.memAuditsAppend(proposal.DocID, audit) if e.pool != nil { if _, err := e.pool.Exec(ctx, ` @@ -230,7 +230,7 @@ func (e *GovernanceEngine) AddComment( g := e.acquireDocLock(docID) defer g.Release() - doc, exists := e.memDocs[docID] + doc, exists := e.memDocsGet(docID) if !exists { var err error doc, err = e.loadDocumentFromDBLocked(ctx, docID) @@ -264,20 +264,11 @@ func (e *GovernanceEngine) AddComment( } if e.pool == nil { - // in-memory: allocate synthetic ids so ResolveComment can address them - var maxID int64 - for _, list := range e.memComms { - for _, c := range list { - if c != nil && c.ID > maxID { - maxID = c.ID - } - } - } - comment.ID = maxID + 1 + e.memCommsAppend(docID, comment, true) + } else { + e.memCommsAppend(docID, comment, false) } - e.memComms[docID] = append(e.memComms[docID], comment) - if e.pool != nil { query := ` INSERT INTO document_comments (doc_id, user_id, user_name, user_role, target_section, comment_type, content, status, created_at) @@ -304,20 +295,7 @@ func (e *GovernanceEngine) ResolveComment(ctx context.Context, commentID int64, e.mu.Lock() defer e.mu.Unlock() - var found *DocumentComment - var docID string - for id, list := range e.memComms { - for _, c := range list { - if c != nil && c.ID == commentID { - found = c - docID = id - break - } - } - if found != nil { - break - } - } + found, docID := e.memCommsFind(commentID) if found == nil && e.pool != nil { var c DocumentComment var role, cType string @@ -334,13 +312,13 @@ func (e *GovernanceEngine) ResolveComment(ctx context.Context, commentID int64, c.CommentType = CommentType(cType) found = &c docID = c.DocID - e.memComms[docID] = append(e.memComms[docID], found) + e.memCommsAppend(docID, found, false) } if found == nil { return nil, fmt.Errorf("docgov: comment not found") } - doc, exists := e.memDocs[docID] + doc, exists := e.memDocsGet(docID) if !exists { var err error doc, err = e.loadDocumentFromDBLocked(ctx, docID) @@ -387,6 +365,6 @@ func (e *GovernanceEngine) addCommentInternal(docID string, userID int64, userNa Status: "open", CreatedAt: now, } - e.memComms[docID] = append(e.memComms[docID], comment) + e.memCommsAppend(docID, comment, false) return nil } diff --git a/internal/docgov/engine_read.go b/internal/docgov/engine_read.go index cb9f0000..d9ba6042 100644 --- a/internal/docgov/engine_read.go +++ b/internal/docgov/engine_read.go @@ -11,7 +11,7 @@ func (e *GovernanceEngine) GetDocument(ctx context.Context, docID string) (*Docu g := e.acquireDocLock(docID) defer g.Release() - if doc, exists := e.memDocs[docID]; exists { + if doc, exists := e.memDocsGet(docID); exists { return doc, nil } if e.pool != nil { @@ -27,7 +27,7 @@ func (e *GovernanceEngine) ListAuditLogs(ctx context.Context, docID string) ([]* g := e.acquireDocLock(docID) defer g.Release() - if len(e.memAudits[docID]) == 0 && e.pool != nil { + if e.memAuditsEmpty(docID) && e.pool != nil { rows, err := e.pool.Query(ctx, ` SELECT id, doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at FROM document_audit_logs @@ -51,7 +51,7 @@ func (e *GovernanceEngine) ListAuditLogs(ctx context.Context, docID string) ([]* if len(detJSON) > 0 { _ = json.Unmarshal(detJSON, &a.Details) } - e.memAudits[docID] = append(e.memAudits[docID], &a) + e.memAuditsAppend(docID, &a) } } if err := rows.Err(); err != nil { @@ -60,10 +60,7 @@ func (e *GovernanceEngine) ListAuditLogs(ctx context.Context, docID string) ([]* } } - logs := e.memAudits[docID] - res := make([]*AuditLog, len(logs)) - copy(res, logs) - return res, nil + return e.memAuditsCopy(docID), nil } // ListComments 列出文档的所有批注 @@ -71,7 +68,7 @@ func (e *GovernanceEngine) ListComments(ctx context.Context, docID string) ([]*D g := e.acquireDocLock(docID) defer g.Release() - if len(e.memComms[docID]) == 0 && e.pool != nil { + if e.memCommsEmpty(docID) && e.pool != nil { rows, err := e.pool.Query(ctx, ` SELECT id, doc_id, user_id, user_name, user_role, target_section, comment_type, content, status, created_at FROM document_comments @@ -90,7 +87,7 @@ func (e *GovernanceEngine) ListComments(ctx context.Context, docID string) ([]*D } c.UserRole = UserRole(role) c.CommentType = CommentType(cType) - e.memComms[docID] = append(e.memComms[docID], &c) + e.memCommsAppend(docID, &c, false) } } if err := rows.Err(); err != nil { @@ -99,10 +96,7 @@ func (e *GovernanceEngine) ListComments(ctx context.Context, docID string) ([]*D } } - comms := e.memComms[docID] - res := make([]*DocumentComment, len(comms)) - copy(res, comms) - return res, nil + return e.memCommsCopy(docID), nil } func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID string) (*DocumentRecord, error) { @@ -162,6 +156,6 @@ func (e *GovernanceEngine) loadDocumentFromDBLocked(ctx context.Context, docID s if len(cfgJSON) > 0 { _ = json.Unmarshal(cfgJSON, &doc.Config) } - e.memDocs[docID] = &doc + e.memDocsPut(docID, &doc) return &doc, nil } diff --git a/internal/docgov/engine_review_flow.go b/internal/docgov/engine_review_flow.go index 7108f50b..c3a360f0 100644 --- a/internal/docgov/engine_review_flow.go +++ b/internal/docgov/engine_review_flow.go @@ -47,7 +47,7 @@ func (e *GovernanceEngine) SubmitForReview(ctx context.Context, docID string, ac return nil, err } - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if err := e.freezeReviewSnapshotLocked(ctx, doc); err != nil { return nil, err } @@ -78,7 +78,7 @@ func (e *GovernanceEngine) ReviewApprove(ctx context.Context, docID string, acto _ = e.addCommentInternal(docID, actorID, actorName, actorRole, "", CommentSuggestion, comment) } - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil { return nil, err } @@ -105,7 +105,7 @@ func (e *GovernanceEngine) ReviewReject(ctx context.Context, docID string, actor // 驳回原因自动持久化为批注记录 _ = e.addCommentInternal(docID, actorID, actorName, actorRole, "", CommentRejection, rejectReason) - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil { return nil, err } @@ -140,7 +140,7 @@ func (e *GovernanceEngine) ApproverApprove(ctx context.Context, docID string, ac _ = e.addCommentInternal(docID, actorID, actorName, actorRole, "", CommentSuggestion, comment) } - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if err := e.persistApproveAndFAV(ctx, doc, audit); err != nil { return nil, err } @@ -181,7 +181,7 @@ func (e *GovernanceEngine) ApproverReject(ctx context.Context, docID string, act _ = e.addCommentInternal(docID, actorID, actorName, actorRole, "", CommentRejection, rejectReason) - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil { return nil, err } @@ -208,7 +208,7 @@ func (e *GovernanceEngine) ReopenForRevision(ctx context.Context, docID string, if err != nil { return nil, err } - e.memAudits[docID] = append(e.memAudits[docID], audit) + e.memAuditsAppend(docID, audit) if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil { return nil, err } diff --git a/internal/docgov/lease.go b/internal/docgov/lease.go index 3cfe4138..d03d0681 100644 --- a/internal/docgov/lease.go +++ b/internal/docgov/lease.go @@ -241,10 +241,11 @@ func (e *GovernanceEngine) FindApprovedProposal(ctx context.Context, docID strin g := e.acquireDocLock(docID) defer g.Release() - for _, p := range e.memProps { - if p.DocID == docID && p.Status == ProposalApprovedByAuthor { - return p, nil - } + p := e.memPropsFind(func(p *ChangeProposal) bool { + return p.DocID == docID && p.Status == ProposalApprovedByAuthor + }) + if p != nil { + return p, nil } if e.pool != nil { var p ChangeProposal @@ -258,7 +259,7 @@ func (e *GovernanceEngine) FindApprovedProposal(ctx context.Context, docID strin `, docID).Scan(&p.ID, &p.ProposalID, &p.DocID, &p.ProjectID, &p.TriggeredBy, &p.ProposedChanges, &p.ImpactAnalysis, &p.RiskAssessment, &stat, &p.AuthorDecisionReason, &p.DecidedAt, &p.CreatedAt) if err == nil { p.Status = ProposalStatus(stat) - e.memProps[p.ProposalID] = &p + e.memPropsPut(p.ProposalID, &p) return &p, nil } } @@ -271,7 +272,7 @@ func (e *GovernanceEngine) MarkProposalApplied(ctx context.Context, proposalID s g := e.acquireDocLock("*proposal:" + proposalID) defer g.Release() - p, ok := e.memProps[proposalID] + p, ok := e.memPropsGet(proposalID) if !ok && e.pool != nil { var loaded ChangeProposal var stat string @@ -284,7 +285,7 @@ func (e *GovernanceEngine) MarkProposalApplied(ctx context.Context, proposalID s } loaded.Status = ProposalStatus(stat) p = &loaded - e.memProps[proposalID] = p + e.memPropsPut(proposalID, p) } if p == nil { return ErrProposalNotFound @@ -468,7 +469,7 @@ func newLeaseFencingToken() string { } func (e *GovernanceEngine) getDocLocked(ctx context.Context, docID string) (*DocumentRecord, error) { - doc, exists := e.memDocs[docID] + doc, exists := e.memDocsGet(docID) if !exists { var err error doc, err = e.loadDocumentFromDBLocked(ctx, docID) @@ -573,7 +574,7 @@ func (e *GovernanceEngine) appendAuditLocked(ctx context.Context, doc *DocumentR Details: details, CreatedAt: now, } - e.memAudits[doc.DocID] = append(e.memAudits[doc.DocID], audit) + e.memAuditsAppend(doc.DocID, audit) if e.pool != nil { detJSON, _ := json.Marshal(details) if _, err := e.pool.Exec(ctx, ` diff --git a/internal/docgov/lock_split.go b/internal/docgov/lock_split.go index 8cff6749..11418f39 100644 --- a/internal/docgov/lock_split.go +++ b/internal/docgov/lock_split.go @@ -57,18 +57,18 @@ func (e *GovernanceEngine) withDocLock(docID string, fn func() error) error { // lookupMemDoc 返回内存中的文档记录（需在 doc 锁内调用）。 func (e *GovernanceEngine) lookupMemDoc(docID string) (*DocumentRecord, bool) { - doc, ok := e.memDocs[docID] - return doc, ok + return e.memDocsGet(docID) } // ensureDocLockedWithDB 在 doc 锁内做内存查找 + DB 兜底加载。 -// 对应原 ensureDocumentLocked，但锁粒度为单文档。 +// 对应原 ensureDocumentLocked，但锁粒度为单文档。map 访问走 memMu，不再重入 e.mu +// （SaveDocumentContent 等调用方可能已经持有 e.mu）。 func (e *GovernanceEngine) ensureDocLockedWithDB(ctx context.Context, docID string) (*DocumentRecord, error) { - if doc, ok := e.memDocs[docID]; ok { + if doc, ok := e.memDocsGet(docID); ok { return doc, nil } if e.pool != nil { - doc, err := e.loadDocumentFromDB(ctx, docID) + doc, err := e.loadDocumentFromDBLocked(ctx, docID) if err == nil && doc != nil { return doc, nil } @@ -76,6 +76,131 @@ func (e *GovernanceEngine) ensureDocLockedWithDB(ctx context.Context, docID stri return nil, ErrDocumentNotFound } +func (e *GovernanceEngine) memDocsGet(docID string) (*DocumentRecord, bool) { + e.memMu.RLock() + defer e.memMu.RUnlock() + doc, ok := e.memDocs[docID] + return doc, ok +} + +func (e *GovernanceEngine) memDocsPut(docID string, doc *DocumentRecord) { + e.memMu.Lock() + defer e.memMu.Unlock() + if e.memDocs == nil { + e.memDocs = map[string]*DocumentRecord{} + } + e.memDocs[docID] = doc +} + +func (e *GovernanceEngine) memDocsFind(match func(*DocumentRecord) bool) *DocumentRecord { + e.memMu.RLock() + defer e.memMu.RUnlock() + for _, doc := range e.memDocs { + if doc != nil && match(doc) { + return doc + } + } + return nil +} + +func (e *GovernanceEngine) forEachMemDoc(fn func(*DocumentRecord)) { + e.memMu.RLock() + defer e.memMu.RUnlock() + for _, doc := range e.memDocs { + fn(doc) + } +} + +func (e *GovernanceEngine) memAuditsEmpty(docID string) bool { + e.memMu.RLock() + defer e.memMu.RUnlock() + return len(e.memAudits[docID]) == 0 +} + +func (e *GovernanceEngine) memAuditsAppend(docID string, audit *AuditLog) { + e.memMu.Lock() + defer e.memMu.Unlock() + e.memAudits[docID] = append(e.memAudits[docID], audit) +} + +func (e *GovernanceEngine) memAuditsCopy(docID string) []*AuditLog { + e.memMu.RLock() + defer e.memMu.RUnlock() + logs := e.memAudits[docID] + res := make([]*AuditLog, len(logs)) + copy(res, logs) + return res +} + +func (e *GovernanceEngine) memCommsEmpty(docID string) bool { + e.memMu.RLock() + defer e.memMu.RUnlock() + return len(e.memComms[docID]) == 0 +} + +func (e *GovernanceEngine) memCommsAppend(docID string, comment *DocumentComment, assignSyntheticID bool) { + e.memMu.Lock() + defer e.memMu.Unlock() + if assignSyntheticID { + var maxID int64 + for _, list := range e.memComms { + for _, c := range list { + if c != nil && c.ID > maxID { + maxID = c.ID + } + } + } + comment.ID = maxID + 1 + } + e.memComms[docID] = append(e.memComms[docID], comment) +} + +func (e *GovernanceEngine) memCommsCopy(docID string) []*DocumentComment { + e.memMu.RLock() + defer e.memMu.RUnlock() + comms := e.memComms[docID] + res := make([]*DocumentComment, len(comms)) + copy(res, comms) + return res +} + +func (e *GovernanceEngine) memCommsFind(commentID int64) (*DocumentComment, string) { + e.memMu.RLock() + defer e.memMu.RUnlock() + for id, list := range e.memComms { + for _, c := range list { + if c != nil && c.ID == commentID { + return c, id + } + } + } + return nil, "" +} + +func (e *GovernanceEngine) memPropsGet(proposalID string) (*ChangeProposal, bool) { + e.memMu.RLock() + defer e.memMu.RUnlock() + p, ok := e.memProps[proposalID] + return p, ok +} + +func (e *GovernanceEngine) memPropsPut(proposalID string, proposal *ChangeProposal) { + e.memMu.Lock() + defer e.memMu.Unlock() + e.memProps[proposalID] = proposal +} + +func (e *GovernanceEngine) memPropsFind(match func(*ChangeProposal) bool) *ChangeProposal { + e.memMu.RLock() + defer e.memMu.RUnlock() + for _, p := range e.memProps { + if p != nil && match(p) { + return p + } + } + return nil +} + // loadDocumentFromDB 在无全局锁状态下从 PG 装载文档记录到内存。 // 这是 T20 B3 的核心：DB RTT 不再进入全局临界区。 func (e *GovernanceEngine) loadDocumentFromDB(ctx context.Context, docID string) (*DocumentRecord, error) { diff --git a/internal/docgov/map_race_test.go b/internal/docgov/map_race_test.go new file mode 100644 index 00000000..a732d5b2 --- /dev/null +++ b/internal/docgov/map_race_test.go @@ -0,0 +1,49 @@ +package docgov + +import ( + "context" + "fmt" + "sync" + "testing" +) + +// TestMemMapsNoRaceAcrossDocs 复现 2026-09-28 dev 崩溃：不同 docID 并发读写共享 map。 +// 修复前 go test -race 会报告 concurrent map read/write（运行时则 fatal）。 +func TestMemMapsNoRaceAcrossDocs(t *testing.T) { + e := NewGovernanceEngine(nil) + ctx := context.Background() + const n = 48 + var wg sync.WaitGroup + wg.Add(n * 3) + for i := 0; i < n; i++ { + id := fmt.Sprintf("race-doc-%d", i) + go func() { + defer wg.Done() + if _, err := e.CreateOrGetDocument(ctx, id, int64(i+1), "task", "title", "stage", "", 1, "author", 2, "reviewer", 3, "approver"); err != nil { + t.Errorf("create %s: %v", id, err) + } + }() + go func() { + defer wg.Done() + _, _ = e.GetDocument(ctx, id) + _, _ = e.ListAuditLogs(ctx, id) + _, _ = e.ListComments(ctx, id) + }() + go func() { + defer wg.Done() + _ = e.addCommentInternal(id, 1, "author", RoleAuthor, "", CommentSuggestion, "note") + }() + } + wg.Wait() + + for i := 0; i < n; i++ { + id := fmt.Sprintf("race-doc-%d", i) + doc, err := e.GetDocument(ctx, id) + if err != nil { + t.Fatalf("get %s: %v", id, err) + } + if doc.DocID != id { + t.Fatalf("doc id = %s, want %s", doc.DocID, id) + } + } +} diff --git a/internal/docgov/materialize.go b/internal/docgov/materialize.go index bbe0fdee..02aa29f4 100644 --- a/internal/docgov/materialize.go +++ b/internal/docgov/materialize.go @@ -111,7 +111,7 @@ func (e *GovernanceEngine) upsertTemplateDoc(ctx context.Context, projectID int6 e.mu.Lock() defer e.mu.Unlock() - if existing, ok := e.memDocs[docID]; ok && existing != nil { + if existing, ok := e.memDocsGet(docID); ok && existing != nil { if !existing.IsTemplateOnly { return nil } @@ -129,10 +129,7 @@ func (e *GovernanceEngine) upsertTemplateDoc(ctx context.Context, projectID int6 existing.UpdatedAt = now return nil } - if e.memDocs == nil { - e.memDocs = map[string]*DocumentRecord{} - } - e.memDocs[docID] = &DocumentRecord{ + e.memDocsPut(docID, &DocumentRecord{ DocID: docID, ProjectID: projectID, TaskKey: taskKey, @@ -152,7 +149,7 @@ func (e *GovernanceEngine) upsertTemplateDoc(ctx context.Context, projectID int6 Config: DefaultDocumentConfig(), CreatedAt: now, UpdatedAt: now, - } + }) return nil } @@ -202,15 +199,16 @@ func (e *GovernanceEngine) hasCanonicalSibling(ctx context.Context, projectID in } e.mu.RLock() defer e.mu.RUnlock() - for _, doc := range e.memDocs { - if doc == nil || doc.ProjectID != projectID || doc.DocID == docID { - continue + found := e.memDocsFind(func(doc *DocumentRecord) bool { + if doc.ProjectID != projectID || doc.DocID == docID { + return false } for _, key := range siblings { if doc.TaskKey == key { - return true, nil + return true } } - } - return false, nil + return false + }) + return found != nil, nil } diff --git a/internal/docgov/project_docs.go b/internal/docgov/project_docs.go index 7af79616..9e83ca1b 100644 --- a/internal/docgov/project_docs.go +++ b/internal/docgov/project_docs.go @@ -108,8 +108,8 @@ func (e *GovernanceEngine) ListWorkflowDocumentsByProject(ctx context.Context, p e.mu.Lock() defer e.mu.Unlock() out := make([]WorkflowDocumentSummary, 0) - for _, doc := range e.memDocs { - if doc.ProjectID == projectID && doc.TaskKey != "" { + e.forEachMemDoc(func(doc *DocumentRecord) { + if doc != nil && doc.ProjectID == projectID && doc.TaskKey != "" { out = append(out, WorkflowDocumentSummary{ DocID: doc.DocID, ProjectID: doc.ProjectID, @@ -133,7 +133,7 @@ func (e *GovernanceEngine) ListWorkflowDocumentsByProject(ctx context.Context, p DutyPolicy: doc.DutyPolicy, }) } - } + }) return collapseWorkflowDocuments(out), nil }