feat(client/consensus/grandpa): BlockImport implementation for GRANDPA (#4762)

Author: timwu20

internal/client/api/backend.go

@@ -254,6 +254,12 @@ type AuxStore interface {
 	GetAux(key []byte) ([]byte, error)
 }
 
+// StorageProvider provides access to storage primitives
+type StorageProvider[H runtime.Hash, N runtime.Number, Hasher runtime.Hasher[H]] interface {
+	// Given a block hash and a key, return the value under the key in that block.
+	Storage(hash H, key storage.StorageKey) (storage.StorageData, error)
+}
+
 // Backend is the client backend.
 //
 // Manages the data layer.

internal/client/consensus/common/block_import.go

@@ -183,3 +183,29 @@ type BlockImportParams[H runtime.Hash, N runtime.Number, E runtime.Extrinsic, He
 	// Cached full header hash (with post-digests applied).
 	PostHash *H
 }
+
+// GetPostHash retrieves the full header hash (with post-digests applied).
+func (b *BlockImportParams[H, N, E, Header]) GetPostHash() H {
+	if b.PostHash != nil {
+		return *b.PostHash
+	}
+	return b.GetPostHeader().Hash()
+}
+
+// GetPostHeader retrieves the post header.
+func (b *BlockImportParams[H, N, E, Header]) GetPostHeader() Header {
+	if len(b.PostDigests) == 0 {
+		return b.Header.Clone().(Header)
+	}
+	hdr := b.Header.Clone().(Header)
+	for _, digestItem := range b.PostDigests {
+		hdr.DigestMut().Push(digestItem)
+	}
+	return hdr
+}
+
+// WithState checks if this block contains state import action
+func (b *BlockImportParams[H, N, E, Header]) WithState() bool {
+	_, ok := b.StateAction.(StateActionApplyChanges)
+	return ok
+}

internal/client/consensus/grandpa/authorities.go

@@ -144,10 +144,11 @@ func (sas *SharedAuthoritySet[H, N]) applyForcedChanges( //nolint:unused
 	bestHash H,
 	bestNumber N,
 	isDescendentOf IsDescendentOf[H],
+	initialSync bool,
 	// TODO: telemtry,
 ) (newSet *appliedChanges[H, N], err error) {
 	authSet := sas.inner.Data()
-	return authSet.applyForcedChanges(bestHash, bestNumber, isDescendentOf)
+	return authSet.applyForcedChanges(bestHash, bestNumber, isDescendentOf, initialSync)
 }
 
 // applyStandardChanges will apply or prune any pending transitions based on a finality trigger. This method ensures
@@ -505,6 +506,7 @@ func (authSet *AuthoritySet[H, N]) currentLimit(min N) (limit *N) {
 func (authSet *AuthoritySet[H, N]) applyForcedChanges(bestHash H, //skipcq:  RVV-B0001
 	bestNumber N,
 	isDescendentOf IsDescendentOf[H],
+	initialSync bool,
 	// TODO: telemetry
 ) (newSet *appliedChanges[H, N], err error) {
 
@@ -541,7 +543,11 @@ func (authSet *AuthoritySet[H, N]) applyForcedChanges(bestHash H, //skipcq:  RVV
 				}
 
 				// apply this change: make the set canonical
-				logger.Infof("👴 Applying authority set change forced at block #%d", change.CanonHeight)
+				l := logger.Infof
+				if initialSync {
+					l = logger.Debugf
+				}
+				l("👴 Applying authority set change forced at block #%d", change.CanonHeight)
 
 				// TODO telemetry
 

internal/client/consensus/grandpa/authorities_test.go

@@ -577,12 +577,13 @@ func TestForceChanges(t *testing.T) {
 		"hash_a10",
 		10,
 		staticIsDescendentOf[string](true),
+		false,
 	)
 	require.NoError(t, err)
 	require.Nil(t, resForced)
 
 	// too late
-	resForced, err = authorities.applyForcedChanges("hash_a16", 16, isDescOfA)
+	resForced, err = authorities.applyForcedChanges("hash_a16", 16, isDescOfA, false)
 	require.NoError(t, err)
 	require.Nil(t, resForced)
 
@@ -602,7 +603,7 @@ func TestForceChanges(t *testing.T) {
 			},
 		},
 	}
-	resForced, err = authorities.applyForcedChanges("hash_a15", 15, isDescOfA)
+	resForced, err = authorities.applyForcedChanges("hash_a15", 15, isDescOfA, false)
 	require.NoError(t, err)
 	require.NotNil(t, resForced)
 	require.Equal(t, exp, *resForced)
@@ -644,6 +645,7 @@ func TestForceChangesWithNoDelay(t *testing.T) {
 		hashA,
 		5,
 		staticIsDescendentOf[string](false),
+		false,
 	)
 	require.NoError(t, err)
 	require.NotNil(t, resForced)
@@ -723,6 +725,7 @@ func TestForceChangesBlockedByStandardChanges(t *testing.T) {
 		"hash_d45",
 		45,
 		staticIsDescendentOf[string](true),
+		false,
 	)
 	require.ErrorIs(t, err, errForcedAuthoritySetChangeDependencyUnsatisfied)
 	require.Equal(t, 0, len(authorities.AuthoritySetChanges))
@@ -748,6 +751,7 @@ func TestForceChangesBlockedByStandardChanges(t *testing.T) {
 		"hash_d45",
 		45,
 		staticIsDescendentOf[string](true),
+		false,
 	)
 	require.ErrorIs(t, err, errForcedAuthoritySetChangeDependencyUnsatisfied)
 	require.Equal(t, expChanges, authorities.AuthoritySetChanges)
@@ -787,6 +791,7 @@ func TestForceChangesBlockedByStandardChanges(t *testing.T) {
 		hashD,
 		45,
 		staticIsDescendentOf[string](true),
+		false,
 	)
 	require.NoError(t, err)
 	require.NotNil(t, resForced)

internal/client/consensus/grandpa/aux_schema.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 
 	"github.com/ChainSafe/gossamer/internal/client/api"
+	shareddata "github.com/ChainSafe/gossamer/internal/client/consensus/common/shared-data"
+	primitives "github.com/ChainSafe/gossamer/internal/primitives/consensus/grandpa"
 	"github.com/ChainSafe/gossamer/internal/primitives/runtime"
 	grandpa "github.com/ChainSafe/gossamer/pkg/finality-grandpa"
 	"github.com/ChainSafe/gossamer/pkg/scale"
@@ -21,12 +23,102 @@ var (
 
 type writeAux func(insertions []api.KeyValue) error
 
+type getGenesisAuthorities func() (primitives.AuthorityList, error)
+
 // Persistent data kept between runs.
 type persistentData[H runtime.Hash, N runtime.Number] struct {
 	authoritySet *SharedAuthoritySet[H, N]
 	setState     *SharedVoterSetState[H, N]
 }
 
+func loadDecoded[T any](store api.AuxStore, key []byte) (*T, error) {
+	encodedValue, err := store.GetAux(key)
+	if err != nil {
+		return nil, err
+	}
+	if encodedValue == nil {
+		return nil, nil
+	}
+
+	var dst T
+	err = scale.Unmarshal(encodedValue, &dst)
+	if err != nil {
+		return nil, err
+	}
+	return &dst, nil
+}
+
+func loadPersistent[H runtime.Hash, N runtime.Number](
+	store api.AuxStore,
+	genesisHash H,
+	genesisNumber N,
+	genesisAuths getGenesisAuthorities,
+) (*persistentData[H, N], error) {
+	genesis := grandpa.HashNumber[H, N]{Hash: genesisHash, Number: genesisNumber}
+	makeGenesisRound := grandpa.NewRoundState[H, N]
+
+	authSet, err := loadDecoded[AuthoritySet[H, N]](store, authoritySetKey)
+	if err != nil {
+		return nil, err
+	}
+	if authSet != nil {
+		var setState voterSetState[H, N]
+		state, err := loadDecoded[voterSetStateVDT[H, N]](store, setStateKey)
+		if err != nil {
+			return nil, err
+		}
+
+		if state != nil && state.inner != nil {
+			setState = state.inner
+		} else {
+			state := makeGenesisRound(genesis)
+			if state.PrevoteGHOST == nil {
+				panic("state is for completed round; completed rounds must have a prevote ghost; qed.")
+			}
+			base := state.PrevoteGHOST
+			setState = newVoterSetStateLive(primitives.SetID(authSet.SetID), *authSet, *base)
+		}
+
+		return &persistentData[H, N]{
+			authoritySet: &SharedAuthoritySet[H, N]{inner: *shareddata.NewSharedData(*authSet)},
+			setState:     &SharedVoterSetState[H, N]{inner: setState},
+		}, nil
+	}
+
+	logger.Info("👴 Loading GRANDPA authority set from genesis on what appears to be first startup")
+	genesisAuthorities, err := genesisAuths()
+	if err != nil {
+		return nil, err
+	}
+	genesisSet, err := NewGenesisAuthoritySet[H, N](genesisAuthorities)
+	if err != nil {
+		panic("genesis authorities is non-empty; all weights are non-zero; qed.")
+	}
+
+	state := makeGenesisRound(genesis)
+	base := state.PrevoteGHOST
+	if base == nil {
+		panic("state is for completed round; completed rounds must have a prevote ghost; qed.")
+	}
+
+	genesisState := newVoterSetStateLive(0, *genesisSet, *base)
+	genesisStateVDT := newVoterSetStateVDT[H, N]()
+	genesisStateVDT.inner = genesisState
+	insert := []api.KeyValue{
+		{Key: authoritySetKey, Value: scale.MustMarshal(*genesisSet)},
+		{Key: setStateKey, Value: scale.MustMarshal(genesisStateVDT)},
+	}
+	err = store.InsertAux(insert, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return &persistentData[H, N]{
+		authoritySet: &SharedAuthoritySet[H, N]{inner: *shareddata.NewSharedData(*genesisSet)},
+		setState:     &SharedVoterSetState[H, N]{inner: genesisState},
+	}, nil
+}
+
 // updateAuthoritySet Update the authority set on disk after a change.
 //
 // If there has just been a handoff, pass a newSet parameter that describes the handoff. set in all cases should

internal/client/consensus/grandpa/grandpa.go

@@ -110,7 +110,7 @@ type ClientForGrandpa[
 	papi.ProvideRuntimeAPI[primitives.GrandpaAPI[H, N]]
 	// api.ExecutorProvider
 	client_common.BlockImport[H, N, E, Header]
-	// api.StorageProvider[H, N, Hasher]
+	api.StorageProvider[H, N, Hasher]
 }
 
 // Something that one can ask to do a block sync request.
@@ -165,7 +165,7 @@ type LinkHalf[
 	persistentData      persistentData[H, N]
 	voterCommandsRx     chan voterCommand
 	justificationSender GrandpaJustificationSender[H, N, Header]
-	justificationStream GrandpaJustificationStream[H, N, Header] //nolint: unused
+	justificationStream GrandpaJustificationStream[H, N, Header]
 }
 
 // Provider for the Grandpa authority set configured on the genesis block.
@@ -174,6 +174,127 @@ type GenesisAuthoritySetProvider interface {
 	Get() (primitives.AuthorityList, error)
 }
 
+// Make block importer and link half necessary to tie the background voter to it.
+//
+// The justificationImportPeriod sets the minimum period on which justifications will be imported.  When importing
+// a block, if it includes a justification it will only be processed if it fits within this period, otherwise it will
+// be ignored (and won't be validated). This is to avoid slowing down sync by a peer serving us unnecessary
+// justifications which aren't trivial to validate.
+func BlockImport[
+	H runtime.Hash,
+	N runtime.Number,
+	Hasher runtime.Hasher[H],
+	Header runtime.Header[N, H],
+	E runtime.Extrinsic,
+](
+	client ClientForGrandpa[H, N, Hasher, Header, E],
+	justificationImportPeriod uint32,
+	genesisAuthoritySetProvider GenesisAuthoritySetProvider,
+	selectChain common.SelectChain[H, N, Header],
+	// TODO: telemetry
+) (*GrandpaBlockImport[H, N, Hasher, Header, E], LinkHalf[H, N, Hasher, Header, E], error) {
+	return blockImportWithAuthoritySetHardForks(
+		client,
+		justificationImportPeriod,
+		genesisAuthoritySetProvider,
+		selectChain,
+		nil,
+	)
+}
+
+// A descriptor for an authority set hard fork. These are authority set changes that are not signalled by the runtime
+// and instead are defined off-chain (hence the hard fork).
+type AuthoritySetHardFork[H, N any] struct {
+	// The new authority set id.
+	SetID SetID
+	// The block hash and number at which the hard fork should be applied.
+	Block HashNumber[H, N]
+	// The authorities in the new set.
+	Authorities primitives.AuthorityList
+	// The latest block number that was finalized before this authority set hard fork. When defined, the authority set
+	// change will be forced, i.e. the node won't wait for the block above to be finalized before enacting the change,
+	// and the given finalized number will be used as a base for voting.
+	LastFinalized *N
+}
+
+// Make block importer and link half necessary to tie the background voter to it. A vector of authority set hard forks
+// can be passed, any authority set change signalled at the given block (either already signalled or in a further block
+// when importing it) will be replaced by a standard change with the given static authorities.
+func blockImportWithAuthoritySetHardForks[
+	H runtime.Hash,
+	N runtime.Number,
+	Hasher runtime.Hasher[H],
+	Header runtime.Header[N, H],
+	E runtime.Extrinsic,
+](
+	client ClientForGrandpa[H, N, Hasher, Header, E],
+	justificationImportPeriod uint32,
+	genesisAuthoritySetProvider GenesisAuthoritySetProvider,
+	selectChain common.SelectChain[H, N, Header],
+	authoritySetHardForks []AuthoritySetHardFork[H, N],
+	// TODO: telemetry
+) (*GrandpaBlockImport[H, N, Hasher, Header, E], LinkHalf[H, N, Hasher, Header, E], error) {
+	chainInfo := client.Info()
+	genesisHash := chainInfo.GenesisHash
+
+	persistentData, err := loadPersistent[H, N](client, genesisHash, 0, genesisAuthoritySetProvider.Get)
+	if err != nil {
+		return nil, LinkHalf[H, N, Hasher, Header, E]{}, err
+	}
+
+	voterCommands := make(chan voterCommand, 100000)
+
+	justificationSender, justificationStream := NewGrandpaJustificationSender[H, N, Header]()
+
+	// create pending change objects with 0 delay for each authority set hard fork.
+	hardForks := make([]struct {
+		SetID
+		PendingChange[H, N]
+	}, len(authoritySetHardForks))
+	for i, fork := range authoritySetHardForks {
+		var kind delayKind
+		if fork.LastFinalized != nil {
+			kind = delayKindBest[N]{MedianLastFinalized: *fork.LastFinalized}
+		} else {
+			kind = delayKindFinalized{}
+		}
+
+		hardForks[i] = struct {
+			SetID
+			PendingChange[H, N]
+		}{
+			SetID: fork.SetID,
+			PendingChange: PendingChange[H, N]{
+				NextAuthorities: fork.Authorities,
+				Delay:           0,
+				CanonHash:       fork.Block.Hash,
+				CanonHeight:     fork.Block.Number,
+				DelayKind:       kind,
+			},
+		}
+	}
+
+	blockImport := newGrandpaBlockImport(
+		client,
+		justificationImportPeriod,
+		selectChain,
+		persistentData.authoritySet,
+		voterCommands,
+		hardForks,
+		justificationSender,
+	)
+
+	linkHalf := LinkHalf[H, N, Hasher, Header, E]{
+		client:              client,
+		selectChain:         selectChain,
+		persistentData:      *persistentData,
+		voterCommandsRx:     voterCommands,
+		justificationSender: justificationSender,
+		justificationStream: justificationStream,
+	}
+	return blockImport, linkHalf, nil
+}
+
 func globalCommunication[
 	H runtime.Hash,
 	N runtime.Number,