graphql, node, rpc: improve HTTP write timeout handling

eth/filters/filter.go

@@ -234,6 +234,9 @@ func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, e
 	var logs []*types.Log
 
 	for ; f.begin <= int64(end); f.begin++ {
+		if f.begin%10 == 0 && ctx.Err() != nil {
+			return logs, ctx.Err()
+		}
 		header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin))
 		if header == nil || err != nil {
 			return logs, err

eth/tracers/api.go

@@ -547,6 +547,9 @@ func (api *API) IntermediateRoots(ctx context.Context, hash common.Hash, config
 		deleteEmptyObjects = chainConfig.IsEIP158(block.Number())
 	)
 	for i, tx := range block.Transactions() {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
 		var (
 			msg, _    = tx.AsMessage(signer, block.BaseFee())
 			txContext = core.NewEVMTxContext(msg)

graphql/graphql_test.go

@@ -321,10 +321,11 @@ func TestGraphQLTransactionLogs(t *testing.T) {
 
 func createNode(t *testing.T) *node.Node {
 	stack, err := node.New(&node.Config{
-		HTTPHost: "127.0.0.1",
-		HTTPPort: 0,
-		WSHost:   "127.0.0.1",
-		WSPort:   0,
+		HTTPHost:     "127.0.0.1",
+		HTTPPort:     0,
+		WSHost:       "127.0.0.1",
+		WSPort:       0,
+		HTTPTimeouts: node.DefaultConfig.HTTPTimeouts,
 	})
 	if err != nil {
 		t.Fatalf("could not create node: %v", err)

node/api_test.go

@@ -252,6 +252,9 @@ func TestStartRPC(t *testing.T) {
 			config := test.cfg
 			// config.Logger = testlog.Logger(t, log.LvlDebug)
 			config.P2P.NoDiscovery = true
+			if config.HTTPTimeouts == (rpc.HTTPTimeouts{}) {
+				config.HTTPTimeouts = rpc.DefaultHTTPTimeouts
+			}
 
 			// Create Node.
 			stack, err := New(&config)

node/node_test.go

@@ -559,13 +559,13 @@ func (test rpcPrefixTest) check(t *testing.T, node *Node) {
 	}
 
 	for _, path := range test.wantHTTP {
-		resp := rpcRequest(t, httpBase+path)
+		resp := rpcRequest(t, httpBase+path, testMethod)
 		if resp.StatusCode != 200 {
 			t.Errorf("Error: %s: bad status code %d, want 200", path, resp.StatusCode)
 		}
 	}
 	for _, path := range test.wantNoHTTP {
-		resp := rpcRequest(t, httpBase+path)
+		resp := rpcRequest(t, httpBase+path, testMethod)
 		if resp.StatusCode != 404 {
 			t.Errorf("Error: %s: bad status code %d, want 404", path, resp.StatusCode)
 		}
@@ -586,10 +586,11 @@ 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,
+		HTTPHost:     "127.0.0.1",
+		HTTPPort:     httpPort,
+		WSHost:       "127.0.0.1",
+		WSPort:       wsPort,
+		HTTPTimeouts: rpc.DefaultHTTPTimeouts,
 	}
 	node, err := New(conf)
 	if err != nil {

node/rpcstack.go

@@ -24,6 +24,7 @@ import (
 	"net"
 	"net/http"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -196,6 +197,7 @@ func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
+
 	// if http-rpc is enabled, try to serve request
 	rpc := h.httpHandler.Load().(*rpcHandler)
 	if rpc != nil {
@@ -462,17 +464,61 @@ var gzPool = sync.Pool{
 }
 
 type gzipResponseWriter struct {
-	io.Writer
-	http.ResponseWriter
+	gz   *gzip.Writer
+	resp http.ResponseWriter
+
+	contentLength uint64 // total length of the uncompressed response
+	written       uint64 // amount of written bytes from the uncompressed response
+	hasLength     bool   // true if uncompressed response had Content-Length
+	inited        bool   // true after initBuffering called first time
+}
+
+// initContentLength checks if the uncompressed response has a content-length header.
+func (w *gzipResponseWriter) initContentLength() {
+	if w.inited {
+		return
+	}
+	w.inited = true
+
+	hdr := w.resp.Header()
+	length := hdr.Get("content-length")
+	hdr.Del("content-length")
+	if len(length) > 0 {
+		if n, err := strconv.ParseUint(length, 10, 64); err != nil {
+			w.hasLength = true
+			w.contentLength = n
+		}
+	}
+}
+
+func (w *gzipResponseWriter) Header() http.Header {
+	return w.resp.Header()
 }
 
 func (w *gzipResponseWriter) WriteHeader(status int) {
-	w.Header().Del("Content-Length")
-	w.ResponseWriter.WriteHeader(status)
+	w.initContentLength()
+	w.resp.WriteHeader(status)
 }
 
 func (w *gzipResponseWriter) Write(b []byte) (int, error) {
-	return w.Writer.Write(b)
+	w.initContentLength()
+
+	n, err := w.gz.Write(b)
+	w.written += uint64(n)
+	if w.hasLength && w.written >= w.contentLength {
+		// The HTTP handler has finished writing the entire uncompressed response. Close
+		// the gzip stream to ensure the footer will be seen by the client if the response
+		// is flushed after this call to write.
+		err = w.gz.Close()
+	}
+	return n, err
+}
+
+func (w *gzipResponseWriter) Flush() {
+	w.gz.Flush()
+	if f, ok := w.resp.(http.Flusher); ok {
+		f.Flush()
+	}
 }
 
 func newGzipHandler(next http.Handler) http.Handler {
@@ -482,15 +528,15 @@ func newGzipHandler(next http.Handler) http.Handler {
 			return
 		}
 
-		w.Header().Set("Content-Encoding", "gzip")
+		w.Header().Set("content-encoding", "gzip")
 
 		gz := gzPool.Get().(*gzip.Writer)
 		defer gzPool.Put(gz)
-
 		gz.Reset(w)
 		defer gz.Close()
 
-		next.ServeHTTP(&gzipResponseWriter{ResponseWriter: w, Writer: gz}, r)
+		wrapper := &gzipResponseWriter{resp: w, gz: gz}
+		next.ServeHTTP(wrapper, r)
 	})
 }
 

node/rpcstack_test.go

@@ -19,6 +19,7 @@ package node
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -34,29 +35,31 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+const testMethod = "rpc_modules"
+
 // TestCorsHandler makes sure CORS are properly handled on the http server.
 func TestCorsHandler(t *testing.T) {
-	srv := createAndStartServer(t, &httpConfig{CorsAllowedOrigins: []string{"test", "test.com"}}, false, &wsConfig{})
+	srv := createAndStartServer(t, &httpConfig{CorsAllowedOrigins: []string{"test", "test.com"}}, false, &wsConfig{}, nil)
 	defer srv.stop()
 	url := "http://" + srv.listenAddr()
 
-	resp := rpcRequest(t, url, "origin", "test.com")
+	resp := rpcRequest(t, url, testMethod, "origin", "test.com")
 	assert.Equal(t, "test.com", resp.Header.Get("Access-Control-Allow-Origin"))
 
-	resp2 := rpcRequest(t, url, "origin", "bad")
+	resp2 := rpcRequest(t, url, testMethod, "origin", "bad")
 	assert.Equal(t, "", resp2.Header.Get("Access-Control-Allow-Origin"))
 }
 
 // TestVhosts makes sure vhosts are properly handled on the http server.
 func TestVhosts(t *testing.T) {
-	srv := createAndStartServer(t, &httpConfig{Vhosts: []string{"test"}}, false, &wsConfig{})
+	srv := createAndStartServer(t, &httpConfig{Vhosts: []string{"test"}}, false, &wsConfig{}, nil)
 	defer srv.stop()
 	url := "http://" + srv.listenAddr()
 
-	resp := rpcRequest(t, url, "host", "test")
+	resp := rpcRequest(t, url, testMethod, "host", "test")
 	assert.Equal(t, resp.StatusCode, http.StatusOK)
 
-	resp2 := rpcRequest(t, url, "host", "bad")
+	resp2 := rpcRequest(t, url, testMethod, "host", "bad")
 	assert.Equal(t, resp2.StatusCode, http.StatusForbidden)
 }
 
@@ -145,7 +148,7 @@ func TestWebsocketOrigins(t *testing.T) {
 		},
 	}
 	for _, tc := range tests {
-		srv := createAndStartServer(t, &httpConfig{}, true, &wsConfig{Origins: splitAndTrim(tc.spec)})
+		srv := createAndStartServer(t, &httpConfig{}, true, &wsConfig{Origins: splitAndTrim(tc.spec)}, nil)
 		url := fmt.Sprintf("ws://%v", srv.listenAddr())
 		for _, origin := range tc.expOk {
 			if err := wsRequest(t, url, "Origin", origin); err != nil {
@@ -231,11 +234,14 @@ func Test_checkPath(t *testing.T) {
 	}
 }
 
-func createAndStartServer(t *testing.T, conf *httpConfig, ws bool, wsConf *wsConfig) *httpServer {
+func createAndStartServer(t *testing.T, conf *httpConfig, ws bool, wsConf *wsConfig, timeouts *rpc.HTTPTimeouts) *httpServer {
 	t.Helper()
 
-	srv := newHTTPServer(testlog.Logger(t, log.LvlDebug), rpc.DefaultHTTPTimeouts)
-	assert.NoError(t, srv.enableRPC(nil, *conf))
+	if timeouts == nil {
+		timeouts = &rpc.DefaultHTTPTimeouts
+	}
+	srv := newHTTPServer(testlog.Logger(t, log.LvlDebug), *timeouts)
+	assert.NoError(t, srv.enableRPC(apis(), *conf))
 	if ws {
 		assert.NoError(t, srv.enableWS(nil, *wsConf))
 	}
@@ -266,16 +272,33 @@ func wsRequest(t *testing.T, url string, extraHeaders ...string) error {
 }
 
 // rpcRequest performs a JSON-RPC request to the given URL.
-func rpcRequest(t *testing.T, url string, extraHeaders ...string) *http.Response {
+func rpcRequest(t *testing.T, url, method string, extraHeaders ...string) *http.Response {
+	t.Helper()
+
+	body := fmt.Sprintf(`{"jsonrpc":"2.0","id":1,"method":"%s","params":[]}`, method)
+	return baseRpcRequest(t, url, body, extraHeaders...)
+}
+
+func batchRpcRequest(t *testing.T, url string, methods []string, extraHeaders ...string) *http.Response {
+	reqs := make([]string, len(methods))
+	for i, m := range methods {
+		reqs[i] = fmt.Sprintf(`{"jsonrpc":"2.0","id":1,"method":"%s","params":[]}`, m)
+	}
+	body := fmt.Sprintf(`[%s]`, strings.Join(reqs, ","))
+	return baseRpcRequest(t, url, body, extraHeaders...)
+}
+
+func baseRpcRequest(t *testing.T, url, bodyStr string, extraHeaders ...string) *http.Response {
 	t.Helper()
 
 	// Create the request.
-	body := bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"rpc_modules","params":[]}`))
+	body := bytes.NewReader([]byte(bodyStr))
 	req, err := http.NewRequest("POST", url, body)
 	if err != nil {
 		t.Fatal("could not create http request:", err)
 	}
 	req.Header.Set("content-type", "application/json")
+	req.Header.Set("accept-encoding", "identity")
 
 	// Apply extra headers.
 	if len(extraHeaders)%2 != 0 {
@@ -315,7 +338,7 @@ func TestJWT(t *testing.T) {
 		return ss
 	}
 	srv := createAndStartServer(t, &httpConfig{jwtSecret: []byte("secret")},
-		true, &wsConfig{Origins: []string{"*"}, jwtSecret: []byte("secret")})
+		true, &wsConfig{Origins: []string{"*"}, jwtSecret: []byte("secret")}, nil)
 	wsUrl := fmt.Sprintf("ws://%v", srv.listenAddr())
 	htUrl := fmt.Sprintf("http://%v", srv.listenAddr())
 
@@ -348,7 +371,7 @@ func TestJWT(t *testing.T) {
 			t.Errorf("test %d-ws, token '%v': expected ok, got %v", i, token, err)
 		}
 		token = tokenFn()
-		if resp := rpcRequest(t, htUrl, "Authorization", token); resp.StatusCode != 200 {
+		if resp := rpcRequest(t, htUrl, testMethod, "Authorization", token); resp.StatusCode != 200 {
 			t.Errorf("test %d-http, token '%v': expected ok, got %v", i, token, resp.StatusCode)
 		}
 	}
@@ -414,10 +437,69 @@ func TestJWT(t *testing.T) {
 		}
 
 		token = tokenFn()
-		resp := rpcRequest(t, htUrl, "Authorization", token)
+		resp := rpcRequest(t, htUrl, testMethod, "Authorization", token)
 		if resp.StatusCode != http.StatusUnauthorized {
 			t.Errorf("tc %d-http, token '%v': expected not to allow,  got %v", i, token, resp.StatusCode)
 		}
 	}
 	srv.stop()
 }
+
+func TestHTTPWriteTimeout(t *testing.T) {
+	const (
+		timeoutRes = `{"jsonrpc":"2.0","id":1,"error":{"code":-32002,"message":"request timed out"}}`
+		greetRes   = `{"jsonrpc":"2.0","id":1,"result":"Hello"}`
+	)
+	// Set-up server
+	timeouts := rpc.DefaultHTTPTimeouts
+	timeouts.WriteTimeout = time.Second
+	srv := createAndStartServer(t, &httpConfig{Modules: []string{"test"}}, false, &wsConfig{}, &timeouts)
+	url := fmt.Sprintf("http://%v", srv.listenAddr())
+
+	// Send normal request
+	t.Run("message", func(t *testing.T) {
+		resp := rpcRequest(t, url, "test_sleep")
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(body) != timeoutRes {
+			t.Errorf("wrong response. have %s, want %s", string(body), timeoutRes)
+		}
+	})
+
+	// Batch request
+	t.Run("batch", func(t *testing.T) {
+		want := fmt.Sprintf("[%s,%s,%s]", greetRes, timeoutRes, timeoutRes)
+		resp := batchRpcRequest(t, url, []string{"test_greet", "test_sleep", "test_greet"})
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(body) != want {
+			t.Errorf("wrong response. have %s, want %s", string(body), want)
+		}
+	})
+}
+
+func apis() []rpc.API {
+	return []rpc.API{
+		{
+			Namespace: "test",
+			Service:   &testService{},
+		},
+	}
+}
+
+type testService struct {
+}
+
+func (s *testService) Greet() string {
+	return "Hello"
+}
+
+func (s *testService) Sleep() {
+	time.Sleep(1500 * time.Millisecond)
+}

p2p/simulations/adapters/inproc.go

@@ -206,7 +206,7 @@ func (sn *SimNode) ServeRPC(conn *websocket.Conn) error {
 	if err != nil {
 		return err
 	}
-	codec := rpc.NewFuncCodec(conn, conn.WriteJSON, conn.ReadJSON)
+	codec := rpc.NewFuncCodec(conn, func(v any, _ bool) error { return conn.WriteJSON(v) }, conn.ReadJSON)
 	handler.ServeCodec(codec, 0)
 	return nil
 }