Merge branch 'master' into illia-malachyn/6635-tests-for-websocket-controller

Author: illia-malachyn

cmd/util/cmd/verify_execution_result/cmd.go

@@ -53,27 +53,33 @@ func run(*cobra.Command, []string) {
 	chainID := flow.ChainID(flagChain)
 	_ = chainID.Chain()
 
+	lg := log.With().
+		Str("chain", string(chainID)).
+		Str("datadir", flagDatadir).
+		Str("chunk_data_pack_dir", flagChunkDataPackDir).
+		Logger()
+
 	if flagFromTo != "" {
 		from, to, err := parseFromTo(flagFromTo)
 		if err != nil {
-			log.Fatal().Err(err).Msg("could not parse from_to")
+			lg.Fatal().Err(err).Msg("could not parse from_to")
 		}
 
-		log.Info().Msgf("verifying range from %d to %d", from, to)
+		lg.Info().Msgf("verifying range from %d to %d", from, to)
 		err = verifier.VerifyRange(from, to, chainID, flagDatadir, flagChunkDataPackDir)
 		if err != nil {
-			log.Fatal().Err(err).Msgf("could not verify range from %d to %d", from, to)
+			lg.Fatal().Err(err).Msgf("could not verify range from %d to %d", from, to)
 		}
-		log.Info().Msgf("successfully verified range from %d to %d", from, to)
+		lg.Info().Msgf("successfully verified range from %d to %d", from, to)
 
 	} else {
-		log.Info().Msgf("verifying last %d sealed blocks", flagLastK)
+		lg.Info().Msgf("verifying last %d sealed blocks", flagLastK)
 		err := verifier.VerifyLastKHeight(flagLastK, chainID, flagDatadir, flagChunkDataPackDir)
 		if err != nil {
-			log.Fatal().Err(err).Msg("could not verify last k height")
+			lg.Fatal().Err(err).Msg("could not verify last k height")
 		}
 
-		log.Info().Msgf("successfully verified last %d sealed blocks", flagLastK)
+		lg.Info().Msgf("successfully verified last %d sealed blocks", flagLastK)
 	}
 }
 

engine/verification/verifier/verifiers.go

