Merge pull request #6180 from oasisprotocol/peternose/feature/block-results

go/consensus/api: Add method GetBlockResults

Author: peternose

.changelog/6180.feature.md

@@ -0,0 +1,4 @@
+go/consensus/api: Add method GetBlockResults
+
+A new method has been added to the consensus backend that returns block
+results for a specified height.

go/consensus/api/api.go

@@ -135,6 +135,9 @@ type Backend interface {
 	// GetBlock returns a consensus block at a specific height.
 	GetBlock(ctx context.Context, height int64) (*Block, error)
 
+	// GetBlockResults returns the consensus block results at a specific height.
+	GetBlockResults(ctx context.Context, height int64) (*BlockResults, error)
+
 	// GetLightBlock returns a light version of the consensus layer block that can be used for light
 	// client verification.
 	GetLightBlock(ctx context.Context, height int64) (*LightBlock, error)
@@ -244,6 +247,17 @@ type Block struct {
 	Meta cbor.RawMessage `json:"meta"`
 }
 
+// BlockResults are consensus block results.
+//
+// While some common fields are provided, most of the structure is dependent on
+// the actual backend implementation.
+type BlockResults struct {
+	// Height contains the block height.
+	Height int64 `json:"height"`
+	// Meta contains the consensus backend specific block results metadata.
+	Meta cbor.RawMessage `json:"meta"`
+}
+
 // NextBlockState has the state of the next block being voted on by validators.
 type NextBlockState struct {
 	Height int64 `json:"height"`

go/consensus/api/grpc.go

@@ -41,6 +41,8 @@ var (
 	methodGetSignerNonce = serviceName.NewMethod("GetSignerNonce", &GetSignerNonceRequest{})
 	// methodGetBlock is the GetBlock method.
 	methodGetBlock = serviceName.NewMethod("GetBlock", int64(0))
+	// methodGetBlockResults is the GetBlockResults method.
+	methodGetBlockResults = serviceName.NewMethod("GetBlockResults", int64(0))
 	// methodGetLightBlock is the GetLightBlock method.
 	methodGetLightBlock = serviceName.NewMethod("GetLightBlock", int64(0))
 	// methodGetLatestHeight is the GetLatestHeight method.
@@ -114,6 +116,10 @@ var (
 				MethodName: methodGetBlock.ShortName(),
 				Handler:    handlerGetBlock,
 			},
+			{
+				MethodName: methodGetBlockResults.ShortName(),
+				Handler:    handlerGetBlockResults,
+			},
 			{
 				MethodName: methodGetLightBlock.ShortName(),
 				Handler:    handlerGetLightBlock,
@@ -369,6 +375,29 @@ func handlerGetBlock(
 	return interceptor(ctx, height, info, handler)
 }
 
+func handlerGetBlockResults(
+	srv any,
+	ctx context.Context,
+	dec func(any) error,
+	interceptor grpc.UnaryServerInterceptor,
+) (any, error) {
+	var height int64
+	if err := dec(&height); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(Services).Core().GetBlockResults(ctx, height)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: methodGetBlockResults.FullName(),
+	}
+	handler := func(ctx context.Context, req any) (any, error) {
+		return srv.(Services).Core().GetBlockResults(ctx, req.(int64))
+	}
+	return interceptor(ctx, height, info, handler)
+}
+
 func handlerGetLightBlock(
 	srv any,
 	ctx context.Context,
@@ -810,6 +839,14 @@ func (c *Client) GetBlock(ctx context.Context, height int64) (*Block, error) {
 	return &rsp, nil
 }
 
+func (c *Client) GetBlockResults(ctx context.Context, height int64) (*BlockResults, error) {
+	var rsp BlockResults
+	if err := c.conn.Invoke(ctx, methodGetBlockResults.FullName(), height, &rsp); err != nil {
+		return nil, err
+	}
+	return &rsp, nil
+}
+
 func (c *Client) GetLightBlock(ctx context.Context, height int64) (*LightBlock, error) {
 	var rsp LightBlock
 	if err := c.conn.Invoke(ctx, methodGetLightBlock.FullName(), height, &rsp); err != nil {

go/consensus/cometbft/api/api.go

@@ -11,7 +11,7 @@ import (
 	cmtquery "github.com/cometbft/cometbft/libs/pubsub/query"
 	cmtp2p "github.com/cometbft/cometbft/p2p"
 	cmtcrypto "github.com/cometbft/cometbft/proto/tendermint/crypto"
-	cmtrpctypes "github.com/cometbft/cometbft/rpc/core/types"
+	cmtcoretypes "github.com/cometbft/cometbft/rpc/core/types"
 	cmttypes "github.com/cometbft/cometbft/types"
 
 	beacon "github.com/oasisprotocol/oasis-core/go/beacon/api"
@@ -20,7 +20,6 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/signature"
 	"github.com/oasisprotocol/oasis-core/go/common/node"
-	"github.com/oasisprotocol/oasis-core/go/common/pubsub"
 	consensus "github.com/oasisprotocol/oasis-core/go/consensus/api"
 	"github.com/oasisprotocol/oasis-core/go/consensus/api/events"
 	"github.com/oasisprotocol/oasis-core/go/consensus/api/transaction"
@@ -157,8 +156,7 @@ func QueryForApp(eventApp string) cmtpubsub.Query {
 	return cmtquery.MustParse(fmt.Sprintf("%s EXISTS", EventTypeForApp(eventApp)))
 }
 
-// BlockMeta is the CometBFT-specific per-block metadata that is
-// exposed via the consensus API.
+// BlockMeta is the CometBFT-specific per-block metadata.
 type BlockMeta struct {
 	// Header is the CometBFT block header.
 	Header *cmttypes.Header `json:"header"`
@@ -199,20 +197,85 @@ func NewBlock(blk *cmttypes.Block) *consensus.Block {
 	}
 }
 
-// Backend is a CometBFT consensus backend.
-type Backend interface {
-	consensus.Backend
+// BlockResults are CometBFT-specific consensus block results.
+type BlockResults struct {
+	// Height contains the block height.
+	Height int64 `json:"height"`
+	// Meta contains the block results metadata.
+	Meta *BlockResultsMeta `json:"meta"`
+}
+
+// BlockResultsMeta is the CometBFT-specific per-block results metadata.
+type BlockResultsMeta struct {
+	TxsResults       []*types.ResponseDeliverTx `json:"txs_results"`
+	BeginBlockEvents []types.Event              `json:"begin_block_events"`
+	EndBlockEvents   []types.Event              `json:"end_block_events"`
+}
+
+// NewBlockResultsMeta converts consensus results into CometBFT-specific block
+// results metadata.
+func NewBlockResultsMeta(results *consensus.BlockResults) (*BlockResultsMeta, error) {
+	var meta BlockResultsMeta
+	if err := cbor.Unmarshal(results.Meta, &meta); err != nil {
+		return nil, fmt.Errorf("malformed block results metadata: %w", err)
+	}
+
+	return &meta, nil
+}
+
+// NewBlockResults converts CometBFT-specific block results into consensus results.
+func NewBlockResults(results *cmtcoretypes.ResultBlockResults) *consensus.BlockResults {
+	meta := BlockResultsMeta{
+		TxsResults:       results.TxsResults,
+		BeginBlockEvents: results.BeginBlockEvents,
+		EndBlockEvents:   results.EndBlockEvents,
+	}
+
+	return &consensus.BlockResults{
+		Height: results.Height,
+		Meta:   cbor.Marshal(meta),
+	}
+}
 
-	// GetCometBFTBlock returns the CometBFT block at the specified height.
-	GetCometBFTBlock(ctx context.Context, height int64) (*cmttypes.Block, error)
+// GetBlockResults returns CometBFT-specific block results at the given height.
+func GetBlockResults(ctx context.Context, height int64, consensus consensus.Backend) (*BlockResults, error) {
+	// Optimize for CometBTF-specific consensus backends.
+	if cmt, ok := consensus.(Backend); ok {
+		results, err := cmt.GetCometBFTBlockResults(ctx, height)
+		if err != nil {
+			return nil, err
+		}
+		return &BlockResults{
+			Height: results.Height,
+			Meta: &BlockResultsMeta{
+				TxsResults:       results.TxsResults,
+				BeginBlockEvents: results.BeginBlockEvents,
+				EndBlockEvents:   results.EndBlockEvents,
+			},
+		}, nil
+
+	}
 
+	results, err := consensus.GetBlockResults(ctx, height)
+	if err != nil {
+		return nil, err
+	}
+	meta, err := NewBlockResultsMeta(results)
+	if err != nil {
+		return nil, err
+	}
+
+	return &BlockResults{
+		Height: results.Height,
+		Meta:   meta,
+	}, nil
+}
+
+// Backend is a CometBFT-specific consensus backend.
+type Backend interface {
 	// GetCometBFTBlockResults returns the ABCI results from processing a block
 	// at a specific height.
-	GetCometBFTBlockResults(ctx context.Context, height int64) (*cmtrpctypes.ResultBlockResults, error)
-
-	// WatchCometBFTBlocks returns a stream of CometBFT blocks as they are
-	// returned via the `EventDataNewBlock` query.
-	WatchCometBFTBlocks() (<-chan *cmttypes.Block, *pubsub.Subscription, error)
+	GetCometBFTBlockResults(ctx context.Context, height int64) (*cmtcoretypes.ResultBlockResults, error)
 }
 
 // HaltHook is a function that gets called when consensus needs to halt for some reason.
@@ -297,8 +360,8 @@ type ServiceClient interface {
 	// ServiceDescriptor returns the consensus service descriptor.
 	ServiceDescriptor() ServiceDescriptor
 
-	// DeliverBlock delivers a new block.
-	DeliverBlock(ctx context.Context, blk *cmttypes.Block) error
+	// DeliverHeight delivers a new block height.
+	DeliverHeight(ctx context.Context, height int64) error
 
 	// DeliverEvent delivers an event emitted by the consensus service.
 	DeliverEvent(ctx context.Context, height int64, tx cmttypes.Tx, ev *types.Event) error
@@ -308,8 +371,8 @@ type ServiceClient interface {
 // all the delivery methods. Implementations should override them as needed.
 type BaseServiceClient struct{}
 
-// DeliverBlock implements ServiceClient.
-func (bsc *BaseServiceClient) DeliverBlock(context.Context, *cmttypes.Block) error {
+// DeliverHeight implements ServiceClient.
+func (bsc *BaseServiceClient) DeliverHeight(context.Context, int64) error {
 	return nil
 }
 

go/consensus/cometbft/beacon/beacon.go

@@ -18,6 +18,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	"github.com/oasisprotocol/oasis-core/go/common/pubsub"
+	consensus "github.com/oasisprotocol/oasis-core/go/consensus/api"
 	"github.com/oasisprotocol/oasis-core/go/consensus/api/events"
 	tmapi "github.com/oasisprotocol/oasis-core/go/consensus/cometbft/api"
 	app "github.com/oasisprotocol/oasis-core/go/consensus/cometbft/apps/beacon"
@@ -33,7 +34,7 @@ type ServiceClient struct {
 
 	logger *logging.Logger
 
-	consensus tmapi.Backend
+	consensus consensus.Backend
 	querier   *app.QueryFactory
 
 	epochNotifier     *pubsub.Broker
@@ -53,7 +54,7 @@ type ServiceClient struct {
 }
 
 // New constructs a new CometBFT backed beacon service client.
-func New(baseEpoch beaconAPI.EpochTime, baseBlock int64, consensus tmapi.Backend, querier *app.QueryFactory) *ServiceClient {
+func New(baseEpoch beaconAPI.EpochTime, baseBlock int64, consensus consensus.Backend, querier *app.QueryFactory) *ServiceClient {
 	return &ServiceClient{
 		logger:            logging.GetLogger("cometbft/beacon"),
 		consensus:         consensus,
@@ -246,12 +247,12 @@ func (sc *ServiceClient) ServiceDescriptor() tmapi.ServiceDescriptor {
 	return tmapi.NewStaticServiceDescriptor("beacon", app.EventType, []cmtpubsub.Query{app.QueryApp})
 }
 
-func (sc *ServiceClient) DeliverBlock(ctx context.Context, blk *cmttypes.Block) error {
+func (sc *ServiceClient) DeliverHeight(ctx context.Context, height int64) error {
 	if sc.initialNotify {
 		return nil
 	}
 
-	q, err := sc.querier.QueryAt(ctx, blk.Height)
+	q, err := sc.querier.QueryAt(ctx, height)
 	if err != nil {
 		return fmt.Errorf("epochtime: failed to query state: %w", err)
 	}

go/consensus/cometbft/full/archive.go

@@ -29,7 +29,7 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/consensus/cometbft/db"
 )
 
-var _ api.Backend = (*archiveService)(nil)
+var _ consensusAPI.Backend = (*archiveService)(nil)
 
 // ArchiveConfig contains configuration parameters for the archive node.
 type ArchiveConfig struct {
@@ -256,16 +256,16 @@ func (srv *archiveService) serviceClientWorker(ctx context.Context, svc api.Serv
 	logger := srv.Logger.With("service", sd.Name())
 	logger.Info("starting command dispatcher")
 
-	latestBlock, err := srv.GetCometBFTBlock(ctx, consensusAPI.HeightLatest)
+	height, err := srv.GetLatestHeight(ctx)
 	if err != nil {
-		logger.Error("failed to fetch latest block",
+		logger.Error("failed to fetch latest height",
 			"err", err,
 		)
 		return
 	}
 
-	if err := svc.DeliverBlock(ctx, latestBlock); err != nil {
-		logger.Error("failed to deliver block to service client",
+	if err := svc.DeliverHeight(ctx, height); err != nil {
+		logger.Error("failed to deliver block height to service client",
 			"err", err,
 		)
 	}