Made protocol state idempotent with respect to inserts

Author: durkmurder

engine/common/follower/core.go

@@ -258,7 +258,7 @@ func (c *Core) processCertifiedBlocks(ctx context.Context, blocks CertifiedBlock
 	// Step 1: add blocks to our PendingTree of certified blocks
 	pendingTreeSpan, _ := c.tracer.StartSpanFromContext(ctx, trace.FollowerExtendPendingTree)
 	connectedBlocks, err := c.pendingTree.AddBlocks(blocks)
-	defer pendingTreeSpan.End()
+	pendingTreeSpan.End()
 	if err != nil {
 		return fmt.Errorf("could not process batch of certified blocks: %w", err)
 	}
@@ -282,7 +282,7 @@ func (c *Core) processCertifiedBlocks(ctx context.Context, blocks CertifiedBlock
 // Is NOT concurrency safe: should be executed by _single dedicated_ goroutine.
 // No errors expected during normal operations.
 func (c *Core) processFinalizedBlock(ctx context.Context, finalized *flow.Header) error {
-	span, ctx := c.tracer.StartSpanFromContext(ctx, trace.FollowerProcessFinalizedBlock)
+	span, _ := c.tracer.StartSpanFromContext(ctx, trace.FollowerProcessFinalizedBlock)
 	defer span.End()
 
 	connectedBlocks, err := c.pendingTree.FinalizeFork(finalized)

state/protocol/badger/mutator.go

@@ -116,6 +116,12 @@ func (m *FollowerState) ExtendCertified(ctx context.Context, candidate *flow.Blo
 	defer span.End()
 
 	blockID := candidate.ID()
+	// check if candidate block has been already processed
+	processed, err := m.checkBlockAlreadyProcessed(blockID)
+	if err != nil || processed {
+		return err
+	}
+
 	// sanity check if certifyingQC actually certifies candidate block
 	if certifyingQC.View != candidate.Header.View {
 		return fmt.Errorf("qc doesn't certify candidate block, expect %d view, got %d", candidate.Header.View, certifyingQC.View)
@@ -125,7 +131,7 @@ func (m *FollowerState) ExtendCertified(ctx context.Context, candidate *flow.Blo
 	}
 
 	// check if the block header is a valid extension of parent block
-	err := m.headerExtend(candidate)
+	err = m.headerExtend(candidate)
 	if err != nil {
 		// since we have a QC for this block, it cannot be an invalid extension
 		return fmt.Errorf("unexpected invalid block (id=%x) with certifying qc (id=%x): %s",
@@ -156,8 +162,14 @@ func (m *ParticipantState) Extend(ctx context.Context, candidate *flow.Block) er
 	span, ctx := m.tracer.StartSpanFromContext(ctx, trace.ProtoStateMutatorExtend)
 	defer span.End()
 
+	// check if candidate block has been already processed
+	processed, err := m.checkBlockAlreadyProcessed(candidate.ID())
+	if err != nil || processed {
+		return err
+	}
+
 	// check if the block header is a valid extension of parent block
-	err := m.headerExtend(candidate)
+	err = m.headerExtend(candidate)
 	if err != nil {
 		return fmt.Errorf("header not compliant with chain state: %w", err)
 	}
@@ -244,6 +256,23 @@ func (m *FollowerState) headerExtend(candidate *flow.Block) error {
 	return nil
 }
 
+// checkBlockAlreadyProcessed checks if block with given blockID has been added to the protocol state.
+// Returns:
+// * (true, nil) - block has been already processed.
+// * (false, nil) - block has not been processed.
+// * (false, error) - unknown error when trying to query protocol state.
+// No errors are expected during normal operation.
+func (m *FollowerState) checkBlockAlreadyProcessed(blockID flow.Identifier) (bool, error) {
+	_, err := m.headers.ByBlockID(blockID)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			return false, nil
+		}
+		return false, fmt.Errorf("could not check if candidate block (%x) has been already processed: %w", blockID, err)
+	}
+	return true, nil
+}
+
 // checkOutdatedExtension checks whether candidate block is
 // valid in the context of the entire state. For this, the block needs to
 // directly connect, through its ancestors, to the last finalized block.

state/protocol/badger/mutator_test.go

@@ -2108,6 +2108,11 @@ func TestExtendInvalidGuarantee(t *testing.T) {
 
 		// now the guarantee has invalid signer indices: the checksum should have 4 bytes, but it only has 1
 		payload.Guarantees[0].SignerIndices = []byte{byte(1)}
+
+		// create new block that has invalid collection guarantee
+		block = unittest.BlockWithParentFixture(head)
+		block.SetPayload(payload)
+
 		err = state.Extend(context.Background(), block)
 		require.True(t, signature.IsInvalidSignerIndicesError(err), err)
 		require.ErrorIs(t, err, signature.ErrInvalidChecksum)
@@ -2291,6 +2296,39 @@ func TestHeaderInvalidTimestamp(t *testing.T) {
 	})
 }
 
+// TestProtocolStateIdempotent tests that both participant and follower states correctly process adding same block twice
+// where second extend doesn't result in an error and effectively is no-op.
+func TestProtocolStateIdempotent(t *testing.T) {
+	rootSnapshot := unittest.RootSnapshotFixture(participants)
+	head, err := rootSnapshot.Head()
+	require.NoError(t, err)
+	t.Run("follower", func(t *testing.T) {
+		util.RunWithFollowerProtocolState(t, rootSnapshot, func(db *badger.DB, state *protocol.FollowerState) {
+			block := unittest.BlockWithParentFixture(head)
+			err := state.ExtendCertified(context.Background(), block, unittest.CertifyBlock(block.Header))
+			require.NoError(t, err)
+
+			// same operation should be no-op
+			err = state.ExtendCertified(context.Background(), block, unittest.CertifyBlock(block.Header))
+			require.NoError(t, err)
+		})
+	})
+	t.Run("participant", func(t *testing.T) {
+		util.RunWithFullProtocolState(t, rootSnapshot, func(db *badger.DB, state *protocol.ParticipantState) {
+			block := unittest.BlockWithParentFixture(head)
+			err := state.Extend(context.Background(), block)
+			require.NoError(t, err)
+
+			// same operation should be no-op
+			err = state.Extend(context.Background(), block)
+			require.NoError(t, err)
+
+			err = state.ExtendCertified(context.Background(), block, unittest.CertifyBlock(block.Header))
+			require.NoError(t, err)
+		})
+	})
+}
+
 func assertEpochEmergencyFallbackTriggered(t *testing.T, state realprotocol.State, expected bool) {
 	triggered, err := state.Params().EpochFallbackTriggered()
 	require.NoError(t, err)