Merge branch 'master' into janez/expose-node-component-management

Author: janezpodhostnik

Makefile

@@ -205,6 +205,7 @@ generate-mocks: install-mock-generators
 	mockery --name 'BlockTracker' --dir="./engine/access/subscription" --case=underscore --output="./engine/access/subscription/mock"  --outpkg="mock"
 	mockery --name 'DataProvider' --dir="./engine/access/rest/websockets/data_providers" --case=underscore --output="./engine/access/rest/websockets/data_providers/mock"  --outpkg="mock"
 	mockery --name 'DataProviderFactory' --dir="./engine/access/rest/websockets/data_providers" --case=underscore --output="./engine/access/rest/websockets/data_providers/mock"  --outpkg="mock"
+	mockery --name 'WebsocketConnection' --dir="./engine/access/rest/websockets" --case=underscore --output="./engine/access/rest/websockets/mock"  --outpkg="mock"
 	mockery --name 'ExecutionDataTracker' --dir="./engine/access/subscription" --case=underscore --output="./engine/access/subscription/mock"  --outpkg="mock"
 	mockery --name 'ConnectionFactory' --dir="./engine/access/rpc/connection" --case=underscore --output="./engine/access/rpc/connection/mock" --outpkg="mock"
 	mockery --name 'Communicator' --dir="./engine/access/rpc/backend" --case=underscore --output="./engine/access/rpc/backend/mock" --outpkg="mock"
@@ -215,6 +216,7 @@ generate-mocks: install-mock-generators
 	mockery --name 'Storage' --dir=module/executiondatasync/tracker --case=underscore --output="module/executiondatasync/tracker/mock" --outpkg="mocktracker"
 	mockery --name 'ScriptExecutor' --dir=module/execution --case=underscore --output="module/execution/mock" --outpkg="mock"
 	mockery --name 'StorageSnapshot' --dir=fvm/storage/snapshot --case=underscore --output="fvm/storage/snapshot/mock" --outpkg="mock"
+	mockery --name 'WebsocketConnection' --dir=engine/access/rest/websockets --case=underscore --output="engine/access/rest/websockets/mock" --outpkg="mock"
 
 	#temporarily make insecure/ a non-module to allow mockery to create mocks
 	mv insecure/go.mod insecure/go2.mod

cmd/bootstrap/utils/md5.go

@@ -2,7 +2,8 @@ package utils
 
 // The google storage API only provides md5 and crc32 hence overriding the linter flag for md5
 import (
-	"crypto/md5" //nolint:gosec
+	// #nosec
+	"crypto/md5"
 	"io"
 	"os"
 )

cmd/util/cmd/verify_execution_result/cmd.go

@@ -18,6 +18,7 @@ var (
 	flagChunkDataPackDir string
 	flagChain            string
 	flagFromTo           string
+	flagWorkerCount      uint // number of workers to verify the blocks concurrently
 )
 
 // # verify the last 100 sealed blocks
@@ -47,33 +48,47 @@ func init() {
 
 	Cmd.Flags().StringVar(&flagFromTo, "from_to", "",
 		"the height range to verify blocks (inclusive), i.e, 1-1000, 1000-2000, 2000-3000, etc.")
+
+	Cmd.Flags().UintVar(&flagWorkerCount, "worker_count", 1,
+		"number of workers to use for verification, default is 1")
+
 }
 
 func run(*cobra.Command, []string) {
 	chainID := flow.ChainID(flagChain)
 	_ = chainID.Chain()
 
+	if flagWorkerCount < 1 {
+		log.Fatal().Msgf("worker count must be at least 1, but got %v", flagWorkerCount)
+	}
+
+	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)
-		err = verifier.VerifyRange(from, to, chainID, flagDatadir, flagChunkDataPackDir)
+		lg.Info().Msgf("verifying range from %d to %d", from, to)
+		err = verifier.VerifyRange(from, to, chainID, flagDatadir, flagChunkDataPackDir, flagWorkerCount)
 		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)
