Egaxhaj figure/adr 038 plugin system with release pio v0.45.x

.gitignore

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

CHANGELOG.md

@@ -37,6 +37,10 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 ## [Unreleased]
 
+### Features
+
+* [\10639](https://github.com/cosmos/cosmos-sdk/pull/10639) Plugin architecture for ADR-038 + FileStreamingService plugin.
+
 # Provenance Specific releases
 
 ## [v0.45-pio-2](https://github.com/provenance-io/cosmos-sdk/releases/tag/v0.45-pio-2) - 2022-02-15

Makefile

@@ -239,6 +239,9 @@ endif
 
 .PHONY: run-tests test test-all $(TEST_TARGETS)
 
+# Sim tests with state listening plugins enabled
+include sim-state-listening.mk
+
 test-sim-nondeterminism:
 	@echo "Running non-determinism test..."
 	@go test -mod=readonly $(SIMAPP) -run TestAppStateDeterminism -Enabled=true \

baseapp/abci.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -196,6 +197,33 @@ 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
+	wg := new(sync.WaitGroup)
+	for _, streamingListener := range app.abciListeners {
+		streamingListener := streamingListener // https://go.dev/doc/faq#closures_and_goroutines
+		if streamingListener.HaltAppOnDeliveryError() {
+			// increment the wait group counter
+			wg.Add(1)
+			go func() {
+				// decrement the counter when the go routine completes
+				defer wg.Done()
+				if err := streamingListener.ListenBeginBlock(app.deliverState.ctx, req, res); err != nil {
+					app.logger.Error("BeginBlock listening hook failed", "height", req.Header.Height, "err", err)
+					app.halt()
+				}
+			}()
+		} else {
+			go func() {
+				if err := streamingListener.ListenBeginBlock(app.deliverState.ctx, req, res); err != nil {
+					app.logger.Error("BeginBlock listening hook failed", "height", req.Header.Height, "err", err)
+				}
+			}()
+		}
+	}
+	// wait for all the listener calls to finish
+	wg.Wait()
+
 	return res
 }
 
@@ -216,6 +244,32 @@ func (app *BaseApp) EndBlock(req abci.RequestEndBlock) (res abci.ResponseEndBloc
 		res.ConsensusParamUpdates = cp
 	}
 
+	// call the hooks with the BeginBlock messages
+	wg := new(sync.WaitGroup)
+	for _, streamingListener := range app.abciListeners {
+		streamingListener := streamingListener // https://go.dev/doc/faq#closures_and_goroutines
+		if streamingListener.HaltAppOnDeliveryError() {
+			// increment the wait group counter
+			wg.Add(1)
+			go func() {
+				// decrement the counter when the go routine completes
+				defer wg.Done()
+				if err := streamingListener.ListenEndBlock(app.deliverState.ctx, req, res); err != nil {
+					app.logger.Error("EndBlock listening hook failed", "height", req.Height, "err", err)
+					app.halt()
+				}
+			}()
+		} else {
+			go func() {
+				if err := streamingListener.ListenEndBlock(app.deliverState.ctx, req, res); err != nil {
+					app.logger.Error("EndBlock listening hook failed", "height", req.Height, "err", err)
+				}
+			}()
+		}
+	}
+	// wait for all the listener calls to finish
+	wg.Wait()
+
 	return res
 }
 
@@ -266,26 +320,55 @@ func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx
 	gInfo := sdk.GasInfo{}
 	resultStr := "successful"
 
+	var abciRes abci.ResponseDeliverTx
 	defer func() {
 		telemetry.IncrCounter(1, "tx", "count")
 		telemetry.IncrCounter(1, "tx", resultStr)
 		telemetry.SetGauge(float32(gInfo.GasUsed), "tx", "gas", "used")
 		telemetry.SetGauge(float32(gInfo.GasWanted), "tx", "gas", "wanted")
+		// call the hooks with the BeginBlock messages
+		wg := new(sync.WaitGroup)
+		for _, streamingListener := range app.abciListeners {
+			streamingListener := streamingListener // https://go.dev/doc/faq#closures_and_goroutines
+			if streamingListener.HaltAppOnDeliveryError() {
+				// increment the wait group counter
+				wg.Add(1)
+				go func() {
+					// decrement the counter when the go routine completes
+					defer wg.Done()
+					if err := streamingListener.ListenDeliverTx(app.deliverState.ctx, req, abciRes); err != nil {
+						app.logger.Error("DeliverTx listening hook failed", "err", err)
+						app.halt()
+					}
+				}()
+			} else {
+				go func() {
+					if err := streamingListener.ListenDeliverTx(app.deliverState.ctx, req, abciRes); err != nil {
+						app.logger.Error("DeliverTx listening hook failed", "err", err)
+					}
+				}()
+			}
+		}
+		// wait for all the listener calls to finish
+		wg.Wait()
 	}()
 
 	gInfo, result, anteEvents, _, err := app.runTx(runTxModeDeliver, req.Tx)
 	if err != nil {
 		resultStr = "failed"
-		return sdkerrors.ResponseDeliverTxWithEvents(err, gInfo.GasWanted, gInfo.GasUsed, anteEvents, app.trace)
+		abciRes = sdkerrors.ResponseDeliverTxWithEvents(err, gInfo.GasWanted, gInfo.GasUsed, anteEvents, app.trace)
+		return abciRes
 	}
 
