Merge branch 'master' into janez/system-chunk-logging-changes

Author: j1010001

storage/locks.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/jordanschalm/lockctx"
@@ -115,6 +116,25 @@ func NewTestingLockManager() lockctx.Manager {
 	return lockctx.NewManager(Locks(), makeLockPolicy())
 }
 
+// HeldOneLock checks that exactly one of the two specified locks is held in the provided lock context.
+func HeldOneLock(lctx lockctx.Proof, lockA string, lockB string) (bool, string) {
+	heldLockA := lctx.HoldsLock(lockA)
+	heldLockB := lctx.HoldsLock(lockB)
+	if heldLockA {
+		if heldLockB {
+			return false, fmt.Sprintf("epxect to hold only one lock, but actually held both locks: %s and %s", lockA, lockB)
+		} else {
+			return true, ""
+		}
+	} else {
+		if heldLockB {
+			return true, ""
+		} else {
+			return false, fmt.Sprintf("expect to hold one of the locks: %s or %s, but actually held none", lockA, lockB)
+		}
+	}
+}
+
 // WithLock is a helper function that creates a new lock context, acquires the specified lock,
 // and executes the provided function within that context.
 // This function passes through any errors returned by fn.

storage/locks_test.go

@@ -0,0 +1,128 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestHeldOneLock(t *testing.T) {
+	lockManager := NewTestingLockManager()
+
+	t.Run("holds only lockA", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+		err := lctx.AcquireLock(LockInsertBlock)
+		require.NoError(t, err)
+
+		held, msg := HeldOneLock(lctx, LockInsertBlock, LockFinalizeBlock)
+		assert.True(t, held)
+		assert.Empty(t, msg)
+	})
+
+	t.Run("holds only lockB", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+		err := lctx.AcquireLock(LockFinalizeBlock)
+		require.NoError(t, err)
+
+		held, msg := HeldOneLock(lctx, LockInsertBlock, LockFinalizeBlock)
+		assert.True(t, held)
+		assert.Empty(t, msg)
+	})
+
+	t.Run("holds both locks", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+		err := lctx.AcquireLock(LockInsertBlock)
+		require.NoError(t, err)
+		err = lctx.AcquireLock(LockFinalizeBlock)
+		require.NoError(t, err)
+
+		held, msg := HeldOneLock(lctx, LockInsertBlock, LockFinalizeBlock)
+		assert.False(t, held)
+		assert.Contains(t, msg, "epxect to hold only one lock, but actually held both locks")
+		assert.Contains(t, msg, LockInsertBlock)
+		assert.Contains(t, msg, LockFinalizeBlock)
+	})
+
+	t.Run("holds neither lock", func(t *testing.T) {
+		// Create a context that doesn't hold any locks
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+
+		held, msg := HeldOneLock(lctx, LockInsertBlock, LockFinalizeBlock)
+		assert.False(t, held)
+		assert.Contains(t, msg, "expect to hold one of the locks")
+		assert.Contains(t, msg, LockInsertBlock)
+		assert.Contains(t, msg, LockFinalizeBlock)
+	})
+
+	t.Run("holds different lock", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+		err := lctx.AcquireLock(LockBootstrapping)
+		require.NoError(t, err)
+
+		held, msg := HeldOneLock(lctx, LockInsertBlock, LockFinalizeBlock)
+		assert.False(t, held)
+		assert.Contains(t, msg, "expect to hold one of the locks")
+		assert.Contains(t, msg, LockInsertBlock)
+		assert.Contains(t, msg, LockFinalizeBlock)
+	})
+
+	t.Run("with different lock combinations", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+		err := lctx.AcquireLock(LockInsertOwnReceipt)
+		require.NoError(t, err)
+
+		held, msg := HeldOneLock(lctx, LockInsertOwnReceipt, LockInsertCollection)
+		assert.True(t, held)
+		assert.Empty(t, msg)
+	})
+
+	t.Run("with cluster block locks", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+		err := lctx.AcquireLock(LockInsertOrFinalizeClusterBlock)
+		require.NoError(t, err)
+
+		held, msg := HeldOneLock(lctx, LockInsertOrFinalizeClusterBlock, LockIndexResultApproval)
+		assert.True(t, held)
+		assert.Empty(t, msg)
+	})
+
+	t.Run("error message format for both locks", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+		err := lctx.AcquireLock(LockInsertBlock)
+		require.NoError(t, err)
+		err = lctx.AcquireLock(LockFinalizeBlock)
+		require.NoError(t, err)
+
+		// Check that both locks are actually held
+		assert.True(t, lctx.HoldsLock(LockInsertBlock))
+		assert.True(t, lctx.HoldsLock(LockFinalizeBlock))
+
+		held, msg := HeldOneLock(lctx, LockInsertBlock, LockFinalizeBlock)
+		assert.False(t, held)
+		require.NotEmpty(t, msg)
+		assert.Contains(t, msg, "epxect to hold only one lock, but actually held both locks")
+		assert.Contains(t, msg, LockInsertBlock)
+		assert.Contains(t, msg, LockFinalizeBlock)
+	})
+
+	t.Run("error message format for no locks", func(t *testing.T) {
+		lctx := lockManager.NewContext()
+		defer lctx.Release()
+
+		held, msg := HeldOneLock(lctx, "lockA", "lockB")
+		assert.False(t, held)
+		require.NotEmpty(t, msg)
+		assert.Contains(t, msg, "expect to hold one of the locks")
+		assert.Contains(t, msg, "lockA")
+		assert.Contains(t, msg, "lockB")
+	})
+}

