fix(dot/digest): create BlockImportHandler and remove channel

dot/core/interfaces.go

@@ -19,6 +19,10 @@ import (
 	"github.com/ChainSafe/gossamer/lib/transaction"
 )
 
+type BlockImportDigestHandler interface {
+	Handle(*types.Header) error
+}
+
 // BlockState interface for block state methods
 type BlockState interface {
 	BestBlockHash() common.Hash

dot/core/messages_integration_test.go

@@ -104,6 +104,11 @@ func TestService_HandleBlockProduced(t *testing.T) {
 		Body: *types.NewBody([]types.Extrinsic{}),
 	}
 
+	onBlockImportHandlerMock := NewMockBlockImportDigestHandler(ctrl)
+	onBlockImportHandlerMock.EXPECT().Handle(&newBlock.Header).Return(nil)
+
+	s.onBlockImport = onBlockImportHandlerMock
+
 	expected := &network.BlockAnnounceMessage{
 		ParentHash:     newBlock.Header.ParentHash,
 		Number:         newBlock.Header.Number,
@@ -172,6 +177,10 @@ func TestService_HandleTransactionMessage(t *testing.T) {
 	currentSlot := currentTimestamp / babeConfig.SlotDuration
 
 	block := buildTestBlockWithoutExtrinsics(t, rt, genHeader, currentSlot, currentTimestamp)
+	onBlockImportDigestHandlerMock := NewMockBlockImportDigestHandler(ctrl)
+	onBlockImportDigestHandlerMock.EXPECT().Handle(&block.Header).Return(nil)
+
+	s.onBlockImport = onBlockImportDigestHandlerMock
 
 	err = s.handleBlock(block, ts)
 	require.NoError(t, err)

dot/core/mocks_generate_test.go

@@ -3,5 +3,5 @@
 
 package core
 
-//go:generate mockgen -destination=mocks_test.go -package $GOPACKAGE . BlockState,StorageState,TransactionState,Network,CodeSubstitutedState,Telemetry
+//go:generate mockgen -destination=mocks_test.go -package $GOPACKAGE . BlockState,StorageState,TransactionState,Network,CodeSubstitutedState,Telemetry,BlockImportDigestHandler
 //go:generate mockgen -destination=mock_runtime_instance_test.go -package $GOPACKAGE github.com/ChainSafe/gossamer/lib/runtime Instance

dot/core/service.go

@@ -52,7 +52,8 @@ type Service struct {
 	codeSubstitutedState CodeSubstitutedState
 
 	// Keystore
-	keys *keystore.GlobalKeystore
+	keys          *keystore.GlobalKeystore
+	onBlockImport BlockImportDigestHandler
 }
 
 // Config holds the configuration for the core Service.
@@ -68,6 +69,7 @@ type Config struct {
 
 	CodeSubstitutes      map[common.Hash]string
 	CodeSubstitutedState CodeSubstitutedState
+	OnBlockImport        BlockImportDigestHandler
 }
 
 // NewService returns a new core service that connects the runtime, BABE
@@ -89,6 +91,7 @@ func NewService(cfg *Config) (*Service, error) {
 		blockAddCh:           blockAddCh,
 		codeSubstitute:       cfg.CodeSubstitutes,
 		codeSubstitutedState: cfg.CodeSubstitutedState,
+		onBlockImport:        cfg.OnBlockImport,
 	}
 
 	return srv, nil
@@ -209,6 +212,11 @@ func (s *Service) handleBlock(block *types.Block, state *rtstorage.TrieState) er
 		}
 	}
 
+	err = s.onBlockImport.Handle(&block.Header)
+	if err != nil {
+		return fmt.Errorf("on block import handle: %w", err)
+	}
+
 	logger.Debugf("imported block %s and stored state trie with root %s",
 		block.Header.Hash(), state.MustRoot())
 

dot/core/service_integration_test.go

@@ -76,6 +76,10 @@ func TestAnnounceBlock(t *testing.T) {
 		Body: *types.NewBody([]types.Extrinsic{}),
 	}
 
+	onBlockImportHandleMock := NewMockBlockImportDigestHandler(ctrl)
+	onBlockImportHandleMock.EXPECT().Handle(&newBlock.Header).Return(nil)
+	s.onBlockImport = onBlockImportHandleMock
+
 	expected := &network.BlockAnnounceMessage{
 		ParentHash:     newBlock.Header.ParentHash,
 		Number:         newBlock.Header.Number,
@@ -481,6 +485,10 @@ func TestService_HandleSubmittedExtrinsic(t *testing.T) {
 
 	block := buildTestBlockWithoutExtrinsics(t, rt, genHeader, currentSlotNumber, currentTimestamp)
 
+	onBlockImportHandlerMock := NewMockBlockImportDigestHandler(ctrl)
+	onBlockImportHandlerMock.EXPECT().Handle(&block.Header).Return(nil)
+	s.onBlockImport = onBlockImportHandlerMock
+
 	err = s.handleBlock(block, ts)
 	require.NoError(t, err)
 

dot/core/service_test.go

@@ -423,9 +423,13 @@ func Test_Service_handleBlock(t *testing.T) {
 		mockBlockState.EXPECT().AddBlock(&block).Return(blocktree.ErrBlockExists)
 		mockBlockState.EXPECT().GetRuntime(block.Header.ParentHash).Return(nil, errTestDummyError)
 
+		onBlockImportHandlerMock := NewMockBlockImportDigestHandler(ctrl)
+		onBlockImportHandlerMock.EXPECT().Handle(&block.Header).Return(nil)
+
 		service := &Service{
-			storageState: mockStorageState,
-			blockState:   mockBlockState,
+			storageState:  mockStorageState,
+			blockState:    mockBlockState,
+			onBlockImport: onBlockImportHandlerMock,
 		}
 		execTest(t, service, &block, trieState, errTestDummyError)
 	})
@@ -448,9 +452,13 @@ func Test_Service_handleBlock(t *testing.T) {
 		mockBlockState.EXPECT().HandleRuntimeChanges(trieState, runtimeMock, block.Header.Hash()).
 			Return(errTestDummyError)
 
+		onBlockImportHandlerMock := NewMockBlockImportDigestHandler(ctrl)
+		onBlockImportHandlerMock.EXPECT().Handle(&block.Header).Return(nil)
+
 		service := &Service{
-			storageState: mockStorageState,
-			blockState:   mockBlockState,
+			storageState:  mockStorageState,
+			blockState:    mockBlockState,
+			onBlockImport: onBlockImportHandlerMock,
 		}
 		execTest(t, service, &block, trieState, errTestDummyError)
 	})
@@ -472,10 +480,13 @@ func Test_Service_handleBlock(t *testing.T) {
 		mockBlockState.EXPECT().GetRuntime(block.Header.ParentHash).Return(runtimeMock, nil)
 		mockBlockState.EXPECT().HandleRuntimeChanges(trieState, runtimeMock, block.Header.Hash()).Return(nil)
 
+		onBlockImportHandlerMock := NewMockBlockImportDigestHandler(ctrl)
+		onBlockImportHandlerMock.EXPECT().Handle(&block.Header).Return(nil)
 		service := &Service{
-			storageState: mockStorageState,
-			blockState:   mockBlockState,
-			ctx:          context.Background(),
+			storageState:  mockStorageState,
+			blockState:    mockBlockState,
+			ctx:           context.Background(),
+			onBlockImport: onBlockImportHandlerMock,
 		}
 		execTest(t, service, &block, trieState, nil)
 	})
