feat: ADR-038 Part 3: State change files auxiliary gRPC server

CHANGELOG.md

@@ -41,6 +41,8 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 * [\#9533](https://github.com/cosmos/cosmos-sdk/pull/9533) Added a new gRPC method, `DenomOwners`, in `x/bank` to query for all account holders of a specific denomination.
 * (bank) [\#9618](https://github.com/cosmos/cosmos-sdk/pull/9618) Update bank.Metadata: add URI and URIHash attributes.
+* (store) [\#8664](https://github.com/cosmos/cosmos-sdk/pull/8664) Implementation of ADR-038 file StreamingService
+* (store) [\#9647](https://github.com/cosmos/cosmos-sdk/pull/9647) Implementation of ADR-038 file auxiliary gRPC server
 
 ### API Breaking Changes
 

baseapp/abci.go

@@ -195,6 +195,14 @@ 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 _, hook := range app.hooks {
+		if err := hook.ListenBeginBlock(app.deliverState.ctx, req, res); err != nil {
+			app.logger.Error("BeginBlock listening hook failed", "height", req.Header.Height, "err", err)
+		}
+	}
+
 	return res
 }
 
@@ -215,6 +223,13 @@ func (app *BaseApp) EndBlock(req abci.RequestEndBlock) (res abci.ResponseEndBloc
 		res.ConsensusParamUpdates = cp
 	}
 
+	// call the streaming service hooks with the EndBlock messages
+	for _, hook := range app.hooks {
+		if err := hook.ListenEndBlock(app.deliverState.ctx, req, res); err != nil {
+			app.logger.Error("EndBlock listening hook failed", "height", req.Height, "err", err)
+		}
+	}
+
 	return res
 }
 
@@ -275,16 +290,32 @@ func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx
 	gInfo, result, err := app.runTx(runTxModeDeliver, req.Tx)
 	if err != nil {
 		resultStr = "failed"
-		return sdkerrors.ResponseDeliverTx(err, gInfo.GasWanted, gInfo.GasUsed, app.trace)
+		res := sdkerrors.ResponseDeliverTx(err, gInfo.GasWanted, gInfo.GasUsed, app.trace)
+		// if we throw and error, be sure to still call the streaming service's hook
+		for _, hook := range app.hooks {
+			if err := hook.ListenDeliverTx(app.deliverState.ctx, req, res); err != nil {
+				app.logger.Error("DeliverTx listening hook failed", "err", err)
+			}
+		}
+		return res
 	}
 
-	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),
 	}
+
+	// call the streaming service hooks with the DeliverTx messages
+	for _, hook := range app.hooks {
+		if err := hook.ListenDeliverTx(app.deliverState.ctx, req, res); err != nil {
+			app.logger.Error("DeliverTx listening hook failed", "err", err)
+		}
+	}
+
+	return res
 }
 
 // Commit implements the ABCI interface. It will commit all state that exists in

baseapp/baseapp.go

@@ -134,6 +134,10 @@ type BaseApp struct { // nolint: maligned
 	// indexEvents defines the set of events in the form {eventType}.{attributeKey},
 	// which informs Tendermint what to index. If empty, all events will be indexed.
 	indexEvents map[string]struct{}
+
+	// hooked services
+	// these hooks will have the ABCI messages routed through them
+	hooks []Hook
 }
 
 // NewBaseApp returns a reference to an initialized BaseApp. It accepts a

baseapp/options.go

@@ -237,3 +237,14 @@ func (app *BaseApp) SetInterfaceRegistry(registry types.InterfaceRegistry) {
 	app.grpcQueryRouter.SetInterfaceRegistry(registry)
 	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 streaming service hooks within the BaseApp
+	// BaseApp will pass BeginBlock, DeliverTx, and EndBlock requests and responses to the streaming services to update their ABCI context using these hooks
+	app.hooks = append(app.hooks, s)
+}

baseapp/streaming.go

