diff --git a/backend/pkg/tools/evidence_receipts.go b/backend/pkg/tools/evidence_receipts.go index 145ba4a9..50c26bb9 100644 --- a/backend/pkg/tools/evidence_receipts.go +++ b/backend/pkg/tools/evidence_receipts.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "hash/fnv" "io" "os" "path/filepath" @@ -95,7 +96,11 @@ type fileEvidenceReceiptRecorder struct { newID func() string } -var evidenceReceiptLocks sync.Map +// Fixed stripes rather than a per-path sync.Map that kept one *sync.Mutex per flow +// for the whole server uptime. +const evidenceReceiptLockShards = 256 + +var evidenceReceiptLocks [evidenceReceiptLockShards]sync.Mutex func newEvidenceReceiptRecorder(dataDir string, flowID int64, enabled bool) evidenceReceiptRecorder { if !enabled { @@ -212,8 +217,9 @@ func evidenceReceiptsPath(dataDir string, flowID int64) (string, error) { } func evidenceReceiptPathLock(path string) *sync.Mutex { - lock, _ := evidenceReceiptLocks.LoadOrStore(path, &sync.Mutex{}) - return lock.(*sync.Mutex) + h := fnv.New32a() + _, _ = h.Write([]byte(path)) + return &evidenceReceiptLocks[h.Sum32()%evidenceReceiptLockShards] } func readLastEvidenceReceiptHash(path string) (string, error) { diff --git a/backend/pkg/tools/evidence_receipts_test.go b/backend/pkg/tools/evidence_receipts_test.go index ee47eba0..e2422fde 100644 --- a/backend/pkg/tools/evidence_receipts_test.go +++ b/backend/pkg/tools/evidence_receipts_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" ) @@ -247,3 +248,47 @@ func readEvidenceReceiptLines(t *testing.T, path string) []evidenceReceipt { return receipts } + +func TestFileEvidenceReceiptRecorderSerializesConcurrentWritesToSamePath(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + const flowID = int64(4242) + const writers = 2 + const perWriter = 40 + + var wg sync.WaitGroup + for w := 0; w < writers; w++ { + recorder := newTestEvidenceReceiptRecorder(dir, flowID) + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < perWriter; i++ { + if err := recorder.RecordFinished(t.Context(), testEvidenceReceiptEvent()); err != nil { + t.Errorf("RecordFinished() error: %v", err) + return + } + } + }() + } + wg.Wait() + + path, err := evidenceReceiptsPath(dir, flowID) + if err != nil { + t.Fatalf("evidenceReceiptsPath() error: %v", err) + } + receipts := readEvidenceReceiptLines(t, path) + + if len(receipts) != writers*perWriter { + t.Fatalf("got %d receipts, want %d", len(receipts), writers*perWriter) + } + if receipts[0].PreviousReceiptHash != "" { + t.Fatalf("first receipt previous hash = %q, want empty", receipts[0].PreviousReceiptHash) + } + for i := 1; i < len(receipts); i++ { + if receipts[i].PreviousReceiptHash != receipts[i-1].ReceiptHash { + t.Fatalf("receipt %d chain broken: previous hash = %q, want %q (a concurrent write was not serialized)", + i, receipts[i].PreviousReceiptHash, receipts[i-1].ReceiptHash) + } + } +}