node: fix after stopping the spy and restarting it immediately, the spy cannot receive the VAA received by the guardian #4447

Author: liuxiaomeiG

node/cmd/guardiand/node.go

@@ -6,12 +6,10 @@ import (
 	"net"
 	_ "net/http/pprof" // #nosec G108 we are using a custom router (`router := mux.NewRouter()`) and thus not automatically expose pprof.
 	"os"
-	"os/signal"
 	"path"
 	"runtime"
 	"slices"
 	"strings"
-	"syscall"
 	"time"
 
 	"github.com/certusone/wormhole/node/pkg/guardiansigner"
@@ -1107,14 +1105,7 @@ func runNode(cmd *cobra.Command, args []string) {
 	rpcMap["ibcLCD"] = *ibcLCD
 	rpcMap["ibcWS"] = *ibcWS
 
-	// Handle SIGTERM
-	sigterm := make(chan os.Signal, 1)
-	signal.Notify(sigterm, syscall.SIGTERM)
-	go func() {
-		<-sigterm
-		logger.Info("Received sigterm. exiting.")
-		rootCtxCancel()
-	}()
+	common.ListenSysExit(logger, rootCtxCancel)
 
 	// log golang version
 	logger.Info("golang version", zap.String("golang_version", runtime.Version()))

node/cmd/spy/spy.go

@@ -343,6 +343,8 @@ func runSpy(cmd *cobra.Command, args []string) {
 	rootCtx, rootCtxCancel = context.WithCancel(context.Background())
 	defer rootCtxCancel()
 
+	common.ListenSysExit(logger, rootCtxCancel)
+
 	// Inbound signed VAAs
 	signedInC := make(chan *gossipv1.SignedVAAWithQuorum, 1024)
 

node/pkg/common/cmd_utils.go

@@ -0,0 +1,21 @@
+package common
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"go.uber.org/zap"
+)
+
+func ListenSysExit(logger *zap.Logger, ctxCancel context.CancelFunc) {
+	// Handle SIGTERM, SIGINT
+	sigterm := make(chan os.Signal, 1)
+	signal.Notify(sigterm, syscall.SIGTERM, syscall.SIGINT)
+	go func() {
+		<-sigterm
+		logger.Info("Received sigterm. exiting.")
+		ctxCancel()
+	}()
+}

node/pkg/common/cmd_utils_test.go

@@ -0,0 +1,119 @@
+package common
+
+import (
+	"context"
+	"syscall"
+	"testing"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+func TestListenSysExit(t *testing.T) {
+	// Create a logger for testing
+	logger := zap.NewNop()
+
+	// Test SIGTERM signal
+	t.Run("SIGTERM signal", func(t *testing.T) {
+		// Create a new context for this test
+		testCtx, testCancel := context.WithCancel(context.Background())
+		defer testCancel()
+
+		// Start ListenSysExit
+		go ListenSysExit(logger, testCancel)
+
+		// Give some time for the signal handler to be set up
+		time.Sleep(100 * time.Millisecond)
+
+		// Send SIGTERM signal to the current process
+		err := syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
+		if err != nil {
+			t.Fatalf("Failed to send SIGTERM signal: %v", err)
+		}
+
+		// Wait for the context to be cancelled
+		select {
+		case <-testCtx.Done():
+			// Context was cancelled as expected
+			t.Log("Context cancelled successfully after SIGTERM")
+		case <-time.After(2 * time.Second):
+			t.Fatal("Context was not cancelled within 2 seconds after SIGTERM")
+		}
+	})
+
+	// Test SIGINT signal
+	t.Run("SIGINT signal", func(t *testing.T) {
+		// Create a new context for this test
+		testCtx, testCancel := context.WithCancel(context.Background())
+		defer testCancel()
+
+		// Start ListenSysExit
+		go ListenSysExit(logger, testCancel)
+
+		// Give some time for the signal handler to be set up
+		time.Sleep(100 * time.Millisecond)
+
+		// Send SIGINT signal to the current process
+		err := syscall.Kill(syscall.Getpid(), syscall.SIGINT)
+		if err != nil {
+			t.Fatalf("Failed to send SIGINT signal: %v", err)
+		}
+
+		// Wait for the context to be cancelled
+		select {
+		case <-testCtx.Done():
+			// Context was cancelled as expected
+			t.Log("Context cancelled successfully after SIGINT")
+		case <-time.After(2 * time.Second):
+			t.Fatal("Context was not cancelled within 2 seconds after SIGINT")
+		}
+	})
+
+	// Test that the function doesn't exit immediately
+	t.Run("no signal sent", func(t *testing.T) {
+		// Create a new context for this test
+		testCtx, testCancel := context.WithCancel(context.Background())
+		defer testCancel()
+
+		// Start ListenSysExit
+		go ListenSysExit(logger, testCancel)
+
+		// Wait a short time and verify context is not cancelled
+		select {
+		case <-testCtx.Done():
+			t.Fatal("Context was unexpectedly cancelled without signal")
+		case <-time.After(500 * time.Millisecond):
+			// Context was not cancelled as expected
+			t.Log("Context remained active as expected when no signal was sent")
+		}
+	})
+}
+
+// TestListenSysExitConcurrent tests that multiple calls to ListenSysExit work correctly
+func TestListenSysExitConcurrent(t *testing.T) {
+	logger := zap.NewNop()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Start multiple ListenSysExit goroutines
+	for i := 0; i < 3; i++ {
+		go ListenSysExit(logger, cancel)
+	}
+
+	// Give some time for all signal handlers to be set up
+	time.Sleep(100 * time.Millisecond)
+
+	// Send SIGTERM signal
+	err := syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
+	if err != nil {
+		t.Fatalf("Failed to send SIGTERM signal: %v", err)
+	}
+
+	// Wait for the context to be cancelled
+	select {
+	case <-ctx.Done():
+		t.Log("Context cancelled successfully with multiple ListenSysExit goroutines")
+	case <-time.After(2 * time.Second):
+		t.Fatal("Context was not cancelled within 2 seconds")
+	}
+}

node/pkg/p2p/p2p.go

@@ -331,6 +331,12 @@ func Run(params *RunParams) func(ctx context.Context) error {
 		}
 
 		defer func() {
+			// close connect
+			for _, conn := range h.Network().Conns() {
+				logger.Info("Closing connection", zap.String("peer", conn.RemotePeer().String()))
+				conn.Close()
+			}
+
 			if err := h.Close(); err != nil {
 				logger.Error("error closing the host", zap.Error(err))
 			}