Data pipeline

baseapp/abci.go

@@ -711,6 +711,8 @@ func (app *BaseApp) VerifyVoteExtension(req *abci.VerifyVoteExtensionRequest) (r
 // must be used.
 func (app *BaseApp) internalFinalizeBlock(ctx context.Context, req *abci.FinalizeBlockRequest) (*abci.FinalizeBlockResponse, error) {
 	var events []abci.Event
+	var publishEvents sdk.PublishEvents
+	var txEventSet []EventSet
 
 	if err := app.checkHalt(req.Height, req.Time); err != nil {
 		return nil, err
@@ -803,6 +805,7 @@ func (app *BaseApp) internalFinalizeBlock(ctx context.Context, req *abci.Finaliz
 		// continue
 	}
 
+	publishEvents = append(publishEvents, app.finalizeBlockState.Context().PublishEventManager().Events()...)
 	events = append(events, beginBlock.Events...)
 
 	// Reset the gas meter so that the AnteHandlers aren't required to
@@ -818,6 +821,8 @@ func (app *BaseApp) internalFinalizeBlock(ctx context.Context, req *abci.Finaliz
 	for _, rawTx := range req.Txs {
 		var response *abci.ExecTxResult
 
+		app.finalizeBlockState.SetContext(app.finalizeBlockState.Context().WithPublishEventManager(sdk.NewPublishEventManager()))
+
 		if _, err := app.txDecoder(rawTx); err == nil {
 			response = app.deliverTx(rawTx)
 		} else {
@@ -841,13 +846,22 @@ func (app *BaseApp) internalFinalizeBlock(ctx context.Context, req *abci.Finaliz
 			// continue
 		}
 
+		filtered, order := filterOutPublishEvents(response.Events)
+		response.Events = filtered
 		txResults = append(txResults, response)
+		txEventSet = append(txEventSet, EventSet{
+			AbciEvents:    response.Events,
+			PublishEvents: app.finalizeBlockState.Context().PublishEventManager().Events(),
+			TrueOrder:     order,
+		})
 	}
 
 	if app.finalizeBlockState.ms.TracingEnabled() {
 		app.finalizeBlockState.ms = app.finalizeBlockState.ms.SetTracingContext(nil).(storetypes.CacheMultiStore)
 	}
 
+	app.finalizeBlockState.SetContext(app.finalizeBlockState.Context().WithPublishEventManager(sdk.NewPublishEventManager()))
+
 	endBlock, err := app.endBlock(app.finalizeBlockState.Context())
 	if err != nil {
 		return nil, err
@@ -861,9 +875,25 @@ func (app *BaseApp) internalFinalizeBlock(ctx context.Context, req *abci.Finaliz
 		// continue
 	}
 
+	publishEvents = append(publishEvents, app.finalizeBlockState.Context().PublishEventManager().Events()...)
+
 	events = append(events, endBlock.Events...)
 	cp := app.GetConsensusParams(app.finalizeBlockState.Context())
 
+	events, trueOrder := filterOutPublishEvents(events)
+	app.flushData = PublishEventFlush{
+		Height:      header.Height,
+		PrevAppHash: header.AppHash,
+		BlockEvents: EventSet{
+			AbciEvents:    events,
+			PublishEvents: publishEvents,
+			TrueOrder:     trueOrder,
+		},
+		TxEvents: txEventSet,
+	}
+
+	fmt.Println("events in finalize block", events)
+
 	return &abci.FinalizeBlockResponse{
 		Events:                events,
 		TxResults:             txResults,
@@ -872,6 +902,22 @@ func (app *BaseApp) internalFinalizeBlock(ctx context.Context, req *abci.Finaliz
 	}, nil
 }
 
+func filterOutPublishEvents(events []abci.Event) ([]abci.Event, []EventType) {
+	var filteredEvents []abci.Event
+	var trueOrder []EventType
+
+	for _, e := range events {
+		if e.Type == sdk.PlaceholderEventType {
+			trueOrder = append(trueOrder, EventTypePublish)
+			continue
+		}
+		filteredEvents = append(filteredEvents, e)
+		trueOrder = append(trueOrder, EventTypeAbci)
+	}
+
+	return filteredEvents, trueOrder
+}
+
 // FinalizeBlock will execute the block proposal provided by FinalizeBlockRequest.
 // Specifically, it will execute an application's BeginBlock (if defined), followed
 // by the transactions in the proposal, finally followed by the application's
@@ -959,7 +1005,10 @@ func (app *BaseApp) Commit() (*abci.CommitResponse, error) {
 		rms.SetCommitHeader(header)
 	}
 
-	app.cms.Commit()
+	commitId := app.cms.Commit()
+
+	app.flushData.NewAppHash = commitId.Hash
+	app.PublishBlockEvents(app.flushData)
 
 	resp := &abci.CommitResponse{
 		RetainHeight: retainHeight,

baseapp/abci_test.go

@@ -564,7 +564,7 @@ func TestABCI_CheckTx(t *testing.T) {
 	anteOpt := func(bapp *baseapp.BaseApp) { bapp.SetAnteHandler(anteHandlerTxTest(t, capKey1, counterKey)) }
 	suite := NewBaseAppSuite(t, anteOpt)
 
-	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, counterKey})
+	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, counterKey, false})
 
 	nTxs := int64(5)
 	_, err := suite.baseApp.InitChain(&abci.InitChainRequest{
@@ -618,7 +618,7 @@ func TestABCI_FinalizeBlock_DeliverTx(t *testing.T) {
 	require.NoError(t, err)
 
 	deliverKey := []byte("deliver-key")
-	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey})
+	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey, false})
 
 	nBlocks := 3
 	txPerHeight := 5
@@ -668,7 +668,7 @@ func TestABCI_FinalizeBlock_MultiMsg(t *testing.T) {
 	require.NoError(t, err)
 
 	deliverKey := []byte("deliver-key")
-	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey})
+	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey, false})
 
 	deliverKey2 := []byte("deliver-key2")
 	baseapptestutil.RegisterCounter2Server(suite.baseApp.MsgServiceRouter(), Counter2ServerImpl{t, capKey1, deliverKey2})

