Merge branch 'minhd-vu/sensor-broadcast' into minhd-vu/rpc-methods

Author: minhd-vu

.github/workflows/ci.yml

@@ -38,7 +38,7 @@ jobs:
     timeout-minutes: 5
     steps:
       - uses: actions/checkout@v6
-      - uses: crate-ci/typos@78bc6fb2c0d734235d57a2d6b9de923cc325ebdd # v1.43.4
+      - uses: crate-ci/typos@57b11c6b7e54c402ccd9cda953f1072ec4f78e33 # v1.43.5
 
   gen-doc:
     name: Check gen-doc generated files

cmd/p2p/sensor/api.go

@@ -1,7 +1,6 @@
 package sensor
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -58,8 +57,7 @@ type apiData struct {
 // handleAPI sets up the API for interacting with the sensor. All endpoints
 // return information about the sensor node and all connected peers, including
 // the types and counts of eth packets sent and received by each peer.
-// The server gracefully shuts down when the context is cancelled.
-func handleAPI(ctx context.Context, server *ethp2p.Server, conns *p2p.Conns) {
+func handleAPI(server *ethp2p.Server, conns *p2p.Conns) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodGet {
@@ -119,16 +117,7 @@ func handleAPI(ctx context.Context, server *ethp2p.Server, conns *p2p.Conns) {
 	})
 
 	addr := fmt.Sprintf(":%d", inputSensorParams.APIPort)
-	httpServer := &http.Server{Addr: addr, Handler: mux}
-
-	go func() {
-		<-ctx.Done()
-		if err := httpServer.Shutdown(context.Background()); err != nil {
-			log.Error().Err(err).Msg("Failed to shutdown API server")
-		}
-	}()
-
-	if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+	if err := http.ListenAndServe(addr, mux); err != nil {
 		log.Error().Err(err).Msg("Failed to start API handler")
 	}
 }

cmd/p2p/sensor/rpc.go

@@ -1,7 +1,6 @@
 package sensor
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -44,8 +43,7 @@ type rpcError struct {
 // It handles eth_sendRawTransaction requests, validates transaction signatures,
 // and broadcasts valid transactions to all connected peers.
 // Supports both single requests and batch requests per JSON-RPC 2.0 specification.
-// The server gracefully shuts down when the context is cancelled.
-func handleRPC(ctx context.Context, conns *p2p.Conns, networkID uint64) {
+func handleRPC(conns *p2p.Conns, networkID uint64) {
 	// Use network ID as chain ID for signature validation
 	chainID := new(big.Int).SetUint64(networkID)
 
@@ -97,17 +95,8 @@ func handleRPC(ctx context.Context, conns *p2p.Conns, networkID uint64) {
 	})
 
 	addr := fmt.Sprintf(":%d", inputSensorParams.RPCPort)
-	server := &http.Server{Addr: addr, Handler: mux}
-
-	go func() {
-		<-ctx.Done()
-		if err := server.Shutdown(context.Background()); err != nil {
-			log.Error().Err(err).Msg("Failed to shutdown RPC server")
-		}
-	}()
-
 	log.Info().Str("addr", addr).Msg("Starting JSON-RPC server")
-	if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+	if err := http.ListenAndServe(addr, mux); err != nil {
 		log.Error().Err(err).Msg("Failed to start RPC server")
 	}
 }

cmd/p2p/sensor/sensor.go

@@ -6,7 +6,6 @@ import (
 	_ "embed"
 	"errors"
 	"fmt"
-	"os"
 	"os/signal"
 	"syscall"
 	"time"
@@ -173,7 +172,10 @@ var SensorCmd = &cobra.Command{
 		return nil
 	},
 	RunE: func(cmd *cobra.Command, args []string) error {
-		db, err := newDatabase(cmd.Context())
+		ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
+		defer stop()
+
+		db, err := newDatabase(ctx)
 		if err != nil {
 			return err
 		}
@@ -214,7 +216,7 @@ var SensorCmd = &cobra.Command{
 		})
 
 		opts := p2p.EthProtocolOptions{
-			Context:                    cmd.Context(),
+			Context:                    ctx,
 			Database:                   db,
 			GenesisHash:                common.HexToHash(inputSensorParams.GenesisHash),
 			RPC:                        inputSensorParams.RPC,
@@ -259,64 +261,36 @@ var SensorCmd = &cobra.Command{
 		if err = server.Start(); err != nil {
 			return err
 		}
-		defer server.Stop()
+		defer stopServer(&server)
 
 		events := make(chan *ethp2p.PeerEvent)
 		sub := server.SubscribeEvents(events)
 		defer sub.Unsubscribe()
 
-		ticker := time.NewTicker(2 * time.Second) // Ticker for recurring tasks every 2 seconds.
-		ticker1h := time.NewTicker(time.Hour)     // Ticker for running DNS discovery every hour.
+		ticker := time.NewTicker(2 * time.Second)
 		defer ticker.Stop()
-		defer ticker1h.Stop()
-
-		dnsLock := make(chan struct{}, 1)
-		signals := make(chan os.Signal, 1)
-		signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
-
-		// Create a cancellable context for graceful shutdown of background goroutines.
-		ctx, cancel := context.WithCancel(cmd.Context())
-		defer cancel()
 
 		if inputSensorParams.ShouldRunPprof {
-			go handlePprof(ctx)
+			go handlePprof()
 		}
 
 		if inputSensorParams.ShouldRunPrometheus {
-			go handlePrometheus(ctx)
+			go handlePrometheus()
 		}
 
-		go handleAPI(ctx, &server, conns)
-
-		// Start the RPC server for receiving transactions
-		go handleRPC(ctx, conns, inputSensorParams.NetworkID)
-
-		// Run DNS discovery immediately at startup.
-		go handleDNSDiscovery(ctx, &server, dnsLock)
+		go handleAPI(&server, conns)
+		go handleRPC(conns, inputSensorParams.NetworkID)
+		go handleDNSDiscovery(&server)
 
 		for {
 			select {
 			case <-ticker.C:
 				peersGauge.Set(float64(server.PeerCount()))
-				db.WritePeers(cmd.Context(), server.Peers(), time.Now())
-
+				db.WritePeers(ctx, server.Peers(), time.Now())
 				metrics.Update(conns.HeadBlock().Block, conns.OldestBlock())
-
-				urls := []string{}
-				for _, peer := range server.Peers() {
-					urls = append(urls, peer.Node().URLv4())
-				}
-
-				if err := p2p.WritePeers(inputSensorParams.NodesFile, urls); err != nil {
-					log.Error().Err(err).Msg("Failed to write nodes to file")
-				}
-			case <-ticker1h.C:
-				go handleDNSDiscovery(ctx, &server, dnsLock)
-			case <-signals:
-				// This gracefully stops the sensor so that the peers can be written to
-				// the nodes file.
+				writePeers(server.Peers())
+			case <-ctx.Done():
 				log.Info().Msg("Stopping sensor...")
-				cancel()
 				return nil
 			case event := <-events:
 				log.Debug().Any("event", event).Send()
@@ -327,70 +301,78 @@ var SensorCmd = &cobra.Command{
 	},
 }
 
-// handlePprof starts a server for performance profiling using pprof on the
-// specified port. This allows for real-time monitoring and analysis of the
-// sensor's performance. The port number is configured through
-// inputSensorParams.PprofPort. The server gracefully shuts down when the
-// context is cancelled.
-func handlePprof(ctx context.Context) {
-	addr := fmt.Sprintf(":%d", inputSensorParams.PprofPort)
-	server := &http.Server{Addr: addr}
+// writePeers writes the enode URLs of connected peers to the nodes file.
+func writePeers(peers []*ethp2p.Peer) {
+	urls := make([]string, 0, len(peers))
+	for _, peer := range peers {
+		urls = append(urls, peer.Node().URLv4())
+	}
+
+	if err := p2p.WritePeers(inputSensorParams.NodesFile, urls); err != nil {
+		log.Error().Err(err).Msg("Failed to write nodes to file")
+	}
+}
+
+// stopServer stops the p2p server with a timeout to avoid hanging on shutdown.
+// This is necessary because go-ethereum's discovery shutdown can deadlock.
+func stopServer(server *ethp2p.Server) {
+	done := make(chan struct{})
 
 	go func() {
-		<-ctx.Done()
-		if err := server.Shutdown(context.Background()); err != nil {
-			log.Error().Err(err).Msg("Failed to shutdown pprof server")
-		}
+		server.Stop()
+		close(done)
 	}()
 
-	if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+	}
+}
+
+// handlePprof starts a server for performance profiling using pprof on the
+// specified port. This allows for real-time monitoring and analysis of the
+// sensor's performance. The port number is configured through
+// inputSensorParams.PprofPort. An error is logged if the server fails to start.
+func handlePprof() {
+	addr := fmt.Sprintf(":%d", inputSensorParams.PprofPort)
+	if err := http.ListenAndServe(addr, nil); err != nil {
 		log.Error().Err(err).Msg("Failed to start pprof")
 	}
 }
 
 // handlePrometheus starts a server to expose Prometheus metrics at the /metrics
 // endpoint. This enables Prometheus to scrape and collect metrics data for
 // monitoring purposes. The port number is configured through
-// inputSensorParams.PrometheusPort. The server gracefully shuts down when the
-// context is cancelled.
-func handlePrometheus(ctx context.Context) {
-	mux := http.NewServeMux()
-	mux.Handle("/metrics", promhttp.Handler())
-
+// inputSensorParams.PrometheusPort. An error is logged if the server fails to
+// start.
+func handlePrometheus() {
+	http.Handle("/metrics", promhttp.Handler())
 	addr := fmt.Sprintf(":%d", inputSensorParams.PrometheusPort)
-	server := &http.Server{Addr: addr, Handler: mux}
-
-	go func() {
-		<-ctx.Done()
-		if err := server.Shutdown(context.Background()); err != nil {
-			log.Error().Err(err).Msg("Failed to shutdown Prometheus server")
-		}
-	}()
-
-	if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+	if err := http.ListenAndServe(addr, nil); err != nil {
 		log.Error().Err(err).Msg("Failed to start Prometheus handler")
 	}
 }
 
 // handleDNSDiscovery performs DNS-based peer discovery and adds new peers to
 // the p2p server. It uses an iterator to discover peers incrementally rather
-// than loading all nodes at once. The lock channel prevents concurrent runs.
-// Discovery stops when the context is cancelled.
-func handleDNSDiscovery(ctx context.Context, server *ethp2p.Server, lock chan struct{}) {
+// than loading all nodes at once. Runs immediately and then hourly.
+func handleDNSDiscovery(server *ethp2p.Server) {
 	if len(inputSensorParams.DiscoveryDNS) == 0 {
 		return
 	}
 
-	select {
-	case lock <- struct{}{}:
-		defer func() { <-lock }()
-	case <-ctx.Done():
-		return
-	default:
-		log.Warn().Msg("DNS discovery already running, skipping")
-		return
+	discoverPeers(server)
+
+	ticker := time.NewTicker(time.Hour)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		discoverPeers(server)
 	}
+}
 
+// discoverPeers performs a single DNS discovery round.
+func discoverPeers(server *ethp2p.Server) {
 	log.Info().
 		Str("discovery-dns", inputSensorParams.DiscoveryDNS).
 		Msg("Starting DNS discovery")
@@ -403,27 +385,13 @@ func handleDNSDiscovery(ctx context.Context, server *ethp2p.Server, lock chan st
 	}
 	defer iter.Close()
 
-	// Add DNS-discovered peers using the iterator.
 	count := 0
 	for iter.Next() {
-		// Check for context cancellation to stop discovery promptly.
-		select {
-		case <-ctx.Done():
-			log.Info().
-				Int("discovered_peers", count).
-				Msg("DNS discovery interrupted")
-			return
-		default:
-		}
-
 		node := iter.Node()
 		log.Debug().
 			Str("enode", node.URLv4()).
 			Msg("Discovered peer through DNS")
 
-		// Add the peer to the static node set. The server itself handles whether to
-		// connect to the peer if it's already connected. If a node is part of the
-		// static peer set, the server will handle reconnecting after disconnects.
 		server.AddPeer(node)
 		count++
 	}

go.mod

@@ -33,7 +33,7 @@ require (
 	golang.org/x/exp v0.0.0-20241009180824-f66d83c29e7c
 	golang.org/x/text v0.34.0
 	golang.org/x/time v0.14.0
-	google.golang.org/api v0.266.0
+	google.golang.org/api v0.267.0
 	google.golang.org/protobuf v1.36.11
 )
 
@@ -163,15 +163,15 @@ require (
 	golang.org/x/sys v0.41.0 // indirect
 	golang.org/x/term v0.40.0 // indirect
 	google.golang.org/genproto v0.0.0-20260128011058-8636f8732409 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260128011058-8636f8732409 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260203192932-546029d2fa20 // indirect
 	google.golang.org/grpc v1.78.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	lukechampine.com/blake3 v1.3.0 // indirect
 )
 
 require (
-	cloud.google.com/go/kms v1.25.0
+	cloud.google.com/go/kms v1.26.0
 	github.com/0xPolygon/cdk-contracts-tooling v0.0.1
 	github.com/chromedp/cdproto v0.0.0-20250724212937-08a3db8b4327
 	github.com/chromedp/chromedp v0.14.2
@@ -180,7 +180,7 @@ require (
 	github.com/fatih/color v1.18.0
 	github.com/gateway-fm/vectorized-poseidon-gold v1.0.0
 	github.com/gdamore/tcell/v2 v2.13.8
-	github.com/go-echarts/go-echarts/v2 v2.6.7
+	github.com/go-echarts/go-echarts/v2 v2.7.0
 	github.com/go-errors/errors v1.5.1
 	github.com/google/tink/go v1.7.0
 	github.com/iden3/go-iden3-crypto v0.0.17