sealing halt on exec fork

Makefile

@@ -121,6 +121,7 @@ generate-mocks:
 	GO111MODULE=on mockery -name 'ConnectionFactory' -dir="./engine/access/rpc/backend" -case=underscore -output="./engine/access/rpc/backend/mock" -outpkg="mock"
 	GO111MODULE=on mockery -name 'IngestRPC' -dir="./engine/execution/ingestion" -case=underscore -tags relic -output="./engine/execution/ingestion/mock" -outpkg="mock"
 	GO111MODULE=on mockery -name '.*' -dir=model/fingerprint -case=underscore -output="./model/fingerprint/mock" -outpkg="mock"
+	GO111MODULE=on mockery -name 'ExecForkActor' --structname 'ExecForkActorMock' -dir=module/mempool/consensus/mock/ -case=underscore -output="./module/mempool/consensus/mock/" -outpkg="mock"
 
 # this ensures there is no unused dependency being added by accident
 .PHONY: tidy

cmd/consensus/main.go

@@ -36,6 +36,7 @@ import (
 	chmodule "github.com/onflow/flow-go/module/chunks"
 	finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
 	"github.com/onflow/flow-go/module/mempool"
+	consensusMempools "github.com/onflow/flow-go/module/mempool/consensus"
 	"github.com/onflow/flow-go/module/mempool/ejectors"
 	"github.com/onflow/flow-go/module/mempool/stdmap"
 	"github.com/onflow/flow-go/module/metrics"
@@ -131,7 +132,11 @@ func main() {
 			// use a custom ejector so we don't eject seals that would break
 			// the chain of seals
 			ejector := ejectors.NewLatestIncorporatedResultSeal(node.Storage.Headers)
-			seals = stdmap.NewIncorporatedResultSeals(sealLimit, stdmap.WithEject(ejector.Eject))
+			seals = stdmap.NewIncorporatedResultSeals(stdmap.WithLimit(sealLimit), stdmap.WithEject(ejector.Eject))
+			seals, err = consensusMempools.NewExecStateForkSuppressor(consensusMempools.LogForkAndCrash(node.Logger), seals, node.DB, node.Logger)
+			if err != nil {
+				return fmt.Errorf("failed to wrap seals mempool into ExecStateForkSuppressor: %w", err)
+			}
 			return nil
 		}).
 		Module("consensus node metrics", func(node *cmd.FlowNodeBuilder) error {

cmd/util/cmd/block_hash_by_height_test.go

@@ -115,7 +115,7 @@ func storeAndIndexHeader(t *testing.T, db *badger.DB, headers *bstorage.Headers,
 }
 
 func storeAndIndexSealFor(t *testing.T, db *badger.DB, seals *bstorage.Seals, h *flow.Header) {
-	seal := unittest.SealFixture()
+	seal := unittest.Seal.Fixture()
 	seal.BlockID = h.ID()
 
 	err := seals.Store(seal)

consensus/integration/nodes_test.go

@@ -187,7 +187,7 @@ func createNode(
 	guarantees, err := stdmap.NewGuarantees(guaranteeLimit)
 	require.NoError(t, err)
 
-	seals := stdmap.NewIncorporatedResultSeals(sealLimit)
+	seals := stdmap.NewIncorporatedResultSeals(stdmap.WithLimit(sealLimit))
 
 	// initialize the block builder
 	build := builder.NewBuilder(metrics, db, state, headersDB, sealsDB, indexDB, guarantees, seals, tracer)

engine/access/rpc/backend/backend_test.go

@@ -661,7 +661,7 @@ func (suite *Suite) TestGetAccount() {
 
 	// setup the latest sealed block
 	header := unittest.BlockHeaderFixture() // create a mock header
-	seal := unittest.SealFixture()          // create a mock seal
+	seal := unittest.Seal.Fixture()         // create a mock seal
 	seal.BlockID = header.ID()              // make the seal point to the header
 
 	suite.snapshot.

engine/consensus/matching/engine.go

@@ -210,12 +210,19 @@ func (e *Engine) onReceipt(originID flow.Identifier, receipt *flow.ExecutionRece
 		Logger()
 
 	resultFinalState, ok := receipt.ExecutionResult.FinalStateCommitment()
-	if !ok { // return
+	if !ok || len(resultFinalState) < 1 { // discard receipt
 		log.Error().Msg("execution receipt without FinalStateCommit received")
 		return engine.NewInvalidInputErrorf("execution receipt without FinalStateCommit: %x", receipt.ID())
 	}
 	log = log.With().Hex("final_state", resultFinalState).Logger()
 
+	resultInitialState, ok := receipt.ExecutionResult.InitialStateCommit()
+	if !ok || len(resultInitialState) < 1 { // discard receipt
+		log.Error().Msg("execution receipt without InitialStateCommit received")
+		return engine.NewInvalidInputErrorf("execution receipt without InitialStateCommit: %x", receipt.ID())
+	}
+	log = log.With().Hex("initial_state", resultInitialState).Logger()
+
 	// CAUTION INCOMPLETE
 	// For many other messages, we check that the message's origin (as established by the
 	// networking layer) is equal to the message's creator as reported by the message itself.
@@ -281,6 +288,7 @@ func (e *Engine) onReceipt(originID flow.Identifier, receipt *flow.ExecutionRece
 	added, err := e.incorporatedResults.Add(flow.NewIncorporatedResult(result.BlockID, result))
 	if err != nil {
 		log.Err(err).Msg("error inserting incorporated result in mempool")
+		return fmt.Errorf("error inserting incorporated result in mempool: %w", err)
 	}
 	if !added {
 		log.Debug().Msg("skipping result already in mempool")
@@ -722,10 +730,13 @@ func (e *Engine) sealResult(incorporatedResult *flow.IncorporatedResult) error {
 	}
 
 	// we don't care if the seal is already in the mempool
-	_ = e.seals.Add(&flow.IncorporatedResultSeal{
+	_, err = e.seals.Add(&flow.IncorporatedResultSeal{
 		IncorporatedResult: incorporatedResult,
 		Seal:               seal,
 	})
+	if err != nil {
+		return fmt.Errorf("failed to store IncorporatedResultSeal in mempool: %w", err)
+	}
 
 	return nil
 }

engine/testutil/nodes.go

@@ -205,7 +205,7 @@ func ConsensusNode(t *testing.T, hub *stub.Hub, identity *flow.Identity, identit
 	approvals, err := stdmap.NewApprovals(1000)
 	require.NoError(t, err)
 
-	seals := stdmap.NewIncorporatedResultSeals(1000)
+	seals := stdmap.NewIncorporatedResultSeals(stdmap.WithLimit(1000))
 
 	// receive collections
 	ingestionEngine, err := consensusingest.New(node.Log, node.Tracer, node.Metrics, node.Metrics, node.Metrics, node.Net, node.State, node.Headers, node.Me, guarantees)

model/flow/block_test.go

@@ -81,7 +81,7 @@ func TestNilProducesSameHashAsEmptySlice(t *testing.T) {
 
 func TestOrderingChangesHash(t *testing.T) {
 
-	seals := unittest.BlockSealsFixture(5)
+	seals := unittest.Seal.Fixtures(5)
 
 	payload1 := flow.Payload{
 		Seals: seals,

model/flow/incorporated_result.go

@@ -1,6 +1,8 @@
 package flow
 
-import "github.com/onflow/flow-go/crypto"
+import (
+	"github.com/onflow/flow-go/crypto"
+)
 
 // IncorporatedResult is a wrapper around an ExecutionResult which contains the
 // ID of the first block on its fork in which it was incorporated.
@@ -35,7 +37,14 @@ func NewIncorporatedResult(incorporatedBlockID Identifier, result *ExecutionResu
 // ID implements flow.Entity.ID for IncorporatedResult to make it capable of
 // being stored directly in mempools and storage.
 func (ir *IncorporatedResult) ID() Identifier {
-	return MakeID(ir)
+	body := struct {
+		IncorporatedBlockID Identifier
+		Result              *ExecutionResult
+	}{
+		IncorporatedBlockID: ir.IncorporatedBlockID,
+		Result:              ir.Result,
+	}
+	return MakeID(body)
 }
 
 // CheckSum implements flow.Entity.CheckSum for IncorporatedResult to make it

module/builder/consensus/builder.go

@@ -4,7 +4,6 @@ package consensus
 
 import (
 	"bytes"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
@@ -234,43 +233,9 @@ func (b *Builder) BuildOn(parentID flow.Identifier, setter func(*flow.Header) er
 	// roadmap (https://github.com/dapperlabs/flow-go/issues/4872)
 
 	// create a mapping of block to seal for all seals in our pool
-	// We consider two seals as inconsistent, if they have different start or end states
-	encounteredInconsistentSealsForSameBlock := false
 	byBlock := make(map[flow.Identifier]*flow.IncorporatedResultSeal)
 	for _, irSeal := range b.sealPool.All() {
-		if len(irSeal.IncorporatedResult.Result.Chunks) < 1 {
-			return nil, fmt.Errorf("ExecutionResult without chunks: %v", irSeal.IncorporatedResult.Result.ID())
-		}
-		if len(irSeal.Seal.FinalState) < 1 {
-			// respective Execution Result should have been rejected by matching engine
-			return nil, fmt.Errorf("seal with empty state commitment: %v", irSeal.ID())
-		}
-		if irSeal2, found := byBlock[irSeal.Seal.BlockID]; found {
-			sc1json, err := json.Marshal(irSeal)
-			if err != nil {
-				return nil, err
-			}
-			sc2json, err := json.Marshal(irSeal2)
-			if err != nil {
-				return nil, err
-			}
-
-			// check whether seals are inconsistent:
-			if !bytes.Equal(irSeal.Seal.FinalState, irSeal2.Seal.FinalState) ||
-				!bytes.Equal(irSeal.IncorporatedResult.Result.Chunks[0].StartState, irSeal2.IncorporatedResult.Result.Chunks[0].StartState) {
-				fmt.Printf("ERROR: inconsistent seals for the same block %v: %s and %s", irSeal.Seal.BlockID, string(sc1json), string(sc2json))
-				encounteredInconsistentSealsForSameBlock = true
-			} else {
-				fmt.Printf("WARNING: multiple seals with different IDs for the same block %v: %s and %s", irSeal.Seal.BlockID, string(sc1json), string(sc2json))
-			}
-
-		} else {
-			byBlock[irSeal.Seal.BlockID] = irSeal
-		}
-	}
-	if encounteredInconsistentSealsForSameBlock {
-		// in case we find inconsistent seals, do not seal anything
-		byBlock = make(map[flow.Identifier]*flow.IncorporatedResultSeal)
+		byBlock[irSeal.Seal.BlockID] = irSeal
 	}
 
 	// get the parent's block seal, which constitutes the beginning of the

module/builder/consensus/builder_test.go

@@ -112,7 +112,11 @@ func (bs *BuilderSuite) createAndRecordBlock(parentBlock *flow.Block) *flow.Bloc
 
 		incorporatedResultForPrevBlock = unittest.IncorporatedResult.Fixture(
 			unittest.IncorporatedResult.WithResult(previousResult),
-			unittest.IncorporatedResult.WithIncorporatedBlockID(block.ID()),
+			unittest.IncorporatedResult.WithIncorporatedBlockID(parentBlock.ID()),
+			// For sealing phase 2, the value for IncorporatedBlockID is the block the result pertains to (here parentBlock).
+			// In later development phases, we will change the logic such that IncorporatedBlockID references the
+			// block which actually incorporates the result:
+			//unittest.IncorporatedResult.WithIncorporatedBlockID(block.ID()),
 		)
 		result := unittest.ExecutionResultFixture(
 			unittest.WithBlock(&block),
@@ -140,21 +144,11 @@ func (bs *BuilderSuite) createAndRecordBlock(parentBlock *flow.Block) *flow.Bloc
 // IncorporatedResultSeal, which ties the seal to the incorporated result it
 // seals, is also recorded for future access.
 func (bs *BuilderSuite) chainSeal(incorporatedResult *flow.IncorporatedResult) {
-
-	finalState, _ := incorporatedResult.Result.FinalStateCommitment()
-
-	seal := &flow.Seal{
-		BlockID:    incorporatedResult.Result.BlockID,
-		ResultID:   incorporatedResult.Result.ID(),
-		FinalState: finalState,
-	}
-	bs.chain = append(bs.chain, seal)
-
-	incorporatedResultSeal := &flow.IncorporatedResultSeal{
-		IncorporatedResult: incorporatedResult,
-		Seal:               seal,
-	}
-
+	incorporatedResultSeal := unittest.IncorporatedResultSeal.Fixture(
+		unittest.IncorporatedResultSeal.WithResult(incorporatedResult.Result),
+		unittest.IncorporatedResultSeal.WithIncorporatedBlockID(incorporatedResult.IncorporatedBlockID),
+	)
+	bs.chain = append(bs.chain, incorporatedResultSeal.Seal)
 	bs.irsMap[incorporatedResultSeal.ID()] = incorporatedResultSeal
 	bs.irsList = append(bs.irsList, incorporatedResultSeal)
 }
@@ -524,23 +518,36 @@ func (bs *BuilderSuite) TestPayloadSealAllValid() {
 	bs.Assert().ElementsMatch(bs.chain, bs.assembled.Seals, "should have included valid chain of seals")
 }
 
-func (bs *BuilderSuite) TestPayloadSealSomeValid() {
-
-	bs.pendingSeals = bs.irsMap
-
-	// add some invalid seals to the mempool
-	for i := 0; i < 8; i++ {
-		invalid := &flow.IncorporatedResultSeal{
-			IncorporatedResult: unittest.IncorporatedResultFixture(),
-			Seal:               unittest.SealFixture(),
+// TestPayloadSealOnlyFork verifies that builder only includes seals for blocks in the respective fork
+// (and _not_ seals for sealable blocks on other forks)
+func (bs *BuilderSuite) TestPayloadSealOnlyFork() {
+	// in the test setup, we already created a single fork
+	//  [first] <- [F0] <- [F1] <- [F2] <- [F3] <- [A0] <- [A1] <- [A2] <- [A3]
+	// Where block
+	//   * [first] is sealed and finalized
+	//   * [F0] ... [F3] are finalized but _not_ sealed
+	//   * [A0] ... [A3] are _not_ finalized and _not_ sealed
+	// We now create an additional fork:  [F3] <- [B0] <- [B1] <- ... <- [B7]
+	var forkHead *flow.Block
+	for f := 0; f < 100; f++ {
+		forkHead = bs.blocks[bs.finalID]
+		for i := 0; i < 8; i++ {
+			forkHead = bs.createAndRecordBlock(forkHead)
+			// Method createAndRecordBlock adds a seal for every block into the mempool.
 		}
-		bs.pendingSeals[invalid.ID()] = invalid
 	}
 
-	_, err := bs.build.BuildOn(bs.parentID, bs.setter)
+	bs.pendingSeals = bs.irsMap
+	_, err := bs.build.BuildOn(forkHead.ID(), bs.setter)
 	bs.Require().NoError(err)
+
+	// expected seals: [F0] <- ... <- [F3] <- [B0] <- ... <- [B7]
+	// Note: bs.chain contains seals for blocks  F0 ... F3 then A0 ... A3 and then B0 ... B7
+	bs.Assert().Equal(12, len(bs.assembled.Seals), "unexpected number of seals")
+	bs.Assert().ElementsMatch(bs.chain[:4], bs.assembled.Seals[:4], "should have included only valid chain of seals")
+	bs.Assert().ElementsMatch(bs.chain[len(bs.chain)-8:], bs.assembled.Seals[4:], "should have included only valid chain of seals")
+
 	bs.Assert().Empty(bs.assembled.Guarantees, "should have no guarantees in payload with empty mempool")
-	bs.Assert().ElementsMatch(bs.chain, bs.assembled.Seals, "should have included only valid chain of seals")
 }
 
 func (bs *BuilderSuite) TestPayloadSealCutoffChain() {

module/mempool/common.go

@@ -0,0 +1,8 @@
+package mempool
+
+import "github.com/onflow/flow-go/model/flow"
+
+// OnEjection is a callback which a mempool executes on ejecting
+// one of its elements. The callbacks are executed from within the thread
+// that serves the mempool. Implementations should be non-blocking.
+type OnEjection func(flow.Entity)

module/mempool/consensus/exec_fork_actor.go

@@ -0,0 +1,28 @@
+package consensus
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/model/flow"
+)
+
+type ExecForkActor func([]*flow.IncorporatedResultSeal)
+
+func LogForkAndCrash(log zerolog.Logger) ExecForkActor {
+	return func(conflictingSeals []*flow.IncorporatedResultSeal) {
+		l := log.Fatal().Int("number conflicting seals", len(conflictingSeals))
+		for i, s := range conflictingSeals {
+			sealAsJson, err := json.Marshal(s)
+			if err != nil {
+				err = fmt.Errorf("failed to marshal candidate seal to json: %w", err)
+				l.Str(fmt.Sprintf("seal_%d", i), err.Error())
+				continue
+			}
+			l = l.Str(fmt.Sprintf("seal_%d", i), string(sealAsJson))
+		}
+		l.Msg("inconsistent seals for the same block")
+	}
+}