Enforce that collection reference blocks are bound to the cluster's operating epoch (#4148)

undefined

Author: jordanschalm

engine/collection/epochmgr/factories/builder.go

@@ -50,6 +50,7 @@ func (f *BuilderFactory) Create(
 	clusterHeaders storage.Headers,
 	clusterPayloads storage.ClusterPayloads,
 	pool mempool.Transactions,
+	epoch uint64,
 ) (module.Builder, *finalizer.Finalizer, error) {
 
 	build, err := builder.NewBuilder(
@@ -60,6 +61,7 @@ func (f *BuilderFactory) Create(
 		clusterPayloads,
 		pool,
 		f.log,
+		epoch,
 		f.opts...,
 	)
 	if err != nil {

engine/collection/epochmgr/factories/cluster_state.go

@@ -47,7 +47,7 @@ func (f *ClusterStateFactory) Create(stateRoot *clusterkv.StateRoot) (
 	}
 	var clusterState *clusterkv.State
 	if isBootStrapped {
-		clusterState, err = clusterkv.OpenState(f.db, f.tracer, headers, payloads, stateRoot.ClusterID())
+		clusterState, err = clusterkv.OpenState(f.db, f.tracer, headers, payloads, stateRoot.ClusterID(), stateRoot.EpochCounter())
 		if err != nil {
 			return nil, nil, nil, nil, fmt.Errorf("could not open cluster state: %w", err)
 		}

engine/collection/epochmgr/factories/epoch.go

@@ -67,7 +67,7 @@ func (factory *EpochComponentsFactory) Create(
 	err error,
 ) {
 
-	counter, err := epoch.Counter()
+	epochCounter, err := epoch.Counter()
 	if err != nil {
 		err = fmt.Errorf("could not get epoch counter: %w", err)
 		return
@@ -81,7 +81,7 @@ func (factory *EpochComponentsFactory) Create(
 	}
 	_, exists := identities.ByNodeID(factory.me.NodeID())
 	if !exists {
-		err = fmt.Errorf("%w (node_id=%x, epoch=%d)", epochmgr.ErrNotAuthorizedForEpoch, factory.me.NodeID(), counter)
+		err = fmt.Errorf("%w (node_id=%x, epoch=%d)", epochmgr.ErrNotAuthorizedForEpoch, factory.me.NodeID(), epochCounter)
 		return
 	}
 
@@ -109,7 +109,7 @@ func (factory *EpochComponentsFactory) Create(
 		blocks   storage.ClusterBlocks
 	)
 
-	stateRoot, err := badger.NewStateRoot(cluster.RootBlock(), cluster.RootQC())
+	stateRoot, err := badger.NewStateRoot(cluster.RootBlock(), cluster.RootQC(), cluster.EpochCounter())
 	if err != nil {
 		err = fmt.Errorf("could not create valid state root: %w", err)
 		return
@@ -123,9 +123,9 @@ func (factory *EpochComponentsFactory) Create(
 	}
 
 	// get the transaction pool for the epoch
-	pool := factory.pools.ForEpoch(counter)
+	pool := factory.pools.ForEpoch(epochCounter)
 
-	builder, finalizer, err := factory.builder.Create(headers, payloads, pool)
+	builder, finalizer, err := factory.builder.Create(headers, payloads, pool, epochCounter)
 	if err != nil {
 		err = fmt.Errorf("could not create builder/finalizer: %w", err)
 		return

engine/collection/test/cluster_switchover_test.go

@@ -1,7 +1,6 @@
 package test
 
 import (
-	"context"
 	"sync"
 	"testing"
 	"time"
@@ -17,7 +16,6 @@ import (
 	"github.com/onflow/flow-go/model/flow/factory"
 	"github.com/onflow/flow-go/model/flow/filter"
 	"github.com/onflow/flow-go/module"
-	"github.com/onflow/flow-go/module/irrecoverable"
 	"github.com/onflow/flow-go/module/util"
 	"github.com/onflow/flow-go/network/channels"
 	"github.com/onflow/flow-go/network/mocknetwork"
@@ -101,14 +99,9 @@ func NewClusterSwitchoverTestCase(t *testing.T, conf ClusterSwitchoverTestConf)
 	tc.root, err = inmem.SnapshotFromBootstrapState(root, result, seal, qc)
 	require.NoError(t, err)
 
-	cancelCtx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	ctx := irrecoverable.NewMockSignalerContext(t, cancelCtx)
-	defer cancel()
-
 	// create a mock node for each collector identity
 	for _, collector := range nodeInfos {
-		node := testutil.CollectionNode(tc.T(), ctx, tc.hub, collector, tc.root)
+		node := testutil.CollectionNode(tc.T(), tc.hub, collector, tc.root)
 		tc.nodes = append(tc.nodes, node)
 	}
 
@@ -274,8 +267,8 @@ func (tc *ClusterSwitchoverTestCase) ExpectTransaction(epochCounter uint64, clus
 }
 
 // ClusterState opens and returns a read-only cluster state for the given node and cluster ID.
-func (tc *ClusterSwitchoverTestCase) ClusterState(node testmock.CollectionNode, clusterID flow.ChainID) cluster.State {
-	state, err := bcluster.OpenState(node.PublicDB, node.Tracer, node.Headers, node.ClusterPayloads, clusterID)
+func (tc *ClusterSwitchoverTestCase) ClusterState(node testmock.CollectionNode, clusterID flow.ChainID, epoch uint64) cluster.State {
+	state, err := bcluster.OpenState(node.PublicDB, node.Tracer, node.Headers, node.ClusterPayloads, clusterID, epoch)
 	require.NoError(tc.T(), err)
 	return state
 }
@@ -371,7 +364,7 @@ func (tc *ClusterSwitchoverTestCase) CheckClusterState(
 	clusterInfo protocol.Cluster,
 ) {
 	node := tc.Collector(identity.NodeID)
-	state := tc.ClusterState(node, clusterInfo.ChainID())
+	state := tc.ClusterState(node, clusterInfo.ChainID(), clusterInfo.EpochCounter())
 	expected := tc.sentTransactions[clusterInfo.EpochCounter()][clusterInfo.Index()]
 	unittest.NewClusterStateChecker(state).
 		ExpectTxCount(len(expected)).

engine/consensus/ingestion/core_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/module/signature"
 	"github.com/onflow/flow-go/module/trace"
+	"github.com/onflow/flow-go/state/cluster"
 	"github.com/onflow/flow-go/state/protocol"
 	mockprotocol "github.com/onflow/flow-go/state/protocol/mock"
 	mockstorage "github.com/onflow/flow-go/storage/mock"
@@ -37,6 +38,9 @@ type IngestionCoreSuite struct {
 
 	finalIdentities flow.IdentityList // identities at finalized state
 	refIdentities   flow.IdentityList // identities at reference block state
+	epochCounter    uint64            // epoch for the cluster originating the guarantee
+	clusterMembers  flow.IdentityList // members of the cluster originating the guarantee
+	clusterID       flow.ChainID      // chain ID of the cluster originating the guarantee
 
 	final *mockprotocol.Snapshot // finalized state snapshot
 	ref   *mockprotocol.Snapshot // state snapshot w.r.t. reference block
@@ -66,7 +70,9 @@ func (suite *IngestionCoreSuite) SetupTest() {
 	suite.execID = exec.NodeID
 	suite.verifID = verif.NodeID
 
-	clusters := flow.IdentityList{coll}
+	suite.epochCounter = 1
+	suite.clusterMembers = flow.IdentityList{coll}
+	suite.clusterID = cluster.CanonicalClusterID(suite.epochCounter, suite.clusterMembers.NodeIDs())
 
 	identities := flow.IdentityList{access, con, coll, exec, verif}
 	suite.finalIdentities = identities.Copy()
@@ -109,8 +115,20 @@ func (suite *IngestionCoreSuite) SetupTest() {
 	)
 	ref.On("Epochs").Return(suite.query)
 	suite.query.On("Current").Return(suite.epoch)
-	cluster.On("Members").Return(clusters)
-	suite.epoch.On("ClusterByChainID", head.ChainID).Return(cluster, nil)
+	cluster.On("Members").Return(suite.clusterMembers)
+	suite.epoch.On("ClusterByChainID", mock.Anything).Return(
+		func(chainID flow.ChainID) protocol.Cluster {
+			if chainID == suite.clusterID {
+				return cluster
+			}
+			return nil
+		},
+		func(chainID flow.ChainID) error {
+			if chainID == suite.clusterID {
+				return nil
+			}
+			return protocol.ErrClusterNotFound
+		})
 
 	state.On("AtBlockID", mock.Anything).Return(ref)
 	ref.On("Identity", mock.Anything).Return(
@@ -234,7 +252,23 @@ func (suite *IngestionCoreSuite) TestOnGuaranteeExpired() {
 	err := suite.core.OnGuarantee(suite.collID, guarantee)
 	suite.Assert().Error(err, "should error with expired collection")
 	suite.Assert().True(engine.IsOutdatedInputError(err))
+}
+
+// TestOnGuaranteeReferenceBlockFromWrongEpoch validates that guarantees which contain a ChainID
+// that is inconsistent with the reference block (ie. the ChainID either refers to a non-existent
+// cluster, or a cluster for a different epoch) should be considered invalid inputs.
+func (suite *IngestionCoreSuite) TestOnGuaranteeReferenceBlockFromWrongEpoch() {
+	// create a guarantee from a cluster in a different epoch
+	guarantee := suite.validGuarantee()
+	guarantee.ChainID = cluster.CanonicalClusterID(suite.epochCounter+1, suite.clusterMembers.NodeIDs())
 
+	// the guarantee is not part of the memory pool
+	suite.pool.On("Has", guarantee.ID()).Return(false)
+
+	// submit the guarantee as if it was sent by a collection node
+	err := suite.core.OnGuarantee(suite.collID, guarantee)
+	suite.Assert().Error(err, "should error with expired collection")
+	suite.Assert().True(engine.IsInvalidInputError(err))
 }
 
 // TestOnGuaranteeInvalidGuarantor verifiers that collections with any _unknown_
@@ -306,7 +340,7 @@ func (suite *IngestionCoreSuite) TestOnGuaranteeUnknownOrigin() {
 // validGuarantee returns a valid collection guarantee based on the suite state.
 func (suite *IngestionCoreSuite) validGuarantee() *flow.CollectionGuarantee {
 	guarantee := unittest.CollectionGuaranteeFixture()
-	guarantee.ChainID = suite.head.ChainID
+	guarantee.ChainID = suite.clusterID
 
 	signerIndices, err := signature.EncodeSignersToIndices(
 		[]flow.Identifier{suite.collID}, []flow.Identifier{suite.collID})

engine/testutil/mock/nodes.go

@@ -134,6 +134,7 @@ func (n CollectionNode) Start(t *testing.T) {
 	go unittest.FailOnIrrecoverableError(t, n.Ctx.Done(), n.Errs)
 	n.IngestionEngine.Start(n.Ctx)
 	n.EpochManagerEngine.Start(n.Ctx)
+	n.ProviderEngine.Start(n.Ctx)
 }
 
 func (n CollectionNode) Ready() <-chan struct{} {

engine/testutil/nodes.go

@@ -274,7 +274,7 @@ func CompleteStateFixture(
 }
 
 // CollectionNode returns a mock collection node.
-func CollectionNode(t *testing.T, ctx irrecoverable.SignalerContext, hub *stub.Hub, identity bootstrap.NodeInfo, rootSnapshot protocol.Snapshot) testmock.CollectionNode {
+func CollectionNode(t *testing.T, hub *stub.Hub, identity bootstrap.NodeInfo, rootSnapshot protocol.Snapshot) testmock.CollectionNode {
 
 	node := GenericNode(t, hub, identity.Identity(), rootSnapshot)
 	privKeys, err := identity.PrivateKeys()
@@ -310,8 +310,6 @@ func CollectionNode(t *testing.T, ctx irrecoverable.SignalerContext, hub *stub.H
 		selector,
 		retrieve)
 	require.NoError(t, err)
-	// TODO: move this start logic to a more generalized test utility (we need all engines to be startable).
-	providerEngine.Start(ctx)
 
 	pusherEngine, err := pusher.New(node.Log, node.Net, node.State, node.Metrics, node.Metrics, node.Me, collections, transactions)
 	require.NoError(t, err)
@@ -404,7 +402,6 @@ func CollectionNode(t *testing.T, ctx irrecoverable.SignalerContext, hub *stub.H
 		heights,
 	)
 	require.NoError(t, err)
-
 	node.ProtocolEvents.AddConsumer(epochManager)
 
 	return testmock.CollectionNode{

integration/tests/collection/suite.go

@@ -320,8 +320,7 @@ func (suite *CollectorSuite) AwaitTransactionsIncluded(txIDs ...flow.Identifier)
 	suite.T().Fatalf("missing transactions: %v", missing)
 }
 
-// Collector returns the collector node with the given index in the
-// given cluster.
+// Collector returns the collector node with the given index in the given cluster.
 func (suite *CollectorSuite) Collector(clusterIdx, nodeIdx uint) *testnet.Container {
 
 	clusters := suite.Clusters()
@@ -335,8 +334,7 @@ func (suite *CollectorSuite) Collector(clusterIdx, nodeIdx uint) *testnet.Contai
 	return suite.net.ContainerByID(node.ID())
 }
 
-// ClusterStateFor returns a cluster state instance for the collector node
-// with the given ID.
+// ClusterStateFor returns a cluster state instance for the collector node with the given ID.
 func (suite *CollectorSuite) ClusterStateFor(id flow.Identifier) *clusterstateimpl.State {
 
 	myCluster, _, ok := suite.Clusters().ByNodeID(id)
@@ -351,9 +349,9 @@ func (suite *CollectorSuite) ClusterStateFor(id flow.Identifier) *clusterstateim
 	require.Nil(suite.T(), err, "could not get node db")
 
 	rootQC := unittest.QuorumCertificateFixture(unittest.QCWithRootBlockID(rootBlock.ID()))
-	clusterStateRoot, err := clusterstateimpl.NewStateRoot(rootBlock, rootQC)
+	clusterStateRoot, err := clusterstateimpl.NewStateRoot(rootBlock, rootQC, setup.Counter)
 	suite.NoError(err)
-	clusterState, err := clusterstateimpl.OpenState(db, nil, nil, nil, clusterStateRoot.ClusterID())
+	clusterState, err := clusterstateimpl.OpenState(db, nil, nil, nil, clusterStateRoot.ClusterID(), clusterStateRoot.EpochCounter())
 	require.NoError(suite.T(), err, "could not get cluster state")
 
 	return clusterState

model/cluster/payload.go

@@ -18,7 +18,9 @@ type Payload struct {
 	// the proposer may choose any reference block, so long as it is finalized
 	// and within the epoch the cluster is associated with. If a cluster was
 	// assigned for epoch E, then all of its reference blocks must have a view
-	// in the range [E.FirstView, E.FinalView].
+	// in the range [E.FirstView, E.FinalView]. However, if epoch fallback is
+	// triggered in epoch E, then any reference block with view ≥ E.FirstView
+	// may be used.
 	//
 	// This determines when the collection expires, using the same expiry rules
 	// as transactions. It is also used as the reference point for committee

module/builder/collection/build_ctx.go

@@ -0,0 +1,53 @@
+package collection
+
+import (
+	"github.com/onflow/flow-go/model/flow"
+)
+
+// blockBuildContext encapsulates required information about the cluster chain and
+// main chain state needed to build a new cluster block proposal.
+type blockBuildContext struct {
+	parent                     *flow.Header     // parent of the block we are building
+	clusterChainFinalizedBlock *flow.Header     // finalized block on the cluster chain
+	refChainFinalizedHeight    uint64           // finalized height on reference chain
+	refChainFinalizedID        flow.Identifier  // finalized block ID on reference chain
+	refEpochFirstHeight        uint64           // first height of this cluster's operating epoch
+	refEpochFinalHeight        *uint64          // last height of this cluster's operating epoch (nil if epoch not ended)
+	refEpochFinalID            *flow.Identifier // ID of last block in this cluster's operating epoch (nil if epoch not ended)
+	config                     Config
+}
+
+// highestPossibleReferenceBlockHeight returns the height of the highest possible valid reference block.
+// It is the highest finalized block which is in this cluster's operating epoch.
+func (ctx *blockBuildContext) highestPossibleReferenceBlockHeight() uint64 {
+	if ctx.refEpochFinalHeight != nil {
+		return *ctx.refEpochFinalHeight
+	}
+	return ctx.refChainFinalizedHeight
+}
+
+// highestPossibleReferenceBlockID returns the ID of the highest possible valid reference block.
+// It is the highest finalized block which is in this cluster's operating epoch.
+func (ctx *blockBuildContext) highestPossibleReferenceBlockID() flow.Identifier {
+	if ctx.refEpochFinalID != nil {
+		return *ctx.refEpochFinalID
+	}
+	return ctx.refChainFinalizedID
+}
+
+// lowestPossibleReferenceBlockHeight returns the height of the lowest possible valid reference block.
+// This is the higher of:
+//   - the first block in this cluster's operating epoch
+//   - the lowest block which could be used as a reference block without being
+//     immediately expired (accounting for the configured expiry buffer)
+func (ctx *blockBuildContext) lowestPossibleReferenceBlockHeight() uint64 {
+	// By default, the lowest possible reference block for a non-expired collection has a height
+	// δ below the latest finalized block, for `δ := flow.DefaultTransactionExpiry - ctx.config.ExpiryBuffer`
+	// However, our current Epoch might not have δ finalized blocks yet, in which case the lowest
+	// possible reference block is the first block in the Epoch.
+	delta := uint64(flow.DefaultTransactionExpiry - ctx.config.ExpiryBuffer)
+	if ctx.refChainFinalizedHeight <= ctx.refEpochFirstHeight+delta {
+		return ctx.refEpochFirstHeight
+	}
+	return ctx.refChainFinalizedHeight - delta
+}