LaconicNetwork/cosmos-sdk
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

refactor: PoC for unordered txs per new spec (#24010)

Browse Source
This commit is contained in:
Tyler
2025-03-24 21:49:52 +00:00
committed by GitHub
parent 6fc8d0dc64
commit 94068be123
17 changed files with 698 additions and 955 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/app.go
+1 -6
View File
@@ -25,7 +25,6 @@ import (
servertypes "github.com/cosmos/cosmos-sdk/server/types"
sdk "github.com/cosmos/cosmos-sdk/types"
"github.com/cosmos/cosmos-sdk/types/module"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
authtx "github.com/cosmos/cosmos-sdk/x/auth/tx"
)
@@ -41,8 +40,7 @@ import (
type App struct {
*baseapp.BaseApp
ModuleManager *module.Manager
UnorderedTxManager *unorderedtx.Manager
ModuleManager *module.Manager
configurator module.Configurator
config *runtimev1alpha1.Module
@@ -160,9 +158,6 @@ func (a *App) Load(loadLatest bool) error {
// PreBlocker application updates every pre block
func (a *App) PreBlocker(ctx sdk.Context, _ *abci.RequestFinalizeBlock) (*sdk.ResponsePreBlock, error) {
if a.UnorderedTxManager != nil {
a.UnorderedTxManager.OnNewBlock(ctx.BlockTime())
}
return a.ModuleManager.PreBlock(ctx)
}
simapp/ante.go
+2 -3
View File
@@ -7,7 +7,6 @@ import (
sdk "github.com/cosmos/cosmos-sdk/types"
"github.com/cosmos/cosmos-sdk/x/auth/ante"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
)
// HandlerOptions are the options required for constructing a default SDK AnteHandler.
@@ -48,8 +47,8 @@ func NewAnteHandler(options HandlerOptions) (sdk.AnteHandler, error) {
ante.NewIncrementSequenceDecorator(options.AccountKeeper),
}
if options.UnorderedTxManager != nil {
anteDecorators = append(anteDecorators, ante.NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, options.UnorderedTxManager, ante.DefaultSha256GasCost))
if options.UnorderedNonceManager != nil {
anteDecorators = append(anteDecorators, ante.NewUnorderedTxDecorator(options.UnorderedNonceManager, options.UnorderedTxOptions...))
}
return sdk.ChainAnteDecorators(anteDecorators...), nil
simapp/app.go
+7 -36
View File
@@ -8,7 +8,6 @@ import (
"io"
"maps"
"os"
"path/filepath"
abci "github.com/cometbft/cometbft/abci/types"
dbm "github.com/cosmos/cosmos-db"
@@ -62,7 +61,6 @@ import (
"github.com/cosmos/cosmos-sdk/version"
"github.com/cosmos/cosmos-sdk/x/auth"
"github.com/cosmos/cosmos-sdk/x/auth/ante"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
authcodec "github.com/cosmos/cosmos-sdk/x/auth/codec"
authkeeper "github.com/cosmos/cosmos-sdk/x/auth/keeper"
"github.com/cosmos/cosmos-sdk/x/auth/posthandler"
@@ -169,8 +167,6 @@ type SimApp struct {
ModuleManager *module.Manager
BasicModuleManager module.BasicManager
UnorderedTxManager *unorderedtx.Manager
// simulation manager
sm *module.SimulationManager
@@ -537,6 +533,7 @@ func NewSimApp(
// NOTE: upgrade module is required to be prioritized
app.ModuleManager.SetOrderPreBlockers(
upgradetypes.ModuleName,
authtypes.ModuleName,
)
// During begin block slashing happens after distr.BeginBlocker so that
// there is nothing left over in the validator fee pool, so as to keep the
@@ -619,25 +616,6 @@ func NewSimApp(
}
app.sm = module.NewSimulationManagerFromAppModules(app.ModuleManager.Modules, overrideModules)
// create, start, and load the unordered tx manager
utxDataDir := filepath.Join(cast.ToString(appOpts.Get(flags.FlagHome)), "data")
app.UnorderedTxManager = unorderedtx.NewManager(utxDataDir)
app.UnorderedTxManager.Start()
if err := app.UnorderedTxManager.OnInit(); err != nil {
panic(fmt.Errorf("failed to initialize unordered tx manager: %w", err))
}
// register custom snapshot extensions (if any)
if manager := app.SnapshotManager(); manager != nil {
err := manager.RegisterExtensions(
unorderedtx.NewSnapshotter(app.UnorderedTxManager),
)
if err != nil {
panic(fmt.Errorf("failed to register snapshot extension: %s", err))
}
}
app.sm.RegisterStoreDecoders()
// initialize stores
@@ -691,12 +669,12 @@ func (app *SimApp) setAnteHandler(txConfig client.TxConfig) {
anteHandler, err := NewAnteHandler(
HandlerOptions{
ante.HandlerOptions{
AccountKeeper: app.AccountKeeper,
BankKeeper: app.BankKeeper,
SignModeHandler: txConfig.SignModeHandler(),
FeegrantKeeper: app.FeeGrantKeeper,
SigGasConsumer: ante.DefaultSigVerificationGasConsumer,
UnorderedTxManager: app.UnorderedTxManager,
UnorderedNonceManager: app.AccountKeeper,
AccountKeeper: app.AccountKeeper,
BankKeeper: app.BankKeeper,
SignModeHandler: txConfig.SignModeHandler(),
FeegrantKeeper: app.FeeGrantKeeper,
SigGasConsumer: ante.DefaultSigVerificationGasConsumer,
},
&app.CircuitKeeper,
},
@@ -720,18 +698,11 @@ func (app *SimApp) setPostHandler() {
app.SetPostHandler(postHandler)
}
// Close implements the Application interface and closes all necessary application
// resources.
func (app *SimApp) Close() error {
return app.UnorderedTxManager.Close()
}
// Name returns the name of the App
func (app *SimApp) Name() string { return app.BaseApp.Name() }
// PreBlocker application updates every pre block
func (app *SimApp) PreBlocker(ctx sdk.Context, _ *abci.RequestFinalizeBlock) (*sdk.ResponsePreBlock, error) {
app.UnorderedTxManager.OnNewBlock(ctx.BlockTime())
return app.ModuleManager.PreBlock(ctx)
}
simapp/app_config.go
+1
View File
@@ -104,6 +104,7 @@ var (
// NOTE: upgrade module is required to be prioritized
PreBlockers: []string{
upgradetypes.ModuleName,
authtypes.ModuleName,
},
// During begin block slashing happens after distr.BeginBlocker so that
// there is nothing left over in the validator fee pool, so as to keep the
simapp/app_di.go
+6 -32
View File
@@ -3,12 +3,9 @@
package simapp
import (
"fmt"
"io"
"path/filepath"
dbm "github.com/cosmos/cosmos-db"
"github.com/spf13/cast"
clienthelpers "cosmossdk.io/client/v2/helpers"
"cosmossdk.io/depinject"
@@ -22,7 +19,6 @@ import (
"github.com/cosmos/cosmos-sdk/baseapp"
"github.com/cosmos/cosmos-sdk/client"
"github.com/cosmos/cosmos-sdk/client/flags"
"github.com/cosmos/cosmos-sdk/codec"
codectypes "github.com/cosmos/cosmos-sdk/codec/types"
"github.com/cosmos/cosmos-sdk/runtime"
@@ -34,7 +30,6 @@ import (
"github.com/cosmos/cosmos-sdk/types/module"
"github.com/cosmos/cosmos-sdk/x/auth"
"github.com/cosmos/cosmos-sdk/x/auth/ante"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
authkeeper "github.com/cosmos/cosmos-sdk/x/auth/keeper"
authsims "github.com/cosmos/cosmos-sdk/x/auth/simulation"
authtypes "github.com/cosmos/cosmos-sdk/x/auth/types"
@@ -68,8 +63,6 @@ type SimApp struct {
txConfig client.TxConfig
interfaceRegistry codectypes.InterfaceRegistry
UnorderedTxManager *unorderedtx.Manager
// essential keepers
AccountKeeper authkeeper.AccountKeeper
BankKeeper bankkeeper.BaseKeeper
@@ -263,25 +256,6 @@ func NewSimApp(
// return app.App.InitChainer(ctx, req)
// })
// create, start, and load the unordered tx manager
utxDataDir := filepath.Join(cast.ToString(appOpts.Get(flags.FlagHome)), "data")
app.UnorderedTxManager = unorderedtx.NewManager(utxDataDir)
app.UnorderedTxManager.Start()
if err := app.UnorderedTxManager.OnInit(); err != nil {
panic(fmt.Errorf("failed to initialize unordered tx manager: %w", err))
}
// register custom snapshot extensions (if any)
if manager := app.SnapshotManager(); manager != nil {
err := manager.RegisterExtensions(
unorderedtx.NewSnapshotter(app.UnorderedTxManager),
)
if err != nil {
panic(fmt.Errorf("failed to register snapshot extension: %w", err))
}
}
// set custom ante handler
app.setAnteHandler(app.txConfig)
@@ -298,12 +272,12 @@ func (app *SimApp) setAnteHandler(txConfig client.TxConfig) {
anteHandler, err := NewAnteHandler(
HandlerOptions{
ante.HandlerOptions{
AccountKeeper: app.AccountKeeper,
BankKeeper: app.BankKeeper,
SignModeHandler: txConfig.SignModeHandler(),
FeegrantKeeper: app.FeeGrantKeeper,
SigGasConsumer: ante.DefaultSigVerificationGasConsumer,
UnorderedTxManager: app.UnorderedTxManager,
UnorderedNonceManager: app.AccountKeeper,
AccountKeeper: app.AccountKeeper,
BankKeeper: app.BankKeeper,
SignModeHandler: txConfig.SignModeHandler(),
FeegrantKeeper: app.FeeGrantKeeper,
SigGasConsumer: ante.DefaultSigVerificationGasConsumer,
},
&app.CircuitKeeper,
},
x/auth/ante/ante.go
+6 -4
View File
@@ -8,7 +8,6 @@ import (
sdk "github.com/cosmos/cosmos-sdk/types"
sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
"github.com/cosmos/cosmos-sdk/types/tx/signing"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
"github.com/cosmos/cosmos-sdk/x/auth/types"
)
@@ -21,7 +20,10 @@ type HandlerOptions struct {
SignModeHandler *txsigning.HandlerMap
SigGasConsumer func(meter storetypes.GasMeter, sig signing.SignatureV2, params types.Params) error
TxFeeChecker TxFeeChecker
UnorderedTxManager *unorderedtx.Manager
// UnorderedNonceManager is an opt-in feature for x/auth.
// When set, this application will be able to receive and process unordered transactions.
UnorderedNonceManager UnorderedNonceManager
UnorderedTxOptions []UnorderedTxDecoratorOptions
}
// NewAnteHandler returns an AnteHandler that checks and increments sequence
@@ -55,8 +57,8 @@ func NewAnteHandler(options HandlerOptions) (sdk.AnteHandler, error) {
NewIncrementSequenceDecorator(options.AccountKeeper),
}
if options.UnorderedTxManager != nil {
anteDecorators = append(anteDecorators, NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, options.UnorderedTxManager, DefaultSha256GasCost))
if options.UnorderedNonceManager != nil {
anteDecorators = append(anteDecorators, NewUnorderedTxDecorator(options.UnorderedNonceManager, options.UnorderedTxOptions...))
}
return sdk.ChainAnteDecorators(anteDecorators...), nil
x/auth/ante/expected_keepers.go
+7
View File
@@ -2,6 +2,7 @@ package ante
import (
"context"
"time"
"cosmossdk.io/core/address"
@@ -19,6 +20,12 @@ type AccountKeeper interface {
AddressCodec() address.Codec
}
// UnorderedNonceManager defines the contract needed for UnorderedNonce management.
type UnorderedNonceManager interface {
RemoveExpiredUnorderedNonces(ctx sdk.Context) error
TryAddUnorderedNonce(ctx sdk.Context, sender []byte, timestamp time.Time) error
}
// FeegrantKeeper defines the expected feegrant keeper.
type FeegrantKeeper interface {
UseGrantedFees(ctx context.Context, granter, grantee sdk.AccAddress, fee sdk.Coins, msgs []sdk.Msg) error
x/auth/ante/unordered.go
+60 -147
View File
@@ -1,65 +1,74 @@
package ante
import (
"bytes"
"crypto/sha256"
"encoding/binary"
"fmt"
"sync"
"time"
"github.com/cosmos/gogoproto/proto"
errorsmod "cosmossdk.io/errors"
sdk "github.com/cosmos/cosmos-sdk/types"
sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
"github.com/cosmos/cosmos-sdk/types/tx/signing"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
authsigning "github.com/cosmos/cosmos-sdk/x/auth/signing"
)
// bufPool is a pool of bytes.Buffer objects to reduce memory allocations.
var bufPool = sync.Pool{
New: func() interface{} {
return new(bytes.Buffer)
},
}
// DefaultSha256GasCost is the suggested default gas cost for Sha256 operations in unordered transaction handling.
const DefaultSha256GasCost = 25
const (
// DefaultMaxTimoutDuration defines a default maximum TTL a transaction can define.
DefaultMaxTimoutDuration = 10 * time.Minute
// DefaultUnorderedTxGasCost defines a default gas cost for unordered transactions.
// We must charge extra gas for unordered transactions
// as they incur extra processing time for cleaning up the expired txs in x/auth PreBlocker.
// Note: this value was chosen by 2x-ing the cost of fetching and removing an unordered nonce entry.
DefaultUnorderedTxGasCost = uint64(2240)
)
var _ sdk.AnteDecorator = (*UnorderedTxDecorator)(nil)
type UnorderedTxDecoratorOptions func(*UnorderedTxDecorator)
// WithTimeoutDuration allows for changing the timeout duration for unordered txs.
func WithTimeoutDuration(duration time.Duration) UnorderedTxDecoratorOptions {
return func(utx *UnorderedTxDecorator) {
utx.maxTxTimeoutDuration = duration
}
}
func WithUnorderedTxGasCost(cost uint64) UnorderedTxDecoratorOptions {
return func(utx *UnorderedTxDecorator) {
utx.txGasCost = cost
}
}
// UnorderedTxDecorator defines an AnteHandler decorator that is responsible for
// checking if a transaction is intended to be unordered and if so, evaluates
// checking if a transaction is intended to be unordered and, if so, evaluates
// the transaction accordingly. An unordered transaction will bypass having its
// nonce incremented, which allows fire-and-forget along with possible parallel
// transaction processing, without having to deal with nonces.
// nonce incremented, which allows fire-and-forget transaction broadcasting,
// removing the necessity of ordering on the sender-side.
//
// The transaction sender must ensure that unordered=true and a timeout_height
// is appropriately set. The AnteHandler will check that the transaction is not
// a duplicate and will evict it from memory when the timeout is reached.
// a duplicate and will evict it from state when the timeout is reached.
//
// The UnorderedTxDecorator should be placed as early as possible in the AnteHandler
// chain to ensure that during DeliverTx, the transaction is added to the UnorderedTxManager.
// chain to ensure that during DeliverTx, the transaction is added to the UnorderedNonceManager.
type UnorderedTxDecorator struct {
// maxUnOrderedTTL defines the maximum TTL a transaction can define.
maxTimeoutDuration time.Duration
txManager *unorderedtx.Manager
sha256GasCost uint64
maxTxTimeoutDuration time.Duration
txGasCost uint64
txManager UnorderedNonceManager
}
func NewUnorderedTxDecorator(
maxDuration time.Duration,
m *unorderedtx.Manager,
sha256GasCost uint64,
utxm UnorderedNonceManager,
opts ...UnorderedTxDecoratorOptions,
) *UnorderedTxDecorator {
return &UnorderedTxDecorator{
maxTimeoutDuration: maxDuration,
txManager: m,
sha256GasCost: sha256GasCost,
utx := &UnorderedTxDecorator{
maxTxTimeoutDuration: DefaultMaxTimoutDuration,
txGasCost: DefaultUnorderedTxGasCost,
txManager: utxm,
}
for _, opt := range opts {
opt(utx)
}
return utx
}
func (d *UnorderedTxDecorator) AnteHandle(
@@ -96,138 +105,42 @@ func (d *UnorderedTxDecorator) ValidateTx(ctx sdk.Context, tx sdk.Tx) error {
"unordered transaction has a timeout_timestamp that has already passed",
)
}
if timeoutTimestamp.After(blockTime.Add(d.maxTimeoutDuration)) {
if timeoutTimestamp.After(blockTime.Add(d.maxTxTimeoutDuration)) {
return errorsmod.Wrapf(
sdkerrors.ErrInvalidRequest,
"unordered tx ttl exceeds %s",
d.maxTimeoutDuration.String(),
d.maxTxTimeoutDuration.String(),
)
}
// consume gas in all exec modes to avoid gas estimation discrepancies
ctx.GasMeter().ConsumeGas(d.sha256GasCost, "consume gas for calculating tx hash")
ctx.GasMeter().ConsumeGas(d.txGasCost, "unordered tx")
// Avoid checking for duplicates and creating the identifier in simulation mode
// This is done to avoid sha256 computation in simulation mode
if ctx.ExecMode() == sdk.ExecModeSimulate {
execMode := ctx.ExecMode()
if execMode == sdk.ExecModeSimulate {
return nil
}
// calculate the tx hash
txHash, err := TxHashFromTimeout(uint64(timeoutTimestamp.Unix()), tx)
signerAddrs, err := extractSignersBytes(tx)
if err != nil {
return err
}
// check for duplicates
if d.txManager.Contains(txHash) {
return errorsmod.Wrap(
sdkerrors.ErrInvalidRequest,
"tx %X is duplicated",
)
}
if ctx.ExecMode() == sdk.ExecModeFinalize {
// a new tx included in the block, add the hash to the unordered tx manager
d.txManager.Add(txHash, timeoutTimestamp)
for _, signerAddr := range signerAddrs {
if err := d.txManager.TryAddUnorderedNonce(ctx, signerAddr, unorderedTx.GetTimeoutTimeStamp()); err != nil {
return errorsmod.Wrapf(
sdkerrors.ErrInvalidRequest,
"failed to add unordered nonce: %s", err,
)
}
}
return nil
}
// TxHashFromTimeout returns a TxHash for an unordered transaction.
func TxHashFromTimeout(timeout uint64, tx sdk.Tx) (unorderedtx.TxHash, error) {
sigTx, ok := tx.(authsigning.Tx)
func extractSignersBytes(tx sdk.Tx) ([][]byte, error) {
sigTx, ok := tx.(authsigning.SigVerifiableTx)
if !ok {
return unorderedtx.TxHash{}, errorsmod.Wrap(sdkerrors.ErrTxDecode, "invalid transaction type")
return nil, errorsmod.Wrap(sdkerrors.ErrTxDecode, "invalid tx type")
}
if sigTx.GetFee().IsZero() {
return unorderedtx.TxHash{}, errorsmod.Wrap(
sdkerrors.ErrInvalidRequest,
"unordered transaction must have a fee",
)
}
buf := bufPool.Get().(*bytes.Buffer)
// Make sure to reset the buffer
buf.Reset()
defer bufPool.Put(buf)
// Add signatures to the transaction identifier
signatures, err := sigTx.GetSignaturesV2()
if err != nil {
return unorderedtx.TxHash{}, err
}
for _, sig := range signatures {
if err := addSignatures(sig.Data, buf); err != nil {
return unorderedtx.TxHash{}, err
}
}
// Use the buffer
for _, msg := range tx.GetMsgs() {
// loop through the messages and write them to the buffer
// encoding the msg to bytes makes it deterministic within the state machine.
// Malleability is not a concern here because the state machine will encode the transaction deterministically.
bz, err := proto.Marshal(msg)
if err != nil {
return unorderedtx.TxHash{}, errorsmod.Wrap(
sdkerrors.ErrInvalidRequest,
"failed to marshal message",
)
}
if _, err := buf.Write(bz); err != nil {
return unorderedtx.TxHash{}, errorsmod.Wrap(
sdkerrors.ErrInvalidRequest,
"failed to write message to buffer",
)
}
}
// write the timeout height to the buffer
if err := binary.Write(buf, binary.LittleEndian, timeout); err != nil {
return unorderedtx.TxHash{}, errorsmod.Wrap(
sdkerrors.ErrInvalidRequest,
"failed to write timeout_height to buffer",
)
}
// write gas to the buffer
if err := binary.Write(buf, binary.LittleEndian, sigTx.GetGas()); err != nil {
return unorderedtx.TxHash{}, errorsmod.Wrap(
sdkerrors.ErrInvalidRequest,
"failed to write unordered to buffer",
)
}
txHash := sha256.Sum256(buf.Bytes())
// Return the Buffer to the pool
return txHash, nil
}
func addSignatures(sig signing.SignatureData, buf *bytes.Buffer) error {
switch data := sig.(type) {
case *signing.SingleSignatureData:
if _, err := buf.Write(data.Signature); err != nil {
return errorsmod.Wrap(
sdkerrors.ErrInvalidRequest,
"failed to write single signature to buffer",
)
}
return nil
case *signing.MultiSignatureData:
for _, sigdata := range data.Signatures {
if err := addSignatures(sigdata, buf); err != nil {
return err
}
}
default:
return fmt.Errorf("unexpected SignatureData %T", data)
}
return nil
return sigTx.GetSigners()
}
x/auth/ante/unordered_test.go
+206 -121
View File
@@ -9,145 +9,202 @@ import (
storetypes "cosmossdk.io/store/types"
cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types"
"github.com/cosmos/cosmos-sdk/runtime"
"github.com/cosmos/cosmos-sdk/testutil"
"github.com/cosmos/cosmos-sdk/testutil/testdata"
sdk "github.com/cosmos/cosmos-sdk/types"
moduletestutil "github.com/cosmos/cosmos-sdk/types/module/testutil"
"github.com/cosmos/cosmos-sdk/types/tx/signing"
"github.com/cosmos/cosmos-sdk/x/auth/ante"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
authcodec "github.com/cosmos/cosmos-sdk/x/auth/codec"
"github.com/cosmos/cosmos-sdk/x/auth/keeper"
"github.com/cosmos/cosmos-sdk/x/auth/types"
)
const gasConsumed = uint64(25)
func TestUnorderedTxDecorator_OrderedTx(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
SetupTestSuite(t, false)
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, txm, ante.DefaultSha256GasCost))
tx, txBz := genUnorderedTx(t, false, time.Time{})
ctx := sdk.Context{}.WithTxBytes(txBz)
_, err := chain(ctx, tx, false)
require.NoError(t, err)
func TestUnorderedAnte(t *testing.T) {
testPK, _, testAddr := testdata.KeyTestPubAddr()
testCases := map[string]struct {
addTxs func() []sdk.Tx
runTx func() sdk.Tx
blockTime time.Time
execMode sdk.ExecMode
expectedErr string
}{
"normal/ordered tx should just skip": {
runTx: func() sdk.Tx {
return genTestTx(t, genTxOptions{})
},
blockTime: time.Unix(0, 0),
execMode: sdk.ExecModeFinalize,
},
"happy case - simple pass": {
runTx: func() sdk.Tx {
return genTestTx(t, genTxOptions{unordered: true, timestamp: time.Unix(10, 0)})
},
blockTime: time.Unix(0, 0),
execMode: sdk.ExecModeFinalize,
},
"zero time should fail": {
runTx: func() sdk.Tx {
return genTestTx(t, genTxOptions{unordered: true})
},
blockTime: time.Unix(0, 0),
execMode: sdk.ExecModeFinalize,
expectedErr: "unordered transaction must have timeout_timestamp set",
},
"timeout before current block time should fail": {
runTx: func() sdk.Tx {
return genTestTx(t, genTxOptions{unordered: true, timestamp: time.Unix(7, 0)})
},
blockTime: time.Unix(10, 1),
execMode: sdk.ExecModeFinalize,
expectedErr: "unordered transaction has a timeout_timestamp that has already passed",
},
"timeout equal to current block time should pass": {
runTx: func() sdk.Tx {
return genTestTx(t, genTxOptions{unordered: true, timestamp: time.Unix(10, 0)})
},
blockTime: time.Unix(10, 0),
execMode: sdk.ExecModeFinalize,
},
"timeout after the max duration should fail": {
runTx: func() sdk.Tx {
return genTestTx(t, genTxOptions{unordered: true, timestamp: time.Unix(10, 1).Add(ante.DefaultMaxTimoutDuration)})
},
blockTime: time.Unix(10, 0),
execMode: sdk.ExecModeFinalize,
expectedErr: "unordered tx ttl exceeds",
},
"fails if manager has duplicate": {
addTxs: func() []sdk.Tx {
tx := genTestTx(
t,
genTxOptions{unordered: true, timestamp: time.Unix(10, 0), pk: testPK, addr: testAddr},
)
return []sdk.Tx{tx}
},
runTx: func() sdk.Tx {
return genTestTx(
t,
genTxOptions{unordered: true, timestamp: time.Unix(10, 0), pk: testPK, addr: testAddr},
)
},
blockTime: time.Unix(5, 0),
execMode: sdk.ExecModeFinalize,
expectedErr: "already used timeout",
},
"duplicate doesn't matter if we're in simulate mode": {
addTxs: func() []sdk.Tx {
tx := genTestTx(
t,
genTxOptions{unordered: true, timestamp: time.Unix(10, 0), pk: testPK, addr: testAddr},
)
return []sdk.Tx{tx}
},
runTx: func() sdk.Tx {
return genTestTx(
t,
genTxOptions{unordered: true, timestamp: time.Unix(10, 0), pk: testPK, addr: testAddr},
)
},
blockTime: time.Unix(5, 0),
execMode: sdk.ExecModeSimulate,
},
}
for name, tc := range testCases {
t.Run(name, func(t *testing.T) {
mockStoreKey := storetypes.NewKVStoreKey("test")
storeService := runtime.NewKVStoreService(mockStoreKey)
ctx := testutil.DefaultContextWithDB(
t,
mockStoreKey,
storetypes.NewTransientStoreKey("transient_test"),
).Ctx.WithBlockTime(tc.blockTime).WithExecMode(tc.execMode)
mgr := keeper.NewAccountKeeper(
moduletestutil.MakeTestEncodingConfig().Codec,
storeService,
types.ProtoBaseAccount,
nil,
authcodec.NewBech32Codec("cosmos"),
"cosmos",
types.NewModuleAddress("gov").String(),
)
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(mgr))
var err error
if tc.addTxs != nil {
txs := tc.addTxs()
for _, tx := range txs {
ctx, err = chain(ctx, tx, false)
require.NoError(t, err)
}
}
_, err = chain(ctx, tc.runTx(), false)
if tc.expectedErr != "" {
require.ErrorContains(t, err, tc.expectedErr)
} else {
require.NoError(t, err)
}
})
}
}
func TestUnorderedTxDecorator_UnorderedTx_NoTTL(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
func TestMultiSignerUnorderedTx(t *testing.T) {
pk1, _, addr1 := testdata.KeyTestPubAddr()
pk2, _, addr2 := testdata.KeyTestPubAddr()
pk3, _, addr3 := testdata.KeyTestPubAddr()
txm.Start()
signerAddrs := []sdk.AccAddress{addr1, addr2, addr3}
SetupTestSuite(t, false)
mockStoreKey := storetypes.NewKVStoreKey("test")
storeService := runtime.NewKVStoreService(mockStoreKey)
ctx := testutil.DefaultContextWithDB(
t,
mockStoreKey,
storetypes.NewTransientStoreKey("transient_test"),
).Ctx.WithBlockTime(time.Unix(9, 0))
mgr := keeper.NewAccountKeeper(
moduletestutil.MakeTestEncodingConfig().Codec,
storeService,
types.ProtoBaseAccount,
nil,
authcodec.NewBech32Codec("cosmos"),
"cosmos",
types.NewModuleAddress("gov").String(),
)
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(mgr))
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, txm, ante.DefaultSha256GasCost))
timeout := time.Unix(10, 0)
tx := genMultiSignedUnorderedTx(t, signerAddrs, timeout, []cryptotypes.PrivKey{pk1, pk2, pk3})
tx, txBz := genUnorderedTx(t, true, time.Time{})
ctx := sdk.Context{}.WithTxBytes(txBz)
_, err := chain(ctx, tx, false)
require.Error(t, err)
}
func TestUnorderedTxDecorator_UnorderedTx_InvalidTTL(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
SetupTestSuite(t, false)
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, txm, ante.DefaultSha256GasCost))
tx, txBz := genUnorderedTx(t, true, time.Now().Add(unorderedtx.DefaultMaxTimeoutDuration+time.Second))
ctx := sdk.Context{}.WithTxBytes(txBz).WithBlockTime(time.Now())
_, err := chain(ctx, tx, false)
require.Error(t, err)
}
func TestUnorderedTxDecorator_UnorderedTx_AlreadyExists(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
SetupTestSuite(t, false)
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, txm, ante.DefaultSha256GasCost))
tx, txBz := genUnorderedTx(t, true, time.Now().Add(time.Minute))
ctx := sdk.Context{}.WithTxBytes(txBz).WithBlockTime(time.Now()).WithGasMeter(storetypes.NewGasMeter(gasConsumed))
bz := [32]byte{}
copy(bz[:], txBz[:32])
txm.Add(bz, time.Now().Add(time.Minute))
_, err := chain(ctx, tx, false)
require.Error(t, err)
}
func TestUnorderedTxDecorator_UnorderedTx_ValidCheckTx(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
SetupTestSuite(t, false)
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, txm, ante.DefaultSha256GasCost))
tx, txBz := genUnorderedTx(t, true, time.Now().Add(time.Minute))
ctx := sdk.Context{}.WithTxBytes(txBz).WithBlockTime(time.Now()).WithExecMode(sdk.ExecModeCheck).WithGasMeter(storetypes.NewGasMeter(gasConsumed))
_, err := chain(ctx, tx, false)
require.NoError(t, err)
}
func TestUnorderedTxDecorator_UnorderedTx_ValidDeliverTx(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
SetupTestSuite(t, false)
chain := sdk.ChainAnteDecorators(ante.NewUnorderedTxDecorator(unorderedtx.DefaultMaxTimeoutDuration, txm, ante.DefaultSha256GasCost))
tx, txBz := genUnorderedTx(t, true, time.Now().Add(time.Minute))
ctx := sdk.Context{}.WithTxBytes(txBz).WithBlockTime(time.Now()).WithExecMode(sdk.ExecModeFinalize).WithGasMeter(storetypes.NewGasMeter(gasConsumed))
_, err := chain(ctx, tx, false)
newCtx, err := chain(ctx, tx, false)
require.NoError(t, err)
bz := [32]byte{}
copy(bz[:], txBz[:32])
require.True(t, txm.Contains(bz))
for _, addr := range signerAddrs {
ok, err := mgr.ContainsUnorderedNonce(newCtx, addr.Bytes(), timeout)
require.NoError(t, err)
require.True(t, ok)
}
}
func genUnorderedTx(t *testing.T, unordered bool, timestamp time.Time) (sdk.Tx, []byte) {
type genTxOptions struct {
unordered bool
timestamp time.Time
pk cryptotypes.PrivKey
addr sdk.AccAddress
}
func genTestTx(t *testing.T, options genTxOptions) sdk.Tx {
t.Helper()
s := SetupTestSuite(t, true)
s.txBuilder = s.clientCtx.TxConfig.NewTxBuilder()
// keys and addresses
priv1, _, addr1 := testdata.KeyTestPubAddr()
priv1 := options.pk
addr1 := options.addr
if options.pk == nil || options.addr == nil {
priv1, _, addr1 = testdata.KeyTestPubAddr()
}
// msg and signatures
msg := testdata.NewTestMsg(addr1)
@@ -157,16 +214,44 @@ func genUnorderedTx(t *testing.T, unordered bool, timestamp time.Time) (sdk.Tx,
s.txBuilder.SetFeeAmount(feeAmount)
s.txBuilder.SetGasLimit(gasLimit)
s.txBuilder.SetUnordered(unordered)
s.txBuilder.SetTimeoutTimestamp(timestamp)
s.txBuilder.SetUnordered(options.unordered)
s.txBuilder.SetTimeoutTimestamp(options.timestamp)
privKeys, accNums, accSeqs := []cryptotypes.PrivKey{priv1}, []uint64{0}, []uint64{0}
tx, err := s.CreateTestTx(s.ctx, privKeys, accNums, accSeqs, s.ctx.ChainID(), signing.SignMode_SIGN_MODE_DIRECT)
require.NoError(t, err)
txBz, err := ante.TxHashFromTimeout(uint64(timestamp.Unix()), tx)
require.NoError(t, err)
return tx
}
func genMultiSignedUnorderedTx(t *testing.T, addrs []sdk.AccAddress, ts time.Time, pks []cryptotypes.PrivKey) sdk.Tx {
t.Helper()
s := SetupTestSuite(t, true)
s.txBuilder = s.clientCtx.TxConfig.NewTxBuilder()
// msg and signatures
msgs := make([]sdk.Msg, 0, len(addrs))
for _, addr := range addrs {
msgs = append(msgs, testdata.NewTestMsg(addr))
}
feeAmount := testdata.NewTestFeeAmount()
gasLimit := testdata.NewTestGasLimit()
require.NoError(t, s.txBuilder.SetMsgs(msgs...))
s.txBuilder.SetFeeAmount(feeAmount)
s.txBuilder.SetGasLimit(gasLimit)
s.txBuilder.SetUnordered(true)
s.txBuilder.SetTimeoutTimestamp(ts)
accNums := make([]uint64, len(pks))
accSeqs := make([]uint64, len(pks))
tx, err := s.CreateTestTx(s.ctx, pks, accNums, accSeqs, s.ctx.ChainID(), signing.SignMode_SIGN_MODE_DIRECT)
require.NoError(t, err)
require.NoError(t, err)
return tx, txBz[:]
return tx
}
x/auth/ante/unorderedtx/manager.go
-286
View File
@@ -1,286 +0,0 @@
package unorderedtx
import (
"bufio"
"bytes"
"context"
"encoding/binary"
"errors"
"fmt"
"io"
"maps"
"os"
"path/filepath"
"slices"
"sync"
"time"
)
const (
// DefaultMaxTimeoutDuration defines the default maximum duration an un-ordered transaction
// can set.
DefaultMaxTimeoutDuration = time.Minute * 40
dirName = "unordered_txs"
fileName = "data"
)
// TxHash defines a transaction hash type alias, which is a fixed array of 32 bytes.
type TxHash [32]byte
// Manager contains the tx hash dictionary for duplicates checking, and expire
// them when block production progresses.
type Manager struct {
// blockTimeCh defines a channel to receive newly committed block time
blockTimeCh chan time.Time
// doneCh allows us to ensure the purgeLoop has gracefully terminated prior to closing
doneCh chan struct{}
// dataDir defines the directory to store unexpired unordered transactions
//
// XXX: Note, ideally we avoid the need to store unexpired unordered transactions
// directly to file. However, store v1 does not allow such a primitive. But,
// once store v2 is fully integrated, we can remove manual file handling and
// store the unexpired unordered transactions directly to SS.
//
// Ref: https://github.com/cosmos/cosmos-sdk/issues/18467
dataDir string
mu sync.RWMutex
// txHashes defines a map from tx hash -> TTL value defined as block time, which is used for duplicate
// checking and replay protection, as well as purging the map when the TTL is
// expired.
txHashes map[TxHash]time.Time
}
func NewManager(dataDir string) *Manager {
path := filepath.Join(dataDir, dirName)
if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
if err = os.MkdirAll(path, os.ModePerm); err != nil {
panic(fmt.Errorf("failed to create unordered txs directory: %w", err))
}
}
m := &Manager{
dataDir: dataDir,
blockTimeCh: make(chan time.Time, 16),
doneCh: make(chan struct{}),
txHashes: make(map[TxHash]time.Time),
}
return m
}
func (m *Manager) Start() {
go m.purgeLoop()
}
// Close must be called when a node gracefully shuts down. Typically, this should
// be called in an application's Close() function, which is called by the server.
// Note, Start() must be called in order for Close() to not hang.
//
// It will free all necessary resources as well as writing all unexpired unordered
// transactions along with their TTL values to file.
func (m *Manager) Close() error {
close(m.blockTimeCh)
<-m.doneCh
m.blockTimeCh = nil
return m.flushToFile()
}
func (m *Manager) Contains(hash TxHash) bool {
m.mu.RLock()
defer m.mu.RUnlock()
_, ok := m.txHashes[hash]
return ok
}
func (m *Manager) Size() int {
m.mu.RLock()
defer m.mu.RUnlock()
return len(m.txHashes)
}
func (m *Manager) Add(txHash TxHash, timestamp time.Time) {
m.mu.Lock()
defer m.mu.Unlock()
m.txHashes[txHash] = timestamp
}
// OnInit must be called when a node starts up. Typically, this should be called
// in an application's constructor, which is called by the server.
func (m *Manager) OnInit() error {
f, err := os.Open(filepath.Join(m.dataDir, dirName, fileName))
if err != nil {
if errors.Is(err, os.ErrNotExist) {
// File does not exist, which we can assume that there are no unexpired
// unordered transactions.
return nil
}
return fmt.Errorf("failed to open unconfirmed txs file: %w", err)
}
defer f.Close()
var (
r = bufio.NewReader(f)
buf = make([]byte, chunkSize)
)
for {
n, err := io.ReadFull(r, buf)
if err != nil {
if errors.Is(err, io.EOF) {
break
} else {
return fmt.Errorf("failed to read unconfirmed txs file: %w", err)
}
}
if n != 32+8 {
return fmt.Errorf("read unexpected number of bytes from unconfirmed txs file: %d", n)
}
var txHash TxHash
copy(txHash[:], buf[:txHashSize])
timeStamp := binary.BigEndian.Uint64(buf[txHashSize:])
m.Add(txHash, time.Unix(int64(timeStamp), 0))
}
return nil
}
// OnNewBlock sends the latest block time to the background purge loop, which
// should be called in ABCI Commit event.
func (m *Manager) OnNewBlock(blockTime time.Time) {
m.blockTimeCh <- blockTime
}
func (m *Manager) exportSnapshot(_ uint64, snapshotWriter func([]byte) error) error {
var buf bytes.Buffer
w := bufio.NewWriter(&buf)
keys := slices.SortedFunc(maps.Keys(m.txHashes), func(i, j TxHash) int { return bytes.Compare(i[:], j[:]) })
for _, txHash := range keys {
timeoutTime := m.txHashes[txHash]
// right now we dont have access block time at this flow, so we would just include the expired txs
// and let it be purge during purge loop
chunk := unorderedTxToBytes(txHash, uint64(timeoutTime.Unix()))
if _, err := w.Write(chunk); err != nil {
return fmt.Errorf("failed to write unordered tx to buffer: %w", err)
}
}
if err := w.Flush(); err != nil {
return fmt.Errorf("failed to flush unordered txs buffer: %w", err)
}
return snapshotWriter(buf.Bytes())
}
// flushToFile writes all unordered transactions (including expired if not pruned yet)
// along with their TTL to file, overwriting the existing file if it exists.
func (m *Manager) flushToFile() error {
f, err := os.Create(filepath.Join(m.dataDir, dirName, fileName))
if err != nil {
return fmt.Errorf("failed to create unordered txs file: %w", err)
}
defer f.Close()
w := bufio.NewWriter(f)
for txHash, timestamp := range m.txHashes {
chunk := unorderedTxToBytes(txHash, uint64(timestamp.Unix()))
if _, err = w.Write(chunk); err != nil {
return fmt.Errorf("failed to write unordered tx to buffer: %w", err)
}
}
if err = w.Flush(); err != nil {
return fmt.Errorf("failed to flush unordered txs buffer: %w", err)
}
return nil
}
// expiredTxs returns expired tx hashes based on the provided block time.
func (m *Manager) expiredTxs(blockTime time.Time) []TxHash {
m.mu.RLock()
defer m.mu.RUnlock()
var result []TxHash
for txHash, timestamp := range m.txHashes {
if blockTime.After(timestamp) {
result = append(result, txHash)
}
}
return result
}
func (m *Manager) purge(txHashes []TxHash) {
m.mu.Lock()
defer m.mu.Unlock()
for _, txHash := range txHashes {
delete(m.txHashes, txHash)
}
}
// purgeLoop removes expired tx hashes in the background
func (m *Manager) purgeLoop() {
for {
latestTime, ok := m.batchReceive()
if !ok {
// channel closed
m.doneCh <- struct{}{}
return
}
hashes := m.expiredTxs(latestTime)
if len(hashes) > 0 {
m.purge(hashes)
}
}
}
// batchReceive receives block time from the channel until the context is done
// or the channel is closed.
func (m *Manager) batchReceive() (time.Time, bool) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var latestTime time.Time
for {
select {
case <-ctx.Done():
return latestTime, true
case blockTime, ok := <-m.blockTimeCh:
if !ok {
// channel is closed
return time.Time{}, false
}
if blockTime.After(latestTime) {
latestTime = blockTime
}
}
}
}
func unorderedTxToBytes(txHash TxHash, ttl uint64) []byte {
chunk := make([]byte, chunkSize)
copy(chunk[:txHashSize], txHash[:])
ttlBz := make([]byte, timeoutSize)
binary.BigEndian.PutUint64(ttlBz, ttl)
copy(chunk[txHashSize:], ttlBz)
return chunk
}
x/auth/ante/unorderedtx/manager_test.go
-145
View File
@@ -1,145 +0,0 @@
package unorderedtx_test
import (
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
)
func TestUnorderedTxManager_Close(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
txm.Start()
require.NoError(t, txm.Close())
require.Panics(t, func() { txm.Close() })
}
func TestUnorderedTxManager_SimpleSize(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
txm.Add(unorderedtx.TxHash{0xFF}, time.Now())
txm.Add(unorderedtx.TxHash{0xAA}, time.Now())
txm.Add(unorderedtx.TxHash{0xCC}, time.Now())
require.Equal(t, 3, txm.Size())
}
func TestUnorderedTxManager_SimpleContains(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
for i := 0; i < 10; i++ {
txHash := unorderedtx.TxHash{byte(i)}
txm.Add(txHash, time.Now())
require.True(t, txm.Contains(txHash))
}
for i := 10; i < 20; i++ {
txHash := unorderedtx.TxHash{byte(i)}
require.False(t, txm.Contains(txHash))
}
}
func TestUnorderedTxManager_InitEmpty(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
require.NoError(t, txm.OnInit())
}
func TestUnorderedTxManager_CloseInit(t *testing.T) {
dataDir := t.TempDir()
txm := unorderedtx.NewManager(dataDir)
txm.Start()
// add a handful of unordered txs
for i := 0; i < 100; i++ {
txm.Add(unorderedtx.TxHash{byte(i)}, time.Now())
}
// close the manager, which should flush all unexpired txs to file
require.NoError(t, txm.Close())
// create a new manager, start it
txm2 := unorderedtx.NewManager(dataDir)
defer func() {
require.NoError(t, txm2.Close())
}()
// start and execute OnInit, which should load the unexpired txs from file
txm2.Start()
require.NoError(t, txm2.OnInit())
require.Equal(t, 100, txm2.Size())
for i := 0; i < 100; i++ {
require.True(t, txm2.Contains(unorderedtx.TxHash{byte(i)}))
}
}
func TestUnorderedTxManager_Flow(t *testing.T) {
txm := unorderedtx.NewManager(t.TempDir())
defer func() {
require.NoError(t, txm.Close())
}()
txm.Start()
currentTime := time.Now()
// Seed the manager with a txs, some of which should eventually be purged and
// the others will remain. Txs with TTL less than or equal to 50 should be purged.
for i := 1; i <= 100; i++ {
txHash := unorderedtx.TxHash{byte(i)}
if i <= 50 {
txm.Add(txHash, currentTime.Add(time.Millisecond*500*time.Duration(i)))
} else {
txm.Add(txHash, currentTime.Add(time.Hour))
}
}
// start a goroutine that mimics new blocks being made every 500ms
doneBlockCh := make(chan bool)
go func() {
ticker := time.NewTicker(time.Millisecond * 500)
defer ticker.Stop()
for t := range ticker.C {
txm.OnNewBlock(t)
if t.After(currentTime.Add(time.Millisecond * 500 * time.Duration(50))) {
doneBlockCh <- true
return
}
}
}()
// Eventually all the txs that should be expired by block 50 should be purged.
// The remaining txs should remain.
require.Eventually(
t,
func() bool {
return txm.Size() == 50
},
2*time.Minute,
5*time.Second,
)
<-doneBlockCh
}
x/auth/ante/unorderedtx/snapshotter.go
-94
View File
@@ -1,94 +0,0 @@
package unorderedtx
import (
"encoding/binary"
"errors"
"io"
"time"
snapshot "cosmossdk.io/store/snapshots/types"
)
const (
txHashSize = 32
timeoutSize = 8
chunkSize = txHashSize + timeoutSize
)
var _ snapshot.ExtensionSnapshotter = &Snapshotter{}
const (
// SnapshotFormat defines the snapshot format of exported unordered transactions.
// No protobuf envelope, no metadata.
SnapshotFormat = 1
// SnapshotName defines the snapshot name of exported unordered transactions.
SnapshotName = "unordered_txs"
)
type Snapshotter struct {
m *Manager
}
func NewSnapshotter(m *Manager) *Snapshotter {
return &Snapshotter{m: m}
}
func (s *Snapshotter) SnapshotName() string {
return SnapshotName
}
func (s *Snapshotter) SnapshotFormat() uint32 {
return SnapshotFormat
}
func (s *Snapshotter) SupportedFormats() []uint32 {
return []uint32{SnapshotFormat}
}
func (s *Snapshotter) SnapshotExtension(height uint64, payloadWriter snapshot.ExtensionPayloadWriter) error {
// export all unordered transactions as a single blob
return s.m.exportSnapshot(height, payloadWriter)
}
func (s *Snapshotter) RestoreExtension(height uint64, format uint32, payloadReader snapshot.ExtensionPayloadReader) error {
if format == SnapshotFormat {
return s.restore(height, payloadReader)
}
return snapshot.ErrUnknownFormat
}
func (s *Snapshotter) restore(height uint64, payloadReader snapshot.ExtensionPayloadReader) error {
// the payload should be the entire set of unordered transactions
payload, err := payloadReader()
if err != nil {
if errors.Is(err, io.EOF) {
return io.ErrUnexpectedEOF
}
return err
}
if len(payload)%chunkSize != 0 {
return errors.New("invalid unordered txs payload length")
}
var i int
for i < len(payload) {
var txHash TxHash
copy(txHash[:], payload[i:i+txHashSize])
timestamp := binary.BigEndian.Uint64(payload[i+txHashSize : i+chunkSize])
// add all txs, we don't care at this point if they are expired,
// we'll let the purge loop handle that
if timestamp != 0 {
s.m.Add(txHash, time.Unix(int64(timestamp), 0))
}
i += chunkSize
}
return nil
}
x/auth/ante/unorderedtx/snapshotter_test.go
-66
View File
@@ -1,66 +0,0 @@
package unorderedtx_test
import (
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/cosmos/cosmos-sdk/x/auth/ante/unorderedtx"
)
func TestSnapshotter(t *testing.T) {
dataDir := t.TempDir()
txm := unorderedtx.NewManager(dataDir)
currentTime := time.Now()
// add a handful of unordered txs
for i := 0; i < 100; i++ {
txm.Add(unorderedtx.TxHash{byte(i)}, currentTime.Add(time.Second*100))
}
var unorderedTxBz []byte
s := unorderedtx.NewSnapshotter(txm)
w := func(bz []byte) error {
unorderedTxBz = bz
return nil
}
err := s.SnapshotExtension(50, w)
require.NoError(t, err)
require.NotEmpty(t, unorderedTxBz)
pr := func() ([]byte, error) {
return unorderedTxBz, nil
}
// restore with an invalid format which should result in an error
err = s.RestoreExtension(50, 2, pr)
require.Error(t, err)
// restore with timestamp > timeout time which should result in all unordered txs synced,
// even the ones that have timed out.
txm2 := unorderedtx.NewManager(dataDir)
s2 := unorderedtx.NewSnapshotter(txm2)
err = s2.RestoreExtension(1, unorderedtx.SnapshotFormat, pr)
require.NoError(t, err)
require.Equal(t, 100, txm2.Size())
// start the manager and wait a bit for the background purge loop to run
txm2.Start()
txm2.OnNewBlock(currentTime.Add(time.Second * 200)) // blocks until channel is read in purge loop
// the loop runs every 5 seconds, so we need to wait for that
require.Eventually(t, func() bool { return txm2.Size() == 0 }, 6*time.Second, 500*time.Millisecond)
// restore with timestamp < timeout time which should result in all unordered txs synced
txm3 := unorderedtx.NewManager(dataDir)
s3 := unorderedtx.NewSnapshotter(txm3)
err = s3.RestoreExtension(uint64(currentTime.Add(time.Second*50).Unix()), unorderedtx.SnapshotFormat, pr)
require.NoError(t, err)
require.Equal(t, 100, txm3.Size())
for i := 0; i < 100; i++ {
require.True(t, txm3.Contains(unorderedtx.TxHash{byte(i)}))
}
}
x/auth/keeper/keeper.go
+64 -15
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"time"
"cosmossdk.io/collections"
"cosmossdk.io/collections/indexes"
@@ -98,10 +99,11 @@ type AccountKeeper struct {
authority string
// State
Schema collections.Schema
Params collections.Item[types.Params]
AccountNumber collections.Sequence
Accounts *collections.IndexedMap[sdk.AccAddress, sdk.AccountI, AccountsIndexes]
Schema collections.Schema
Params collections.Item[types.Params]
AccountNumber collections.Sequence
Accounts *collections.IndexedMap[sdk.AccAddress, sdk.AccountI, AccountsIndexes]
UnorderedNonces collections.KeySet[collections.Pair[int64, []byte]]
}
var _ AccountKeeperI = &AccountKeeper{}
@@ -124,16 +126,17 @@ func NewAccountKeeper(
sb := collections.NewSchemaBuilder(storeService)
ak := AccountKeeper{
addressCodec: ac,
bech32Prefix: bech32Prefix,
storeService: storeService,
proto: proto,
cdc: cdc,
permAddrs: permAddrs,
authority: authority,
Params: collections.NewItem(sb, types.ParamsKey, "params", codec.CollValue[types.Params](cdc)),
AccountNumber: collections.NewSequence(sb, types.GlobalAccountNumberKey, "account_number"),
Accounts: collections.NewIndexedMap(sb, types.AddressStoreKeyPrefix, "accounts", sdk.AccAddressKey, codec.CollInterfaceValue[sdk.AccountI](cdc), NewAccountIndexes(sb)),
addressCodec: ac,
bech32Prefix: bech32Prefix,
storeService: storeService,
proto: proto,
cdc: cdc,
permAddrs: permAddrs,
authority: authority,
Params: collections.NewItem(sb, types.ParamsKey, "params", codec.CollValue[types.Params](cdc)),
AccountNumber: collections.NewSequence(sb, types.GlobalAccountNumberKey, "account_number"),
Accounts: collections.NewIndexedMap(sb, types.AddressStoreKeyPrefix, "accounts", sdk.AccAddressKey, codec.CollInterfaceValue[sdk.AccountI](cdc), NewAccountIndexes(sb)),
UnorderedNonces: collections.NewKeySet(sb, types.UnorderedNoncesKey, "unordered_nonces", collections.PairKeyCodec(collections.Int64Key, collections.BytesKey)),
}
schema, err := sb.Build()
if err != nil {
@@ -165,7 +168,6 @@ func (ak AccountKeeper) GetPubKey(ctx context.Context, addr sdk.AccAddress) (cry
if acc == nil {
return nil, errorsmod.Wrapf(sdkerrors.ErrUnknownAddress, "account %s does not exist", addr)
}
return acc.GetPubKey(), nil
}
@@ -277,3 +279,50 @@ func (ak AccountKeeper) GetParams(ctx context.Context) (params types.Params) {
}
return params
}
// -------------------------------------
// Unordered Nonce management methods
// -------------------------------------
// ContainsUnorderedNonce reports whether the sender has used this timeout already.
func (ak AccountKeeper) ContainsUnorderedNonce(ctx sdk.Context, sender []byte, timeout time.Time) (bool, error) {
return ak.UnorderedNonces.Has(ctx, collections.Join(timeout.UnixNano(), sender))
}
// TryAddUnorderedNonce tries to add a new unordered nonce for the sender.
// If the sender already has an entry with the provided timeout, an error is returned.
func (ak AccountKeeper) TryAddUnorderedNonce(ctx sdk.Context, sender []byte, timeout time.Time) error {
alreadyHas, err := ak.ContainsUnorderedNonce(ctx, sender, timeout)
if err != nil {
return fmt.Errorf("failed to check unordered nonce in storage: %w", err)
}
if alreadyHas {
return fmt.Errorf("sender %s has already used timeout %d", sdk.AccAddress(sender).String(), timeout.UnixNano())
}
return ak.UnorderedNonces.Set(ctx, collections.Join(timeout.UnixNano(), sender))
}
// RemoveExpiredUnorderedNonces removes all unordered nonces that have a timeout value before
// the current block time.
func (ak AccountKeeper) RemoveExpiredUnorderedNonces(ctx sdk.Context) error {
blkTime := ctx.BlockTime().UnixNano()
it, err := ak.UnorderedNonces.Iterate(ctx, collections.NewPrefixUntilPairRange[int64, []byte](blkTime))
if err != nil {
return err
}
defer it.Close()
keys, err := it.Keys()
if err != nil {
return err
}
for _, key := range keys {
if err := ak.UnorderedNonces.Remove(ctx, key); err != nil {
return err
}
}
return nil
}
x/auth/keeper/unordered_tx_test.go
+324
View File
@@ -0,0 +1,324 @@
package keeper_test
import (
"testing"
"time"
"github.com/stretchr/testify/require"
storetypes "cosmossdk.io/store/types"
"github.com/cosmos/cosmos-sdk/runtime"
"github.com/cosmos/cosmos-sdk/testutil"
sdk "github.com/cosmos/cosmos-sdk/types"
moduletestutil "github.com/cosmos/cosmos-sdk/types/module/testutil"
authcodec "github.com/cosmos/cosmos-sdk/x/auth/codec"
"github.com/cosmos/cosmos-sdk/x/auth/keeper"
"github.com/cosmos/cosmos-sdk/x/auth/types"
)
func TestManager(t *testing.T) {
var (
mgr keeper.AccountKeeper
ctx sdk.Context
)
encCfg := moduletestutil.MakeTestEncodingConfig()
reset := func() {
mockStoreKey := storetypes.NewKVStoreKey("test")
storeService := runtime.NewKVStoreService(mockStoreKey)
ctx = testutil.DefaultContextWithDB(t, mockStoreKey, storetypes.NewTransientStoreKey("transient_test")).Ctx
mgr = keeper.NewAccountKeeper(
encCfg.Codec,
storeService,
types.ProtoBaseAccount,
nil,
authcodec.NewBech32Codec("cosmos"),
"cosmos",
types.NewModuleAddress("gov").String(),
)
}
type utxSequence struct {
sender []byte
timeout time.Time
}
testCases := map[string]struct {
addFunc []utxSequence
blockTime time.Time
expectContains []utxSequence
expectNotContains []utxSequence
}{
"transactions are not removed when block time is before every utx": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(10, 0),
},
{
[]byte("cosmos3"),
time.Unix(10, 0),
},
},
blockTime: time.Unix(5, 0),
expectContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(10, 0),
},
{
[]byte("cosmos3"),
time.Unix(10, 0),
},
},
},
"transactions are removed if their timeout is equal to the block time": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(10, 0),
},
{
[]byte("cosmos3"),
time.Unix(10, 0),
},
},
blockTime: time.Unix(10, 10),
expectNotContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(10, 0),
},
{
[]byte("cosmos3"),
time.Unix(10, 0),
},
},
},
"only some txs are removed": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(15, 0),
},
{
[]byte("cosmos3"),
time.Unix(20, 0),
},
},
blockTime: time.Unix(16, 10),
expectContains: []utxSequence{
{
[]byte("cosmos3"),
time.Unix(20, 0),
},
},
expectNotContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(15, 0),
},
},
},
"empty state - no transactions to remove": {
addFunc: []utxSequence{},
blockTime: time.Unix(10, 0),
expectContains: []utxSequence{},
expectNotContains: []utxSequence{},
},
"multiple senders with same timestamp": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(10, 0),
},
},
blockTime: time.Unix(10, 1),
expectNotContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos2"),
time.Unix(10, 0),
},
},
},
"same sender with multiple timestamps": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos1"),
time.Unix(15, 0),
},
{
[]byte("cosmos1"),
time.Unix(20, 0),
},
},
blockTime: time.Unix(16, 0),
expectContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(20, 0),
},
},
expectNotContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 0),
},
{
[]byte("cosmos1"),
time.Unix(15, 0),
},
},
},
"nanosecond precision boundary test": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 999999998),
},
{
[]byte("cosmos2"),
time.Unix(10, 999999999),
},
{
[]byte("cosmos3"),
time.Unix(11, 0),
},
},
blockTime: time.Unix(10, 999999999),
expectContains: []utxSequence{
{
[]byte("cosmos3"),
time.Unix(11, 0),
},
},
expectNotContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(10, 999999998),
},
{
[]byte("cosmos2"),
time.Unix(10, 999999999),
},
},
},
"zero timestamp test": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(0, 0),
},
},
blockTime: time.Unix(1, 0),
expectNotContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(0, 0),
},
},
},
"far future timestamp": {
addFunc: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(2^30, 0), // Very far in the future
},
},
blockTime: time.Unix(10, 0),
expectContains: []utxSequence{
{
[]byte("cosmos1"),
time.Unix(2^30, 0),
},
},
},
}
for name, tc := range testCases {
reset()
t.Run(name, func(t *testing.T) {
ctx = ctx.WithBlockTime(tc.blockTime)
for _, seq := range tc.addFunc {
err := mgr.TryAddUnorderedNonce(ctx, seq.sender, seq.timeout)
t.Logf("added transaction: %d/%s", seq.timeout.UnixNano(), seq.sender)
require.NoError(t, err)
}
t.Logf("removing txs. block_time: %d", tc.blockTime.UnixNano())
err := mgr.RemoveExpiredUnorderedNonces(ctx)
require.NoError(t, err)
for _, seq := range tc.expectNotContains {
has, err := mgr.ContainsUnorderedNonce(ctx, seq.sender, seq.timeout)
require.NoError(t, err)
require.False(t, has, "should not contain %s", seq.sender)
}
for _, seq := range tc.expectContains {
has, err := mgr.ContainsUnorderedNonce(ctx, seq.sender, seq.timeout)
require.NoError(t, err)
require.True(t, has, "expected to contain %d/%s", uint64(seq.timeout.UnixNano()), seq.sender)
}
})
}
}
func TestCannotAddDuplicate(t *testing.T) {
mockStoreKey := storetypes.NewKVStoreKey("test")
storeService := runtime.NewKVStoreService(mockStoreKey)
ctx := testutil.DefaultContextWithDB(t, mockStoreKey, storetypes.NewTransientStoreKey("transient_test")).Ctx
mgr := keeper.NewAccountKeeper(
moduletestutil.MakeTestEncodingConfig().Codec,
storeService,
types.ProtoBaseAccount,
nil,
authcodec.NewBech32Codec("cosmos"),
"cosmos",
types.NewModuleAddress("gov").String(),
)
addUser := []byte("foo")
timeout := time.Unix(10, 0)
err := mgr.TryAddUnorderedNonce(ctx, addUser, timeout)
require.NoError(t, err)
err = mgr.TryAddUnorderedNonce(ctx, addUser, timeout)
require.ErrorContains(t, err, "already used timeout")
}
x/auth/module.go
+11
View File
@@ -36,6 +36,7 @@ var (
_ module.AppModuleSimulation = AppModule{}
_ module.HasGenesis = AppModule{}
_ module.HasServices = AppModule{}
_ appmodule.HasPreBlocker = AppModule{}
_ appmodule.AppModule = AppModule{}
)
@@ -94,6 +95,16 @@ type AppModule struct {
legacySubspace exported.Subspace
}
// PreBlock cleans up expired unordered transaction nonces from state.
// Please ensure to add `x/auth`'s module name to the OrderPreBlocker list in your application.
func (am AppModule) PreBlock(ctx context.Context) (appmodule.ResponsePreBlock, error) {
err := am.accountKeeper.RemoveExpiredUnorderedNonces(sdk.UnwrapSDKContext(ctx))
if err != nil {
return nil, err
}
return &sdk.ResponsePreBlock{ConsensusParamsChanged: false}, nil
}
// IsOnePerModuleType implements the depinject.OnePerModuleType interface.
func (am AppModule) IsOnePerModuleType() {}
x/auth/types/keys.go
+3
View File
@@ -28,4 +28,7 @@ var (
// AccountNumberStoreKeyPrefix prefix for account-by-id store
AccountNumberStoreKeyPrefix = collections.NewPrefix("accountNumber")
// UnorderedNoncesKey prefix for the unordered sequence storage.
UnorderedNoncesKey = collections.NewPrefix(90)
)