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

fix(docgov): 保护共享内存 map，避免并发读写把进程打崩

Browse Source 
per-doc 锁只串行同一文档，不同 docID 仍裸读写 memDocs 等 map。GetDocument 与另一文档写入撞上时运行时 fatal，代理把全部 API 变成 500。
This commit is contained in:
weichen
2026-09-28 08:52:44 +00:00
parent a3db288a7a
commit a90cce73b9
12 changed files with 254 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/docgov/duty_project.go
+9 -8
View File
@@ -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
}
internal/docgov/engine.go
+5 -1
View File
@@ -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
internal/docgov/engine_claim.go
+6 -10
View File
@@ -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)
}
}
internal/docgov/engine_document.go
+4 -4
View File
@@ -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
}
internal/docgov/engine_proposal.go
+14 -36
View File
@@ -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
}
internal/docgov/engine_read.go
+8 -14
View File
@@ -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
}
internal/docgov/engine_review_flow.go
+6 -6
View File
@@ -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
}
internal/docgov/lease.go
+10 -9
View File
@@ -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, `
internal/docgov/lock_split.go
+130 -5
View File
@@ -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) {
internal/docgov/map_race_test.go
+49
View File
@@ -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)
}
}
}
internal/docgov/materialize.go
+10 -12
View File
@@ -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
}
internal/docgov/project_docs.go
+3 -3
View File
@@ -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
}