refactor inserting execution fork evidence

Author: zhangchiqing

module/mempool/consensus/exec_fork_suppressor.go

@@ -46,6 +46,7 @@ type ExecForkSuppressor struct {
 	execForkDetected      atomic.Bool
 	onExecFork            ExecForkActor
 	execForkEvidenceStore storage.ExecutionForkEvidence
+	lockManager           storage.LockManager
 	log                   zerolog.Logger
 }
 
@@ -61,6 +62,7 @@ func NewExecStateForkSuppressor(
 	seals mempool.IncorporatedResultSeals,
 	onExecFork ExecForkActor,
 	db storage.DB,
+	lockManager storage.LockManager,
 	log zerolog.Logger,
 ) (*ExecForkSuppressor, error) {
 	executionForkEvidenceStore := store.NewExecutionForkEvidence(db)
@@ -83,6 +85,7 @@ func NewExecStateForkSuppressor(
 		execForkDetected:      *atomic.NewBool(execForkDetectedFlag),
 		onExecFork:            onExecFork,
 		execForkEvidenceStore: executionForkEvidenceStore,
+		lockManager:           lockManager,
 		log:                   log.With().Str("mempool", "ExecForkSuppressor").Logger(),
 	}
 

storage/execution_fork_evidence.go

@@ -1,6 +1,9 @@
 package storage
 
-import "github.com/onflow/flow-go/model/flow"
+import (
+	"github.com/jordanschalm/lockctx"
+	"github.com/onflow/flow-go/model/flow"
+)
 
 // ExecutionForkEvidence represents persistent storage for execution fork evidence.
 // CAUTION: Not safe for concurrent use by multiple goroutines.
@@ -9,8 +12,9 @@ type ExecutionForkEvidence interface {
 	// if no execution fork evidence is currently stored in the database.
 	// This function is a no-op if evidence is already stored, because
 	// only one execution fork evidence can be stored at a time.
+	// The caller must hold the [storage.LockInsertExecutionForkEvidence] lock.
 	// No errors are expected during normal operations.
-	StoreIfNotExists(conflictingSeals []*flow.IncorporatedResultSeal) error
+	StoreIfNotExists(lctx lockctx.Proof, conflictingSeals []*flow.IncorporatedResultSeal) error
 
 	// Retrieve reads conflicting seals from the database.
 	// No error is returned if database record doesn't exist.

storage/locks.go

@@ -29,6 +29,8 @@ const (
 	LockBootstrapping = "lock_bootstrapping"
 	// LockInsertChunkDataPack protects the insertion of chunk data packs (not yet used anywhere
 	LockInsertChunkDataPack = "lock_insert_chunk_data_pack"
+	// LockInsertExecutionForkEvidence protects the insertion of execution fork evidence
+	LockInsertExecutionForkEvidence = "lock_insert_execution_fork_evidence"
 )
 
 // Locks returns a list of all named locks used by the storage layer.
@@ -42,6 +44,7 @@ func Locks() []string {
 		LockInsertCollection,
 		LockBootstrapping,
 		LockInsertChunkDataPack,
+		LockInsertExecutionForkEvidence,
 	}
 }
 

storage/operation/execution_fork_evidence.go

@@ -1,6 +1,9 @@
 package operation
 
 import (
+	"fmt"
+
+	"github.com/jordanschalm/lockctx"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/storage"
 )
@@ -28,8 +31,24 @@ func RemoveExecutionForkEvidence(w storage.Writer) error {
 }
 
 // InsertExecutionForkEvidence upserts conflicting seals to the database.
-// If a record already exists, it is overwritten; otherwise a new record is created.
+// If a record already exists, it is NOT overwritten, the new record is ignored.
+// The caller must hold the [storage.LockInsertExecutionForkEvidence] lock.
 // No errors are expected during normal operations.
-func InsertExecutionForkEvidence(w storage.Writer, conflictingSeals []*flow.IncorporatedResultSeal) error {
-	return UpsertByKey(w, MakePrefix(codeExecutionFork), conflictingSeals)
+func InsertExecutionForkEvidence(lctx lockctx.Proof, rw storage.ReaderBatchWriter, conflictingSeals []*flow.IncorporatedResultSeal) error {
+	if !lctx.HoldsLock(storage.LockInsertExecutionForkEvidence) {
+		return fmt.Errorf("InsertExecutionForkEvidence requires LockInsertBlock to be held")
+	}
+	key := MakePrefix(codeExecutionFork)
+	exist, err := KeyExists(rw.GlobalReader(), MakePrefix(codeExecutionFork))
+	if err != nil {
+		return fmt.Errorf("failed to check if execution fork evidence exists: %w", err)
+	}
+
+	if exist {
+		// Some evidence about execution fork already stored;
+		// We only keep the first evidence => nothing more to do
+		return nil
+	}
+
+	return UpsertByKey(rw.Writer(), key, conflictingSeals)
 }

storage/store/execution_fork_evidence.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/jordanschalm/lockctx"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/operation"
@@ -23,28 +24,11 @@ func NewExecutionForkEvidence(db storage.DB) *ExecutionForkEvidence {
 
 // StoreIfNotExists stores the given conflictingSeals to the database.
 // This is a no-op if there is already a record in the database with the same key.
+// The caller must hold the [storage.LockInsertExecutionForkEvidence] lock.
 // No errors are expected during normal operations.
-// CAUTION: This function is not safe for concurrent use by multiple goroutines.  For safety,
-// we rely on the fact that Execution Fork Evidence has a very small surface area of use
-// in the Execution Fork Suppressor.  The Execution Fork Suppressor is responsible for
-// ensuring mutually exclusive access to this function.
-func (efe *ExecutionForkEvidence) StoreIfNotExists(conflictingSeals []*flow.IncorporatedResultSeal) error {
+func (efe *ExecutionForkEvidence) StoreIfNotExists(lctx lockctx.Proof, conflictingSeals []*flow.IncorporatedResultSeal) error {
 	return efe.db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-		found, err := operation.HasExecutionForkEvidence(rw.GlobalReader())
-		if err != nil {
-			return fmt.Errorf("failed to check if evidence about execution fork exists: %w", err)
-		}
-		if found {
-			// Some evidence about execution fork already stored;
-			// We only keep the first evidence => nothing more to do
-			return nil
-		}
-
-		err = operation.InsertExecutionForkEvidence(rw.Writer(), conflictingSeals)
-		if err != nil {
-			return fmt.Errorf("failed to store evidence about execution fork: %w", err)
-		}
-		return nil
+		return operation.InsertExecutionForkEvidence(lctx, rw, conflictingSeals)
 	})
 }