Merge branch 'master' into fix-testnet

CHANGELOG.md

@@ -46,6 +46,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 * [\#9776](https://github.com/cosmos/cosmos-sdk/pull/9776) Add flag `staking-bond-denom` to specify the staking bond denomination value when initializing a new chain.
 * [\#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
 * [\#9837](https://github.com/cosmos/cosmos-sdk/issues/9837) `--generate-only` flag will accept the keyname now.
 * [\#10045](https://github.com/cosmos/cosmos-sdk/pull/10045) Revert [#8549](https://github.com/cosmos/cosmos-sdk/pull/8549). Do not route grpc queries through Tendermint.
 * [\#10326](https://github.com/cosmos/cosmos-sdk/pull/10326) `x/authz` add query all grants by granter query.

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 _, streamingListener := range app.abciListeners {
+		if err := streamingListener.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 _, streamingListener := range app.abciListeners {
+		if err := streamingListener.ListenEndBlock(app.deliverState.ctx, req, res); err != nil {
+			app.logger.Error("EndBlock listening hook failed", "height", req.Height, "err", err)
+		}
+	}
+
 	return res
 }
 
@@ -262,15 +277,25 @@ func (app *BaseApp) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
 func (app *BaseApp) DeliverTx(req abci.RequestDeliverTx) abci.ResponseDeliverTx {
 	defer telemetry.MeasureSince(time.Now(), "abci", "deliver_tx")
 
+	var res abci.ResponseDeliverTx
+	defer func() {
+		for _, streamingListener := range app.abciListeners {
+			if err := streamingListener.ListenDeliverTx(app.deliverState.ctx, req, res); err != nil {
+				app.logger.Error("DeliverTx listening hook failed", "err", err)
+			}
+		}
+	}()
 	tx, err := app.txDecoder(req.Tx)
 	if err != nil {
-		return sdkerrors.ResponseDeliverTx(err, 0, 0, app.trace)
+		res = sdkerrors.ResponseDeliverTx(err, 0, 0, app.trace)
+		return res
 	}
 
 	ctx := app.getContextForTx(runTxModeDeliver, req.Tx)
-	res, err := app.txHandler.DeliverTx(ctx, tx, req)
+	res, err = app.txHandler.DeliverTx(ctx, tx, req)
 	if err != nil {
-		return sdkerrors.ResponseDeliverTx(err, uint64(res.GasUsed), uint64(res.GasWanted), app.trace)
+		res = sdkerrors.ResponseDeliverTx(err, uint64(res.GasUsed), uint64(res.GasWanted), app.trace)
+		return res
 	}
 
 	return res

baseapp/baseapp.go

@@ -125,6 +125,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{}
+
+	// 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

@@ -229,3 +229,14 @@ func (app *BaseApp) SetInterfaceRegistry(registry types.InterfaceRegistry) {
 	app.interfaceRegistry = registry
 	app.grpcQueryRouter.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)
+}

baseapp/streaming.go

@@ -0,0 +1,33 @@
+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
+}
+
+// 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
+}

client/keys/add_ledger_test.go

@@ -1,4 +1,5 @@
-//+build ledger test_ledger_mock
+//go:build ledger || test_ledger_mock
+// +build ledger test_ledger_mock
 
 package keys
 
@@ -194,7 +195,7 @@ func Test_runAddCmdLedgerDryRun(t *testing.T) {
 			} else {
 				_, err = kb.Key("testkey")
 				require.Error(t, err)
-				require.Equal(t, "testkey: key not found", err.Error())
+				require.Equal(t, "testkey.info: key not found", err.Error())
 			}
 		})
 	}

client/keys/add_test.go

@@ -8,7 +8,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-
 	"github.com/tendermint/tendermint/libs/cli"
 
 	"github.com/cosmos/cosmos-sdk/client"
@@ -226,7 +225,7 @@ func Test_runAddCmdDryRun(t *testing.T) {
 			} else {
 				_, err = kb.Key("testkey")
 				require.Error(t, err)
-				require.Equal(t, "testkey: key not found", err.Error())
+				require.Equal(t, "testkey.info: key not found", err.Error())
 			}
 		})
 	}

client/keys/delete.go

