From a4a1388b5e50e47b74a45dc970b8ed0f471db46f Mon Sep 17 00:00:00 2001
From: Maciej Kulawik <mkulawik@offchainlabs.com>
Date: Fri, 17 Nov 2023 16:20:58 +0000
Subject: [PATCH] trie/triedb/hashdb: lock db lock when accessing dirties cache

Add read locking of db lock around access to dirties cache in hashdb.Database to prevent data race versus hashdb.Database.dereference which can modify the dirities map by deleting an item.

Fixes #28541
---
 trie/triedb/hashdb/database.go | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/trie/triedb/hashdb/database.go b/trie/triedb/hashdb/database.go
index 764ab24ec8..a90c100ebc 100644
--- a/trie/triedb/hashdb/database.go
+++ b/trie/triedb/hashdb/database.go
@@ -352,20 +352,25 @@ func (db *Database) Cap(limit common.StorageSize) error {
 	// outside code doesn't see an inconsistent state (referenced data removed from
 	// memory cache during commit but not yet in persistent storage). This is ensured
 	// by only uncaching existing data when the database write finalizes.
-	nodes, storage, start := len(db.dirties), db.dirtiesSize, time.Now()
+	start := time.Now()
 	batch := db.diskdb.NewBatch()
+	db.lock.RLock()
+	nodes, storage := len(db.dirties), db.dirtiesSize
 
 	// db.dirtiesSize only contains the useful data in the cache, but when reporting
 	// the total memory consumption, the maintenance metadata is also needed to be
 	// counted.
 	size := db.dirtiesSize + common.StorageSize(len(db.dirties)*cachedNodeSize)
+	db.lock.RUnlock()
 	size += db.childrenSize
 
 	// Keep committing nodes from the flush-list until we're below allowance
 	oldest := db.oldest
 	for size > limit && oldest != (common.Hash{}) {
 		// Fetch the oldest referenced node and push into the batch
+		db.lock.RLock()
 		node := db.dirties[oldest]
+		db.lock.RUnlock()
 		rawdb.WriteLegacyTrieNode(batch, oldest, node.node)
 
 		// If we exceeded the ideal batch size, commit and reset
@@ -436,7 +441,9 @@ func (db *Database) Commit(node common.Hash, report bool) error {
 	batch := db.diskdb.NewBatch()
 
 	// Move the trie itself into the batch, flushing if enough data is accumulated
+	db.lock.RLock()
 	nodes, storage := len(db.dirties), db.dirtiesSize
+	db.lock.RUnlock()
 
 	uncacher := &cleaner{db}
 	if err := db.commit(node, batch, uncacher); err != nil {
@@ -478,7 +485,9 @@ func (db *Database) Commit(node common.Hash, report bool) error {
 // commit is the private locked version of Commit.
 func (db *Database) commit(hash common.Hash, batch ethdb.Batch, uncacher *cleaner) error {
 	// If the node does not exist, it's a previously committed node
+	db.lock.RLock()
 	node, ok := db.dirties[hash]
+	db.lock.RUnlock()
 	if !ok {
 		return nil
 	}