Merge branch 'master' into gregor/callbacks/process-fallback

Author: j1010001

cmd/util/cmd/read-light-block/read_light_block_test.go

@@ -25,14 +25,14 @@ func TestReadClusterRange(t *testing.T) {
 		err := unittest.WithLock(t, lockManager, storage.LockInsertOrFinalizeClusterBlock, func(lctx lockctx.Context) error {
 			// add parent as boundary
 			err := db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-				return operation.IndexClusterBlockHeight(lctx, rw.Writer(), parent.ChainID, parent.Height, parent.ID())
+				return operation.IndexClusterBlockHeight(lctx, rw, parent.ChainID, parent.Height, parent.ID())
 			})
 			if err != nil {
 				return err
 			}
 
 			return db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-				return operation.UpsertClusterFinalizedHeight(lctx, rw.Writer(), parent.ChainID, parent.Height)
+				return operation.BootstrapClusterFinalizedHeight(lctx, rw, parent.ChainID, parent.Height)
 			})
 		})
 		require.NoError(t, err)

state/cluster/badger/mutator_test.go

@@ -389,7 +389,7 @@ func (suite *MutatorSuite) TestExtend_Success() {
 
 	// the block should be indexed by its parent
 	var childIDs flow.IdentifierList
-	err = procedure.LookupBlockChildren(r, suite.genesis.ID(), &childIDs)
+	err = operation.RetrieveBlockChildren(r, suite.genesis.ID(), &childIDs)
 	suite.Assert().Nil(err)
 	suite.Require().Len(childIDs, 1)
 	suite.Assert().Equal(proposal.Block.ID(), childIDs[0])

state/cluster/badger/snapshot.go

@@ -1,30 +1,46 @@
 package badger
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/onflow/flow-go/model/cluster"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/irrecoverable"
 	clusterState "github.com/onflow/flow-go/state/cluster"
+	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/operation"
 	"github.com/onflow/flow-go/storage/procedure"
 )
 
-// Snapshot represents a snapshot of chain state anchored at a particular
-// reference block.
+// Snapshot pertains to a specific fork of the collector cluster consensus. Specifically,
+// it references one block denoted as the `Head`. This Snapshot type is for collector
+// clusters, so we are referencing a cluster block, aka collection, here.
+//
+// This implementation must be used for KNOWN reference BLOCKs only.
 type Snapshot struct {
-	err     error
 	state   *State
 	blockID flow.Identifier
 }
 
 var _ clusterState.Snapshot = (*Snapshot)(nil)
 
