Add ExecutedBlock type (ava-labs#1601)

Author: aaronbuchwald

api/indexer/tx_indexer.go

@@ -25,8 +25,8 @@ const (
 )
 
 var (
-	_ event.SubscriptionFactory[*chain.StatefulBlock] = (*subscriptionFactory)(nil)
-	_ event.Subscription[*chain.StatefulBlock]        = (*txDBIndexer)(nil)
+	_ event.SubscriptionFactory[*chain.ExecutedBlock] = (*subscriptionFactory)(nil)
+	_ event.Subscription[*chain.ExecutedBlock]        = (*txDBIndexer)(nil)
 )
 
 type Config struct {
@@ -47,7 +47,7 @@ func OptionFunc(v *vm.VM, config Config) error {
 	if !config.Enabled {
 		return nil
 	}
-	dbPath := filepath.Join(v.DataDir, "indexer", "db")
+	dbPath := filepath.Join(v.DataDir, Namespace)
 	db, _, err := pebble.New(dbPath, pebble.NewDefaultConfig())
 	if err != nil {
 		return err
@@ -77,26 +77,24 @@ type subscriptionFactory struct {
 	indexer *txDBIndexer
 }
 
-func (s *subscriptionFactory) New() (event.Subscription[*chain.StatefulBlock], error) {
+func (s *subscriptionFactory) New() (event.Subscription[*chain.ExecutedBlock], error) {
 	return s.indexer, nil
 }
 
 type txDBIndexer struct {
 	db database.Database
 }
 
-func (t *txDBIndexer) Accept(blk *chain.StatefulBlock) error {
+func (t *txDBIndexer) Accept(blk *chain.ExecutedBlock) error {
 	batch := t.db.NewBatch()
 	defer batch.Reset()
 
-	timestamp := blk.GetTimestamp()
-	results := blk.Results()
-	for j, tx := range blk.Txs {
-		result := results[j]
+	for j, tx := range blk.Block.Txs {
+		result := blk.Results[j]
 		if err := t.storeTransaction(
 			batch,
 			tx.ID(),
-			timestamp,
+			blk.Block.Tmstmp,
 			result.Success,
 			result.Units,
 			result.Fee,

api/ws/client.go

@@ -151,7 +151,11 @@ func (c *WebSocketClient) ListenBlock(
 ) (*chain.StatelessBlock, []*chain.Result, fees.Dimensions, error) {
 	select {
 	case msg := <-c.pendingBlocks:
-		return UnpackBlockMessage(msg, parser)
+		executedBlock, err := chain.UnmarshalExecutedBlock(msg, parser)
+		if err != nil {
+			return nil, nil, fees.Dimensions{}, err
+		}
+		return executedBlock.Block, executedBlock.Results, executedBlock.UnitPrices, nil
 	case <-c.readStopped:
 		return nil, nil, fees.Dimensions{}, c.err
 	case <-ctx.Done():

api/ws/packer.go

@@ -11,57 +11,13 @@ import (
 	"github.com/ava-labs/hypersdk/chain"
 	"github.com/ava-labs/hypersdk/codec"
 	"github.com/ava-labs/hypersdk/consts"
-	"github.com/ava-labs/hypersdk/fees"
 )
 
 const (
 	BlockMode byte = 0
 	TxMode    byte = 1
 )
 
-func PackBlockMessage(b *chain.StatefulBlock) ([]byte, error) {
-	results := b.Results()
-	size := codec.BytesLen(b.Bytes()) + consts.IntLen + codec.CummSize(results) + fees.DimensionsLen
-	p := codec.NewWriter(size, consts.MaxInt)
-	p.PackBytes(b.Bytes())
-	mresults, err := chain.MarshalResults(results)
-	if err != nil {
-		return nil, err
-	}
-	p.PackBytes(mresults)
-	p.PackFixedBytes(b.FeeManager().UnitPrices().Bytes())
-	return p.Bytes(), p.Err()
-}
-
-func UnpackBlockMessage(
-	msg []byte,
-	parser chain.Parser,
-) (*chain.StatelessBlock, []*chain.Result, fees.Dimensions, error) {
-	p := codec.NewReader(msg, consts.MaxInt)
-	var blkMsg []byte
-	p.UnpackBytes(-1, true, &blkMsg)
-	blk, err := chain.UnmarshalBlock(blkMsg, parser)
-	if err != nil {
-		return nil, nil, fees.Dimensions{}, err
-	}
-	var resultsMsg []byte
-	p.UnpackBytes(-1, true, &resultsMsg)
-	results, err := chain.UnmarshalResults(resultsMsg)
-	if err != nil {
-		return nil, nil, fees.Dimensions{}, err
-	}
-	pricesMsg := make([]byte, fees.DimensionsLen)
-	p.UnpackFixedBytes(fees.DimensionsLen, &pricesMsg)
-	prices, err := fees.UnpackDimensions(pricesMsg)
-	if err != nil {
-		return nil, nil, fees.Dimensions{}, err
-	}
-	if !p.Empty() {
-		return nil, nil, fees.Dimensions{}, chain.ErrInvalidObject
-	}
-	return blk, results, prices, p.Err()
-}
-
 // Could be a better place for these methods
 // Packs an accepted block message
 func PackAcceptedTxMessage(txID ids.ID, result *chain.Result) ([]byte, error) {

api/ws/server.go

@@ -72,8 +72,8 @@ func OptionFunc(v *vm.VM, config Config) error {
 		},
 	}
 
-	blockSubscription := event.SubscriptionFuncFactory[*chain.StatefulBlock]{
-		AcceptF: func(event *chain.StatefulBlock) error {
+	blockSubscription := event.SubscriptionFuncFactory[*chain.ExecutedBlock]{
+		AcceptF: func(event *chain.ExecutedBlock) error {
 			return server.AcceptBlock(event)
 		},
 	}
@@ -195,9 +195,9 @@ func (w *WebSocketServer) setMinTx(t int64) error {
 	return nil
 }
 
-func (w *WebSocketServer) AcceptBlock(b *chain.StatefulBlock) error {
+func (w *WebSocketServer) AcceptBlock(b *chain.ExecutedBlock) error {
 	if w.blockListeners.Len() > 0 {
-		bytes, err := PackBlockMessage(b)
+		bytes, err := b.Marshal()
 		if err != nil {
 			return err
 		}
@@ -209,8 +209,8 @@ func (w *WebSocketServer) AcceptBlock(b *chain.StatefulBlock) error {
 
 	w.txL.Lock()
 	defer w.txL.Unlock()
-	results := b.Results()
-	for i, tx := range b.Txs {
+	results := b.Results
+	for i, tx := range b.Block.Txs {
 		txID := tx.ID()
 		listeners, ok := w.txListeners[txID]
 		if !ok {
@@ -227,7 +227,7 @@ func (w *WebSocketServer) AcceptBlock(b *chain.StatefulBlock) error {
 		delete(w.txListeners, txID)
 		// [expiringTxs] will be cleared eventually (does not support removal)
 	}
-	return w.setMinTx(b.Tmstmp)
+	return w.setMinTx(b.Block.Tmstmp)
 }
 
 func (w *WebSocketServer) MessageCallback() pubsub.Callback {

chain/block.go

@@ -22,11 +22,13 @@ import (
 
 	"github.com/ava-labs/hypersdk/codec"
 	"github.com/ava-labs/hypersdk/consts"
-	"github.com/ava-labs/hypersdk/internal/fees"
+	"github.com/ava-labs/hypersdk/fees"
 	"github.com/ava-labs/hypersdk/internal/window"
 	"github.com/ava-labs/hypersdk/internal/workers"
 	"github.com/ava-labs/hypersdk/state"
 	"github.com/ava-labs/hypersdk/utils"
+
+	internalfees "github.com/ava-labs/hypersdk/internal/fees"
 )
 
 var (
@@ -164,7 +166,7 @@ type StatefulBlock struct {
 	txsSet   set.Set[ids.ID]
 
 	results    []*Result
-	feeManager *fees.Manager
+	feeManager *internalfees.Manager
 
 	vm   VM
 	view merkledb.View
@@ -291,7 +293,7 @@ func (b *StatefulBlock) initializeBuilt(
 	ctx context.Context,
 	view merkledb.View,
 	results []*Result,
-	feeManager *fees.Manager,
+	feeManager *internalfees.Manager,
 ) error {
 	_, span := b.vm.Tracer().Start(ctx, "StatefulBlock.initializeBuilt")
 	defer span.End()
@@ -487,7 +489,7 @@ func (b *StatefulBlock) innerVerify(ctx context.Context, vctx VerifyContext) err
 	if err != nil {
 		return err
 	}
-	parentFeeManager := fees.NewManager(feeRaw)
+	parentFeeManager := internalfees.NewManager(feeRaw)
 	feeManager, err := parentFeeManager.ComputeNext(b.Tmstmp, r)
 	if err != nil {
 		return err
@@ -901,10 +903,86 @@ func (b *StatefulBlock) Results() []*Result {
 	return b.results
 }
 
-func (b *StatefulBlock) FeeManager() *fees.Manager {
+func (b *StatefulBlock) FeeManager() *internalfees.Manager {
 	return b.feeManager
 }
 
+type ExecutedBlock struct {
+	BlockID    ids.ID          `json:"blockID"`
+	Block      *StatelessBlock `json:"block"`
+	Results    []*Result       `json:"results"`
+	UnitPrices fees.Dimensions `json:"unitPrices"`
+}
+
+func NewExecutedBlockFromStateful(b *StatefulBlock) *ExecutedBlock {
+	return &ExecutedBlock{
+		BlockID:    b.ID(),
+		Block:      b.StatelessBlock,
+		Results:    b.results,
+		UnitPrices: b.feeManager.UnitPrices(),
+	}
+}
+
+func NewExecutedBlock(statelessBlock *StatelessBlock, results []*Result, unitPrices fees.Dimensions) (*ExecutedBlock, error) {
+	blkID, err := statelessBlock.ID()
+	if err != nil {
+		return nil, err
+	}
+	return &ExecutedBlock{
+		BlockID:    blkID,
+		Block:      statelessBlock,
+		Results:    results,
+		UnitPrices: unitPrices,
+	}, nil
+}
+
+func (b *ExecutedBlock) Marshal() ([]byte, error) {
+	blockBytes, err := b.Block.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	size := codec.BytesLen(blockBytes) + codec.CummSize(b.Results) + fees.DimensionsLen
+	writer := codec.NewWriter(size, consts.NetworkSizeLimit)
+
+	writer.PackBytes(blockBytes)
+	resultBytes, err := MarshalResults(b.Results)
+	if err != nil {
+		return nil, err
+	}
+	writer.PackBytes(resultBytes)
+	writer.PackFixedBytes(b.UnitPrices.Bytes())
+
+	return writer.Bytes(), writer.Err()
+}
+
+func UnmarshalExecutedBlock(bytes []byte, parser Parser) (*ExecutedBlock, error) {
+	reader := codec.NewReader(bytes, consts.NetworkSizeLimit)
+
+	var blkMsg []byte
+	reader.UnpackBytes(-1, true, &blkMsg)
+	blk, err := UnmarshalBlock(blkMsg, parser)
+	if err != nil {
+		return nil, err
+	}
+	var resultsMsg []byte
+	reader.UnpackBytes(-1, true, &resultsMsg)
+	results, err := UnmarshalResults(resultsMsg)
+	if err != nil {
+		return nil, err
+	}
+	unitPricesBytes := make([]byte, fees.DimensionsLen)
+	reader.UnpackFixedBytes(fees.DimensionsLen, &unitPricesBytes)
+	prices, err := fees.UnpackDimensions(unitPricesBytes)
+	if err != nil {
+		return nil, err
+	}
+	if !reader.Empty() {
+		return nil, ErrInvalidObject
+	}
+	return NewExecutedBlock(blk, results, prices)
+}
+
 type SyncableBlock struct {
 	*StatefulBlock
 }

extension/externalsubscriber/client.go

@@ -17,7 +17,7 @@ import (
 	pb "github.com/ava-labs/hypersdk/proto/pb/externalsubscriber"
 )
 
-var _ event.Subscription[*chain.StatefulBlock] = (*ExternalSubscriberClient)(nil)
+var _ event.Subscription[*chain.ExecutedBlock] = (*ExternalSubscriberClient)(nil)
 
 type ExternalSubscriberClient struct {
 	conn   *grpc.ClientConn
@@ -55,23 +55,18 @@ func NewExternalSubscriberClient(
 	}, nil
 }
 
-func (e *ExternalSubscriberClient) Accept(blk *chain.StatefulBlock) error {
+func (e *ExternalSubscriberClient) Accept(blk *chain.ExecutedBlock) error {
 	blockBytes, err := blk.Marshal()
 	if err != nil {
 		return err
 	}
-	resultsMarshaled, err := chain.MarshalResults(blk.Results())
-	if err != nil {
-		return err
-	}
 
 	req := &pb.BlockRequest{
 		BlockData: blockBytes,
-		Results:   resultsMarshaled,
 	}
 	e.log.Debug("sending accepted block to server",
-		zap.Stringer("blockID", blk.ID()),
-		zap.Uint64("blockHeight", blk.Hght),
+		zap.Stringer("blockID", blk.BlockID),
+		zap.Uint64("blockHeight", blk.Block.Hght),
 	)
 	_, err = e.client.AcceptBlock(context.TODO(), req)
 	return err

extension/externalsubscriber/option.go

@@ -42,8 +42,8 @@ func OptionFunc(v *vm.VM, config Config) error {
 		return err
 	}
 
-	blockSubscription := event.SubscriptionFuncFactory[*chain.StatefulBlock]{
-		AcceptF: func(blk *chain.StatefulBlock) error {
+	blockSubscription := event.SubscriptionFuncFactory[*chain.ExecutedBlock]{
+		AcceptF: func(blk *chain.ExecutedBlock) error {
 			return server.Accept(blk)
 		},
 	}