storage/operation/proposal_signatures.go

@@ -1,13 +1,31 @@
 package operation
 
 import (
+	"fmt"
+
+	"github.com/jordanschalm/lockctx"
+
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/storage"
 )
 
 // InsertProposalSignature inserts a proposal signature by block ID.
-// Returns storage.ErrAlreadyExists if a proposal signature has already been inserted for the block.
-func InsertProposalSignature(w storage.Writer, blockID flow.Identifier, sig *[]byte) error {
+//
+// CAUTION:
+//   - The caller must acquire either the lock [storage.LockInsertBlock] or [storage.LockInsertOrFinalizeClusterBlock] (but not both)
+//     and hold it until the database write has been committed.
+//   - OVERWRITES existing data (potential for data corruption):
+//     The lock proof serves as a reminder that the CALLER is responsible to ensure that the DEDUPLICATION CHECK
+//     is done elsewhere ATOMICALLY with this write operation. It is intended that this function is called only for new
+//     blocks, i.e. no signature was previously persisted for it.
+//
+// No errors are expected during normal operation.
+func InsertProposalSignature(lctx lockctx.Proof, w storage.Writer, blockID flow.Identifier, sig *[]byte) error {
+	held, errStr := storage.HeldOneLock(lctx, storage.LockInsertBlock, storage.LockInsertOrFinalizeClusterBlock)
+	if !held {
+		return fmt.Errorf("%s", errStr)
+	}
+
 	return UpsertByKey(w, MakePrefix(codeBlockIDToProposalSignature, blockID), sig)
 }
 

storage/procedure/cluster.go

@@ -53,7 +53,7 @@ func InsertClusterBlock(lctx lockctx.Proof, rw storage.ReaderBatchWriter, propos
 	}
 
 	// insert the block payload; without further overwrite checks (see above for explanation)
-	err = operation.InsertProposalSignature(rw.Writer(), blockID, &proposal.ProposerSigData)
+	err = operation.InsertProposalSignature(lctx, rw.Writer(), blockID, &proposal.ProposerSigData)
 	if err != nil {
 		return fmt.Errorf("could not insert proposer signature: %w", err)
 	}

storage/store/headers.go

@@ -83,7 +83,7 @@ func (h *Headers) storeTx(
 		return err
 	}
 
-	return h.sigs.storeTx(rw, blockID, proposalSig)
+	return h.sigs.storeTx(lctx, rw, blockID, proposalSig)
 }
 
 func (h *Headers) retrieveTx(blockID flow.Identifier) (*flow.Header, error) {

storage/store/proposal_signatures.go

@@ -1,6 +1,8 @@
 package store
 
 import (
+	"github.com/jordanschalm/lockctx"
+
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/metrics"
@@ -21,8 +23,8 @@ type proposalSignatures struct {
 // newProposalSignatures creates a proposalSignatures instance which is a database of block proposal signatures
 // which supports storing, caching and retrieving by block ID.
 func newProposalSignatures(collector module.CacheMetrics, db storage.DB) *proposalSignatures {
-	store := func(rw storage.ReaderBatchWriter, blockID flow.Identifier, sig []byte) error {
-		return operation.InsertProposalSignature(rw.Writer(), blockID, &sig)
+	store := func(lctx lockctx.Proof, rw storage.ReaderBatchWriter, blockID flow.Identifier, sig []byte) error {
+		return operation.InsertProposalSignature(lctx, rw.Writer(), blockID, &sig)
 	}
 
 	retrieve := func(r storage.Reader, blockID flow.Identifier) ([]byte, error) {
@@ -35,13 +37,24 @@ func newProposalSignatures(collector module.CacheMetrics, db storage.DB) *propos
 		db: db,
 		cache: newCache(collector, metrics.ResourceProposalSignature,
 			withLimit[flow.Identifier, []byte](4*flow.DefaultTransactionExpiry),
-			withStore(store),
+			withStoreWithLock(store),
 			withRetrieve(retrieve)),
 	}
 }
 
-func (h *proposalSignatures) storeTx(rw storage.ReaderBatchWriter, blockID flow.Identifier, sig []byte) error {
-	return h.cache.PutTx(rw, blockID, sig)
+// storeTx persists the given `sig` as the proposer's signature for the specified block.
+//
+// CAUTION:
+//   - The caller must acquire either the lock [storage.LockInsertBlock] or [storage.LockInsertOrFinalizeClusterBlock]
+//     but not both and hold the lock until the database write has been committed.
+//   - OVERWRITES existing data (potential for data corruption):
+//     The lock proof serves as a reminder that the CALLER is responsible to ensure that the DEDUPLICATION CHECK
+//     is done elsewhere ATOMICALLY with this write operation. It is intended that this function is called only for new
+//     blocks, i.e. no signature was previously persisted for it.
+//
+// No error returns expected during normal operations.
+func (h *proposalSignatures) storeTx(lctx lockctx.Proof, rw storage.ReaderBatchWriter, blockID flow.Identifier, sig []byte) error {
+	return h.cache.PutWithLockTx(lctx, rw, blockID, sig)
 }
 
 func (h *proposalSignatures) retrieveTx(blockID flow.Identifier) ([]byte, error) {