fixup! feat(jsonrpc): add support for JSON-RPC 2.0 batch requests

Author: renatomaia

internal/jsonrpc/batchcalls_test.go

@@ -283,6 +283,45 @@ func TestJSONRPCBatchStopsBetweenEntriesWhenContextIsCanceled(t *testing.T) {
 	}
 }
 
+func TestJSONRPCBatchReturnsErrorsForIDDRequestsAfterDeadline(t *testing.T) {
+	s := newBatchTestService()
+	var calls atomic.Int32
+	const method = "test_deadline_batch"
+	withTestRPCHandler(t, method, func(_ *Service, _ *http.Request, _ RPCRequest) (any, error) {
+		calls.Add(1)
+		return true, nil
+	})
+
+	body := []byte(fmt.Sprintf(`[
+		{"jsonrpc":"2.0","method":%q,"id":1},
+		{"jsonrpc":"2.0","method":%q},
+		{"jsonrpc":"2.0","method":%q,"id":"three"},
+		false,
+		{"jsonrpc":"2.0","method":%q,"id":null}
+	]`, method, method, method, method))
+	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
+	defer cancel()
+	req := httptest.NewRequest(http.MethodPost, "/rpc", bytes.NewReader(body)).WithContext(ctx)
+	rr := httptest.NewRecorder()
+	s.handleRPC(rr, req)
+
+	require.Zero(t, calls.Load(), "expired batch entries must not be dispatched")
+	responses := decodeRPCBatch(t, rr.Body.Bytes())
+	require.Len(t, responses, 5, "not all entries receive deadline errors")
+	requireRPCError(t, responses[0], float64(1), JSONRPC_RATE_LIMIT_EXCEEDED)
+	requireRPCError(t, responses[1], nil, JSONRPC_RATE_LIMIT_EXCEEDED)
+	requireRPCError(t, responses[2], "three", JSONRPC_RATE_LIMIT_EXCEEDED)
+	requireRPCError(t, responses[3], nil, JSONRPC_INVALID_REQUEST)
+	requireRPCError(t, responses[4], nil, JSONRPC_RATE_LIMIT_EXCEEDED)
+	for i, response := range responses {
+		if i == 3 {
+			require.Equal(t, "invalid request", response.Error.Message)
+		} else {
+			require.Equal(t, "batch deadline exceeded", response.Error.Message)
+		}
+	}
+}
+
 func TestJSONRPCBatchUsesOneAdmissionPermit(t *testing.T) {
 	s := newBatchTestService()
 	s.admission = service.NewSemaphoreAdmission(1)

internal/jsonrpc/jsonrpc.go

@@ -42,10 +42,11 @@ const (
 	// but the requested entity does not; unknown applications use
 	// JSONRPC_APPLICATION_NOT_FOUND. For forward-looking keys, this can be the
 	// "not created yet" signal and may be safe to poll depending on the method.
-	JSONRPC_RESOURCE_NOT_FOUND int = -32001 //nolint: revive
+	JSONRPC_RESOURCE_NOT_FOUND int = -31001 //nolint: revive
 	// Application not found: the application identifier itself is unknown to
 	// this node. A configuration error that will not resolve by retrying.
-	JSONRPC_APPLICATION_NOT_FOUND int = -32002 //nolint: revive
+	JSONRPC_APPLICATION_NOT_FOUND int = -31002 //nolint: revive
+	JSONRPC_RATE_LIMIT_EXCEEDED   int = -32002 //nolint: revive
 	JSONRPC_PARSE_ERROR           int = -32700 //nolint: revive
 	JSONRPC_INVALID_REQUEST       int = -32600 //nolint: revive
 	JSONRPC_METHOD_NOT_FOUND      int = -32601 //nolint: revive
@@ -195,40 +196,44 @@ func (s *Service) handleRPC(w http.ResponseWriter, r *http.Request) {
 		}
 
 		budgetResp := newBudgetWriter(w, MAX_RESPONSE_SIZE)
-	reqLoop:
 		for i, rawReq := range reqSeq {
 
-			switch r.Context().Err() {
-			case context.DeadlineExceeded:
-				s.Logger.Warn("RPC method dispatch timeout")
-				fallthrough
-			case context.Canceled:
-				break reqLoop
-			}
-
 			if i > 0 && !s.writeByte(w, ',') {
 				return
 			}
 
 			var responded bool
 			var req RPCRequest
-			if err := json.Unmarshal(rawReq, &req); err != nil {
-				responded = s.writeRPCError(w, nil, JSONRPC_INVALID_REQUEST, "invalid request")
-			} else {
-				s.Logger.Debug(fmt.Sprintf("Dispatching RPC request: %s", req.Method))
-				buffer := budgetResp.NewLimitedWriter()
-				if buffer == nil {
-					responded = s.writeRPCError(w, req.ID, JSONRPC_INVALID_REQUEST, "response size budget exceeded")
+
+			switch r.Context().Err() {
+			case context.Canceled:
+				return
+			case context.DeadlineExceeded:
+				s.Logger.Warn("RPC method dispatch timeout")
+				if err := json.Unmarshal(rawReq, &req); err != nil {
+					responded = s.writeRPCError(w, nil, JSONRPC_INVALID_REQUEST, "invalid request")
+				} else {
+					responded = s.writeRPCError(w, req.ID, JSONRPC_RATE_LIMIT_EXCEEDED, "batch deadline exceeded")
+				}
+			default:
+				if err := json.Unmarshal(rawReq, &req); err != nil {
+					responded = s.writeRPCError(w, nil, JSONRPC_INVALID_REQUEST, "invalid request")
 				} else {
-					err := s.dispatchOneRequest(buffer, r, req)
-					switch {
-					case err == nil:
-						responded = s.handleResponseResult(w, buffer.Flush())
-					case errors.Is(err, io.ErrShortBuffer):
+					s.Logger.Debug(fmt.Sprintf("Dispatching RPC request: %s", req.Method))
+					buffer := budgetResp.NewLimitedWriter()
+					if buffer == nil {
 						responded = s.writeRPCError(w, req.ID, JSONRPC_INVALID_REQUEST, "response size budget exceeded")
-					default:
-						s.Logger.Error("RPC method response encode failed", "method", req.Method, "error", err)
-						responded = s.writeRPCError(w, req.ID, JSONRPC_INTERNAL_ERROR, "Internal server error")
+					} else {
+						err := s.dispatchOneRequest(buffer, r, req)
+						switch {
+						case err == nil:
+							responded = s.handleResponseResult(w, buffer.Flush())
+						case errors.Is(err, io.ErrShortBuffer):
+							responded = s.writeRPCError(w, req.ID, JSONRPC_INVALID_REQUEST, "response size budget exceeded")
+						default:
+							s.Logger.Error("RPC method response encode failed", "method", req.Method, "error", err)
+							responded = s.writeRPCError(w, req.ID, JSONRPC_INTERNAL_ERROR, "Internal server error")
+						}
 					}
 				}
 			}