-	return abci.ResponseDeliverTx{
+	abciRes = 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 abciRes
 }
 
 // Commit implements the ABCI interface. It will commit all state that exists in

baseapp/baseapp.go

@@ -6,20 +6,19 @@ import (
 	"reflect"
 	"strings"
 
-	"github.com/gogo/protobuf/proto"
-	abci "github.com/tendermint/tendermint/abci/types"
-	"github.com/tendermint/tendermint/crypto/tmhash"
-	"github.com/tendermint/tendermint/libs/log"
-	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
-	dbm "github.com/tendermint/tm-db"
-
 	"github.com/cosmos/cosmos-sdk/codec/types"
 	"github.com/cosmos/cosmos-sdk/snapshots"
 	"github.com/cosmos/cosmos-sdk/store"
 	"github.com/cosmos/cosmos-sdk/store/rootmulti"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 	"github.com/cosmos/cosmos-sdk/x/auth/legacy/legacytx"
+	"github.com/gogo/protobuf/proto"
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/crypto/tmhash"
+	"github.com/tendermint/tendermint/libs/log"
+	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
+	dbm "github.com/tendermint/tm-db"
 )
 
 const (
@@ -135,6 +134,10 @@ type BaseApp struct { // nolint: maligned
 	indexEvents map[string]struct{}
 
 	feeHandler sdk.FeeHandler
+
+	// abciListeners for hooking into the ABCI message processing of the BaseApp
+	// and exposing the requests and responses to external consumers
+	abciListeners []ABCIListener
 }
 
 // NewBaseApp returns a reference to an initialized BaseApp. It accepts a

baseapp/options.go

@@ -2,14 +2,12 @@ package baseapp
 
 import (
 	"fmt"
-	"io"
-
-	dbm "github.com/tendermint/tm-db"
-
 	"github.com/cosmos/cosmos-sdk/codec/types"
 	"github.com/cosmos/cosmos-sdk/snapshots"
 	"github.com/cosmos/cosmos-sdk/store"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	dbm "github.com/tendermint/tm-db"
+	"io"
 )
 
 // File for storing in-package BaseApp optional functions,
@@ -251,3 +249,14 @@ func (app *BaseApp) SetFeeHandler(feeHandler sdk.FeeHandler) {
 
 	app.feeHandler = feeHandler
 }
+
+// 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)
+}

baseapp/streaming.go

@@ -0,0 +1,40 @@
+package baseapp
+
+import (
+	"io"
+	"sync"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+
+	store "github.com/cosmos/cosmos-sdk/store/types"
+	"github.com/cosmos/cosmos-sdk/types"
+)
+
+// ABCIListener interface used to hook into the ABCI message processing of the BaseApp
+type ABCIListener interface {
+	// ListenBeginBlock updates the streaming service with the latest BeginBlock messages
+	ListenBeginBlock(ctx types.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error
+	// ListenEndBlock updates the steaming service with the latest EndBlock messages
+	ListenEndBlock(ctx types.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error
+	// ListenDeliverTx updates the steaming service with the latest DeliverTx messages
+	ListenDeliverTx(ctx types.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error
+	// HaltAppOnDeliveryError returns true if the application has been configured to halt when
+	// ListenBeginBlock, ListenEndBlock, ListenDeliverTx fail to process messages and false when
+	// the application has been configured to send messages to ListenBeginBlock, ListenEndBlock, ListenDeliverTx
+	// in fire-and-forget fashion.
+	//
+	// This behavior is controlled by a corresponding app config setting.
+	HaltAppOnDeliveryError() bool
+}
+
+// StreamingService interface for registering WriteListeners with the BaseApp and updating the service with the ABCI messages using the hooks
+type StreamingService interface {
+	// Stream is the streaming service loop, awaits kv pairs and writes them to some destination stream or file
+	Stream(wg *sync.WaitGroup) error
+	// Listeners returns the streaming service's listeners for the BaseApp to register
+	Listeners() map[store.StoreKey][]store.WriteListener
+	// ABCIListener interface for hooking into the ABCI messages from inside the BaseApp
+	ABCIListener
+	// Closer interface
+	io.Closer
+}