Merge branch 'master' into leo/refactor-storage-collections-for-an

Author: zhangchiqing

.github/workflows/flaky-test-monitor.yml

@@ -55,7 +55,7 @@ jobs:
       matrix:
         targets: ${{ fromJSON(needs.create-dynamic-test-matrix.outputs.dynamic-matrix)}}
     # need to set image explicitly due to GitHub logging issue as described in https://github.com/onflow/flow-go/pull/3087#issuecomment-1234383202
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout repo
         uses: actions/checkout@v4

access/api.go

@@ -203,41 +203,20 @@ type API interface {
 	//
 	// If invalid parameters will be supplied SubscribeBlockDigestsFromLatest will return a failed subscription.
 	SubscribeBlockDigestsFromLatest(ctx context.Context, blockStatus flow.BlockStatus) subscription.Subscription
-	// SubscribeTransactionStatusesFromStartBlockID subscribes to transaction status updates for a given transaction ID.
-	// Monitoring begins from the specified block ID. The subscription streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
-	// these final statuses, the subscription will automatically terminate.
+	// SubscribeTransactionStatuses subscribes to transaction status updates for a given transaction ID. Monitoring starts
+	// from the latest block to obtain the current transaction status. If the transaction is already in the final state
+	// ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]), all statuses will be prepared and sent to the client
+	// sequentially. If the transaction is not in the final state, the subscription will stream status updates until the transaction
+	// reaches the final state. Once a final state is reached, the subscription will automatically terminate.
 	//
 	// Parameters:
-	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
-	//   - txID: The identifier of the transaction to monitor.
-	//   - startBlockID: The block ID from which to start monitoring.
-	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
-	SubscribeTransactionStatusesFromStartBlockID(ctx context.Context, txID flow.Identifier, startBlockID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
-	// SubscribeTransactionStatusesFromStartHeight subscribes to transaction status updates for a given transaction ID.
-	// Monitoring begins from the specified block height. The subscription streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
-	// these final statuses, the subscription will automatically terminate.
-	//
-	// Parameters:
-	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
+	//   - ctx: Context to manage the subscription's lifecycle, including cancellation.
 	//   - txID: The unique identifier of the transaction to monitor.
-	//   - startHeight: The block height from which to start monitoring.
 	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
-	SubscribeTransactionStatusesFromStartHeight(ctx context.Context, txID flow.Identifier, startHeight uint64, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
-	// SubscribeTransactionStatusesFromLatest subscribes to transaction status updates for a given transaction ID.
-	// Monitoring begins from the latest block. The subscription streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). When the transaction reaches one of
-	// these final statuses, the subscription will automatically terminate.
-	//
-	// Parameters:
-	//   - ctx: The context to manage the subscription's lifecycle, including cancellation.
-	//   - txID: The unique identifier of the transaction to monitor.
-	//   - requiredEventEncodingVersion: The version of event encoding required for the subscription.
-	SubscribeTransactionStatusesFromLatest(ctx context.Context, txID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
-	// SendAndSubscribeTransactionStatuses sends a transaction to the network and subscribes to its status updates.
+	SubscribeTransactionStatuses(ctx context.Context, txID flow.Identifier, requiredEventEncodingVersion entities.EventEncodingVersion) subscription.Subscription
+	// SendAndSubscribeTransactionStatuses sends a transaction to the execution node and subscribes to its status updates.
 	// Monitoring begins from the reference block saved in the transaction itself and streams status updates until the transaction
-	// reaches a final state (TransactionStatusSealed or TransactionStatusExpired). Once a final status is reached, the subscription
+	// reaches the final state ([flow.TransactionStatusSealed] or [flow.TransactionStatusExpired]). Once the final status has been reached, the subscription
 	// automatically terminates.
 	//
 	// Parameters:
@@ -261,6 +240,14 @@ type TransactionResult struct {
 	BlockHeight   uint64
 }
 
+func (r *TransactionResult) IsExecuted() bool {
+	return r.Status == flow.TransactionStatusExecuted || r.Status == flow.TransactionStatusSealed
+}
+
+func (r *TransactionResult) IsFinal() bool {
+	return r.Status == flow.TransactionStatusSealed || r.Status == flow.TransactionStatusExpired
+}
+
 func TransactionResultToMessage(result *TransactionResult) *access.TransactionResultResponse {
 	return &access.TransactionResultResponse{
 		Status:        entities.TransactionStatus(result.Status),
@@ -314,10 +301,17 @@ type CompatibleRange struct {
 
 // NodeVersionInfo contains information about node, such as semver, commit, sporkID, protocolVersion, etc
 type NodeVersionInfo struct {
-	Semver               string
-	Commit               string
-	SporkId              flow.Identifier
-	ProtocolVersion      uint64
+	Semver  string
+	Commit  string
+	SporkId flow.Identifier
+	// ProtocolVersion is the deprecated protocol version number.
+	// Deprecated: Previously this referred to the major software version as of the most recent spork.
+	// Replaced by protocol_state_version.
+	ProtocolVersion uint64
+	// ProtocolStateVersion is the Protocol State version as of the latest finalized block.
+	// This tracks the schema version of the Protocol State and is used to coordinate breaking changes in the Protocol.
+	// Version numbers are monotonically increasing.
+	ProtocolStateVersion uint64
 	SporkRootBlockHeight uint64
 	NodeRootBlockHeight  uint64
 	CompatibleRange      *CompatibleRange

access/handler.go

@@ -98,6 +98,7 @@ func (h *Handler) GetNodeVersionInfo(
 			Commit:               nodeVersionInfo.Commit,
 			SporkId:              nodeVersionInfo.SporkId[:],
 			ProtocolVersion:      nodeVersionInfo.ProtocolVersion,
+			ProtocolStateVersion: nodeVersionInfo.ProtocolStateVersion,
 			SporkRootBlockHeight: nodeVersionInfo.SporkRootBlockHeight,
 			NodeRootBlockHeight:  nodeVersionInfo.NodeRootBlockHeight,
 			CompatibleRange:      CompatibleRangeToMessage(nodeVersionInfo.CompatibleRange),

access/mock/api.go

@@ -120,3 +120,8 @@ curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"
 ```
 curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"commandName": "backfill-tx-error-messages", "data": { "start-height": 340, "end-height": 343, "execution-node-ids":["ec7b934df29248d574ae1cc33ae77f22f0fcf96a79e009224c46374d1837824e", "8cbdc8d24a28899a33140cb68d4146cd6f2f6c18c57f54c299f26351d126919e"] }}'
 ```
+
+### Trigger chunk data pack pebble database checkpoint
+```
+curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"commandName": "create-chunk-data-packs-checkpoint" }'
+```

admin/README.md

@@ -7,7 +7,6 @@ import (
 	"github.com/onflow/flow-go/admin"
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/engine/access/ingestion/tx_error_messages"
-	commonrpc "github.com/onflow/flow-go/engine/common/rpc"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
 	"github.com/onflow/flow-go/state/protocol"
@@ -177,7 +176,7 @@ func (b *BackfillTxErrorMessagesCommand) parseExecutionNodeIds(executionNodeIdsI
 		if len(executionNodeIds) == 0 {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", "must be a non empty list of strings", executionNodeIdsIn)
 		}
-		requestedENIdentifiers, err := commonrpc.IdentifierList(executionNodeIds)
+		requestedENIdentifiers, err := flow.IdentifierListFromHex(executionNodeIds)
 		if err != nil {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", err.Error(), executionNodeIdsIn)
 		}

admin/commands/storage/backfill_tx_error_messages.go

@@ -0,0 +1,57 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"path"
+	"time"
+
+	"github.com/cockroachdb/pebble"
+	"github.com/rs/zerolog/log"
+
+	"github.com/onflow/flow-go/admin"
+	"github.com/onflow/flow-go/admin/commands"
+)
+
+var _ commands.AdminCommand = (*ChunksCheckpointCommand)(nil)
+
+// ChunksCheckpointCommand creates a checkpoint for pebble database for querying the data
+// while keeping the node alive.
+type ChunksCheckpointCommand struct {
+	checkpointDir  string
+	chunkDataPacks *pebble.DB
+}
+
+func NewChunksCheckpointCommand(dir string, chunkDataPacks *pebble.DB) commands.AdminCommand {
+	return &ChunksCheckpointCommand{
+		checkpointDir:  dir,
+		chunkDataPacks: chunkDataPacks,
+	}
+}
+
+func (c *ChunksCheckpointCommand) Handler(ctx context.Context, req *admin.CommandRequest) (interface{}, error) {
+	log.Info().Msgf("admintool: creating chunkDataPacks database checkpoint")
+
+	targetDir := nextTmpFolder(c.checkpointDir)
+
+	log.Info().Msgf("admintool: creating chunkDataPacks database checkpoint at: %v", targetDir)
+
+	err := c.chunkDataPacks.Checkpoint(targetDir)
+	if err != nil {
+		return nil, admin.NewInvalidAdminReqErrorf("failed to create checkpoint at %v: %w", targetDir, err)
+	}
+
+	log.Info().Msgf("admintool: successfully created chunkDataPacks database checkpoint at: %v", targetDir)
+
+	return fmt.Sprintf("successfully created chunkDataPacks db checkpoint at %v", targetDir), nil
+}
+
+func (c *ChunksCheckpointCommand) Validator(req *admin.CommandRequest) error {
+	return nil
+}
+
+func nextTmpFolder(dir string) string {
+	// use timestamp as folder name
+	folderName := time.Now().Format("2006-01-02_15-04-05")
+	return path.Join(dir, folderName)
+}

admin/commands/storage/chunk_data_packs_db_checkpoint.go

@@ -349,6 +349,11 @@ type FlowAccessNodeBuilder struct {
 	Transactions                 storage.Transactions
 	Collections                  storage.Collections
 
+	// storage
+	events                         storage.Events
+	lightTransactionResults        storage.LightTransactionResults
+	transactionResultErrorMessages storage.TransactionResultErrorMessages
+
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
 	// Hence, a factory function that needs to be called just before creating the sync engine
@@ -597,7 +602,9 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 			}
 
 			if executionDataDBMode == execution_data.ExecutionDataDBModePebble {
-				builder.ExecutionDatastoreManager, err = edstorage.NewPebbleDatastoreManager(datastoreDir, nil)
+				builder.ExecutionDatastoreManager, err = edstorage.NewPebbleDatastoreManager(
+					node.Logger.With().Str("pebbledb", "endata").Logger(),
+					datastoreDir, nil)
 				if err != nil {
 					return fmt.Errorf("could not create PebbleDatastoreManager for execution data: %w", err)
 				}
@@ -883,15 +890,17 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				return nil
 			}).
 			Module("transaction results storage", func(node *cmd.NodeConfig) error {
-				builder.Storage.LightTransactionResults = bstorage.NewLightTransactionResults(node.Metrics.Cache, node.DB, bstorage.DefaultCacheSize)
+				builder.lightTransactionResults = store.NewLightTransactionResults(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
 				return nil
 			}).
 			DependableComponent("execution data indexer", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 				// Note: using a DependableComponent here to ensure that the indexer does not block
 				// other components from starting while bootstrapping the register db since it may
 				// take hours to complete.
 
-				pdb, err := pstorage.OpenRegisterPebbleDB(builder.registersDBPath)
+				pdb, err := pstorage.OpenRegisterPebbleDB(
+					node.Logger.With().Str("pebbledb", "registers").Logger(),
+					builder.registersDBPath)
 				if err != nil {
 					return nil, fmt.Errorf("could not open registers db: %w", err)
 				}
@@ -972,13 +981,13 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				indexerCore, err := indexer.New(
 					builder.Logger,
 					metrics.NewExecutionStateIndexerCollector(),
-					builder.DB,
+					builder.ProtocolDB,
 					builder.Storage.RegisterIndex,
 					builder.Storage.Headers,
-					builder.Storage.Events,
+					notNil(builder.events),
 					notNil(builder.Collections),
 					notNil(builder.Transactions),
-					builder.Storage.LightTransactionResults,
+					notNil(builder.lightTransactionResults),
 					builder.RootChainID.Chain(),
 					indexerDerivedChainData,
 					builder.collectionExecutedMetric,
@@ -1852,19 +1861,19 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			return nil
 		}).
 		Module("events storage", func(node *cmd.NodeConfig) error {
-			builder.Storage.Events = bstorage.NewEvents(node.Metrics.Cache, node.DB)
+			builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
 			return nil
 		}).
 		Module("reporter", func(node *cmd.NodeConfig) error {
 			builder.Reporter = index.NewReporter()
 			return nil
 		}).
 		Module("events index", func(node *cmd.NodeConfig) error {
-			builder.EventsIndex = index.NewEventsIndex(builder.Reporter, builder.Storage.Events)
+			builder.EventsIndex = index.NewEventsIndex(builder.Reporter, builder.events)
 			return nil
 		}).
 		Module("transaction result index", func(node *cmd.NodeConfig) error {
-			builder.TxResultsIndex = index.NewTransactionResultsIndex(builder.Reporter, builder.Storage.LightTransactionResults)
+			builder.TxResultsIndex = index.NewTransactionResultsIndex(builder.Reporter, builder.lightTransactionResults)
 			return nil
 		}).
 		Module("processed finalized block height consumer progress", func(node *cmd.NodeConfig) error {
@@ -1888,7 +1897,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		}).
 		Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
 			if builder.storeTxResultErrorMessages {
-				builder.Storage.TransactionResultErrorMessages = bstorage.NewTransactionResultErrorMessages(node.Metrics.Cache, node.DB, bstorage.DefaultCacheSize)
+				builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
 			}
 
 			return nil
@@ -2020,12 +2029,12 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			}
 
-			preferredENIdentifiers, err := commonrpc.IdentifierList(backendConfig.PreferredExecutionNodeIDs)
+			preferredENIdentifiers, err := flow.IdentifierListFromHex(backendConfig.PreferredExecutionNodeIDs)
 			if err != nil {
 				return nil, fmt.Errorf("failed to convert node id string to Flow Identifier for preferred EN map: %w", err)
 			}
 
-			fixedENIdentifiers, err := commonrpc.IdentifierList(backendConfig.FixedExecutionNodeIDs)
+			fixedENIdentifiers, err := flow.IdentifierListFromHex(backendConfig.FixedExecutionNodeIDs)
 			if err != nil {
 				return nil, fmt.Errorf("failed to convert node id string to Flow Identifier for fixed EN map: %w", err)
 			}
@@ -2048,7 +2057,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				Transactions:          notNil(builder.Transactions),
 				ExecutionReceipts:     node.Storage.Receipts,
 				ExecutionResults:      node.Storage.Results,
-				TxResultErrorMessages: node.Storage.TransactionResultErrorMessages,
+				TxResultErrorMessages: builder.transactionResultErrorMessages,
 				ChainID:               node.RootChainID,
 				AccessMetrics:         builder.AccessMetrics,
 				ConnFactory:           connFactory,
@@ -2134,7 +2143,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				builder.TxResultErrorMessagesCore = tx_error_messages.NewTxErrorMessagesCore(
 					node.Logger,
 					builder.nodeBackend,
-					node.Storage.TransactionResultErrorMessages,
+					builder.transactionResultErrorMessages,
 					builder.ExecNodeIdentitiesProvider,
 				)
 			}