-		err := verifier.VerifyLastKHeight(flagLastK, chainID, flagDatadir, flagChunkDataPackDir)
+		lg.Info().Msgf("verifying last %d sealed blocks", flagLastK)
+		err := verifier.VerifyLastKHeight(flagLastK, chainID, flagDatadir, flagChunkDataPackDir, flagWorkerCount)
 		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/access/rest/websockets/config.go

@@ -4,6 +4,33 @@ import (
 	"time"
 )
 
+const (
+	// PingPeriod defines the interval at which ping messages are sent to the client.
+	// This value must be less than pongWait, cause it that case the server ensures it sends a ping well before the PongWait
+	// timeout elapses. Each new pong message resets the server's read deadline, keeping the connection alive as long as
+	// the client is responsive.
+	//
+	// Example:
+	// At t=9, the server sends a ping, initial read deadline is t=10 (for the first message)
+	// At t=10, the client responds with a pong. The server resets its read deadline to t=20.
+	// At t=18, the server sends another ping. If the client responds with a pong at t=19, the read deadline is extended to t=29.
+	//
+	// In case of failure:
+	// If the client stops responding, the server will send a ping at t=9 but won't receive a pong by t=10. The server then closes the connection.
+	PingPeriod = (PongWait * 9) / 10
+
+	// PongWait specifies the maximum time to wait for a pong response message from the peer
+	// after sending a ping
+	PongWait = 10 * time.Second
+
+	// WriteWait specifies a timeout for the write operation. If the write
+	// isn't completed within this duration, it fails with a timeout error.
+	// SetWriteDeadline ensures the write operation does not block indefinitely
+	// if the client is slow or unresponsive. This prevents resource exhaustion
+	// and allows the server to gracefully handle timeouts for delayed writes.
+	WriteWait = 10 * time.Second
+)
+
 type Config struct {
 	MaxSubscriptionsPerConnection uint64
 	MaxResponsesPerSecond         uint64

engine/access/rest/websockets/connection.go

@@ -0,0 +1,57 @@
+package websockets
+
+import (
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+type WebsocketConnection interface {
+	ReadJSON(v interface{}) error
+	WriteJSON(v interface{}) error
+	WriteControl(messageType int, deadline time.Time) error
+	Close() error
+	SetReadDeadline(deadline time.Time) error
+	SetWriteDeadline(deadline time.Time) error
+	SetPongHandler(h func(string) error)
+}
+
+type WebsocketConnectionImpl struct {
+	conn *websocket.Conn
+}
+
+func NewWebsocketConnection(conn *websocket.Conn) *WebsocketConnectionImpl {
+	return &WebsocketConnectionImpl{
+		conn: conn,
+	}
+}
+
+var _ WebsocketConnection = (*WebsocketConnectionImpl)(nil)
+
+func (c *WebsocketConnectionImpl) ReadJSON(v interface{}) error {
+	return c.conn.ReadJSON(v)
+}
+
+func (c *WebsocketConnectionImpl) WriteJSON(v interface{}) error {
+	return c.conn.WriteJSON(v)
+}
+
+func (c *WebsocketConnectionImpl) WriteControl(messageType int, deadline time.Time) error {
+	return c.conn.WriteControl(messageType, nil, deadline)
+}
+
+func (c *WebsocketConnectionImpl) Close() error {
+	return c.conn.Close()
+}
+
+func (c *WebsocketConnectionImpl) SetReadDeadline(deadline time.Time) error {
+	return c.conn.SetReadDeadline(deadline)
+}
+
+func (c *WebsocketConnectionImpl) SetWriteDeadline(deadline time.Time) error {
+	return c.conn.SetWriteDeadline(deadline)
+}
+
+func (c *WebsocketConnectionImpl) SetPongHandler(h func(string) error) {
+	c.conn.SetPongHandler(h)
+}