Merge branch 'master' into peter/7830-scheduled-tx-index

Author: peterargue

cmd/collection/main.go

@@ -636,6 +636,7 @@ func main() {
 				node.Logger,
 				node.Me,
 				node.ProtocolDB,
+				node.StorageLockMgr,
 				node.State,
 				node.Metrics.Engine,
 				node.Metrics.Mempool,

cmd/consensus/main.go

@@ -396,6 +396,7 @@ func main() {
 				multipleReceiptsFilterMempool,
 				consensusMempools.LogForkAndCrash(node.Logger),
 				node.ProtocolDB,
+				node.StorageLockMgr,
 				node.Logger,
 			)
 			if err != nil {
@@ -621,7 +622,7 @@ func main() {
 			notifier.AddFollowerConsumer(followerDistributor)
 
 			// initialize the persister
-			persist, err := persister.New(node.ProtocolDB, node.RootChainID)
+			persist, err := persister.New(node.ProtocolDB, node.RootChainID, node.StorageLockMgr)
 			if err != nil {
 				return nil, err
 			}

cmd/scaffold.go

@@ -303,7 +303,7 @@ func (fnb *FlowNodeBuilder) EnqueuePingService() {
 		var hotstuffViewFunc func() (uint64, error)
 		// Setup consensus nodes to report their HotStuff view
 		if fnb.BaseConfig.NodeRole == flow.RoleConsensus.String() {
-			hotstuffReader, err := persister.NewReader(node.ProtocolDB, node.RootChainID)
+			hotstuffReader, err := persister.NewReader(node.ProtocolDB, node.RootChainID, node.StorageLockMgr)
 			if err != nil {
 				return nil, err
 			}

cmd/util/cmd/read-hotstuff/cmd/get_liveness.go

@@ -22,8 +22,10 @@ func init() {
 
 func runGetLivenessData(*cobra.Command, []string) {
 	err := common.WithStorage(flagDatadir, func(db storage.DB) error {
+		lockManager := storage.NewTestingLockManager()
+
 		chainID := flow.ChainID(flagChain)
-		reader, err := persister.NewReader(db, chainID)
+		reader, err := persister.NewReader(db, chainID, lockManager)
 		if err != nil {
 			log.Fatal().Err(err).Msg("could not create reader from db")
 		}

cmd/util/cmd/read-hotstuff/cmd/get_safety.go

@@ -22,9 +22,10 @@ func init() {
 
 func runGetSafetyData(*cobra.Command, []string) {
 	err := common.WithStorage(flagDatadir, func(db storage.DB) error {
+		lockManager := storage.NewTestingLockManager()
 
 		chainID := flow.ChainID(flagChain)
-		reader, err := persister.NewReader(db, chainID)
+		reader, err := persister.NewReader(db, chainID, lockManager)
 		if err != nil {
 			log.Fatal().Err(err).Msg("could not create reader from db")
 		}

consensus/hotstuff/persister/persister.go

@@ -3,6 +3,8 @@ package persister
 import (
 	"fmt"
 
+	"github.com/jordanschalm/lockctx"
+
 	"github.com/onflow/flow-go/consensus/hotstuff"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/storage"
@@ -15,8 +17,9 @@ import (
 // SafetyData and LivenessData, for each distinct chain ID. This bootstrapping must be complete
 // before constructing a Persister instance with New (otherwise it will return an error).
 type Persister struct {
-	db      storage.DB
-	chainID flow.ChainID
+	db          storage.DB
+	chainID     flow.ChainID
+	lockManager lockctx.Manager
 }
 
 var _ hotstuff.Persister = (*Persister)(nil)
@@ -26,15 +29,16 @@ var _ hotstuff.PersisterReader = (*Persister)(nil)
 // Persister depends on protocol.State and cluster.State bootstrapping to set initial values for
 // SafetyData and LivenessData, for each distinct chain ID. This bootstrapping must be completed
 // before first using a Persister instance.
-func New(db storage.DB, chainID flow.ChainID) (*Persister, error) {
+func New(db storage.DB, chainID flow.ChainID, lockManager lockctx.Manager) (*Persister, error) {
 	err := ensureSafetyDataAndLivenessDataAreBootstrapped(db, chainID)
 	if err != nil {
 		return nil, fmt.Errorf("fail to check persister was properly bootstrapped: %w", err)
 	}
 
 	p := &Persister{
-		db:      db,
-		chainID: chainID,
+		db:          db,
+		chainID:     chainID,
+		lockManager: lockManager,
 	}
 	return p, nil
 }
@@ -62,8 +66,8 @@ func ensureSafetyDataAndLivenessDataAreBootstrapped(db storage.DB, chainID flow.
 }
 
 // NewReader returns a new Persister as a PersisterReader type (only read methods accessible).
-func NewReader(db storage.DB, chainID flow.ChainID) (hotstuff.PersisterReader, error) {
-	return New(db, chainID)
+func NewReader(db storage.DB, chainID flow.ChainID, lockManager lockctx.Manager) (hotstuff.PersisterReader, error) {
+	return New(db, chainID, lockManager)
 }
 
 // GetSafetyData will retrieve last persisted safety data.
@@ -85,15 +89,19 @@ func (p *Persister) GetLivenessData() (*hotstuff.LivenessData, error) {
 // PutSafetyData persists the last safety data.
 // During normal operations, no errors are expected.
 func (p *Persister) PutSafetyData(safetyData *hotstuff.SafetyData) error {
-	return p.db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-		return operation.UpsertSafetyData(rw.Writer(), p.chainID, safetyData)
+	return storage.WithLock(p.lockManager, storage.LockInsertSafetyData, func(lctx lockctx.Context) error {
+		return p.db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
+			return operation.UpsertSafetyData(lctx, rw, p.chainID, safetyData)
+		})
 	})
 }
 
 // PutLivenessData persists the last liveness data.
 // During normal operations, no errors are expected.
 func (p *Persister) PutLivenessData(livenessData *hotstuff.LivenessData) error {
-	return p.db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-		return operation.UpsertLivenessData(rw.Writer(), p.chainID, livenessData)
+	return storage.WithLock(p.lockManager, storage.LockInsertLivenessData, func(lctx lockctx.Context) error {
+		return p.db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
+			return operation.UpsertLivenessData(lctx, rw, p.chainID, livenessData)
+		})
 	})
 }

consensus/integration/nodes_test.go

@@ -563,7 +563,7 @@ func createNode(
 
 	signer := verification.NewCombinedSigner(me, beaconKeyStore)
 
-	persist, err := persister.New(db, rootHeader.ChainID)
+	persist, err := persister.New(db, rootHeader.ChainID, lockManager)
 	require.NoError(t, err)
 
 	livenessData, err := persist.GetLivenessData()

engine/collection/epochmgr/factories/hotstuff.go

@@ -32,6 +32,7 @@ type HotStuffFactory struct {
 	baseLogger     zerolog.Logger
 	me             module.Local
 	db             storage.DB
+	lockManager    storage.LockManager
 	protoState     protocol.State
 	engineMetrics  module.EngineMetrics
 	mempoolMetrics module.MempoolMetrics
@@ -43,6 +44,7 @@ func NewHotStuffFactory(
 	log zerolog.Logger,
 	me module.Local,
 	db storage.DB,
+	lockManager storage.LockManager,
 	protoState protocol.State,
 	engineMetrics module.EngineMetrics,
 	mempoolMetrics module.MempoolMetrics,
@@ -54,6 +56,7 @@ func NewHotStuffFactory(
 		baseLogger:     log,
 		me:             me,
 		db:             db,
+		lockManager:    lockManager,
 		protoState:     protoState,
 		engineMetrics:  engineMetrics,
 		mempoolMetrics: mempoolMetrics,
@@ -155,7 +158,7 @@ func (f *HotStuffFactory) CreateModules(
 		return nil, nil, err
 	}
 
-	persist, err := persister.New(f.db, cluster.ChainID())
+	persist, err := persister.New(f.db, cluster.ChainID(), f.lockManager)
 	if err != nil {
 		return nil, nil, err
 	}

engine/testutil/nodes.go

@@ -377,6 +377,7 @@ func CollectionNode(t *testing.T, hub *stub.Hub, identity bootstrap.NodeInfo, ro
 		node.Log,
 		node.Me,
 		db,
+		node.LockManager,
 		node.State,
 		node.Metrics,
 		node.Metrics,

module/mempool/consensus/exec_fork_suppressor.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/jordanschalm/lockctx"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"go.uber.org/atomic"
@@ -46,6 +47,7 @@ type ExecForkSuppressor struct {
 	execForkDetected      atomic.Bool
 	onExecFork            ExecForkActor
 	execForkEvidenceStore storage.ExecutionForkEvidence
+	lockManager           storage.LockManager
 	log                   zerolog.Logger
 }
 
@@ -61,6 +63,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 +86,7 @@ func NewExecStateForkSuppressor(
 		execForkDetected:      *atomic.NewBool(execForkDetectedFlag),
 		onExecFork:            onExecFork,
 		execForkEvidenceStore: executionForkEvidenceStore,
+		lockManager:           lockManager,
 		log:                   log.With().Str("mempool", "ExecForkSuppressor").Logger(),
 	}
 
@@ -365,9 +369,13 @@ func (s *ExecForkSuppressor) filterConflictingSeals(sealsByBlockID map[flow.Iden
 				s.execForkDetected.Store(true)
 				s.Clear()
 				conflictingSeals = append(sealsList{candidateSeal}, conflictingSeals...)
-				err := s.execForkEvidenceStore.StoreIfNotExists(conflictingSeals)
+
+				// Acquire lock and store execution fork evidence
+				err := storage.WithLock(s.lockManager, storage.LockInsertExecutionForkEvidence, func(lctx lockctx.Context) error {
+					return s.execForkEvidenceStore.StoreIfNotExists(lctx, conflictingSeals)
+				})
 				if err != nil {
-					panic("failed to store execution fork evidence")
+					s.log.Fatal().Msg("failed to store execution fork evidence")
 				}
 				s.onExecFork(conflictingSeals)
 				return nil