fix: add WebSocket endpoint registration and message type tests (#20) (#356)

* ci: add workflow to auto-add issues to project board

Automatically adds newly opened issues to the EVStack Evolve
project board (orgs/evstack/projects/7).

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* swarm: add WebSocket endpoint registration and message type handling tests (#20)

Critical fix: RegisterRPCFuncs does not register a WebSocket endpoint.
ev-abci was missing /websocket — clients could not connect via WebSocket at
all. Fix adds an explicit WebsocketManager registration in startRPC.

Documentation added to startRPC explaining RFC 6455 §11.8 message type
handling as delegated to CometBFT's wsConnection (gorilla/websocket).

New pkg/rpc/server_test.go with 7 integration tests:
- TestWebSocket_TextValidJSON: health request over WS returns success
- TestWebSocket_TextInvalidJSON: malformed JSON yields parse error (-32700)
- TestWebSocket_BinaryFrame: binary frames yield parse error, not crash
- TestWebSocket_CloseFrame: normal close (1000) closes connection cleanly
- TestWebSocket_PingPong: server-initiated pings received within timeout
- TestWebSocket_LargeMessage: exceeding read limit yields CloseMessageTooBig
- TestRPCServer_WebSocketEndpointRegistered: end-to-end /websocket works

Note: client-initiated ping test uses server-initiated pings instead to
avoid a pre-existing data race in CometBFT's ws_handler.go (writeRoutine
sets SetPingHandler concurrent with readRoutine's first advanceFrame call).

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>

* swarm: fix LargeMessage test assertion for close code specificity (#20)

IsUnexpectedCloseError(err, 1009) returns true when the code is NOT
1009, so the OR with IsCloseError made the assertion pass for any close
frame. Replace with IsCloseError(err, CloseMessageTooBig) alone to
enforce the specific code 1009.

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>

* build: promote gorilla/websocket to direct dependency

go mod tidy moves gorilla/websocket from indirect to direct now that
the WebSocket test code imports it explicitly.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: wire OnDisconnect and ReadLimit into WebSocket manager

Without OnDisconnect, disconnected WebSocket clients leave dangling
subscriptions on the EventBus. Pass the EventBus to RPCServer so the
WebSocket manager can call UnsubscribeAll on disconnect, matching
CometBFT's node.go pattern. Also set ReadLimit from RPC config.

Closes #18

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* test: add OnDisconnect subscription cleanup test

Verify that when a WebSocket client subscribes to events and then
disconnects, the OnDisconnect callback fires and removes all
subscriptions for that client from the EventBus.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

---------

Co-authored-by: Claude Opus 4.6 <[email protected]>

Author: tac0turtle

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/evstack/ev-node/core v1.0.0-rc.1.0.20260216131057-1da76345e4b9
 	github.com/go-kit/kit v0.13.0
 	github.com/golang/protobuf v1.5.4
+	github.com/gorilla/websocket v1.5.3
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/hashicorp/go-metrics v0.5.4
 	github.com/ipfs/go-datastore v0.9.1
@@ -154,7 +155,6 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/handlers v1.5.2 // indirect
 	github.com/gorilla/mux v1.8.1 // indirect
-	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7 // indirect
 	github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect

pkg/rpc/server.go

@@ -11,7 +11,9 @@ import (
 	"cosmossdk.io/log"
 	cmtcfg "github.com/cometbft/cometbft/config"
 	cmtlog "github.com/cometbft/cometbft/libs/log"
+	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
 	rpcserver "github.com/cometbft/cometbft/rpc/jsonrpc/server"
+	cmttypes "github.com/cometbft/cometbft/types"
 	servercmtlog "github.com/cosmos/cosmos-sdk/server/log"
 	"github.com/rs/cors"
 	"golang.org/x/net/netutil"
@@ -23,11 +25,13 @@ import (
 func NewRPCServer(
 	cfg *cmtcfg.RPCConfig,
 	logger log.Logger,
+	eventBus *cmttypes.EventBus,
 ) *RPCServer {
 	cmtLogger := servercmtlog.CometLoggerWrapper{Logger: logger}
 	return &RPCServer{
-		config: cfg,
-		logger: cmtLogger,
+		config:   cfg,
+		logger:   cmtLogger,
+		eventBus: eventBus,
 	}
 }
 
@@ -38,13 +42,31 @@ type RPCServer struct {
 	httpHandler http.Handler
 	server      http.Server
 	logger      cmtlog.Logger
+	eventBus    *cmttypes.EventBus
 }
 
 // Start starts the RPC server.
 func (r *RPCServer) Start() error {
 	return r.startRPC()
 }
 
+// startRPC starts the RPC server and registers all HTTP, JSON-RPC, and
+// WebSocket endpoints.
+//
+// WebSocket support (issue #20, RFC 6455 §11.8):
+//
+// CometBFT's RegisterRPCFuncs only registers:
+//   - GET/POST /<method>  — per-method HTTP handlers
+//   - POST /              — batch JSON-RPC handler (HTTP only; no WS upgrade)
+//
+// WebSocket connectivity is provided by a separate WebsocketManager registered
+// explicitly at /websocket. Message-type handling is delegated to CometBFT's
+// wsConnection (gorilla/websocket-based):
+//   - Text   (0x1): JSON-RPC request → method dispatch → JSON response
+//   - Binary (0x2): JSON decode attempted; fails with a parse-error response
+//   - Close  (0x8): connection closed cleanly via IsCloseError check
+//   - Ping   (0x9): pong sent automatically by gorilla/websocket default handler
+//   - Pong   (0xA): resets the read deadline (handled via SetPongHandler)
 func (r *RPCServer) startRPC() error {
 	if r.config.ListenAddress == "" {
 		r.logger.Info("listen address not specified - RPC will not be exposed")
@@ -64,6 +86,23 @@ func (r *RPCServer) startRPC() error {
 
 	mux := http.NewServeMux()
 	rpcserver.RegisterRPCFuncs(mux, core.Routes, r.logger)
+
+	// RegisterRPCFuncs does not register a WebSocket endpoint.
+	// We must register it separately so clients can connect via /websocket.
+	wmLogger := r.logger.With("protocol", "websocket")
+	wm := rpcserver.NewWebsocketManager(
+		core.Routes,
+		rpcserver.OnDisconnect(func(remoteAddr string) {
+			err := r.eventBus.UnsubscribeAll(context.Background(), remoteAddr)
+			if err != nil && err != cmtpubsub.ErrSubscriptionNotFound {
+				wmLogger.Error("Failed to unsubscribe addr from events", "addr", remoteAddr, "err", err)
+			}
+		}),
+		rpcserver.ReadLimit(r.config.MaxBodyBytes),
+	)
+	wm.SetLogger(wmLogger)
+	mux.HandleFunc("/websocket", wm.WebsocketHandler)
+
 	r.httpHandler = mux
 
 	if r.config.MaxOpenConnections != 0 {