@@ -531,12 +542,15 @@ func Test_Service_HandleBlockProduced(t *testing.T) {
 		mockBlockState.EXPECT().HandleRuntimeChanges(trieState, runtimeMock, block.Header.Hash()).Return(nil)
 		mockNetwork := NewMockNetwork(ctrl)
 		mockNetwork.EXPECT().GossipMessage(msg)
+		onBlockImportHandlerMock := NewMockBlockImportDigestHandler(ctrl)
+		onBlockImportHandlerMock.EXPECT().Handle(&block.Header).Return(nil)
 
 		service := &Service{
-			storageState: mockStorageState,
-			blockState:   mockBlockState,
-			net:          mockNetwork,
-			ctx:          context.Background(),
+			storageState:  mockStorageState,
+			blockState:    mockBlockState,
+			net:           mockNetwork,
+			ctx:           context.Background(),
+			onBlockImport: onBlockImportHandlerMock,
 		}
 		execTest(t, service, &block, trieState, nil)
 	})

dot/digest/block_import.go

@@ -0,0 +1,146 @@
+// Copyright 2023 ChainSafe Systems (ON)
+// SPDX-License-Identifier: LGPL-3.0-only
+
+package digest
+
+import (
+	"fmt"
+
+	"github.com/ChainSafe/gossamer/dot/types"
+	"github.com/ChainSafe/gossamer/pkg/scale"
+)
+
+type BlockImportHandler struct {
+	epochState   EpochState
+	grandpaState GrandpaState
+}
+
+func NewBlockImportHandler(epochState EpochState, grandpaState GrandpaState) *BlockImportHandler {
+	return &BlockImportHandler{
+		epochState:   epochState,
+		grandpaState: grandpaState,
+	}
+}
+
+func (h *BlockImportHandler) Handle(importedBlockHeader *types.Header) error {
+	err := h.handleDigests(importedBlockHeader)
+	if err != nil {
+		return fmt.Errorf("while handling digests: %w", err)
+	}
+
+	// TODO: move to core handleBlock
+	// https://github.com/ChainSafe/gossamer/issues/3330
+	err = h.grandpaState.ApplyForcedChanges(importedBlockHeader)
+	if err != nil {
+		return fmt.Errorf("while apply forced changes: %s", err)
+	}
+
+	return nil
+}
+
+// HandleDigests handles consensus digests for an imported block
+func (h *BlockImportHandler) handleDigests(header *types.Header) error {
+	consensusDigests := toConsensusDigests(header.Digest.Types)
+	consensusDigests, err := checkForGRANDPAForcedChanges(consensusDigests)
+	if err != nil {
+		return fmt.Errorf("failed while checking GRANDPA digests: %w", err)
+	}
+
+	for i := range consensusDigests {
+		// avoiding implicit memory aliasing in for loop, since:
+		// for _, digest := range consensusDigests { &digest }
+		// is using the address of a loop variable
+		digest := consensusDigests[i]
+		err := h.handleConsensusDigest(&digest, header)
+		if err != nil {
+			logger.Errorf("cannot handle consensus digest: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func (h *BlockImportHandler) handleConsensusDigest(d *types.ConsensusDigest, header *types.Header) error {
+	switch d.ConsensusEngineID {
+	case types.GrandpaEngineID:
+		data := types.NewGrandpaConsensusDigest()
+		err := scale.Unmarshal(d.Data, &data)
+		if err != nil {
+			return err
+		}
+
+		return h.grandpaState.HandleGRANDPADigest(header, data)
+	case types.BabeEngineID:
+		data := types.NewBabeConsensusDigest()
+		err := scale.Unmarshal(d.Data, &data)
+		if err != nil {
+			return err
+		}
+
+		return h.epochState.HandleBABEDigest(header, data)
+	default:
+		return fmt.Errorf("%w: 0x%x", ErrUnknownConsensusEngineID, d.ConsensusEngineID.ToBytes())
+	}
+}
+
+// toConsensusDigests converts a slice of scale.VaryingDataType to a slice of types.ConsensusDigest.
+func toConsensusDigests(scaleVaryingTypes []scale.VaryingDataType) []types.ConsensusDigest {
+	consensusDigests := make([]types.ConsensusDigest, 0, len(scaleVaryingTypes))
+
+	for _, d := range scaleVaryingTypes {
+		digestValue, err := d.Value()
+		if err != nil {
+			logger.Error(err.Error())
+			continue
+		}
+		digest, ok := digestValue.(types.ConsensusDigest)
+		if !ok {
+			continue
+		}
+
+		switch digest.ConsensusEngineID {
+		case types.GrandpaEngineID, types.BabeEngineID:
+			consensusDigests = append(consensusDigests, digest)
+		}
+	}
+
+	return consensusDigests
+}
+
+// checkForGRANDPAForcedChanges removes any GrandpaScheduledChange in the presence of a
+// GrandpaForcedChange in the same block digest, returning a new slice of types.ConsensusDigest
+func checkForGRANDPAForcedChanges(digests []types.ConsensusDigest) ([]types.ConsensusDigest, error) {
+	var hasForcedChange bool
+	digestsWithoutScheduled := make([]types.ConsensusDigest, 0, len(digests))
+	for i, digest := range digests {
+		if digest.ConsensusEngineID != types.GrandpaEngineID {
+			digestsWithoutScheduled = append(digestsWithoutScheduled, digest)
+			continue
+		}
+
+		data := types.NewGrandpaConsensusDigest()
+		err := scale.Unmarshal(digest.Data, &data)
+		if err != nil {
+			return nil, fmt.Errorf("cannot unmarshal GRANDPA consensus digest: %w", err)
+		}
+
+		dataValue, err := data.Value()
+		if err != nil {
+			return nil, fmt.Errorf("getting value of digest type at index %d: %w", i, err)
+		}
+		switch dataValue.(type) {
+		case types.GrandpaScheduledChange:
+		case types.GrandpaForcedChange:
+			hasForcedChange = true
+			digestsWithoutScheduled = append(digestsWithoutScheduled, digest)
+		default:
+			digestsWithoutScheduled = append(digestsWithoutScheduled, digest)
+		}
+	}
+
+	if hasForcedChange {
+		return digestsWithoutScheduled, nil
+	}
+
+	return digests, nil
+}