Fix JSON-RPC notification responses

internal/server/http_server/handlers.go

@@ -152,18 +152,30 @@ func (j *JsonRpcHandler) GetRequestType() protocol.RequestType {
 }
 
 func (j *JsonRpcHandler) RequestCount() int {
-	return len(j.jsonRpcRequests)
+	count := 0
+	for _, request := range j.jsonRpcRequests {
+		if len(bytes.TrimSpace(request.Id)) > 0 {
+			count++
+		}
+	}
+	return count
 }
 
 func (j *JsonRpcHandler) RequestDecode(ctx context.Context) (*Request, error) {
 	upstreamRequests := make([]protocol.RequestHolder, 0)
 
-	for i, jsonRpcReq := range j.jsonRpcRequests {
+	responseOrder := 0
+	for _, jsonRpcReq := range j.jsonRpcRequests {
 		id, err := uuid.NewUUID()
 		if err != nil {
 			return nil, err
 		}
-		j.idMap[id.String()] = lo.T2(jsonRpcReq.Id, i)
+		order := -1
+		if len(bytes.TrimSpace(jsonRpcReq.Id)) > 0 {
+			order = responseOrder
+			responseOrder++
+		}
+		j.idMap[id.String()] = lo.T2(jsonRpcReq.Id, order)
 
 		specName := chains.GetMethodSpecNameByChainName(j.preReq.Chain)
 		isSub := j.isWsCtx && specs.IsSubscribeMethod(specName, jsonRpcReq.Method)
@@ -194,5 +206,6 @@ func (j *JsonRpcHandler) ResponseEncode(response protocol.ResponseHolder) *Respo
 	return &Response{
 		ResponseReader: response.EncodeResponse(realId),
 		Order:          order,
+		Suppress:       ok && len(bytes.TrimSpace(idPair.A)) == 0,
 	}
 }

internal/server/http_server/handlers_test.go

@@ -1,6 +1,7 @@
 package http_server_test
 
 import (
+	"bytes"
 	"context"
 	"io"
 	"net/http"
@@ -16,6 +17,77 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestJsonRpcHandlerSuppressesNotificationResponses(t *testing.T) {
+	handler, err := http_server.NewJsonRpcHandler(
+		&http_server.Request{Chain: "ethereum"},
+		strings.NewReader(`{"jsonrpc":"2.0","method":"net_version","params":[]}`),
+		true,
+	)
+	require.NoError(t, err)
+	request, err := handler.RequestDecode(context.Background())
+	require.NoError(t, err)
+	require.Len(t, request.UpstreamRequests, 1)
+
+	response := handler.ResponseEncode(protocol.NewSimpleHttpUpstreamResponse(
+		request.UpstreamRequests[0].Id(), []byte(`"1"`), protocol.JsonRpc,
+	))
+
+	assert.True(t, response.Suppress)
+	assert.Equal(t, 0, handler.RequestCount())
+}
+
+func TestJsonRpcHandlerRespondsToExplicitNullId(t *testing.T) {
+	handler, err := http_server.NewJsonRpcHandler(
+		&http_server.Request{Chain: "ethereum"},
+		strings.NewReader(`{"jsonrpc":"2.0","id":null,"method":"net_version","params":[]}`),
+		true,
+	)
+	require.NoError(t, err)
+	request, err := handler.RequestDecode(context.Background())
+	require.NoError(t, err)
+
+	response := handler.ResponseEncode(protocol.NewSimpleHttpUpstreamResponse(
+		request.UpstreamRequests[0].Id(), []byte(`"1"`), protocol.JsonRpc,
+	))
+	body, err := io.ReadAll(response.ResponseReader)
+	require.NoError(t, err)
+
+	assert.False(t, response.Suppress)
+	assert.Equal(t, 1, handler.RequestCount())
+	assert.JSONEq(t, `{"id":null,"jsonrpc":"2.0","result":"1"}`, string(body))
+}
+
+func TestJsonRpcHandlerOrdersMixedBatchWithoutNotifications(t *testing.T) {
+	handler, err := http_server.NewJsonRpcHandler(
+		&http_server.Request{Chain: "ethereum"},
+		strings.NewReader(`[
+			{"jsonrpc":"2.0","method":"net_version","params":[]},
+			{"jsonrpc":"2.0","id":7,"method":"eth_chainId","params":[]}
+		]`),
+		false,
+	)
+	require.NoError(t, err)
+	request, err := handler.RequestDecode(context.Background())
+	require.NoError(t, err)
+	require.Len(t, request.UpstreamRequests, 2)
+
+	notification := handler.ResponseEncode(protocol.NewSimpleHttpUpstreamResponse(
+		request.UpstreamRequests[0].Id(), []byte(`"1"`), protocol.JsonRpc,
+	))
+	regular := handler.ResponseEncode(protocol.NewSimpleHttpUpstreamResponse(
+		request.UpstreamRequests[1].Id(), []byte(`"0x1"`), protocol.JsonRpc,
+	))
+
+	assert.True(t, notification.Suppress)
+	assert.Equal(t, -1, notification.Order)
+	assert.False(t, regular.Suppress)
+	assert.Equal(t, 0, regular.Order)
+	assert.Equal(t, 1, handler.RequestCount())
+	regularBody, err := io.ReadAll(regular.ResponseReader)
+	require.NoError(t, err)
+	assert.True(t, bytes.Contains(regularBody, []byte(`"id":7`)))
+}
+
 // TestMain loads a dedicated test_specs/ fixture so the REST parser tests
 // can exercise wildcard / multi-verb / literal-beats-wildcard cases without
 // depending on whatever the embedded production specs happen to declare.

internal/server/http_server/http_server.go

@@ -61,6 +61,7 @@ type Request struct {
 type Response struct {
 	ResponseReader io.Reader
 	Order          int
+	Suppress       bool
 }
 
 type FastJSONSerializer struct{}
@@ -204,10 +205,24 @@ func handleResponse(
 	code := http.StatusOK
 	httpResponse := reqCtx.Response()
 	if !requestHandler.IsSingle() {
-		responses := utils.Map(handleResp.responseWrappers, func(wrapper *protocol.ResponseHolderWrapper) *Response {
-			return requestHandler.ResponseEncode(wrapper.Response)
-		})
-		responseReader = ArraySortingStream(ctx, responses, requestHandler.RequestCount())
+		expected := requestHandler.RequestCount()
+		if expected == 0 {
+			for range handleResp.responseWrappers {
+			}
+			code = http.StatusNoContent
+		} else {
+			responses := make(chan *Response)
+			go func() {
+				defer close(responses)
+				for wrapper := range handleResp.responseWrappers {
+					response := requestHandler.ResponseEncode(wrapper.Response)
+					if !response.Suppress {
+						responses <- response
+					}
+				}
+			}()
+			responseReader = ArraySortingStream(ctx, responses, expected)
+		}
 	} else {
 		select {
 		case <-ctx.Done():
@@ -219,11 +234,16 @@ func handleResponse(
 			)
 		case responseWrapper, ok := <-handleResp.responseWrappers:
 			if ok {
-				httpResponse.Header().Set("response-provider", responseWrapper.UpstreamId)
-				code = protocol.ToHttpCode(responseWrapper.Response)
-				responseReader = requestHandler.ResponseEncode(responseWrapper.Response).ResponseReader
-
-				copyUpstreamResponseHeaders(httpResponse.Header(), responseWrapper.Response)
+				response := requestHandler.ResponseEncode(responseWrapper.Response)
+				if response.Suppress {
+					code = http.StatusNoContent
+				} else {
+					httpResponse.Header().Set("response-provider", responseWrapper.UpstreamId)
+					code = protocol.ToHttpCode(responseWrapper.Response)
+					responseReader = response.ResponseReader
+
+					copyUpstreamResponseHeaders(httpResponse.Header(), responseWrapper.Response)
+				}
 			}
 		}
 	}
@@ -266,6 +286,10 @@ func setCorsHeaders(reqCtx echo.Context, corsOrigins []string) {
 }
 
 func writeResponse(httpResponse *echo.Response, code int, responseReader io.Reader) error {
+	if responseReader == nil {
+		httpResponse.WriteHeader(code)
+		return nil
+	}
 	if httpResponse.Header().Get(echo.HeaderContentType) == "" {
 		httpResponse.Header().Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
 	}

internal/server/http_server/http_server_test.go

@@ -7,17 +7,102 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/drpcorg/nodecore/internal/server/http_server"
 	servernodecore "github.com/drpcorg/nodecore/internal/server/server_ctx"
 	"github.com/drpcorg/nodecore/pkg/chains"
 	"github.com/drpcorg/nodecore/pkg/test_utils"
 	"github.com/drpcorg/nodecore/pkg/test_utils/mocks"
+	"github.com/gorilla/websocket"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 )
 
+func TestHttpJsonRpcNotificationReturnsNoContent(t *testing.T) {
+	upSup := mocks.NewUpstreamSupervisorMock()
+	authProc := mocks.NewMockAuthProcessor()
+	appCtx := servernodecore.NewApplicationServerContext(upSup, nil, nil, authProc, nil, nil, nil, nil, nil, nil)
+	server := http_server.NewHttpServer(context.Background(), appCtx)
+	ts := httptest.NewServer(server)
+	defer ts.Close()
+
+	authProc.On("Authenticate", mock.Anything, mock.Anything).Return(nil)
+	authProc.On("PreKeyValidate", mock.Anything, mock.Anything).Return(nil, nil)
+	upSup.On("GetChainSupervisor", chains.POLYGON).Return(nil)
+
+	request, err := http.NewRequest(http.MethodPost, ts.URL+"/queries/polygon", strings.NewReader(
+		`{"jsonrpc":"2.0","method":"net_version","params":[]}`,
+	))
+	require.NoError(t, err)
+	response, err := http.DefaultClient.Do(request)
+	require.NoError(t, err)
+	defer response.Body.Close()
+	body, err := io.ReadAll(response.Body)
+	require.NoError(t, err)
+
+	assert.Equal(t, http.StatusNoContent, response.StatusCode)
+	assert.Empty(t, body)
+}
+
+func TestWebsocketJsonRpcNotificationIsSilentAndConnectionStaysOpen(t *testing.T) {
+	upSup := mocks.NewUpstreamSupervisorMock()
+	authProc := mocks.NewMockAuthProcessor()
+	appCtx := servernodecore.NewApplicationServerContext(upSup, nil, nil, authProc, nil, nil, nil, nil, nil, nil)
+	server := http_server.NewHttpServer(context.Background(), appCtx)
+	ts := httptest.NewServer(server)
+	defer ts.Close()
+
+	authProc.On("Authenticate", mock.Anything, mock.Anything).Return(nil)
+	authProc.On("PreKeyValidate", mock.Anything, mock.Anything).Return(nil, nil).Twice()
+	upSup.On("GetChainSupervisor", chains.POLYGON).Return(nil).Twice()
+
+	wsURL, err := url.Parse(ts.URL)
+	require.NoError(t, err)
+	wsURL.Scheme = "ws"
+	wsURL.Path = "/queries/polygon"
+	conn, _, err := websocket.DefaultDialer.Dial(wsURL.String(), nil)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(
+		`{"jsonrpc":"2.0","method":"net_version","params":[]}`,
+	)))
+	messages := make(chan []byte, 1)
+	readErrors := make(chan error, 1)
+	go func() {
+		_, message, readErr := conn.ReadMessage()
+		if readErr != nil {
+			readErrors <- readErr
+			return
+		}
+		messages <- message
+	}()
+	select {
+	case message := <-messages:
+		t.Fatalf("notification unexpectedly produced a websocket frame: %s", message)
+	case err := <-readErrors:
+		t.Fatalf("websocket closed after notification: %v", err)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(
+		`{"jsonrpc":"2.0","id":7,"method":"net_version","params":[]}`,
+	)))
+	select {
+	case message := <-messages:
+		assert.JSONEq(t, `{"id":7,"jsonrpc":"2.0","error":{"message":"no available upstreams to process a request","code":1}}`, string(message))
+	case err := <-readErrors:
+		t.Fatalf("websocket closed before regular response: %v", err)
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for regular websocket response")
+	}
+}
+
 func TestHttpServerOptionsRequest(t *testing.T) {
 	tests := []struct {
 		name    string

internal/server/http_server/ws_server.go

@@ -131,15 +131,18 @@ loop:
 								return
 							}
 						}
+						encoded := requestHandler.ResponseEncode(response.Response)
+						if encoded.Suppress {
+							continue
+						}
 						writeEvent := func() {
 							wsLock.Lock()
 							defer wsLock.Unlock()
 							writer, err := conn.NextWriter(message.messageType)
 							if err != nil {
 								log.Error().Err(err).Msg("couldn't get writer to send a response")
 							} else {
-								resp := requestHandler.ResponseEncode(response.Response)
-								if _, err = io.Copy(writer, resp.ResponseReader); err != nil {
+								if _, err = io.Copy(writer, encoded.ResponseReader); err != nil {
 									log.Error().Err(err).Msg("couldn't copy message")
 								}
 								if err = writer.Close(); err != nil {