feat(krakatoa): operate_exclusively toggles krakatoa features

evmd/app.go

@@ -185,7 +185,7 @@ type EVMD struct {
 	FeeMarketKeeper feemarketkeeper.Keeper
 	EVMKeeper       *evmkeeper.Keeper
 	Erc20Keeper     erc20keeper.Keeper
-	EVMMempool      *evmmempool.ExperimentalEVMMempool
+	EVMMempool      sdkmempool.ExtMempool
 
 	// the module manager
 	ModuleManager      *module.Manager

evmd/mempool.go

@@ -17,10 +17,6 @@ import (
 	sdkmempool "github.com/cosmos/cosmos-sdk/types/mempool"
 )
 
-// enables abci.InsertTx & abci.ReapTxs to be used exclusively by the mempool.
-// @see evmmempool.ExperimentalEVMMempool.OperateExclusively
-const mempoolOperateExclusively = true
-
 // configureEVMMempool sets up the EVM mempool and related handlers using viper configuration.
 func (app *EVMD) configureEVMMempool(appOpts servertypes.AppOptions, logger log.Logger) error {
 	if evmtypes.GetChainConfig() == nil {
@@ -34,65 +30,97 @@ func (app *EVMD) configureEVMMempool(appOpts servertypes.AppOptions, logger log.
 		return nil
 	}
 
-	mempoolConfig, err := app.createMempoolConfig(appOpts, logger)
-	if err != nil {
-		return fmt.Errorf("failed to get mempool config: %w", err)
+	if server.GetShouldOperateExclusively(appOpts, logger) {
+		logger.Info("app-side mempool is operating exclusively, setting up Krakatoa mempool")
+
+		krakatoaConfig := app.createKrakatoaMempoolConfig(appOpts, logger)
+		txEncoder := evmmempool.NewTxEncoder(app.txConfig)
+		evmRechecker := evmmempool.NewTxRechecker(krakatoaConfig.AnteHandler, txEncoder)
+		cosmosRechecker := evmmempool.NewTxRechecker(krakatoaConfig.AnteHandler, txEncoder)
+
+		krakatoaMempool := evmmempool.NewKrakatoaMempool(
+			app.CreateQueryContext,
+			logger,
+			app.EVMKeeper,
+			app.FeeMarketKeeper,
+			app.txConfig,
+			evmRechecker,
+			cosmosRechecker,
+			krakatoaConfig,
+			cosmosPoolMaxTx,
+		)
+
+		app.SetInsertTxHandler(app.NewInsertTxHandler(krakatoaMempool))
+		app.SetReapTxsHandler(app.NewReapTxsHandler(krakatoaMempool))
+
+		txVerifier := NewNoCheckProposalTxVerifier(app.BaseApp)
+		abciProposalHandler := baseapp.NewDefaultProposalHandler(krakatoaMempool, txVerifier)
+		abciProposalHandler.SetSignerExtractionAdapter(
+			evmmempool.NewEthSignerExtractionAdapter(
+				sdkmempool.NewDefaultSignerExtractionAdapter(),
+			),
+		)
+		app.SetPrepareProposal(abciProposalHandler.PrepareProposalHandler())
+
+		app.EVMMempool = krakatoaMempool
+		app.SetMempool(krakatoaMempool)
+	} else {
+		logger.Info("app-side mempool is not operating exclusively, setting up default EVM mempool")
+
+		evmMempool := evmmempool.NewExperimentalEVMMempool(
+			app.CreateQueryContext,
+			logger,
+			app.EVMKeeper,
+			app.FeeMarketKeeper,
+			app.txConfig,
+			app.createMempoolConfig(appOpts, logger),
+			cosmosPoolMaxTx,
+		)
+
+		app.SetCheckTxHandler(evmmempool.NewCheckTxHandler(evmMempool))
+
+		abciProposalHandler := baseapp.NewDefaultProposalHandler(evmMempool, app)
+		abciProposalHandler.SetSignerExtractionAdapter(
+			evmmempool.NewEthSignerExtractionAdapter(
+				sdkmempool.NewDefaultSignerExtractionAdapter(),
+			),
+		)
+		app.SetPrepareProposal(abciProposalHandler.PrepareProposalHandler())
+
+		app.EVMMempool = evmMempool
+		app.SetMempool(evmMempool)
 	}
 
-	txEncoder := evmmempool.NewTxEncoder(app.txConfig)
-	evmRechecker := evmmempool.NewTxRechecker(mempoolConfig.AnteHandler, txEncoder)
-	cosmosRechecker := evmmempool.NewTxRechecker(mempoolConfig.AnteHandler, txEncoder)
-
-	evmMempool := evmmempool.NewExperimentalEVMMempool(
-		app.CreateQueryContext,
-		logger,
-		app.EVMKeeper,
-		app.FeeMarketKeeper,
-		app.txConfig,
-		txEncoder,
-		evmRechecker,
-		cosmosRechecker,
-		mempoolConfig,
-		cosmosPoolMaxTx,
-	)
-	app.EVMMempool = evmMempool
-	app.SetMempool(evmMempool)
-	checkTxHandler := evmmempool.NewCheckTxHandler(evmMempool)
-	app.SetCheckTxHandler(checkTxHandler)
-	app.SetInsertTxHandler(app.NewInsertTxHandler(evmMempool))
-	app.SetReapTxsHandler(app.NewReapTxsHandler(evmMempool))
-
-	txVerifier := NewNoCheckProposalTxVerifier(app.BaseApp)
-	abciProposalHandler := baseapp.NewDefaultProposalHandler(evmMempool, txVerifier)
-	abciProposalHandler.SetSignerExtractionAdapter(
-		evmmempool.NewEthSignerExtractionAdapter(
-			sdkmempool.NewDefaultSignerExtractionAdapter(),
-		),
-	)
-	app.SetPrepareProposal(abciProposalHandler.PrepareProposalHandler())
-
 	return nil
 }
 
 // createMempoolConfig creates a new EVMMempoolConfig with the default configuration
 // and overrides it with values from appOpts if they exist and are non-zero.
-func (app *EVMD) createMempoolConfig(appOpts servertypes.AppOptions, logger log.Logger) (*evmmempool.EVMMempoolConfig, error) {
+func (app *EVMD) createMempoolConfig(appOpts servertypes.AppOptions, logger log.Logger) *evmmempool.EVMMempoolConfig {
 	return &evmmempool.EVMMempoolConfig{
-		AnteHandler:              app.GetAnteHandler(),
-		LegacyPoolConfig:         server.GetLegacyPoolConfig(appOpts, logger),
-		BlockGasLimit:            server.GetBlockGasLimit(appOpts, logger),
-		MinTip:                   server.GetMinTip(appOpts, logger),
-		OperateExclusively:       mempoolOperateExclusively,
+		AnteHandler:      app.GetAnteHandler(),
+		LegacyPoolConfig: server.GetLegacyPoolConfig(appOpts, logger),
+		BlockGasLimit:    server.GetBlockGasLimit(appOpts, logger),
+		MinTip:           server.GetMinTip(appOpts, logger),
+	}
+}
+
+// createKrakatoaMempoolConfig creates a new KrakatoaMempoolConfig with the default configuration
+// and overrides it with values from appOpts if they exist and are non-zero.
+func (app *EVMD) createKrakatoaMempoolConfig(appOpts servertypes.AppOptions, logger log.Logger) *evmmempool.KrakatoaMempoolConfig {
+	mempoolConfig := app.createMempoolConfig(appOpts, logger)
+	return &evmmempool.KrakatoaMempoolConfig{
+		EVMMempoolConfig:         *mempoolConfig,
 		PendingTxProposalTimeout: server.GetPendingTxProposalTimeout(appOpts, logger),
 		InsertQueueSize:          server.GetMempoolInsertQueueSize(appOpts, logger),
-	}, nil
+	}
 }
 
 const (
 	CodeTypeNoRetry = 1
 )
 
-func (app *EVMD) NewInsertTxHandler(evmMempool *evmmempool.ExperimentalEVMMempool) sdk.InsertTxHandler {
+func (app *EVMD) NewInsertTxHandler(evmMempool *evmmempool.KrakatoaMempool) sdk.InsertTxHandler {
 	return func(req *abci.RequestInsertTx) (*abci.ResponseInsertTx, error) {
 		txBytes := req.GetTx()
 
@@ -121,7 +149,7 @@ func (app *EVMD) NewInsertTxHandler(evmMempool *evmmempool.ExperimentalEVMMempoo
 	}
 }
 
-func (app *EVMD) NewReapTxsHandler(evmMempool *evmmempool.ExperimentalEVMMempool) sdk.ReapTxsHandler {
+func (app *EVMD) NewReapTxsHandler(evmMempool *evmmempool.KrakatoaMempool) sdk.ReapTxsHandler {
 	return func(req *abci.RequestReapTxs) (*abci.ResponseReapTxs, error) {
 		maxBytes, maxGas := req.GetMaxBytes(), req.GetMaxGas()
 		txs, err := evmMempool.ReapNewValidTxs(maxBytes, maxGas)

evmd/tests/integration/create_app.go

@@ -7,7 +7,6 @@ import (
 	dbm "github.com/cosmos/cosmos-db"
 	"github.com/cosmos/evm"
 	"github.com/cosmos/evm/evmd"
-	evmmempool "github.com/cosmos/evm/mempool"
 	srvflags "github.com/cosmos/evm/server/flags"
 	"github.com/cosmos/evm/testutil/constants"
 	feemarkettypes "github.com/cosmos/evm/x/feemarket/types"
@@ -23,9 +22,8 @@ import (
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 )
 
-// CreateEvmd creates an evm app for regular integration tests (non-mempool)
-// This version uses a noop mempool to avoid state issues during transaction processing
-func CreateEvmd(chainID string, evmChainID uint64, customBaseAppOptions ...func(*baseapp.BaseApp)) evm.EvmApp {
+// CreateEvmd creates an evm app for integration tests
+func CreateEvmd(chainID string, evmChainID uint64, exclusiveMempool bool, customBaseAppOptions ...func(*baseapp.BaseApp)) evm.EvmApp {
 	// A temporary home directory is created and used to prevent race conditions
 	// related to home directory locks in chains that use the WASM module.
 	defaultNodeHome, err := os.MkdirTemp("", "evmd-temp-homedir")
@@ -36,7 +34,7 @@ func CreateEvmd(chainID string, evmChainID uint64, customBaseAppOptions ...func(
 	db := dbm.NewMemDB()
 	logger := log.NewNopLogger()
 	loadLatest := true
-	appOptions := NewAppOptionsWithFlagHomeAndChainID(defaultNodeHome, evmChainID)
+	appOptions := NewAppOptionsWithFlagHomeAndChainID(defaultNodeHome, evmChainID, exclusiveMempool)
 
 	baseAppOptions := append(customBaseAppOptions, baseapp.SetChainID(chainID))
 
@@ -55,8 +53,8 @@ func CreateEvmd(chainID string, evmChainID uint64, customBaseAppOptions ...func(
 		WithHeight(1).
 		WithTxConfig(app.GetTxConfig())
 
-	// Get the mempool and set the client context
-	if m, ok := app.GetMempool().(*evmmempool.ExperimentalEVMMempool); ok && m != nil {
+	// Get the mempool and set the client context if supported
+	if m, ok := app.GetMempool().(interface{ SetClientCtx(client.Context) }); ok && m != nil {
 		m.SetClientCtx(clientCtx)
 	}
 
@@ -76,7 +74,7 @@ func SetupEvmd() (ibctesting.TestingApp, map[string]json.RawMessage) {
 		dbm.NewMemDB(),
 		nil,
 		true,
-		NewAppOptionsWithFlagHomeAndChainID(defaultNodeHome, constants.EighteenDecimalsChainID),
+		NewAppOptionsWithFlagHomeAndChainID(defaultNodeHome, constants.EighteenDecimalsChainID, false),
 	)
 	// disable base fee for testing
 	genesisState := app.DefaultGenesis()
@@ -93,11 +91,12 @@ func SetupEvmd() (ibctesting.TestingApp, map[string]json.RawMessage) {
 	return app, genesisState
 }
 
-func NewAppOptionsWithFlagHomeAndChainID(home string, evmChainID uint64) simutils.AppOptionsMap {
+func NewAppOptionsWithFlagHomeAndChainID(home string, evmChainID uint64, exlcusiveMempool bool) simutils.AppOptionsMap {
 	return simutils.AppOptionsMap{
-		flags.FlagHome:                            home,
-		srvflags.EVMChainID:                       evmChainID,
-		srvflags.EVMMempoolInsertQueueSize:        5000,
+		flags.FlagHome:                              home,
+		srvflags.EVMChainID:                         evmChainID,
+		srvflags.EVMMempoolOperateExclusively:       exlcusiveMempool,
+		srvflags.EVMMempoolInsertQueueSize:          5000,
 		srvflags.EVMMempoolPendingTxProposalTimeout: "250ms",
 	}
 }

evmd/tests/integration/mempool/mempool_test.go

@@ -9,9 +9,15 @@ import (
 	"github.com/cosmos/evm/evmd/tests/integration"
 	"github.com/cosmos/evm/tests/integration/mempool"
 	testapp "github.com/cosmos/evm/testutil/app"
+	"github.com/cosmos/evm/testutil/integration/evm/network"
 )
 
 func TestMempoolIntegrationTestSuite(t *testing.T) {
 	create := testapp.ToEvmAppCreator[evm.IntegrationNetworkApp](integration.CreateEvmd, "evm.IntegrationNetworkApp")
 	suite.Run(t, mempool.NewMempoolIntegrationTestSuite(create))
 }
+
+func TestKrakatoaMempoolIntegrationTestSuite(t *testing.T) {
+	create := testapp.ToEvmAppCreator[evm.IntegrationNetworkApp](integration.CreateEvmd, "evm.IntegrationNetworkApp")
+	suite.Run(t, mempool.NewMempoolIntegrationTestSuite(create, network.WithExclusiveMempool()))
+}

go.mod

@@ -115,6 +115,7 @@ require (
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/chigopher/pathlib v0.19.1 // indirect
 	github.com/chzyer/readline v1.5.1 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5 // indirect
@@ -195,12 +196,14 @@ require (
 	github.com/hdevalence/ed25519consensus v0.2.0 // indirect
 	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
 	github.com/huandu/skiplist v1.2.1 // indirect
+	github.com/huandu/xstrings v1.4.0 // indirect
 	github.com/huin/goupnp v1.3.0 // indirect
 	github.com/iancoleman/strcase v0.3.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/ipfs/go-cid v0.5.0 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/jinzhu/copier v0.4.0 // indirect
 	github.com/jmhodges/levigo v1.0.0 // indirect
 	github.com/klauspost/compress v1.18.4 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
@@ -306,6 +309,7 @@ require (
 	github.com/tklauser/numcpus v0.11.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ulikunitz/xz v0.5.15 // indirect
+	github.com/vektra/mockery/v2 v2.53.6 // indirect
 	github.com/wlynxg/anet v0.0.5 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	github.com/zondax/golem v0.27.0 // indirect

go.sum

@@ -201,6 +201,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chenzhuoyu/base64x v0.0.0-20211019084208-fb5309c8db06/go.mod h1:DH46F32mSOjUmXrMHnKwZdA8wcEefY7UVqBKYGjpdQY=
 github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311/go.mod h1:b583jCggY9gE99b6G5LEC39OIiVsWj+R97kbl5odCEk=
+github.com/chigopher/pathlib v0.19.1 h1:RoLlUJc0CqBGwq239cilyhxPNLXTK+HXoASGyGznx5A=
+github.com/chigopher/pathlib v0.19.1/go.mod h1:tzC1dZLW8o33UQpWkNkhvPwL5n4yyFRFm/jL1YGWFvY=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/logex v1.2.1 h1:XHDu3E6q+gdHgsdTPH6ImJMIp436vR6MPtH8gP05QzM=
 github.com/chzyer/logex v1.2.1/go.mod h1:JLbx6lG2kDbNRFnfkgvh4eRJRPX1QCoOIWomwysCBrQ=
@@ -610,6 +612,8 @@ github.com/huandu/go-assert v1.1.5 h1:fjemmA7sSfYHJD7CUqs9qTwwfdNAx7/j2/ZlHXzNB3
 github.com/huandu/go-assert v1.1.5/go.mod h1:yOLvuqZwmcHIC5rIzrBhT7D3Q9c3GFnd0JrPVhn/06U=
 github.com/huandu/skiplist v1.2.1 h1:dTi93MgjwErA/8idWTzIw4Y1kZsMWx35fmI2c8Rij7w=
 github.com/huandu/skiplist v1.2.1/go.mod h1:7v3iFjLcSAzO4fN5B8dvebvo/qsfumiLiDXMrPiHF9w=
+github.com/huandu/xstrings v1.4.0 h1:D17IlohoQq4UcpqD7fDk80P7l+lwAmlFaBHgOipl2FU=
+github.com/huandu/xstrings v1.4.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
 github.com/hudl/fargo v1.3.0/go.mod h1:y3CKSmjA+wD2gak7sUSXTAoopbhU08POFhmITJgmKTg=
 github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc=
 github.com/huin/goupnp v1.3.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8=
@@ -639,6 +643,8 @@ github.com/jhump/protoreflect v1.18.0 h1:TOz0MSR/0JOZ5kECB/0ufGnC2jdsgZ123Rd/k4Z
 github.com/jhump/protoreflect v1.18.0/go.mod h1:ezWcltJIVF4zYdIFM+D/sHV4Oh5LNU08ORzCGfwvTz8=
 github.com/jhump/protoreflect/v2 v2.0.0-beta.1 h1:Dw1rslK/VotaUGYsv53XVWITr+5RCPXfvvlGrM/+B6w=
 github.com/jhump/protoreflect/v2 v2.0.0-beta.1/go.mod h1:D9LBEowZyv8/iSu97FU2zmXG3JxVTmNw21mu63niFzU=
+github.com/jinzhu/copier v0.4.0 h1:w3ciUoD19shMCRargcpm0cm91ytaBhDvuRpz1ODO/U8=
+github.com/jinzhu/copier v0.4.0/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmhodges/levigo v1.0.0 h1:q5EC36kV79HWeTBWsod3mG11EgStG3qArTKcvlksN1U=
 github.com/jmhodges/levigo v1.0.0/go.mod h1:Q6Qx+uH3RAqyK4rFQroq9RL7mdkABMcfhEI+nNuzMJQ=
@@ -1117,6 +1123,8 @@ github.com/urfave/cli v1.22.1 h1:+mkCCcOFKPnCmVYVcURKps1Xe+3zP90gSYGNfRkjoIY=
 github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/urfave/cli/v2 v2.27.5 h1:WoHEJLdsXr6dDWoJgMq/CboDmyY/8HMMH1fTECbih+w=
 github.com/urfave/cli/v2 v2.27.5/go.mod h1:3Sevf16NykTbInEnD0yKkjDAeZDS0A6bzhBH5hrMvTQ=
+github.com/vektra/mockery/v2 v2.53.6 h1:qfUB6saauu652ZlMF/mEdlj7B/A0fw2XR0XBACBrf7Y=
+github.com/vektra/mockery/v2 v2.53.6/go.mod h1:fjxC+mskIZqf67+z34pHxRRyyZnPnWNA36Cirf01Pkg=
 github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
 github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
 github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=

mempool/blockchain_test.go

@@ -45,7 +45,7 @@ func TestBlockchainRaceCondition(t *testing.T) {
 	logger := log.NewNopLogger()
 
 	// Create mock keepers using generated mocks
-	mockVMKeeper := mocks.NewVMKeeper(t)
+	mockVMKeeper := mocks.NewVMKeeperI(t)
 	mockFeeMarketKeeper := mocks.NewFeeMarketKeeper(t)
 
 	ethCfg := vmtypes.DefaultChainConfig(constants.EighteenDecimalsChainID)

mempool/interface.go

@@ -13,6 +13,17 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
+// NotifiedMempool is the set of methods that a mempool must implement in order
+// to be notified of new blocks by this Keeper via the EndBlocker.
+type NotifiedMempool interface {
+	// HasEventBus returns true if the mempool has an event bus configured to
+	// get cometbft events
+	HasEventBus() bool
+
+	// GetBlockchain returns the mempools blockchain representation.
+	GetBlockchain() *Blockchain
+}
+
 type VMKeeperI interface {
 	GetBaseFee(ctx sdk.Context) *big.Int
 	GetParams(ctx sdk.Context) (params vmtypes.Params)
@@ -29,7 +40,7 @@ type VMKeeperI interface {
 	SetCode(ctx sdk.Context, codeHash []byte, code []byte)
 	DeleteAccount(ctx sdk.Context, addr common.Address) error
 	KVStoreKeys() map[string]storetypes.StoreKey
-	SetEvmMempool(evmMempool *ExperimentalEVMMempool)
+	SetEvmMempool(evmMempool NotifiedMempool)
 }
 
 type FeeMarketKeeperI interface {

mempool/internal/queue/queue.go

@@ -8,8 +8,6 @@ import (
 
 	"github.com/gammazero/deque"
 
-	"cosmossdk.io/log/v2"
-
 	"github.com/cosmos/cosmos-sdk/telemetry"
 )
 
@@ -40,18 +38,16 @@ type Queue[Tx any] struct {
 	// rejecting new additions
 	maxSize int
 
-	logger log.Logger
-	done   chan struct{}
+	done chan struct{}
 }
 
 var ErrQueueFull = errors.New("queue full")
 
 // New creates a new queue.
-func New[Tx any](insert func(txs []*Tx) []error, maxSize int, logger log.Logger) *Queue[Tx] {
+func New[Tx any](insert func(txs []*Tx) []error, maxSize int) *Queue[Tx] {
 	iq := &Queue[Tx]{
 		insert:  insert,
 		maxSize: maxSize,
-		logger:  logger,
 		signal:  make(chan struct{}, 1),
 		done:    make(chan struct{}),
 	}