@@ -24,6 +24,7 @@ import (
 // VerifyLastKHeight verifies the last k sealed blocks by verifying all chunks in the results.
 // It assumes the latest sealed block has been executed, and the chunk data packs have not been
 // pruned.
+// Note, it returns nil if certain block is not executed, in this case warning will be logged
 func VerifyLastKHeight(k uint64, chainID flow.ChainID, protocolDataDir string, chunkDataPackDir string) (err error) {
 	closer, storages, chunkDataPacks, state, verifier, err := initStorages(chainID, protocolDataDir, chunkDataPackDir)
 	if err != nil {
@@ -73,6 +74,7 @@ func VerifyLastKHeight(k uint64, chainID flow.ChainID, protocolDataDir string, c
 }
 
 // VerifyRange verifies all chunks in the results of the blocks in the given range.
+// Note, it returns nil if certain block is not executed, in this case warning will be logged
 func VerifyRange(
 	from, to uint64,
 	chainID flow.ChainID,
@@ -124,7 +126,8 @@ func initStorages(chainID flow.ChainID, dataDir string, chunkDataPackDir string)
 		return nil, nil, nil, nil, nil, fmt.Errorf("could not init protocol state: %w", err)
 	}
 
-	chunkDataPackDB, err := storagepebble.OpenDefaultPebbleDB(chunkDataPackDir)
+	// require the chunk data pack data must exist before returning the storage module
+	chunkDataPackDB, err := storagepebble.MustOpenDefaultPebbleDB(chunkDataPackDir)
 	if err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("could not open chunk data pack DB: %w", err)
 	}
@@ -147,6 +150,8 @@ func initStorages(chainID flow.ChainID, dataDir string, chunkDataPackDir string)
 	return closer, storages, chunkDataPacks, state, verifier, nil
 }
 
+// verifyHeight verifies all chunks in the results of the block at the given height.
+// Note: it returns nil if the block is not executed.
 func verifyHeight(
 	height uint64,
 	headers storage.Headers,
@@ -164,6 +169,11 @@ func verifyHeight(
 
 	result, err := results.ByBlockID(blockID)
 	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			log.Warn().Uint64("height", height).Hex("block_id", blockID[:]).Msg("execution result not found")
+			return nil
+		}
+
 		return fmt.Errorf("could not get execution result by block ID %s: %w", blockID, err)
 	}
 	snapshot := state.AtBlockID(blockID)

storage/pebble/open.go

@@ -3,6 +3,8 @@ package pebble
 import (
 	"errors"
 	"fmt"
+	"os"
+	"path/filepath"
 
 	"github.com/cockroachdb/pebble"
 	"github.com/hashicorp/go-multierror"
@@ -54,6 +56,8 @@ func OpenRegisterPebbleDB(dir string) (*pebble.DB, error) {
 
 // OpenDefaultPebbleDB opens a pebble database using default options,
 // such as cache size and comparer
+// If the pebbleDB is not bootstrapped at this folder, it will auto-bootstrap it,
+// use MustOpenDefaultPebbleDB if you want to return error instead
 func OpenDefaultPebbleDB(dir string) (*pebble.DB, error) {
 	cache := pebble.NewCache(DefaultPebbleCacheSize)
 	defer cache.Unref()
@@ -66,6 +70,57 @@ func OpenDefaultPebbleDB(dir string) (*pebble.DB, error) {
 	return db, nil
 }
 
+// MustOpenDefaultPebbleDB returns error if the pebbleDB is not bootstrapped at this folder
+// if bootstrapped, then open the pebbleDB
+func MustOpenDefaultPebbleDB(dir string) (*pebble.DB, error) {
+	err := IsPebbleInitialized(dir)
+	if err != nil {
+		return nil, fmt.Errorf("pebble db is not initialized: %w", err)
+	}
+
+	return OpenDefaultPebbleDB(dir)
+}
+
+// IsPebbleInitialized checks if the given folder contains a valid Pebble DB.
+func IsPebbleInitialized(folderPath string) error {
+	// Check if the folder exists
+	info, err := os.Stat(folderPath)
+	if os.IsNotExist(err) {
+		return fmt.Errorf("directory does not exist: %s", folderPath)
+	}
+	if !info.IsDir() {
+		return fmt.Errorf("not a directory: %s", folderPath)
+	}
+
+	// Look for Pebble-specific files
+	requiredFiles := []string{"CURRENT", "MANIFEST-*"}
+	for _, pattern := range requiredFiles {
+		matches, err := filepath.Glob(filepath.Join(folderPath, pattern))
+		if err != nil {
+			return fmt.Errorf("error checking for files: %v", err)
+		}
+		if len(matches) == 0 {
+			return fmt.Errorf("missing required file: %s", pattern)
+		}
+	}
+
+	// Optionally, validate the CURRENT file references a MANIFEST file
+	currentPath := filepath.Join(folderPath, "CURRENT")
+	currentFile, err := os.Open(currentPath)
+	if err != nil {
+		return fmt.Errorf("error reading CURRENT file: %v", err)
+	}
+	defer currentFile.Close()
+
+	// Basic validation by ensuring the CURRENT file is non-empty
+	stat, err := currentFile.Stat()
+	if err != nil || stat.Size() == 0 {
+		return fmt.Errorf("CURRENT file is invalid")
+	}
+
+	return nil
+}
+
 // ReadHeightsFromBootstrappedDB reads the first and latest height from a bootstrapped register db
 // If the register db is not bootstrapped, it returns storage.ErrNotBootstrapped
 // If the register db is corrupted, it returns an error

storage/pebble/open_test.go

@@ -2,6 +2,7 @@ package pebble
 
 import (
 	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -74,3 +75,29 @@ func TestNewBootstrappedRegistersWithPath(t *testing.T) {
 		require.NoError(t, db2.Close())
 	})
 }
+
+func TestMustOpenDefaultPebbleDB(t *testing.T) {
+	t.Parallel()
+	unittest.RunWithTempDir(t, func(dir string) {
+		// verify error is returned when the db is not bootstrapped
+		_, err := MustOpenDefaultPebbleDB(dir)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "not initialized")
+
+		// bootstrap the db
+		db, err := OpenDefaultPebbleDB(dir)
+		require.NoError(t, err)
+		require.NoError(t, initHeights(db, uint64(10)))
+		require.NoError(t, db.Close())
+		fmt.Println(dir)
+
+		// verify no error is returned when the db is bootstrapped
+		db, err = MustOpenDefaultPebbleDB(dir)
+		require.NoError(t, err)
+
+		h, err := latestStoredHeight(db)
+		require.NoError(t, err)
+		require.Equal(t, uint64(10), h)
+		require.NoError(t, db.Close())
+	})
+}