-func (s *Snapshot) Collection() (*flow.Collection, error) {
-	if s.err != nil {
-		return nil, s.err
+// newSnapshot instantiates a new snapshot for the given collection ID.
+// CAUTION: This constructor must be called for KNOWN blocks.
+// For unknown blocks, please use `invalid.NewSnapshot` or `invalid.NewSnapshotf`.
+func newSnapshot(state *State, blockID flow.Identifier) *Snapshot {
+	return &Snapshot{
+		state:   state,
+		blockID: blockID,
 	}
+}
 
+// Collection returns the collection designated as the reference for this
+// snapshot. Technically, this is a portion of the payload of a cluster block.
+//
+// By contract of the constructor, the blockID must correspond to a known collection in the database.
+// No error returns are expected during normal operation.
+func (s *Snapshot) Collection() (*flow.Collection, error) {
 	// get the payload
 	var payload cluster.Payload
 	err := procedure.RetrieveClusterPayload(s.state.db.Reader(), s.blockID, &payload)
@@ -36,38 +52,50 @@ func (s *Snapshot) Collection() (*flow.Collection, error) {
 	return &collection, nil
 }
 
+// Head returns the header of the collection that designated as the reference for this
+// snapshot.
+//
+// By contract of the constructor, the blockID must correspond to a known collection in the database.
+// No error returns are expected during normal operation.
 func (s *Snapshot) Head() (*flow.Header, error) {
-	if s.err != nil {
-		return nil, s.err
-	}
-
 	var head flow.Header
-	err := s.head(&head)
+	err := operation.RetrieveHeader(s.state.db.Reader(), s.blockID, &head)
+	if err != nil {
+		// `storage.ErrNotFound` is the only error that the storage layer may return other than exceptions.
+		// In the context of this call, `s.blockID` should correspond to a known block, so receiving a
+		// `storage.ErrNotFound` is an exception here.
+		return nil, irrecoverable.NewExceptionf("could not retrieve header for block (%s): %w", s.blockID, err)
+	}
 	return &head, err
 }
 
+// Pending returns the IDs of all collections descending from the snapshot's head collection.
+// The result is ordered such that parents are included before their children. While only valid
+// descendants will be returned, note that the descendants may not be finalized yet.
+// By contract of the constructor, the blockID must correspond to a known collection in the database.
+// No error returns are expected during normal operation.
 func (s *Snapshot) Pending() ([]flow.Identifier, error) {
-	if s.err != nil {
-		return nil, s.err
-	}
 	return s.pending(s.blockID)
 }
 
-// head finds the header referenced by the snapshot.
-func (s *Snapshot) head(head *flow.Header) error {
-	// get the snapshot header
-	err := operation.RetrieveHeader(s.state.db.Reader(), s.blockID, head)
-	if err != nil {
-		return fmt.Errorf("could not retrieve header for block (%s): %w", s.blockID, err)
-	}
-	return nil
-}
-
+// pending returns a slice with all blocks descending from the given blockID (children, grandchildren, etc).
+// CAUTION: this function behaves only correctly for known blocks, which should always be the case as
+// required by the constructor.
+// No error returns are expected during normal operation.
 func (s *Snapshot) pending(blockID flow.Identifier) ([]flow.Identifier, error) {
 	var pendingIDs flow.IdentifierList
-	err := procedure.LookupBlockChildren(s.state.db.Reader(), blockID, &pendingIDs)
+	err := operation.RetrieveBlockChildren(s.state.db.Reader(), blockID, &pendingIDs)
 	if err != nil {
-		return nil, fmt.Errorf("could not get pending children: %w", err)
+		if !errors.Is(err, storage.ErrNotFound) {
+			return nil, fmt.Errorf("could not get pending block %v: %w", blockID, err)
+		}
+
+		// The low-level storage returns `storage.ErrNotFound` in two cases:
+		// 1. the block/collection is unknown
+		// 2. the block/collection is known but no children have been indexed yet
+		// By contract of the constructor, the blockID must correspond to a known collection in the database.
+		// A snapshot with s.err == nil is only created for known blocks. Hence, only case 2 is
+		// possible here, and we just return an empty list.
 	}
 
 	for _, pendingID := range pendingIDs {

state/cluster/badger/state.go

@@ -10,7 +10,9 @@ import (
 	clustermodel "github.com/onflow/flow-go/model/cluster"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/state"
 	"github.com/onflow/flow-go/state/cluster"
+	"github.com/onflow/flow-go/state/cluster/invalid"
 	"github.com/onflow/flow-go/storage"
 	"github.com/onflow/flow-go/storage/operation"
 	"github.com/onflow/flow-go/storage/procedure"
@@ -22,6 +24,8 @@ type State struct {
 	epoch     uint64       // the operating epoch for the cluster
 }
 
+var _ cluster.State = (*State)(nil)
+
 // Bootstrap initializes the persistent cluster state with a genesis block.
 // The genesis block must have height 0, a parent hash of 32 zero bytes,
 // and an empty collection as payload.
@@ -62,13 +66,12 @@ func Bootstrap(db storage.DB, lockManager lockctx.Manager, stateRoot *StateRoot)
 			return fmt.Errorf("could not insert genesis block: %w", err)
 		}
 		// insert block height -> ID mapping
-		err = operation.IndexClusterBlockHeight(lctx, rw.Writer(), chainID, genesis.Height, genesis.ID())
+		err = operation.IndexClusterBlockHeight(lctx, rw, chainID, genesis.Height, genesis.ID())
 		if err != nil {
 			return fmt.Errorf("failed to map genesis block height to block: %w", err)
 		}
 		// insert boundary
-		err = operation.UpsertClusterFinalizedHeight(lctx, rw.Writer(), chainID, genesis.Height)
-		// insert started view for hotstuff
+		err = operation.BootstrapClusterFinalizedHeight(lctx, rw, chainID, genesis.Height)
 		if err != nil {
 			return fmt.Errorf("could not insert genesis boundary: %w", err)
 		}
@@ -79,7 +82,7 @@ func Bootstrap(db storage.DB, lockManager lockctx.Manager, stateRoot *StateRoot)
 		}
 
 		livenessData := &hotstuff.LivenessData{
-			CurrentView: genesis.View + 1,
+			CurrentView: genesis.View + 1, // starting view for hotstuff
 			NewestQC:    rootQC,
 		}
 		// insert safety data
@@ -131,42 +134,38 @@ func (s *State) Params() cluster.Params {
 }
 
 func (s *State) Final() cluster.Snapshot {
-	// get the finalized block ID
-	var blockID flow.Identifier
-	err := (func(r storage.Reader) error {
-		var boundary uint64
-		err := operation.RetrieveClusterFinalizedHeight(r, s.clusterID, &boundary)
-		if err != nil {
-			return fmt.Errorf("could not retrieve finalized boundary: %w", err)
-		}
-
-		err = operation.LookupClusterBlockHeight(r, s.clusterID, boundary, &blockID)
-		if err != nil {
-			return fmt.Errorf("could not retrieve finalized ID: %w", err)
-		}
-
-		return nil
-	})(s.db.Reader())
-
+	// get height of latest finalized collection and then the ID of the collection with the corresponding height
+	r := s.db.Reader()
+	var latestFinalizedClusterHeight uint64
+	err := operation.RetrieveClusterFinalizedHeight(r, s.clusterID, &latestFinalizedClusterHeight)
 	if err != nil {
-		return &Snapshot{
-			err: err,
-		}
+		return invalid.NewSnapshotf("could not retrieve finalized boundary: %w", err)
 	}
 
-	snapshot := &Snapshot{
-		state:   s,
-		blockID: blockID,
+	var blockID flow.Identifier
+	err = operation.LookupClusterBlockHeight(r, s.clusterID, latestFinalizedClusterHeight, &blockID)
+	if err != nil {
+		return invalid.NewSnapshotf("could not retrieve finalized ID: %w", err)
 	}
-	return snapshot
+
+	return newSnapshot(s, blockID)
 }
 
+// AtBlockID returns the snapshot of the persistent cluster at the given
+// block ID. It is available for any block that was introduced into the
+// cluster state, and can thus represent an ambiguous state that was or
+// will never be finalized.
+// If the block is unknown, it returns an invalid snapshot, which returns
+// state.ErrUnknownSnapshotReference for all methods
 func (s *State) AtBlockID(blockID flow.Identifier) cluster.Snapshot {
-	snapshot := &Snapshot{
-		state:   s,
-		blockID: blockID,
+	exists, err := operation.BlockExists(s.db.Reader(), blockID)
+	if err != nil {
+		return invalid.NewSnapshotf("could not check existence of reference block: %w", err)
+	}
+	if !exists {
+		return invalid.NewSnapshotf("unknown block %x: %w", blockID, state.ErrUnknownSnapshotReference)
 	}
-	return snapshot
+	return newSnapshot(s, blockID)
 }
 
 // IsBootstrapped returns whether the database contains a bootstrapped state.

state/cluster/badger/state_test.go

@@ -0,0 +1,90 @@
+package badger
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/state"
+	"github.com/onflow/flow-go/storage"
+	"github.com/onflow/flow-go/storage/operation/dbtest"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// TestUnknownSnapshotReference verifies that AtBlockID returns a snapshot that
+// returns state.ErrUnknownSnapshotReference for all methods when given an unknown block ID.
+func TestUnknownSnapshotReference(t *testing.T) {
+	dbtest.RunWithDB(t, func(t *testing.T, db storage.DB) {
+		lockManager := storage.NewTestingLockManager()
+
+		// Setup
+		genesis, err := unittest.ClusterBlock.Genesis()
+		require.NoError(t, err)
+
+		root := unittest.RootSnapshotFixture(unittest.IdentityListFixture(5, unittest.WithAllRoles()))
+		epochCounter := root.Encodable().SealingSegment.LatestProtocolStateEntry().EpochEntry.EpochCounter()
+
+		clusterStateRoot, err := NewStateRoot(genesis, unittest.QuorumCertificateFixture(), epochCounter)
+		require.NoError(t, err)
+		clusterState, err := Bootstrap(db, lockManager, clusterStateRoot)
+		require.NoError(t, err)
+
+		// Test
+		unknownBlockID := unittest.IdentifierFixture()
+		snapshot := clusterState.AtBlockID(unknownBlockID)
+
+		// Verify that Collection() returns state.ErrUnknownSnapshotReference
+		_, err = snapshot.Collection()
+		assert.Error(t, err)
+		assert.ErrorIs(t, err, state.ErrUnknownSnapshotReference)
+
+		// Verify that Head() returns state.ErrUnknownSnapshotReference
+		_, err = snapshot.Head()
+		assert.Error(t, err)
+		assert.ErrorIs(t, err, state.ErrUnknownSnapshotReference)
+
+		// Verify that Pending() returns state.ErrUnknownSnapshotReference
+		_, err = snapshot.Pending()
+		assert.Error(t, err)
+		assert.ErrorIs(t, err, state.ErrUnknownSnapshotReference)
+	})
+}
+
+// TestValidSnapshotReference verifies that AtBlockID returns a working snapshot
+// when given a valid block ID (the genesis block).
+func TestValidSnapshotReference(t *testing.T) {
+	dbtest.RunWithDB(t, func(t *testing.T, db storage.DB) {
+		lockManager := storage.NewTestingLockManager()
+
+		// Setup
+		genesis, err := unittest.ClusterBlock.Genesis()
+		require.NoError(t, err)
+
+		root := unittest.RootSnapshotFixture(unittest.IdentityListFixture(5, unittest.WithAllRoles()))
+		epochCounter := root.Encodable().SealingSegment.LatestProtocolStateEntry().EpochEntry.EpochCounter()
+
+		clusterStateRoot, err := NewStateRoot(genesis, unittest.QuorumCertificateFixture(), epochCounter)
+		require.NoError(t, err)
+		clusterState, err := Bootstrap(db, lockManager, clusterStateRoot)
+		require.NoError(t, err)
+
+		// Test with valid block ID (genesis block)
+		snapshot := clusterState.AtBlockID(genesis.ID())
+
+		// Verify that Collection() works correctly
+		collection, err := snapshot.Collection()
+		assert.NoError(t, err)
+		assert.Equal(t, &genesis.Payload.Collection, collection)
+
+		// Verify that Head() works correctly
+		head, err := snapshot.Head()
+		assert.NoError(t, err)
+		assert.Equal(t, genesis.ToHeader().ID(), head.ID())
+
+		// Verify that Pending() works correctly (should return empty list for genesis)
+		pending, err := snapshot.Pending()
+		assert.NoError(t, err)
+		assert.Empty(t, pending)
+	})
+}

state/cluster/invalid/snapshot.go

@@ -0,0 +1,47 @@
+package invalid
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/state"
+	"github.com/onflow/flow-go/state/cluster"
+)
+
+// Snapshot represents a snapshot that does not exist or could not be queried.
+type Snapshot struct {
+	err error
+}
+
+// NewSnapshot returns a new invalid snapshot, containing an error describing why the
+// snapshot could not be retrieved. The following are typical
+// errors resulting in the construction of an invalid Snapshot:
+//   - state.ErrUnknownSnapshotReference if the reference point for the snapshot
+//     (height or block ID) does not resolve to a queriable block in the state.
+//   - generic error in case of unexpected state inconsistencies or bugs
+func NewSnapshot(err error) *Snapshot {
+	if errors.Is(err, state.ErrUnknownSnapshotReference) {
+		return &Snapshot{err: err}
+	}
+	return &Snapshot{fmt.Errorf("critical unexpected error querying snapshot: %w", err)}
+}
+
+var _ cluster.Snapshot = (*Snapshot)(nil)
+
+// NewSnapshotf is NewSnapshot with ergonomic error formatting.
+func NewSnapshotf(msg string, args ...interface{}) *Snapshot {
+	return NewSnapshot(fmt.Errorf(msg, args...))
+}
+
+func (u *Snapshot) Collection() (*flow.Collection, error) {
+	return nil, u.err
+}
+
+func (u *Snapshot) Head() (*flow.Header, error) {
+	return nil, u.err
+}
+
+func (u *Snapshot) Pending() ([]flow.Identifier, error) {
+	return nil, u.err
+}