fix: allow mempool interrupt while it is still ongoing (#495)

* add context cancel

* simplify channel cases

* add changelog

* revert config change

* add minimal change set for sigint

* rename

* offload server shutdown go goroutine

* remove goroutine in favor of timeout

* use short timeout

Author: almk-dev

CHANGELOG.md

@@ -6,9 +6,10 @@
 
 ### BUG FIXES
 
-- [\#471](https://github.com/cosmos/evm/pull/471) Notify new block for mempool in time.
+- [\#471](https://github.com/cosmos/evm/pull/471) Notify new block for mempool in time
 - [\#492](https://github.com/cosmos/evm/pull/492) Duplicate case switch to avoid empty execution block
-- [\#509](https://github.com/cosmos/evm/pull/509) Allow value with slashes when query token_pairs.
+- [\#509](https://github.com/cosmos/evm/pull/509) Allow value with slashes when query token_pairs
+- [\#495](https://github.com/cosmos/evm/pull/495) Allow immediate SIGINT interrupt when mempool is not empty
 
 ### IMPROVEMENTS
 

evmd/app.go

@@ -1146,19 +1146,22 @@ func (app *EVMD) SetClientCtx(clientCtx client.Context) {
 	app.clientCtx = clientCtx
 }
 
-// Close unsubscribes from the CometBFT event bus (if set) and closes the underlying BaseApp.
+// Close unsubscribes from the CometBFT event bus (if set) and closes the mempool and underlying BaseApp.
 func (app *EVMD) Close() error {
 	var err error
 	if m, ok := app.GetMempool().(*evmmempool.ExperimentalEVMMempool); ok {
+		app.Logger().Info("Shutting down mempool")
 		err = m.Close()
 	}
-	err = errors.Join(err, app.BaseApp.Close())
+
 	msg := "Application gracefully shutdown"
+	err = errors.Join(err, app.BaseApp.Close())
 	if err == nil {
 		app.Logger().Info(msg)
 	} else {
 		app.Logger().Error(msg, "error", err)
 	}
+
 	return err
 }
 

mempool/mempool.go

@@ -396,12 +396,20 @@ func (m *ExperimentalEVMMempool) SetEventBus(eventBus *cmttypes.EventBus) {
 	}()
 }
 
-// Close unsubscribes from the CometBFT event bus.
+// Close unsubscribes from the CometBFT event bus and shuts down the mempool.
 func (m *ExperimentalEVMMempool) Close() error {
+	var errs []error
 	if m.eventBus != nil {
-		return m.eventBus.Unsubscribe(context.Background(), SubscriberName, stream.NewBlockHeaderEvents)
+		if err := m.eventBus.Unsubscribe(context.Background(), SubscriberName, stream.NewBlockHeaderEvents); err != nil {
+			errs = append(errs, fmt.Errorf("failed to unsubscribe from event bus: %w", err))
+		}
 	}
-	return nil
+
+	if err := m.txPool.Close(); err != nil {
+		errs = append(errs, fmt.Errorf("failed to close txpool: %w", err))
+	}
+
+	return errors.Join(errs...)
 }
 
 // getEVMMessage validates that the transaction contains exactly one message and returns it if it's an EVM message.

server/json_rpc.go

@@ -25,7 +25,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/server"
 )
 
-const shutdownTimeout = 5 * time.Second
+const shutdownTimeout = 200 * time.Millisecond
 
 type AppWithPendingTxStream interface {
 	RegisterPendingTxListener(listener func(common.Hash))
@@ -111,14 +111,13 @@ func StartJSONRPC(
 		case <-ctx.Done():
 			// The calling process canceled or closed the provided context, so we must
 			// gracefully stop the JSON-RPC server.
-			logger.Info("stopping JSON-RPC server...", "address", config.JSONRPC.Address)
+			logger.Info("stopping JSON-RPC server...", "address", config.JSONRPC.Address, "timeout", shutdownTimeout)
 			ctxShutdown, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
 			defer cancel()
 			if err := httpSrv.Shutdown(ctxShutdown); err != nil {
 				logger.Error("failed to shutdown JSON-RPC server", "error", err.Error())
 			}
 			return nil
-
 		case err := <-errCh:
 			if err == http.ErrServerClosed {
 				close(httpSrvDone)

server/start.go

@@ -132,8 +132,8 @@ which accepts a path for the resulting pprof file.
 				return err
 			}
 
-			withTM, _ := cmd.Flags().GetBool(srvflags.WithCometBFT)
-			if !withTM {
+			withbft, _ := cmd.Flags().GetBool(srvflags.WithCometBFT)
+			if !withbft {
 				serverCtx.Logger.Info("starting ABCI without CometBFT")
 				return wrapCPUProfile(serverCtx, func() error {
 					return startStandAlone(serverCtx, clientCtx, opts)
@@ -400,7 +400,7 @@ func startInProcess(svrCtx *server.Context, clientCtx client.Context, opts Start
 	genDocProvider := GenDocProvider(cfg)
 
 	var (
-		tmNode   *node.Node
+		bftNode  *node.Node
 		gRPCOnly = svrCtx.Viper.GetBool(srvflags.GRPCOnly)
 	)
 
@@ -412,7 +412,7 @@ func startInProcess(svrCtx *server.Context, clientCtx client.Context, opts Start
 		logger.Info("starting node with ABCI CometBFT in-process")
 
 		cmtApp := server.NewCometABCIWrapper(app)
-		tmNode, err = node.NewNode(
+		bftNode, err = node.NewNode(
 			cfg,
 			pvm.LoadOrGenFilePV(cfg.PrivValidatorKeyFile(), cfg.PrivValidatorStateFile()),
 			nodeKey,
@@ -427,26 +427,26 @@ func startInProcess(svrCtx *server.Context, clientCtx client.Context, opts Start
 			return err
 		}
 
-		if err := tmNode.Start(); err != nil {
+		if err := bftNode.Start(); err != nil {
 			logger.Error("failed start CometBFT server", "error", err.Error())
 			return err
 		}
 
 		if m, ok := evmApp.GetMempool().(*evmmempool.ExperimentalEVMMempool); ok {
-			m.SetEventBus(tmNode.EventBus())
+			m.SetEventBus(bftNode.EventBus())
 		}
 		defer func() {
-			if tmNode.IsRunning() {
-				_ = tmNode.Stop()
+			if bftNode.IsRunning() {
+				_ = bftNode.Stop()
 			}
 		}()
 	}
 
 	// Add the tx service to the gRPC router. We only need to register this
 	// service if API or gRPC or JSONRPC is enabled, and avoid doing so in the general
 	// case, because it spawns a new local CometBFT RPC client.
-	if (config.API.Enable || config.GRPC.Enable || config.JSONRPC.Enable || config.JSONRPC.EnableIndexer) && tmNode != nil {
-		clientCtx = clientCtx.WithClient(local.New(tmNode))
+	if (config.API.Enable || config.GRPC.Enable || config.JSONRPC.Enable || config.JSONRPC.EnableIndexer) && bftNode != nil {
+		clientCtx = clientCtx.WithClient(local.New(bftNode))
 
 		app.RegisterTxService(clientCtx)
 		app.RegisterTendermintService(clientCtx)