@@ -0,0 +1,30 @@
+package baseapp
+
+import (
+	"sync"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+
+	store "github.com/cosmos/cosmos-sdk/store/types"
+	"github.com/cosmos/cosmos-sdk/types"
+)
+
+// Hook interface used to hook into the ABCI message processing of the BaseApp
+type Hook interface {
+	// update the streaming service with the latest BeginBlock messages
+	ListenBeginBlock(ctx types.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error
+	// update the steaming service with the latest EndBlock messages
+	ListenEndBlock(ctx types.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error
+	// update the steaming service with the latest DeliverTx messages
+	ListenDeliverTx(ctx types.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error
+}
+
+// StreamingService interface for registering WriteListeners with the BaseApp and updating the service with the ABCI messages using the hooks
+type StreamingService interface {
+	// streaming service loop, awaits kv pairs and writes them to some destination stream or file
+	Stream(wg *sync.WaitGroup, quitChan <-chan struct{})
+	// returns the streaming service's listeners for the BaseApp to register
+	Listeners() map[types.StoreKey][]store.WriteListener
+	// interface for hooking into the ABCI messages from inside the BaseApp
+	Hook
+}

docs/architecture/adr-038-state-listening.md

@@ -32,7 +32,7 @@ In a new file, `store/types/listening.go`, we will create a `WriteListener` inte
 type WriteListener interface {
 	// if value is nil then it was deleted
 	// storeKey indicates the source KVStore, to facilitate using the the same WriteListener across separate KVStores
-	// set bool indicates if it was a set; true: set, false: delete
+	// delete bool indicates if it was a delete; true: delete, false: set
     OnWrite(storeKey StoreKey, key []byte, value []byte, delete bool) error
 }
 ```
@@ -48,10 +48,10 @@ and determine the source of each KV pair.
 
 ```protobuf
 message StoreKVPair {
-  optional string store_key = 1; // the store key for the KVStore this pair originates from
-  required bool set = 2; // true indicates a set operation, false indicates a delete operation
-  required bytes key = 3;
-  required bytes value = 4;
+  string store_key = 1; // the store key for the KVStore this pair originates from
+  bool set = 2; // true indicates a set operation, false indicates a delete operation
+  bytes key = 3;
+  bytes value = 4;
 }
 ```
 
@@ -209,9 +209,9 @@ We will introduce a new `StreamingService` interface for exposing `WriteListener
 ```go
 // Hook interface used to hook into the ABCI message processing of the BaseApp
 type Hook interface {
-	ListenBeginBlock(ctx sdk.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) // update the streaming service with the latest BeginBlock messages
-	ListenEndBlock(ctx sdk.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) // update the steaming service with the latest EndBlock messages
-	ListenDeliverTx(ctx sdk.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) // update the steaming service with the latest DeliverTx messages
+	ListenBeginBlock(ctx sdk.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error // update the streaming service with the latest BeginBlock messages
+	ListenEndBlock(ctx sdk.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error// update the steaming service with the latest EndBlock messages
+	ListenDeliverTx(ctx sdk.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error // update the steaming service with the latest DeliverTx messages
 }
 
 // StreamingService interface for registering WriteListeners with the BaseApp and updating the service with the ABCI messages using the hooks
@@ -384,8 +384,8 @@ using the provided `AppOptions` and TOML configuration fields.
 We will add a new method to the `BaseApp` to enable the registration of `StreamingService`s:
 
 ```go
-// RegisterStreamingService is used to register a streaming service with the BaseApp
-func (app *BaseApp) RegisterHooks(s StreamingService) {
+// SetStreamingService is used to register a streaming service with the BaseApp
+func (app *BaseApp) SetStreamingService(s StreamingService) {
 	// set the listeners for each StoreKey
 	for key, lis := range s.Listeners() {
 		app.cms.AddListeners(key, lis)
@@ -482,60 +482,60 @@ We will also provide a mapping of the TOML `store.streamers` "file" configuratio
 streaming service. In the future, as other streaming services are added, their constructors will be added here as well.
 
 ```go
-// StreamingServiceConstructor is used to construct a streaming service
-type StreamingServiceConstructor func(opts servertypes.AppOptions, keys []sdk.StoreKey) (StreamingService, error)
+// ServiceConstructor is used to construct a streaming service
+type ServiceConstructor func(opts serverTypes.AppOptions, keys []sdk.StoreKey, marshaller codec.BinaryMarshaler) (sdk.StreamingService, error)
 
-// StreamingServiceType enum for specifying the type of StreamingService
-type StreamingServiceType int
+// ServiceType enum for specifying the type of StreamingService
+type ServiceType int
 
 const (
-	Unknown StreamingServiceType = iota
-	File
-	// add more in the future
+  Unknown ServiceType = iota
+  File
+  // add more in the future
 )
 
-// NewStreamingServiceType returns the StreamingServiceType corresponding to the provided name
-func NewStreamingServiceType(name string) StreamingServiceType {
-	switch strings.ToLower(name) {
-	case "file", "f":
-		return File
-	default:
-		return Unknown
-	}
-}
-
-// String returns the string name of a StreamingServiceType
-func (sst StreamingServiceType) String() string {
-	switch sst {
-	case File:
-		return "file"
-	default:
-		return ""
-	}
-}
-
-// StreamingServiceConstructorLookupTable is a mapping of StreamingServiceTypes to StreamingServiceConstructors
-var StreamingServiceConstructorLookupTable = map[StreamingServiceType]StreamingServiceConstructor{
-	File: FileStreamingConstructor,
-}
-
-// NewStreamingServiceConstructor returns the StreamingServiceConstructor corresponding to the provided name
-func NewStreamingServiceConstructor(name string) (StreamingServiceConstructor, error) {
-	ssType := NewStreamingServiceType(name)
-	if ssType == Unknown {
-		return nil, fmt.Errorf("unrecognized streaming service name %s", name)
-	}
-	if constructor, ok := StreamingServiceConstructorLookupTable[ssType]; ok {
-		return constructor, nil
-	}
-	return nil, fmt.Errorf("streaming service constructor of type %s not found", ssType.String())
-}
-
-// FileStreamingConstructor is the StreamingServiceConstructor function for creating a FileStreamingService
-func FileStreamingConstructor(opts servertypes.AppOptions, keys []sdk.StoreKey) (StreamingService, error) {
-	filePrefix := cast.ToString(opts.Get("streamers.file.prefix"))
-	fileDir := cast.ToString(opts.Get("streamers.file.writeDir"))
-	return streaming.NewFileStreamingService(fileDir, filePrefix, keys), nil
+// NewStreamingServiceType returns the streaming.ServiceType corresponding to the provided name
+func NewStreamingServiceType(name string) ServiceType {
+  switch strings.ToLower(name) {
+  case "file", "f":
+    return File
+  default:
+    return Unknown
+  }
+}
+
+// String returns the string name of a streaming.ServiceType
+func (sst ServiceType) String() string {
+  switch sst {
+  case File:
+    return "file"
+  default:
+    return ""
+  }
+}
+
+// ServiceConstructorLookupTable is a mapping of streaming.ServiceTypes to streaming.ServiceConstructors
+var ServiceConstructorLookupTable = map[ServiceType]ServiceConstructor{
+  File: FileStreamingConstructor,
+}
+
+// NewServiceConstructor returns the streaming.ServiceConstructor corresponding to the provided name
+func NewServiceConstructor(name string) (ServiceConstructor, error) {
+  ssType := NewStreamingServiceType(name)
+  if ssType == Unknown {
+    return nil, fmt.Errorf("unrecognized streaming service name %s", name)
+  }
+  if constructor, ok := ServiceConstructorLookupTable[ssType]; ok {
+    return constructor, nil
+  }
+  return nil, fmt.Errorf("streaming service constructor of type %s not found", ssType.String())
+}
+
+// FileStreamingConstructor is the streaming.ServiceConstructor function for creating a FileStreamingService
+func FileStreamingConstructor(opts serverTypes.AppOptions, keys []sdk.StoreKey, marshaller codec.BinaryMarshaler) (sdk.StreamingService, error) {
+  filePrefix := cast.ToString(opts.Get("streamers.file.prefix"))
+  fileDir := cast.ToString(opts.Get("streamers.file.writeDir"))
+  return file.NewStreamingService(fileDir, filePrefix, keys, marshaller)
 }
 ```
 
@@ -564,8 +564,8 @@ func NewSimApp(
 	listeners := cast.ToStringSlice(appOpts.Get("store.streamers"))
 	for _, listenerName := range listeners {
 		// get the store keys allowed to be exposed for this streaming service/state listeners
-		exposeKeyStrs := cast.ToStringSlice(appOpts.Get(fmt.Sprintf("streamers.%s.keys", listenerName))
-		exposeStoreKeys = make([]storeTypes.StoreKey, 0, len(exposeKeyStrs))
+		exposeKeyStrs := cast.ToStringSlice(appOpts.Get(fmt.Sprintf("streamers.%s.keys", listenerName)))
+		exposeStoreKeys := make([]storeTypes.StoreKey, 0, len(exposeKeyStrs))
 		for _, keyStr := range exposeKeyStrs {
 			if storeKey, ok := keys[keyStr]; ok {
 				exposeStoreKeys = append(exposeStoreKeys, storeKey)
@@ -577,15 +577,15 @@ func NewSimApp(
 			tmos.Exit(err.Error()) // or continue?
 		}
 		// generate the streaming service using the constructor, appOptions, and the StoreKeys we want to expose
-		streamingService, err := constructor(appOpts, exposeStoreKeys)
+		streamingService, err := constructor(appOpts, exposeStoreKeys, appCodec)
 		if err != nil {
 			tmos.Exit(err.Error())
 		}
 		// register the streaming service with the BaseApp
 		bApp.RegisterStreamingService(streamingService)
 		// waitgroup and quit channel for optional shutdown coordination of the streaming service
 		wg := new(sync.WaitGroup)
-		quitChan := new(chan struct{}))
+		quitChan := make(chan struct{}))
 		// kick off the background streaming service loop
 		streamingService.Stream(wg, quitChan) // maybe this should be done from inside BaseApp instead?
 	}