diff --git a/node/node.go b/node/node.go index 41c9971fe8..d9d80bc1fd 100644 --- a/node/node.go +++ b/node/node.go @@ -521,8 +521,10 @@ func (n *Node) wsServerForPort(port int, authenticated bool) *httpServer { httpServer, wsServer = n.httpAuth, n.wsAuth } if n.config.HTTPHost == "" || httpServer.port == port { + n.log.Info("CONFIGURED HTTP SERVER INSTEAD OF WS SERVER") return httpServer } + n.log.Info("PROPERLY RETURNED THE WS SERVER") return wsServer } @@ -683,7 +685,7 @@ func (n *Node) HTTPEndpoint() string { // WSEndpoint returns the current JSON-RPC over WebSocket endpoint. func (n *Node) WSEndpoint() string { - if n.http.wsAllowed() { + if n.http.wsAllowed("WSEndpoint") { return "ws://" + n.http.listenAddr() + n.http.wsConfig.prefix } return "ws://" + n.ws.listenAddr() + n.ws.wsConfig.prefix @@ -696,7 +698,7 @@ func (n *Node) HTTPAuthEndpoint() string { // WSAuthEndpoint returns the current authenticated JSON-RPC over WebSocket endpoint. func (n *Node) WSAuthEndpoint() string { - if n.httpAuth.wsAllowed() { + if n.httpAuth.wsAllowed("WSAuthEndpoint") { return "ws://" + n.httpAuth.listenAddr() + n.httpAuth.wsConfig.prefix } return "ws://" + n.wsAuth.listenAddr() + n.wsAuth.wsConfig.prefix diff --git a/node/node_test.go b/node/node_test.go index 04810a815b..2e86496796 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -24,12 +24,15 @@ import ( "net/http" "reflect" "strings" + "sync" "testing" + "time" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rpc" + "github.com/gorilla/websocket" "github.com/stretchr/testify/assert" ) @@ -479,6 +482,93 @@ func TestWebsocketHTTPOnSeparatePort_WSRequest(t *testing.T) { } } +func TestWebSocketRouting(t *testing.T) { + node := startHTTP(t, 8544, 8545) + defer node.Close() + // Start both the HTTP and WS servers + httpServer := node.http + wsServer := node.ws + + // Define the number of requests you'll be sending to each server + numRequests := 1000 + + // Use wait groups to synchronize the end of our goroutines + var wg sync.WaitGroup + + hs := "http://" + httpServer.endpoint + + // HTTP requests + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < numRequests; i++ { + resp, err := http.Get(hs) + if err != nil { + t.Error(err) + } + // Add additional checks on the response if necessary + resp.Body.Close() + } + }() + + s := "ws://" + wsServer.endpoint + + // WebSocket requests + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < numRequests; i++ { + // Create a new WebSocket connection and make sure it's successful + conn, _, err := websocket.DefaultDialer.Dial(s, nil) + if err != nil { + t.Error(err) + } + conn.Close() + } + }() + + wg.Wait() +} + +func TestWebSocketStartup(t *testing.T) { + // Define the number of startup cycles + numCycles := 1000 + + for cycle := 0; cycle < numCycles; cycle++ { + testFunc(t, cycle) + fmt.Println("Cycle", cycle, "complete") + } +} + +func testFunc(t *testing.T, cycle int) { + // Start the WS server + node := startHTTP(t, 8544, 8545) + defer node.Close() + // Start both the HTTP and WS servers + // httpServer := node.http + wsServer := node.ws + ws := "ws://" + wsServer.endpoint + + // Introduce a slight delay before sending the first request + time.Sleep(50 * time.Millisecond) // Adjust this delay as necessary + + // Define the number of requests you'll be sending to the WS server during each cycle + numRequests := 2 + + // Send requests to the WebSocket server + for i := 0; i < numRequests; i++ { + // Create a new WebSocket connection and make sure it's successful + conn, _, err := websocket.DefaultDialer.Dial(ws, nil) + if err != nil { + t.Errorf("Error during cycle %d, request %d: %v", cycle, i, err) + } + conn.Close() + } + + // Stop the WS server + node.Close() +} + type rpcPrefixTest struct { httpPrefix, wsPrefix string // These lists paths on which JSON-RPC should be served / not served. @@ -586,11 +676,13 @@ func (test rpcPrefixTest) check(t *testing.T, node *Node) { func createNode(t *testing.T, httpPort, wsPort int) *Node { conf := &Config{ - HTTPHost: "127.0.0.1", - HTTPPort: httpPort, - WSHost: "127.0.0.1", - WSPort: wsPort, - HTTPTimeouts: rpc.DefaultHTTPTimeouts, + HTTPHost: "0.0.0.0", + HTTPPort: httpPort, + HTTPVirtualHosts: []string{"*"}, + WSHost: "0.0.0.0", + WSPort: wsPort, + WSOrigins: []string{"*"}, + HTTPTimeouts: rpc.DefaultHTTPTimeouts, } node, err := New(conf) if err != nil { diff --git a/node/rpcstack.go b/node/rpcstack.go index b33c238051..2395e89454 100644 --- a/node/rpcstack.go +++ b/node/rpcstack.go @@ -64,6 +64,7 @@ type rpcHandler struct { } type httpServer struct { + name string log log.Logger timeouts rpc.HTTPTimeouts mux http.ServeMux // registered handlers go here @@ -97,10 +98,23 @@ func newHTTPServer(log log.Logger, timeouts rpc.HTTPTimeouts) *httpServer { h := &httpServer{log: log, timeouts: timeouts, handlerNames: make(map[string]string)} h.httpHandler.Store((*rpcHandler)(nil)) - h.wsHandler.Store((*rpcHandler)(nil)) + h.storeWsHandler((*rpcHandler)(nil), "newHTTPServer") return h } +func (h *httpServer) storeWsHandler(val any, where string) { + h.wsHandler.Store(val) + if val != nil { + h.log.Info(fmt.Sprintf("storeWsHandler %s", where), "ws", val, "h", h) + } +} + +func (h *httpServer) loadWsHandler(where string) *rpcHandler { + ws := h.wsHandler.Load().(*rpcHandler) + h.log.Info(fmt.Sprintf("loadWsHandler %s", where), "ws", ws, "h", h) + return ws +} + // setListenAddr configures the listening address of the server. // The address can only be set while the server isn't running. func (h *httpServer) setListenAddr(host string, port int) error { @@ -131,6 +145,7 @@ func (h *httpServer) listenAddr() string { func (h *httpServer) start() error { h.mu.Lock() defer h.mu.Unlock() + h.log.Info("RUN start()") if h.endpoint == "" || h.listener != nil { return nil // already running or not configured @@ -156,19 +171,31 @@ func (h *httpServer) start() error { return err } h.listener = listener - go h.server.Serve(listener) + // go func() { + // h.log.Info("STARTING RPC SERVER") + // h.server.Serve(listener) + // }() - if h.wsAllowed() { + if h.wsAllowed("start") { url := fmt.Sprintf("ws://%v", listener.Addr()) if h.wsConfig.prefix != "" { url += h.wsConfig.prefix } h.log.Info("WebSocket enabled", "url", url) + h.name = "ws" + // go func() { + // h.log.Info("STARTING RPC SERVER") + // h.server.Serve(listener) + // }() + } else { + h.name = "http" } - // if server is websocket only, return after logging - if !h.rpcAllowed() { - return nil - } + + go func() { + h.log.Info("STARTING RPC SERVER") + h.server.Serve(listener) + }() + // Log http endpoint. h.log.Info("HTTP server started", "endpoint", listener.Addr(), "auth", (h.httpConfig.jwtSecret != nil), @@ -177,6 +204,18 @@ func (h *httpServer) start() error { "vhosts", strings.Join(h.httpConfig.Vhosts, ","), ) + // if server is websocket only, return after logging + if !h.rpcAllowed() { + return nil + } + // // Log http endpoint. + // h.log.Info("HTTP server started", + // "endpoint", listener.Addr(), "auth", (h.httpConfig.jwtSecret != nil), + // "prefix", h.httpConfig.prefix, + // "cors", strings.Join(h.httpConfig.CorsAllowedOrigins, ","), + // "vhosts", strings.Join(h.httpConfig.Vhosts, ","), + // ) + // Log all handlers mounted on server. var paths []string for path := range h.handlerNames { @@ -191,17 +230,34 @@ func (h *httpServer) start() error { logged[name] = true } } + return nil } -func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { - // check if ws request and serve if ws enabled - ws := h.wsHandler.Load().(*rpcHandler) - if ws != nil && isWebsocket(r) { - if checkPath(r, h.wsConfig.prefix) { - ws.ServeHTTP(w, r) +func printHeaders(h *httpServer, headers http.Header) { + for key, values := range headers { + for _, value := range values { + h.log.Info("HEADER", key, value) } - return + } +} + +func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + + h.log.Info("REQUEST", "r", r) + ws := h.loadWsHandler("ServeHTTP") + h.log.Info("TRIAGE", "name", h.name, "ws", ws, "h", h, "isWebsocket", isWebsocket(r), "upgrade", r.Header.Get("Upgrade"), "connection", r.Header.Get("Connection")) + printHeaders(h, r.Header) + // check if ws request and serve if ws enabled + // if ws != nil && isWebsocket(r) { + if isWebsocket(r) { + if ws != nil { + if checkPath(r, h.wsConfig.prefix) { + ws.ServeHTTP(w, r) + } + return + } + panic("TRIAGE FAILED WS") } // if http-rpc is enabled, try to serve request @@ -266,13 +322,13 @@ func (h *httpServer) doStop() { // Shut down the server. httpHandler := h.httpHandler.Load().(*rpcHandler) - wsHandler := h.wsHandler.Load().(*rpcHandler) + wsHandler := h.loadWsHandler("doStop") if httpHandler != nil { h.httpHandler.Store((*rpcHandler)(nil)) httpHandler.server.Stop() } if wsHandler != nil { - h.wsHandler.Store((*rpcHandler)(nil)) + h.storeWsHandler((*rpcHandler)(nil), "doStop") wsHandler.server.Stop() } @@ -330,7 +386,7 @@ func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error { h.mu.Lock() defer h.mu.Unlock() - if h.wsAllowed() { + if h.wsAllowed("enableWS") { return fmt.Errorf("JSON-RPC over WebSocket is already enabled") } // Create RPC server and handler. @@ -340,10 +396,14 @@ func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error { return err } h.wsConfig = config - h.wsHandler.Store(&rpcHandler{ + + h.storeWsHandler(&rpcHandler{ Handler: NewWSHandlerStack(srv.WebsocketHandler(config.Origins), config.jwtSecret), server: srv, - }) + }, "enableWS") + // ws := h.loadWsHandler("enableWS") + // h.log.Info("enableWS wsHandler = YES", "ws", ws) + // h.startRoutineToCheckWsHandler() return nil } @@ -359,11 +419,21 @@ func (h *httpServer) stopWS() { } } +// func (h *httpServer) startRoutineToCheckWsHandler() { +// // Goroutine to check the value every second +// go func() { +// for range time.Tick(1 * time.Second) { +// ws := h.loadWsHandler() +// h.log.Info("check wsHandler", "ws", ws, "h", h) +// } +// }() +// } + // disableWS disables the WebSocket handler. This is internal, the caller must hold h.mu. func (h *httpServer) disableWS() bool { - ws := h.wsHandler.Load().(*rpcHandler) + ws := h.loadWsHandler("disableWS") if ws != nil { - h.wsHandler.Store((*rpcHandler)(nil)) + h.storeWsHandler((*rpcHandler)(nil), "disableWS") ws.server.Stop() } return ws != nil @@ -375,8 +445,8 @@ func (h *httpServer) rpcAllowed() bool { } // wsAllowed returns true when JSON-RPC over WebSocket is enabled. -func (h *httpServer) wsAllowed() bool { - return h.wsHandler.Load().(*rpcHandler) != nil +func (h *httpServer) wsAllowed(where string) bool { + return h.loadWsHandler(fmt.Sprintf("wsAllowed %s", where)) != nil } // isWebsocket checks the header of an http request for a websocket upgrade request.