baseapp/baseapp.go

@@ -205,6 +205,10 @@ type BaseApp struct {
 	EnableStreamer bool
 	StreamEvents   chan StreamEvents
 
+	EnablePublish bool
+	PublishEvents chan PublishEventFlush
+	flushData     PublishEventFlush
+
 	traceFlightRecorder *metrics.TraceRecorder
 }
 
@@ -227,6 +231,7 @@ func NewBaseApp(
 		sigverifyTx:      true,
 		queryGasLimit:    math.MaxUint64,
 		StreamEvents:     make(chan StreamEvents),
+		PublishEvents:    make(chan PublishEventFlush),
 	}
 
 	for _, option := range options {

baseapp/baseapp_test.go

@@ -219,7 +219,7 @@ func TestAnteHandlerGasMeter(t *testing.T) {
 	require.NoError(t, err)
 
 	deliverKey := []byte("deliver-key")
-	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey})
+	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey, false})
 
 	tx := newTxCounter(t, suite.txConfig, 0, 0)
 	txBytes, err := suite.txConfig.TxEncoder()(tx)
@@ -562,7 +562,7 @@ func TestBaseAppAnteHandler(t *testing.T) {
 	suite := NewBaseAppSuite(t, anteOpt)
 
 	deliverKey := []byte("deliver-key")
-	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey})
+	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, deliverKey, false})
 
 	_, err := suite.baseApp.InitChain(&abci.InitChainRequest{
 		ConsensusParams: &cmtproto.ConsensusParams{},
@@ -636,7 +636,7 @@ func TestBaseAppPostHandler(t *testing.T) {
 	}
 
 	suite := NewBaseAppSuite(t, anteOpt)
-	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, []byte("foo")})
+	baseapptestutil.RegisterCounterServer(suite.baseApp.MsgServiceRouter(), CounterServerImpl{t, capKey1, []byte("foo"), false})
 
 	_, err := suite.baseApp.InitChain(&abci.InitChainRequest{
 		ConsensusParams: &cmtproto.ConsensusParams{},

baseapp/chain_stream.go

@@ -1,8 +1,9 @@
 package baseapp
 
 import (
-	abci "github.com/cometbft/cometbft/abci/types"
 	"time"
+
+	abci "github.com/cometbft/cometbft/abci/types"
 )
 
 type StreamEvents struct {

baseapp/publish_event.go

@@ -0,0 +1,33 @@
+package baseapp
+
+import (
+	abci "github.com/cometbft/cometbft/abci/types"
+	types "github.com/cosmos/cosmos-sdk/types"
+)
+
+type EventType byte
+
+const (
+	EventTypeAbci EventType = iota
+	EventTypePublish
+)
+
+type EventSet struct {
+	AbciEvents    []abci.Event
+	PublishEvents types.PublishEvents
+	TrueOrder     []EventType
+}
+
+type PublishEventFlush struct {
+	Height      int64
+	PrevAppHash []byte
+	NewAppHash  []byte
+	BlockEvents EventSet
+	TxEvents    []EventSet
+}
+
+func (app *BaseApp) PublishBlockEvents(flush PublishEventFlush) {
+	if app.EnablePublish {
+		app.PublishEvents <- flush
+	}
+}