From 94068be1232659a1c4a92d44550c285ba919eb26 Mon Sep 17 00:00:00 2001 From: Tyler <48813565+technicallyty@users.noreply.github.com> Date: Mon, 24 Mar 2025 14:49:52 -0700 Subject: [PATCH] refactor: PoC for unordered txs per new spec (#24010) --- runtime/app.go | 7 +- simapp/ante.go | 5 +- simapp/app.go | 43 +-- simapp/app_config.go | 1 + simapp/app_di.go | 38 +-- x/auth/ante/ante.go | 10 +- x/auth/ante/expected_keepers.go | 7 + x/auth/ante/unordered.go | 207 ++++--------- x/auth/ante/unordered_test.go | 327 ++++++++++++-------- x/auth/ante/unorderedtx/manager.go | 286 ----------------- x/auth/ante/unorderedtx/manager_test.go | 145 --------- x/auth/ante/unorderedtx/snapshotter.go | 94 ------ x/auth/ante/unorderedtx/snapshotter_test.go | 66 ---- x/auth/keeper/keeper.go | 79 ++++- x/auth/keeper/unordered_tx_test.go | 324 +++++++++++++++++++ x/auth/module.go | 11 + x/auth/types/keys.go | 3 + 17 files changed, 698 insertions(+), 955 deletions(-) delete mode 100644 x/auth/ante/unorderedtx/manager.go delete mode 100644 x/auth/ante/unorderedtx/manager_test.go delete mode 100644 x/auth/ante/unorderedtx/snapshotter.go delete mode 100644 x/auth/ante/unorderedtx/snapshotter_test.go create mode 100644 x/auth/keeper/unordered_tx_test.go diff --git a/runtime/app.go b/runtime/app.go index 00e3617f2e..1eaa2e293f 100644 --- a/runtime/app.go +++ b/runtime/app.go @@ -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) } diff --git a/simapp/ante.go b/simapp/ante.go index 6ccb5df268..c33e88ef69 100644 --- a/simapp/ante.go +++ b/simapp/ante.go @@ -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 diff --git a/simapp/app.go b/simapp/app.go index ae88239e74..32b932e8c4 100644 --- a/simapp/app.go +++ b/simapp/app.go @@ -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) } diff --git a/simapp/app_config.go b/simapp/app_config.go index 51b5cb9208..c2cc96ba34 100644 --- a/simapp/app_config.go +++ b/simapp/app_config.go @@ -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 diff --git a/simapp/app_di.go b/simapp/app_di.go index 2a3407032c..7565b1098d 100644 --- a/simapp/app_di.go +++ b/simapp/app_di.go @@ -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, }, diff --git a/x/auth/ante/ante.go b/x/auth/ante/ante.go index f414767233..755ff8f385 100644 --- a/x/auth/ante/ante.go +++ b/x/auth/ante/ante.go @@ -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 diff --git a/x/auth/ante/expected_keepers.go b/x/auth/ante/expected_keepers.go index 139204cedf..45ded21626 100644 --- a/x/auth/ante/expected_keepers.go +++ b/x/auth/ante/expected_keepers.go @@ -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 diff --git a/x/auth/ante/unordered.go b/x/auth/ante/unordered.go index a97878959e..28b2f927ce 100644 --- a/x/auth/ante/unordered.go +++ b/x/auth/ante/unordered.go @@ -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() } diff --git a/x/auth/ante/unordered_test.go b/x/auth/ante/unordered_test.go index 99d7c90f09..f47901fe89 100644 --- a/x/auth/ante/unordered_test.go +++ b/x/auth/ante/unordered_test.go @@ -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 } diff --git a/x/auth/ante/unorderedtx/manager.go b/x/auth/ante/unorderedtx/manager.go deleted file mode 100644 index d41399deca..0000000000 --- a/x/auth/ante/unorderedtx/manager.go +++ /dev/null @@ -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 -} diff --git a/x/auth/ante/unorderedtx/manager_test.go b/x/auth/ante/unorderedtx/manager_test.go deleted file mode 100644 index 271a2627d4..0000000000 --- a/x/auth/ante/unorderedtx/manager_test.go +++ /dev/null @@ -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 -} diff --git a/x/auth/ante/unorderedtx/snapshotter.go b/x/auth/ante/unorderedtx/snapshotter.go deleted file mode 100644 index 4c855c9e31..0000000000 --- a/x/auth/ante/unorderedtx/snapshotter.go +++ /dev/null @@ -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 -} diff --git a/x/auth/ante/unorderedtx/snapshotter_test.go b/x/auth/ante/unorderedtx/snapshotter_test.go deleted file mode 100644 index a912d69aea..0000000000 --- a/x/auth/ante/unorderedtx/snapshotter_test.go +++ /dev/null @@ -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)})) - } -} diff --git a/x/auth/keeper/keeper.go b/x/auth/keeper/keeper.go index 393ed99907..f03f45af8a 100644 --- a/x/auth/keeper/keeper.go +++ b/x/auth/keeper/keeper.go @@ -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 +} diff --git a/x/auth/keeper/unordered_tx_test.go b/x/auth/keeper/unordered_tx_test.go new file mode 100644 index 0000000000..6576fe9fa8 --- /dev/null +++ b/x/auth/keeper/unordered_tx_test.go @@ -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") +} diff --git a/x/auth/module.go b/x/auth/module.go index c8f46a9efe..47495ba6ff 100644 --- a/x/auth/module.go +++ b/x/auth/module.go @@ -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() {} diff --git a/x/auth/types/keys.go b/x/auth/types/keys.go index fb3295511f..0f96301972 100644 --- a/x/auth/types/keys.go +++ b/x/auth/types/keys.go @@ -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) )