feat(server/v2/stf): delayed marshalling of typed event

server/v2/cometbft/abci.go

@@ -62,9 +62,9 @@ type consensus[T transaction.Tx] struct {
 	streamingManager streaming.Manager
 	mempool          mempool.Mempool[T]
 
-	cfg           Config
-	chainID       string
-	indexedEvents map[string]struct{}
+	cfg               Config
+	chainID           string
+	indexedABCIEvents map[string]struct{}
 
 	initialHeight uint64
 	// this is only available after this node has committed a block (in FinalizeBlock),
@@ -105,9 +105,12 @@ func (c *consensus[T]) CheckTx(ctx context.Context, req *abciproto.CheckTxReques
 			return nil, err
 		}
 
-		events, err := intoABCIEvents(resp.Events, c.indexedEvents)
-		if err != nil {
-			return nil, err
+		events := make([]abci.Event, 0)
+		if !c.cfg.AppTomlConfig.DisableABCIEvents {
+			events, err = intoABCIEvents(resp.Events, c.indexedABCIEvents)
+			if err != nil {
+				return nil, err
+			}
 		}
 
 		cometResp := &abciproto.CheckTxResponse{
@@ -116,6 +119,7 @@ func (c *consensus[T]) CheckTx(ctx context.Context, req *abciproto.CheckTxReques
 			GasUsed:   uint64ToInt64(resp.GasUsed),
 			Events:    events,
 		}
+
 		if resp.Error != nil {
 			space, code, log := errorsmod.ABCIInfo(resp.Error, c.cfg.AppTomlConfig.Trace)
 			cometResp.Code = code
@@ -557,7 +561,7 @@ func (c *consensus[T]) FinalizeBlock(
 		return nil, err
 	}
 
-	return finalizeBlockResponse(resp, cp, appHash, c.indexedEvents, c.cfg.AppTomlConfig.Trace)
+	return finalizeBlockResponse(resp, cp, appHash, c.indexedABCIEvents, c.cfg.AppTomlConfig.DisableABCIEvents, c.cfg.AppTomlConfig.Trace)
 }
 
 func (c *consensus[T]) internalFinalizeBlock(

server/v2/cometbft/config.go

@@ -16,7 +16,6 @@ type Config struct {
 func DefaultAppTomlConfig() *AppTomlConfig {
 	return &AppTomlConfig{
 		MinRetainBlocks: 0,
-		IndexEvents:     make([]string, 0),
 		HaltHeight:      0,
 		HaltTime:        0,
 		Address:         "tcp://127.0.0.1:26658",
@@ -28,22 +27,25 @@ func DefaultAppTomlConfig() *AppTomlConfig {
 			Target:            make(map[string]indexer.Config),
 			ChannelBufferSize: 1024,
 		},
+		IndexABCIEvents:   make([]string, 0),
+		DisableABCIEvents: false,
 	}
 }
 
 type AppTomlConfig struct {
-	MinRetainBlocks uint64   `mapstructure:"min-retain-blocks" toml:"min-retain-blocks" comment:"min-retain-blocks defines the minimum block height offset from the current block being committed, such that all blocks past this offset are pruned from CometBFT. A value of 0 indicates that no blocks should be pruned."`
-	IndexEvents     []string `mapstructure:"index-events" toml:"index-events" comment:"index-events defines the set of events in the form {eventType}.{attributeKey}, which informs CometBFT what to index. If empty, all events will be indexed."`
-	HaltHeight      uint64   `mapstructure:"halt-height" toml:"halt-height" comment:"halt-height contains a non-zero block height at which a node will gracefully halt and shutdown that can be used to assist upgrades and testing."`
-	HaltTime        uint64   `mapstructure:"halt-time" toml:"halt-time" comment:"halt-time contains a non-zero minimum block time (in Unix seconds) at which a node will gracefully halt and shutdown that can be used to assist upgrades and testing."`
-	Address         string   `mapstructure:"address" toml:"address" comment:"address defines the CometBFT RPC server address to bind to."`
-	Transport       string   `mapstructure:"transport" toml:"transport" comment:"transport defines the CometBFT RPC server transport protocol: socket, grpc"`
-	Trace           bool     `mapstructure:"trace" toml:"trace" comment:"trace enables the CometBFT RPC server to output trace information about its internal operations."`
-	Standalone      bool     `mapstructure:"standalone" toml:"standalone" comment:"standalone starts the application without the CometBFT node. The node should be started separately."`
+	MinRetainBlocks uint64 `mapstructure:"min-retain-blocks" toml:"min-retain-blocks" comment:"min-retain-blocks defines the minimum block height offset from the current block being committed, such that all blocks past this offset are pruned from CometBFT. A value of 0 indicates that no blocks should be pruned."`
+	HaltHeight      uint64 `mapstructure:"halt-height" toml:"halt-height" comment:"halt-height contains a non-zero block height at which a node will gracefully halt and shutdown that can be used to assist upgrades and testing."`
+	HaltTime        uint64 `mapstructure:"halt-time" toml:"halt-time" comment:"halt-time contains a non-zero minimum block time (in Unix seconds) at which a node will gracefully halt and shutdown that can be used to assist upgrades and testing."`
+	Address         string `mapstructure:"address" toml:"address" comment:"address defines the CometBFT RPC server address to bind to."`
+	Transport       string `mapstructure:"transport" toml:"transport" comment:"transport defines the CometBFT RPC server transport protocol: socket, grpc"`
+	Trace           bool   `mapstructure:"trace" toml:"trace" comment:"trace enables the CometBFT RPC server to output trace information about its internal operations."`
+	Standalone      bool   `mapstructure:"standalone" toml:"standalone" comment:"standalone starts the application without the CometBFT node. The node should be started separately."`
 
 	// Sub configs
-	Mempool mempool.Config         `mapstructure:"mempool" toml:"mempool" comment:"mempool defines the configuration for the SDK built-in app-side mempool implementations."`
-	Indexer indexer.IndexingConfig `mapstructure:"indexer" toml:"indexer" comment:"indexer defines the configuration for the SDK built-in indexer implementation."`
+	Mempool           mempool.Config         `mapstructure:"mempool" toml:"mempool" comment:"mempool defines the configuration for the SDK built-in app-side mempool implementations."`
+	Indexer           indexer.IndexingConfig `mapstructure:"indexer" toml:"indexer" comment:"indexer defines the configuration for the SDK built-in indexer implementation."`
+	IndexABCIEvents   []string               `mapstructure:"index-abci-events" toml:"index-abci-events" comment:"index-abci-events defines the set of events in the form {eventType}.{attributeKey}, which informs CometBFT what to index. If empty, all events will be indexed."`
+	DisableABCIEvents bool                   `mapstructure:"disable-abci-events" toml:"disable-abci-events" comment:"disable-abci-events disables the ABCI event indexing. It is useful when relying on the indexer for event indexing."`
 }
 
 // CfgOption is a function that allows to overwrite the default server configuration.

server/v2/cometbft/query.go

@@ -61,7 +61,7 @@ func (c *consensus[T]) handleQueryApp(ctx context.Context, path []string, req *a
 			return nil, errorsmod.Wrap(err, "failed to simulate tx")
 		}
 
-		bz, err := intoABCISimulationResponse(txResult, c.indexedEvents)
+		bz, err := intoABCISimulationResponse(txResult, c.indexedABCIEvents)
 		if err != nil {
 			return nil, errorsmod.Wrap(err, "failed to marshal txResult")
 		}

server/v2/cometbft/server.go

@@ -2,7 +2,6 @@ package cometbft
 
 import (
 	"context"
-	"cosmossdk.io/server/v2/cometbft/oe"
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
@@ -35,6 +34,7 @@ import (
 	"cosmossdk.io/server/v2/appmanager"
 	cometlog "cosmossdk.io/server/v2/cometbft/log"
 	"cosmossdk.io/server/v2/cometbft/mempool"
+	"cosmossdk.io/server/v2/cometbft/oe"
 	"cosmossdk.io/server/v2/cometbft/types"
 	"cosmossdk.io/store/v2/snapshots"
 
@@ -122,9 +122,9 @@ func New[T transaction.Tx](
 		}
 	}
 
-	indexEvents := make(map[string]struct{}, len(srv.config.AppTomlConfig.IndexEvents))
-	for _, e := range srv.config.AppTomlConfig.IndexEvents {
-		indexEvents[e] = struct{}{}
+	indexedABCIEvents := make(map[string]struct{}, len(srv.config.AppTomlConfig.IndexABCIEvents))
+	for _, e := range srv.config.AppTomlConfig.IndexABCIEvents {
+		indexedABCIEvents[e] = struct{}{}
 	}
 
 	ss := store.GetStateStorage().(snapshots.StorageSnapshotter)
@@ -185,7 +185,7 @@ func New[T transaction.Tx](
 		checkTxHandler:         srv.serverOptions.CheckTxHandler,
 		extendVote:             srv.serverOptions.ExtendVoteHandler,
 		chainID:                chainID,
-		indexedEvents:          indexEvents,
+		indexedABCIEvents:      indexedABCIEvents,
 		initialHeight:          0,
 		queryHandlersMap:       queryHandlers,
 		getProtoRegistry:       sync.OnceValues(gogoproto.MergedRegistry),

server/v2/cometbft/utils.go

@@ -2,9 +2,12 @@ package cometbft
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"math"
+	"slices"
 	"strings"
 	"time"
 
@@ -70,16 +73,20 @@ func finalizeBlockResponse(
 	cp *cmtproto.ConsensusParams,
 	appHash []byte,
 	indexSet map[string]struct{},
+	disableABCIEvents,
 	debug bool,
 ) (*abci.FinalizeBlockResponse, error) {
-	allEvents := append(in.BeginBlockEvents, in.EndBlockEvents...)
+	events := make([]abci.Event, 0)
 
-	events, err := intoABCIEvents(allEvents, indexSet)
-	if err != nil {
-		return nil, err
+	if !disableABCIEvents {
+		var err error
+		events, err = intoABCIEvents(append(in.BeginBlockEvents, in.EndBlockEvents...), indexSet)
+		if err != nil {
+			return nil, err
+		}
 	}
 
-	txResults, err := intoABCITxResults(in.TxResults, indexSet, debug)
+	txResults, err := intoABCITxResults(in.TxResults, indexSet, disableABCIEvents, debug)
 	if err != nil {
 		return nil, err
 	}
@@ -91,6 +98,7 @@ func finalizeBlockResponse(
 		AppHash:               appHash,
 		ConsensusParamUpdates: cp,
 	}
+
 	return resp, nil
 }
 
@@ -108,12 +116,21 @@ func intoABCIValidatorUpdates(updates []appmodulev2.ValidatorUpdate) []abci.Vali
 	return valsetUpdates
 }
 
-func intoABCITxResults(results []server.TxResult, indexSet map[string]struct{}, debug bool) ([]*abci.ExecTxResult, error) {
+func intoABCITxResults(
+	results []server.TxResult,
+	indexSet map[string]struct{},
+	disableABCIEvents, debug bool,
+) ([]*abci.ExecTxResult, error) {
 	res := make([]*abci.ExecTxResult, len(results))
 	for i := range results {
-		events, err := intoABCIEvents(results[i].Events, indexSet)
-		if err != nil {
-			return nil, err
+		var err error
+		events := make([]abci.Event, 0)
+
+		if !disableABCIEvents {
+			events, err = intoABCIEvents(results[i].Events, indexSet)
+			if err != nil {
+				return nil, err
+			}
 		}
 
 		res[i] = responseExecTxResultWithEvents(
@@ -132,16 +149,42 @@ func intoABCIEvents(events []event.Event, indexSet map[string]struct{}) ([]abci.
 	indexAll := len(indexSet) == 0
 	abciEvents := make([]abci.Event, len(events))
 	for i, e := range events {
-		attributes, err := e.Attributes()
-		if err != nil {
-			return nil, err
+		attrs := make([]event.Attribute, 0)
+
+		if e.Data != nil {
+			resp, err := e.Data()
+			if err != nil {
+				return nil, fmt.Errorf("failed to marshal event data: %w", err)
+			}
+
+			var attrMap map[string]json.RawMessage
+			if err := json.Unmarshal(resp, &attrMap); err != nil {
+				return nil, fmt.Errorf("failed to unmarshal event data: %w", err)
+			}
+
+			// sort the keys to ensure the order is always the same
+			keys := slices.Sorted(maps.Keys(attrMap))
+			for _, k := range keys {
+				v := attrMap[k]
+				attrs = append(attrs, event.Attribute{
+					Key:   k,
+					Value: string(v),
+				})
+			}
+		} else {
+			var err error
+			attrs, err = e.Attributes()
+			if err != nil {
+				return nil, err
+			}
 		}
+
 		abciEvents[i] = abci.Event{
 			Type:       e.Type,
-			Attributes: make([]abci.EventAttribute, len(attributes)),
+			Attributes: make([]abci.EventAttribute, len(attrs)),
 		}
 
-		for j, attr := range attributes {
+		for j, attr := range attrs {
 			_, index := indexSet[fmt.Sprintf("%s.%s", e.Type, attr.Key)]
 			abciEvents[i].Attributes[j] = abci.EventAttribute{
 				Key:   attr.Key,
@@ -154,26 +197,9 @@ func intoABCIEvents(events []event.Event, indexSet map[string]struct{}) ([]abci.
 }
 
 func intoABCISimulationResponse(txRes server.TxResult, indexSet map[string]struct{}) ([]byte, error) {
-	indexAll := len(indexSet) == 0
-	abciEvents := make([]abci.Event, len(txRes.Events))
-	for i, e := range txRes.Events {
-		attributes, err := e.Attributes()
-		if err != nil {
-			return nil, err
-		}
-		abciEvents[i] = abci.Event{
-			Type:       e.Type,
-			Attributes: make([]abci.EventAttribute, len(attributes)),
-		}
-
-		for j, attr := range attributes {
-			_, index := indexSet[fmt.Sprintf("%s.%s", e.Type, attr.Key)]
-			abciEvents[i].Attributes[j] = abci.EventAttribute{
-				Key:   attr.Key,
-				Value: attr.Value,
-				Index: index || indexAll,
-			}
-		}
+	abciEvents, err := intoABCIEvents(txRes.Events, indexSet)
+	if err != nil {
+		return nil, err
 	}
 
 	msgResponses := make([]*gogoany.Any, len(txRes.Resp))

server/v2/stf/core_event_service.go

@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"maps"
-	"slices"
 
 	"github.com/cosmos/gogoproto/jsonpb"
 	gogoproto "github.com/cosmos/gogoproto/proto"
@@ -39,12 +37,20 @@ type eventManager struct {
 // Emit emits an typed event that is defined in the protobuf file.
 // In the future these events will be added to consensus.
 func (em *eventManager) Emit(tev transaction.Msg) error {
-	res, err := TypedEventToEvent(tev)
-	if err != nil {
-		return err
+	event := event.Event{
+		Type: gogoproto.MessageName(tev),
+		Data: func() (json.RawMessage, error) {
+			buf := new(bytes.Buffer)
+			jm := &jsonpb.Marshaler{OrigName: true, EmitDefaults: true, AnyResolver: nil}
+			if err := jm.Marshal(buf, tev); err != nil {
+				return nil, err
+			}
+
+			return buf.Bytes(), nil
+		},
 	}
 
-	em.executionContext.events = append(em.executionContext.events, res)
+	em.executionContext.events = append(em.executionContext.events, event)
 	return nil
 }
 
@@ -53,34 +59,3 @@ func (em *eventManager) EmitKV(eventType string, attrs ...event.Attribute) error
 	em.executionContext.events = append(em.executionContext.events, event.NewEvent(eventType, attrs...))
 	return nil
 }
-
-// TypedEventToEvent takes typed event and converts to Event object
-func TypedEventToEvent(tev transaction.Msg) (event.Event, error) {
-	evtType := gogoproto.MessageName(tev)
-	buf := new(bytes.Buffer)
-	jm := &jsonpb.Marshaler{OrigName: true, EmitDefaults: true, AnyResolver: nil}
-	if err := jm.Marshal(buf, tev); err != nil {
-		return event.Event{}, err
-	}
-
-	var attrMap map[string]json.RawMessage
-	if err := json.Unmarshal(buf.Bytes(), &attrMap); err != nil {
-		return event.Event{}, err
-	}
-
-	// sort the keys to ensure the order is always the same
-	keys := slices.Sorted(maps.Keys(attrMap))
-	attrs := make([]event.Attribute, 0, len(attrMap))
-	for _, k := range keys {
-		v := attrMap[k]
-		attrs = append(attrs, event.Attribute{
-			Key:   k,
-			Value: string(v),
-		})
-	}
-
-	return event.Event{
-		Type:       evtType,
-		Attributes: func() ([]event.Attribute, error) { return attrs, nil },
-	}, nil
-}