feat: Adds ADR 038 - go plugin system

.gitignore

@@ -45,6 +45,8 @@ vagrant
 # IDE
 .idea
 *.iml
+*.ipr
+*.iws
 .dir-locals.el
 .vscode
 

CHANGELOG.md

@@ -39,7 +39,9 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 ## Unreleased
 
-* nothing
+### Features
+
+* [#404](https://github.com/provenance-io/cosmos-sdk/pull/404) Add ADR-038 State Listening with Go Plugin System
 
 ---
 

Makefile

@@ -273,6 +273,12 @@ test-sim-nondeterminism:
 	@go test -mod=readonly $(SIMAPP) -run TestAppStateDeterminism -Enabled=true \
 		-NumBlocks=100 -BlockSize=200 -Commit=true -Period=0 -v -timeout 24h
 
+test-sim-nondeterminism-streaming:
+	@echo "Running non-determinism-streaming test..."
+	@go test -mod=readonly $(SIMAPP) -run TestAppStateDeterminism -Enabled=true \
+		-NumBlocks=100 -BlockSize=200 -Commit=true -Period=0 -v -timeout 24h \
+		-EnableStreaming=true
+
 test-sim-custom-genesis-fast:
 	@echo "Running custom genesis simulation..."
 	@echo "By default, ${HOME}/.gaiad/config/genesis.json will be used."

baseapp/abci.go

@@ -12,11 +12,13 @@ import (
 	"time"
 
 	"github.com/gogo/protobuf/proto"
-	abci "github.com/tendermint/tendermint/abci/types"
-	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
+	"github.com/hashicorp/go-plugin"
 	"google.golang.org/grpc/codes"
 	grpcstatus "google.golang.org/grpc/status"
 
+	abci "github.com/tendermint/tendermint/abci/types"
+	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
+
 	"github.com/cosmos/cosmos-sdk/codec"
 	snapshottypes "github.com/cosmos/cosmos-sdk/snapshots/types"
 	"github.com/cosmos/cosmos-sdk/telemetry"
@@ -186,10 +188,12 @@ func (app *BaseApp) BeginBlock(req abci.RequestBeginBlock) (res abci.ResponseBeg
 	// set the signed validators for addition to context in deliverTx
 	app.voteInfos = req.LastCommitInfo.GetVotes()
 
-	// call the hooks with the BeginBlock messages
-	for _, streamingListener := range app.abciListeners {
-		if err := streamingListener.ListenBeginBlock(app.deliverState.ctx, req, res); err != nil {
-			panic(fmt.Errorf("BeginBlock listening hook failed, height: %d, err: %w", req.Header.Height, err))
+	// call the streaming service hook with the BeginBlock messages
+	for _, abciListener := range app.streamingManager.AbciListeners {
+		ctx := app.deliverState.ctx
+		blockHeight := ctx.BlockHeight()
+		if err := abciListener.ListenBeginBlock(ctx, req, res); err != nil {
+			app.logger.Error("BeginBlock listening hook failed", "height", blockHeight, "err", err)
 		}
 	}
 
@@ -213,10 +217,12 @@ func (app *BaseApp) EndBlock(req abci.RequestEndBlock) (res abci.ResponseEndBloc
 		res.ConsensusParamUpdates = cp
 	}
 
-	// call the streaming service hooks with the EndBlock messages
-	for _, streamingListener := range app.abciListeners {
-		if err := streamingListener.ListenEndBlock(app.deliverState.ctx, req, res); err != nil {
-			panic(fmt.Errorf("EndBlock listening hook failed, height: %d, err: %w", req.Height, err))
+	// call the streaming service hook with the EndBlock messages
+	for _, abciListener := range app.streamingManager.AbciListeners {
+		ctx := app.deliverState.ctx
+		blockHeight := ctx.BlockHeight()
+		if err := abciListener.ListenEndBlock(ctx, req, res); err != nil {
+			app.logger.Error("EndBlock listening hook failed", "height", blockHeight, "err", err)
 		}
 	}
 
@@ -263,14 +269,18 @@ func (app *BaseApp) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
 // Otherwise, the ResponseDeliverTx will contain releveant error information.
 // Regardless of tx execution outcome, the ResponseDeliverTx will contain relevant
 // gas execution context.
-func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) (res abci.ResponseDeliverTx) {
+func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx {
 	gInfo := sdk.GasInfo{}
 	resultStr := "successful"
 
+	var res abci.ResponseDeliverTx
 	defer func() {
-		for _, streamingListener := range app.abciListeners {
-			if err := streamingListener.ListenDeliverTx(app.deliverState.ctx, req, res); err != nil {
-				panic(fmt.Errorf("DeliverTx listening hook failed: %w", err))
+		// call the streaming service hook with the EndBlock messages
+		for _, abciListener := range app.streamingManager.AbciListeners {
+			ctx := app.deliverState.ctx
+			blockHeight := ctx.BlockHeight()
+			if err := abciListener.ListenDeliverTx(ctx, req, res); err != nil {
+				app.logger.Error("DeliverTx listening hook failed", "height", blockHeight, "err", err)
 			}
 		}
 	}()
@@ -288,13 +298,15 @@ func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) (res abci.ResponseDeliv
 		return sdkerrors.ResponseDeliverTxWithEvents(err, gInfo.GasWanted, gInfo.GasUsed, sdk.MarkEventsToIndex(anteEvents, app.indexEvents), app.trace)
 	}
 
-	return abci.ResponseDeliverTx{
+	res = abci.ResponseDeliverTx{
 		GasWanted: int64(gInfo.GasWanted), // TODO: Should type accept unsigned ints?
 		GasUsed:   int64(gInfo.GasUsed),   // TODO: Should type accept unsigned ints?
 		Log:       result.Log,
 		Data:      result.Data,
 		Events:    sdk.MarkEventsToIndex(result.Events, app.indexEvents),
 	}
+
+	return res
 }
 
 // Commit implements the ABCI interface. It will commit all state that exists in
@@ -305,6 +317,7 @@ func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) (res abci.ResponseDeliv
 // against that height and gracefully halt if it matches the latest committed
 // height.
 func (app *BaseApp) Commit() abci.ResponseCommit {
+
 	header := app.deliverState.ctx.BlockHeader()
 	retainHeight := app.GetBlockRetentionHeight(header.Height)
 
@@ -319,11 +332,18 @@ func (app *BaseApp) Commit() abci.ResponseCommit {
 		RetainHeight: retainHeight,
 	}
 
-	// call the hooks with the Commit message
-	for _, streamingListener := range app.abciListeners {
-		if err := streamingListener.ListenCommit(app.deliverState.ctx, res); err != nil {
-			panic(fmt.Errorf("Commit listening hook failed, height: %d, err: %w", header.Height, err))
+	// call the streaming service hook with the EndBlock messages
+	abciListeners := app.streamingManager.AbciListeners
+	if len(abciListeners) > 0 {
+		ctx := app.deliverState.ctx
+		blockHeight := ctx.BlockHeight()
+		changeSet := app.cms.PopStateCache()
+		for _, abciListener := range abciListeners {
+			if err := abciListener.ListenCommit(ctx, res, changeSet); err != nil {
+				app.logger.Error("ListenCommit listening hook failed", "height", blockHeight, "err", err)
+			}
 		}
+		changeSet = nil
 	}
 
 	app.logger.Info("commit synced", "commit", fmt.Sprintf("%X", commitID))
@@ -376,6 +396,9 @@ func (app *BaseApp) halt() {
 		}
 	}
 
+	// Stop any running streaming plugins
+	plugin.CleanupClients()
+
 	// Resort to exiting immediately if the process could not be found or killed
 	// via SIGINT/SIGTERM signals.
 	app.logger.Info("failed to send SIGINT/SIGTERM; exiting...")

baseapp/baseapp.go

@@ -107,9 +107,8 @@ type BaseApp struct { // nolint: maligned
 	// which informs Tendermint what to index. If empty, all events will be indexed.
 	indexEvents map[string]struct{}
 
-	// abciListeners for hooking into the ABCI message processing of the BaseApp
-	// and exposing the requests and responses to external consumers
-	abciListeners []ABCIListener
+	// streamingManager for managing instances and configuration of ABCIListener services
+	streamingManager storetypes.StreamingManager
 
 	feeHandler sdk.FeeHandler
 

baseapp/baseapp_test.go

@@ -2265,3 +2265,82 @@ func TestDefaultStoreLoader(t *testing.T) {
 		require.Equal(t, storetypes.StoreTypeIAVL, store.GetStoreType())
 	}
 }
+
+var (
+	distKey1 = storetypes.NewKVStoreKey("distKey1")
+)
+
+func TestABCI_MultiListener_StateChanges(t *testing.T) {
+	// increment the tx counter
+	anteKey := []byte("ante-key")
+	anteOpt := func(bapp *BaseApp) { bapp.SetAnteHandler(anteHandlerTxTest(t, capKey1, anteKey)) }
+
+	// increment the msg counter
+	deliverKey := []byte("deliver-key")
+	deliverKey2 := []byte("deliver-key2")
+	routerOpt := func(bapp *BaseApp) {
+		r1 := sdk.NewRoute(routeMsgCounter, handlerMsgCounter(t, capKey1, deliverKey))
+		r2 := sdk.NewRoute(routeMsgCounter2, handlerMsgCounter(t, capKey1, deliverKey2))
+		bapp.Router().AddRoute(r1)
+		bapp.Router().AddRoute(r2)
+	}
+
+	distOpt := func(bapp *BaseApp) { bapp.MountStores(distKey1) }
+	mockListener1 := NewMockABCIListener("lis_1")
+	mockListener2 := NewMockABCIListener("lis_2")
+	streamingManager := storetypes.StreamingManager{AbciListeners: []storetypes.ABCIListener{&mockListener1, &mockListener2}}
+	streamingManagerOpt := func(bapp *BaseApp) { bapp.SetStreamingManager(streamingManager) }
+	addListenerOpt := func(bapp *BaseApp) { bapp.CommitMultiStore().AddListeners([]storetypes.StoreKey{distKey1}) }
+
+	app := setupBaseApp(t, anteOpt, routerOpt, distOpt, streamingManagerOpt, addListenerOpt)
+	app.InitChain(abci.RequestInitChain{})
+
+	// Create same codec used in txDecoder
+	codec := codec.NewLegacyAmino()
+	registerTestCodec(codec)
+
+	nBlocks := 3
+	txPerHeight := 5
+
+	for blockN := 0; blockN < nBlocks; blockN++ {
+		header := tmproto.Header{Height: int64(blockN) + 1}
+		app.BeginBlock(abci.RequestBeginBlock{Header: header})
+		var expectedChangeSet []*storetypes.StoreKVPair
+
+		for i := 0; i < txPerHeight; i++ {
+			counter := int64(blockN*txPerHeight + i)
+			tx := newTxCounter(counter, counter)
+
+			txBytes, err := codec.Marshal(tx)
+			require.NoError(t, err)
+
+			sKey := []byte(fmt.Sprintf("distKey%d", i))
+			sVal := []byte(fmt.Sprintf("distVal%d", i))
+
+			ctx := app.getState(runTxModeDeliver).ctx
+			store := ctx.KVStore(distKey1)
+			store.Set(sKey, sVal)
+
+			expectedChangeSet = append(expectedChangeSet, &storetypes.StoreKVPair{
+				StoreKey: distKey1.Name(),
+				Delete:   false,
+				Key:      sKey,
+				Value:    sVal,
+			})
+
+			res := app.DeliverTx(abci.RequestDeliverTx{Tx: txBytes})
+			require.True(t, res.IsOK(), fmt.Sprintf("%v", res))
+
+			events := res.GetEvents()
+			require.Len(t, events, 3, "should contain ante handler, message type and counter events respectively")
+			require.Equal(t, sdk.MarkEventsToIndex(counterEvent("ante_handler", counter).ToABCIEvents(), map[string]struct{}{})[0], events[0], "ante handler event")
+			require.Equal(t, sdk.MarkEventsToIndex(counterEvent(sdk.EventTypeMessage, counter).ToABCIEvents(), map[string]struct{}{})[0], events[2], "msg handler update counter event")
+		}
+
+		app.EndBlock(abci.RequestEndBlock{})
+		app.Commit()
+
+		require.Equal(t, expectedChangeSet, mockListener1.ChangeSet, "should contain the same changeSet")
+		require.Equal(t, expectedChangeSet, mockListener2.ChangeSet, "should contain the same changeSet")
+	}
+}

baseapp/options.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/snapshots"
 	snapshottypes "github.com/cosmos/cosmos-sdk/snapshots/types"
 	"github.com/cosmos/cosmos-sdk/store"
+	storetypes "github.com/cosmos/cosmos-sdk/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
@@ -234,17 +235,6 @@ func (app *BaseApp) SetInterfaceRegistry(registry types.InterfaceRegistry) {
 	app.msgServiceRouter.SetInterfaceRegistry(registry)
 }
 
-// SetStreamingService is used to set a streaming service into the BaseApp hooks and load the listeners into the multistore
-func (app *BaseApp) SetStreamingService(s StreamingService) {
-	// add the listeners for each StoreKey
-	for key, lis := range s.Listeners() {
-		app.cms.AddListeners(key, lis)
-	}
-	// register the StreamingService within the BaseApp
-	// BaseApp will pass BeginBlock, DeliverTx, and EndBlock requests and responses to the streaming services to update their ABCI context
-	app.abciListeners = append(app.abciListeners, s)
-}
-
 // SetQueryMultiStore set a alternative MultiStore implementation to support grpc query service.
 //
 // Ref: https://github.com/cosmos/cosmos-sdk/issues/13317
@@ -272,3 +262,8 @@ func (app *BaseApp) SetAggregateEventsFunc(aggregateEventsFunc func(resultEvents
 
 	app.aggregateEventsFunc = aggregateEventsFunc
 }
+
+// SetStreamingManager sets the streaming manager for the BaseApp.test
+func (app *BaseApp) SetStreamingManager(manager storetypes.StreamingManager) {
+	app.streamingManager = manager
+}