@@ -3,11 +3,11 @@ package keys
 import (
 	"bufio"
 
+	"github.com/spf13/cobra"
+
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/client/input"
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
-
-	"github.com/spf13/cobra"
 )
 
 const (

client/keys/delete_test.go

@@ -53,7 +53,7 @@ func Test_runDeleteCmd(t *testing.T) {
 
 	err = cmd.ExecuteContext(ctx)
 	require.Error(t, err)
-	require.EqualError(t, err, "blah: key not found")
+	require.EqualError(t, err, "blah.info: key not found")
 
 	// User confirmation missing
 	cmd.SetArgs([]string{

client/keys/rename.go

@@ -4,10 +4,11 @@ import (
 	"bufio"
 	"fmt"
 
+	"github.com/spf13/cobra"
+
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/client/input"
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
-	"github.com/spf13/cobra"
 )
 
 // RenameKeyCommand renames a key from the key store.

client/keys/rename_test.go

@@ -49,7 +49,7 @@ func Test_runRenameCmd(t *testing.T) {
 	cmd.SetArgs([]string{"blah", "blaah", fmt.Sprintf("--%s=%s", flags.FlagHome, kbHome)})
 	err = cmd.ExecuteContext(ctx)
 	require.Error(t, err)
-	require.EqualError(t, err, "blah: key not found")
+	require.EqualError(t, err, "blah.info: key not found")
 
 	// User confirmation missing
 	cmd.SetArgs([]string{

cosmovisor/upgrade_test.go

@@ -1,3 +1,4 @@
+//go:build linux
 // +build linux
 
 package cosmovisor_test
@@ -208,18 +209,18 @@ func (s *upgradeTestSuite) TestDownloadBinary() {
 			canDownload: false,
 		},
 		"get zipped directory": {
-			url:         "./testdata/repo/zip_directory/autod.zip",
+			url:         "./testdata/repo/chain3-zip_dir/autod.zip",
 			canDownload: true,
 			validBinary: true,
 		},
 		"get zipped directory with valid checksum": {
-			// sha256sum ./testdata/repo/zip_directory/autod.zip
-			url:         "./testdata/repo/zip_directory/autod.zip?checksum=sha256:3784e4574cad69b67e34d4ea4425eff140063a3870270a301d6bb24a098a27ae",
+			// sha256sum ./testdata/repo/chain3-zip_dir/autod.zip
+			url:         "./testdata/repo/chain3-zip_dir/autod.zip?checksum=sha256:8951f52a0aea8617de0ae459a20daf704c29d259c425e60d520e363df0f166b4",
 			canDownload: true,
 			validBinary: true,
 		},
 		"get zipped directory with invalid checksum": {
-			url:         "./testdata/repo/zip_directory/autod.zip?checksum=sha256:73e2bd6cbb99261733caf137015d5cc58e3f96248d8b01da68be8564989dd906",
+			url:         "./testdata/repo/chain3-zip_dir/autod.zip?checksum=sha256:73e2bd6cbb99261733caf137015d5cc58e3f96248d8b01da68be8564989dd906",
 			canDownload: false,
 		},
 		"invalid url": {
@@ -255,15 +256,15 @@ func (s *upgradeTestSuite) TestDownloadBinary() {
 		err = cosmovisor.DownloadBinary(cfg, info)
 		if !tc.canDownload {
 			s.Require().Error(err)
-			return
-		}
-		s.Require().NoError(err)
-
-		err = cosmovisor.EnsureBinary(cfg.UpgradeBin(upgrade))
-		if tc.validBinary {
-			s.Require().NoError(err)
 		} else {
-			s.Require().Error(err)
+			s.Require().NoError(err)
+
+			err = cosmovisor.EnsureBinary(cfg.UpgradeBin(upgrade))
+			if tc.validBinary {
+				s.Require().NoError(err)
+			} else {
+				s.Require().Error(err)
+			}
 		}
 	}
 }

crypto/keyring/keyring.go

@@ -11,7 +11,6 @@ import (
 	"strings"
 
 	"github.com/99designs/keyring"
-	"github.com/cosmos/go-bip39"
 	"github.com/pkg/errors"
 	"github.com/tendermint/crypto/bcrypt"
 	tmcrypto "github.com/tendermint/tendermint/crypto"
@@ -24,6 +23,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/crypto/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
+	"github.com/cosmos/go-bip39"
 )
 
 // Backend options for Keyring
@@ -440,6 +440,8 @@ func (ks keystore) Rename(oldName, newName string) error {
 	return nil
 }
 
+// Delete deletes a key in the keyring. `uid` represents the key name, without
+// the `.info` suffix.
 func (ks keystore) Delete(uid string) error {
 	k, err := ks.Key(uid)
 	if err != nil {
@@ -456,7 +458,7 @@ func (ks keystore) Delete(uid string) error {
 		return err
 	}
 
-	err = ks.db.Remove(uid)
+	err = ks.db.Remove(infoKey(uid))
 	if err != nil {
 		return err
 	}
@@ -770,7 +772,7 @@ func (ks keystore) writeRecord(k *Record) error {
 		return err
 	}
 
-	key := k.Name
+	key := infoKey(k.Name)
 
 	exists, err := ks.existsInDb(addr, key)
 	if err != nil {
@@ -877,6 +879,9 @@ func (ks keystore) MigrateAll() (bool, error) {
 
 // migrate converts keyring.Item from amino to proto serialization format.
 func (ks keystore) migrate(key string) (*Record, bool, error) {
+	if !(strings.HasSuffix(key, infoSuffix)) && !(strings.HasPrefix(key, sdk.Bech32PrefixAccAddr)) {
+		key = infoKey(key)
+	}
 	item, err := ks.db.Get(key)
 	if err != nil {
 		return nil, false, wrapKeyNotFound(err, key)

crypto/keyring/keyring_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	"github.com/99designs/keyring"
-	bip39 "github.com/cosmos/go-bip39"
+	"github.com/cosmos/go-bip39"
 	"github.com/stretchr/testify/require"
 
 	"github.com/cosmos/cosmos-sdk/codec"
@@ -432,7 +432,7 @@ func TestKeyringKeybaseExportImportPrivKey(t *testing.T) {
 
 	// try export non existing key
 	_, err = kb.ExportPrivKeyArmor("john3", "wrongpassword")
-	require.EqualError(t, err, "john3: key not found")
+	require.EqualError(t, err, "john3.info: key not found")
 }
 
 func TestInMemoryLanguage(t *testing.T) {

crypto/keyring/legacy_info.go

@@ -43,6 +43,8 @@ type legacyLocalInfo struct {
 	Algo         hd.PubKeyType      `json:"algo"`
 }
 
+func infoKey(name string) string { return fmt.Sprintf("%s.%s", name, infoSuffix) }
+
 // GetType implements Info interface
 func (i legacyLocalInfo) GetType() KeyType {
 	return TypeLocal

crypto/keyring/types.go

@@ -37,6 +37,7 @@ const (
 	// bits of entropy to draw when creating a mnemonic
 	defaultEntropySize = 256
 	addressSuffix      = "address"
+	infoSuffix         = "info"
 )
 
 // KeyType reflects a human-readable type for key listing.