From bf03916b30db7e40256ca0c2aa7b8501ce6d6cd9 Mon Sep 17 00:00:00 2001 From: Ri-go Date: Tue, 6 Oct 2026 03:17:56 +0800 Subject: [PATCH] Fix incomplete upstream response handling --- internal/forwarder/http.go | 49 ++++- internal/forwarder/http_test.go | 16 +- internal/forwarder/truncation_test.go | 238 ++++++++++++++++++++ internal/server/capture.go | 8 + internal/server/cmt_rpc/truncation_test.go | 240 +++++++++++++++++++++ internal/server/cmt_rpc/websocket.go | 9 + internal/server/eth_rpc/server.go | 18 ++ internal/server/eth_rpc/truncation_test.go | 199 +++++++++++++++++ 8 files changed, 765 insertions(+), 12 deletions(-) create mode 100644 internal/forwarder/truncation_test.go create mode 100644 internal/server/cmt_rpc/truncation_test.go create mode 100644 internal/server/eth_rpc/truncation_test.go diff --git a/internal/forwarder/http.go b/internal/forwarder/http.go index 4825ffa..6b04a21 100644 --- a/internal/forwarder/http.go +++ b/internal/forwarder/http.go @@ -68,11 +68,23 @@ func (e *ErrAllAttemptsFailed) Error() string { func (e *ErrAllAttemptsFailed) Unwrap() error { return e.Last } +// bufferedResponse has not committed anything to a client. ResetResponse +// discards an attempt's status and body and restores the pre-attempt headers, +// allowing a retry or a complete error response without leaking partial data. +type bufferedResponse interface { + ResetResponse(http.Header) +} + // Forward proxies r to upstream, choosing an endpoint URL per candidate by // joining the candidate's protocol-specific base with r.URL.Path + r.URL.RawQuery. // // The original r.Body is buffered once so that retries can replay it. func (f *HTTP) Forward(w http.ResponseWriter, r *http.Request, key types.RouteKey) { + buffered, canReset := w.(bufferedResponse) + var initialHeaders http.Header + if canReset { + initialHeaders = w.Header().Clone() + } candidates := f.selector.Candidates(key) if len(candidates) == 0 { writeJSONError(w, http.StatusServiceUnavailable, "no eligible backend") @@ -203,20 +215,43 @@ func (f *HTTP) Forward(w http.ResponseWriter, r *http.Request, key types.RouteKe cancel() if body.err != nil { - // Upstream died mid-body. Headers (and part of the body) are - // already written, so the response cannot be salvaged — debit - // the breaker and stop. The request still terminated through - // this backend: count it in the primary traffic metrics. + if canReset { + buffered.ResetResponse(initialHeaders) + } + if r.Context().Err() != nil { + // A downstream cancellation can interrupt the upstream read. + // It says nothing about the backend's health. + f.circuit.Release(b.Name, key.Protocol) + if canReset { + writeJSONError(w, http.StatusBadGateway, "request canceled") + return + } + panic(http.ErrAbortHandler) + } f.circuit.Record(b.Name, key.Protocol, false) metrics.RelayTruncated.WithLabelValues(b.Name, string(key.Protocol)).Inc() - metrics.RequestsTotal.WithLabelValues(string(key.Protocol), key.Class.String(), b.Name, statusBucket(resp.StatusCode)).Inc() - metrics.RequestDuration.WithLabelValues(string(key.Protocol), key.Class.String(), b.Name).Observe(dur.Seconds()) log.FromCtx(r.Context()).Warn("upstream body truncated mid-relay", "backend", b.Name, "err", log.ErrorMessage(body.err), "attempt", attempts, ) - return + if !canReset { + // Only the streamed response actually exposed this status. + // Buffered attempts count below as failures, with a single + // final request outcome recorded on success or exhaustion. + metrics.RequestsTotal.WithLabelValues(string(key.Protocol), key.Class.String(), b.Name, statusBucket(resp.StatusCode)).Inc() + metrics.RequestDuration.WithLabelValues(string(key.Protocol), key.Class.String(), b.Name).Observe(time.Since(started).Seconds()) + // The real HTTP writer may already have sent headers/body. + // Abort the connection (or HTTP/2 stream), so net/http cannot + // turn this partial response into a cleanly finished success. + panic(http.ErrAbortHandler) + } + lastErr = body.err + lastHistorical = nil + metrics.FailoverAttempts.WithLabelValues(b.Name, "next", classifyErr(body.err)).Inc() + // The existing candidate loop enforces coverage, idempotency, + // cancellation and MaxAttempts. No bytes have left the buffer. + continue } if copyErr != nil { // Only the client-side write failed; the upstream served fine. diff --git a/internal/forwarder/http_test.go b/internal/forwarder/http_test.go index f2a81cd..0510f3c 100644 --- a/internal/forwarder/http_test.go +++ b/internal/forwarder/http_test.go @@ -280,16 +280,22 @@ func TestForwardTruncatedUpstreamRecordsFailure(t *testing.T) { cm := newTestCircuit() fwd := newForwarderWithCircuit(stubSelector{cands: []*backend.Backend{mkBackend("trunc", upstream.URL)}}, cm, 1) + front := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fwd.Forward(w, r, types.RouteKey{Protocol: types.ProtoRPC, Method: "status", Class: types.ClassLatest, Idempotent: true}) + })) + defer front.Close() before := testutil.ToFloat64(metrics.RelayTruncated.WithLabelValues("trunc", string(types.ProtoRPC))) reqBefore := testutil.ToFloat64(metrics.RequestsTotal.WithLabelValues(string(types.ProtoRPC), types.ClassLatest.String(), "trunc", "2xx")) durBefore := requestDurationCount(t, string(types.ProtoRPC), types.ClassLatest.String(), "trunc") for i := 0; i < 2; i++ { - rec := httptest.NewRecorder() - r := httptest.NewRequest(http.MethodGet, "/status", nil) - fwd.Forward(rec, r, types.RouteKey{Protocol: types.ProtoRPC, Method: "status", Class: types.ClassLatest, Idempotent: true}) - if rec.Code != 200 { - t.Fatalf("headers were already relayed; expected 200, got %d", rec.Code) + resp, err := front.Client().Get(front.URL + "/status") + if err == nil { + _, readErr := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if readErr == nil { + t.Fatal("truncated relay finished without a transport error") + } } } diff --git a/internal/forwarder/truncation_test.go b/internal/forwarder/truncation_test.go new file mode 100644 index 0000000..ee571d4 --- /dev/null +++ b/internal/forwarder/truncation_test.go @@ -0,0 +1,238 @@ +package forwarder + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + + "github.com/InjectiveLabs/stitch/internal/backend" + "github.com/InjectiveLabs/stitch/internal/circuit" + "github.com/InjectiveLabs/stitch/internal/metrics" + "github.com/InjectiveLabs/stitch/internal/pool" + "github.com/InjectiveLabs/stitch/internal/server" + "github.com/InjectiveLabs/stitch/internal/types" +) + +func TestStreamedTruncationAbortsHTTP1AndHTTP2(t *testing.T) { + for _, h2 := range []bool{false, true} { + name := "HTTP1" + if h2 { + name = "HTTP2" + } + t.Run(name, func(t *testing.T) { + prefix := []byte(strings.Repeat("x", 128<<10)) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/healthy" { + _, _ = w.Write([]byte(`{"ok":true}`)) + return + } + _, _ = w.Write(prefix) + w.(http.Flusher).Flush() + panic(http.ErrAbortHandler) + })) + defer upstream.Close() + var retryCalls atomic.Int32 + next := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + retryCalls.Add(1) + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer next.Close() + fwd := newForwarder(stubSelector{cands: []*backend.Backend{mkBackend("stream", upstream.URL), mkBackend("next", next.URL)}}) + front := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fwd.Forward(w, r, types.RouteKey{Protocol: types.ProtoRPC, Idempotent: true}) + })) + front.EnableHTTP2 = h2 + front.StartTLS() + defer front.Close() + resp, err := front.Client().Get(front.URL + "/broken") + if err != nil { + t.Fatal(err) // The large prefix commits headers before the abort. + } + body, readErr := io.ReadAll(resp.Body) + _ = resp.Body.Close() + wantProto := 1 + if h2 { + wantProto = 2 + } + if resp.ProtoMajor != wantProto || readErr == nil || len(body) == 0 { + t.Fatalf("expected partial body plus transport error over %s: protocol=%s bytes=%d err=%v", name, resp.Proto, len(body), readErr) + } + if retryCalls.Load() != 0 { + t.Fatal("must not append a retry to a committed response") + } + resp, err = front.Client().Get(front.URL + "/healthy") + if err != nil { + t.Fatal(err) + } + body, err = io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil || string(body) != `{"ok":true}` { + t.Fatalf("server unusable after abort: body=%q err=%v", body, err) + } + }) + } +} + +func TestBufferedNonIdempotentTruncationDoesNotRetry(t *testing.T) { + var nextCalls atomic.Int32 + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", "1000") + w.Header().Set("Content-Encoding", "identity") + _, _ = w.Write([]byte(`{"result":`)) + })) + defer bad.Close() + next := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + nextCalls.Add(1) + _, _ = w.Write([]byte(`{"result":true}`)) + })) + defer next.Close() + fwd := newForwarder(stubSelector{cands: []*backend.Backend{mkBackend("write", bad.URL), mkBackend("next-write", next.URL)}}) + front := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + captured := server.NewCapture(w.Header()) + fwd.Forward(captured, r, types.RouteKey{Protocol: types.ProtoRPC, Idempotent: false}) + captured.FlushTo(w) + })) + defer front.Close() + resp, err := front.Client().Post(front.URL, "application/json", strings.NewReader(`{"method":"write"}`)) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil || resp.StatusCode != http.StatusBadGateway || !json.Valid(body) || bytes.Contains(body, []byte(`"result"`)) { + t.Fatalf("response: status=%d body=%q err=%v", resp.StatusCode, body, err) + } + if resp.Header.Get("Content-Encoding") != "" || nextCalls.Load() != 0 { + t.Fatalf("stale encoding=%q retries=%d", resp.Header.Get("Content-Encoding"), nextCalls.Load()) + } +} + +type notifyingCapture struct { + *server.Capture + written chan struct{} +} + +func (c *notifyingCapture) Write(body []byte) (int, error) { + n, err := c.Capture.Write(body) + select { + case c.written <- struct{}{}: + default: + } + return n, err +} + +func TestBufferedBodyCancellationReleasesCircuitAndStopsRetries(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"result":`)) + w.(http.Flusher).Flush() + <-r.Context().Done() + })) + defer upstream.Close() + var nextCalls atomic.Int32 + next := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + nextCalls.Add(1) + })) + defer next.Close() + cm := circuit.NewManager(circuit.Policy{ErrorThreshold: 0.5, MinRequests: 2, OpenDuration: time.Nanosecond}) + fwd := NewHTTP(stubSelector{cands: []*backend.Backend{mkBackend("canceled", upstream.URL), mkBackend("untried", next.URL)}}, pool.NewHTTPPool(), cm, Policy{MaxAttempts: 2, PerAttemptTimeout: time.Second}) + for attempt := range 3 { + wantState := circuit.StateClosed + if attempt == 2 { + cm.Record("canceled", types.ProtoRPC, false) + cm.Record("canceled", types.ProtoRPC, false) + wantState = circuit.StateHalfOpen + } + ctx, cancel := context.WithCancel(context.Background()) + req := httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx) + captured := ¬ifyingCapture{Capture: server.NewCapture(make(http.Header)), written: make(chan struct{}, 1)} + done := make(chan struct{}) + go func() { + defer close(done) + fwd.Forward(captured, req, types.RouteKey{Protocol: types.ProtoRPC, Idempotent: true}) + }() + select { + case <-captured.written: + case <-time.After(3 * time.Second): + cancel() + t.Fatal("upstream body did not reach the capture") + } + cancel() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("forwarder did not stop on cancellation") + } + if bytes.Contains(captured.BodyBytes(), []byte(`"result"`)) { + t.Fatalf("canceled body survived: %q", captured.BodyBytes()) + } + if cm.State("canceled", types.ProtoRPC) != wantState || nextCalls.Load() != 0 { + t.Fatalf("cancellation changed circuit or retried: state=%s want=%s retries=%d", cm.State("canceled", types.ProtoRPC), wantState, nextCalls.Load()) + } + } + if !cm.Acquire("canceled", types.ProtoRPC) { + t.Fatal("body cancellation did not release the half-open canary") + } + cm.Release("canceled", types.ProtoRPC) +} + +func TestBufferedTruncationMetricsCountOnlyFinalOutcome(t *testing.T) { + for _, attempts := range []int{1, 2} { + t.Run(fmt.Sprint(attempts), func(t *testing.T) { + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", "1000") + _, _ = w.Write([]byte(`{"result":`)) + })) + defer bad.Close() + good := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"result":true}`)) + })) + defer good.Close() + badName, goodName := "metrics-bad-"+t.Name(), "metrics-good-"+t.Name() + fwd := newForwarderWithCircuit(stubSelector{cands: []*backend.Backend{mkBackend(badName, bad.URL), mkBackend(goodName, good.URL)}}, newTestCircuit(), attempts) + front := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + captured := server.NewCapture(w.Header()) + fwd.Forward(captured, r, types.RouteKey{Protocol: types.ProtoRPC, Class: types.ClassLatest, Idempotent: true}) + captured.FlushTo(w) + })) + defer front.Close() + counter := func(name, outcome string) float64 { + return testutil.ToFloat64(metrics.RequestsTotal.WithLabelValues(string(types.ProtoRPC), types.ClassLatest.String(), name, outcome)) + } + badBefore, goodBefore, failedBefore := counter(badName, "2xx"), counter(goodName, "2xx"), counter("-", "all_failed") + badDurationBefore := requestDurationCount(t, string(types.ProtoRPC), types.ClassLatest.String(), badName) + truncatedBefore := testutil.ToFloat64(metrics.RelayTruncated.WithLabelValues(badName, string(types.ProtoRPC))) + resp, err := front.Client().Get(front.URL) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil || !json.Valid(body) { + t.Fatalf("invalid buffered outcome: body=%q err=%v", body, err) + } + wantGood, wantFailed, wantStatus := float64(0), float64(1), http.StatusBadGateway + if attempts == 2 { + wantGood, wantFailed, wantStatus = 1, 0, http.StatusOK + } + if resp.StatusCode != wantStatus || counter(goodName, "2xx")-goodBefore != wantGood || counter("-", "all_failed")-failedBefore != wantFailed { + t.Fatalf("final outcome must be counted exactly once: status=%d success=%v failed=%v", resp.StatusCode, counter(goodName, "2xx")-goodBefore, counter("-", "all_failed")-failedBefore) + } + if counter(badName, "2xx") != badBefore || requestDurationCount(t, string(types.ProtoRPC), types.ClassLatest.String(), badName) != badDurationBefore { + t.Fatal("discarded buffered attempt counted as a completed response") + } + if got := testutil.ToFloat64(metrics.RelayTruncated.WithLabelValues(badName, string(types.ProtoRPC))) - truncatedBefore; got != 1 { + t.Fatalf("truncated attempt metric delta=%v, want 1", got) + } + }) + } +} diff --git a/internal/server/capture.go b/internal/server/capture.go index 1ef1991..824243e 100644 --- a/internal/server/capture.go +++ b/internal/server/capture.go @@ -31,6 +31,14 @@ func (c *Capture) Header() http.Header { return c.header } func (c *Capture) WriteHeader(code int) { c.status = code } func (c *Capture) Write(b []byte) (int, error) { return c.body.Write(b) } +// ResetResponse discards a failed upstream attempt before anything is flushed. +// Restore caller-owned headers, excluding headers copied from that attempt. +func (c *Capture) ResetResponse(headers http.Header) { + c.header = headers.Clone() + c.status = http.StatusOK + c.body.Reset() +} + // Status returns the captured status code (200 if WriteHeader was never // called). func (c *Capture) Status() int { return c.status } diff --git a/internal/server/cmt_rpc/truncation_test.go b/internal/server/cmt_rpc/truncation_test.go new file mode 100644 index 0000000..93feab8 --- /dev/null +++ b/internal/server/cmt_rpc/truncation_test.go @@ -0,0 +1,240 @@ +package cmt_rpc + +import ( + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + + "github.com/InjectiveLabs/stitch/internal/backend" + "github.com/InjectiveLabs/stitch/internal/cache" + "github.com/InjectiveLabs/stitch/internal/circuit" + "github.com/InjectiveLabs/stitch/internal/forwarder" + "github.com/InjectiveLabs/stitch/internal/health" + "github.com/InjectiveLabs/stitch/internal/pool" + "github.com/InjectiveLabs/stitch/internal/selector" + "github.com/InjectiveLabs/stitch/internal/types" +) + +const blockResultsPath = "/block_results?height=75" + +func completeBlockResults() []byte { + return []byte(`{"jsonrpc":"2.0","id":-1,"result":{"height":"75","txs_results":[],"data":"` + strings.Repeat("x", 2<<20) + `"}}`) +} + +func truncationBackend(name, endpoint string, lower int64, weight int) *backend.Backend { + return &backend.Backend{ + Name: name, Weight: weight, + Coverage: backend.Coverage{Kind: backend.CovOpen, Lower: lower}, + Endpoints: map[types.Protocol]string{types.ProtoRPC: endpoint}, + } +} + +func newTruncationServer(t *testing.T, bs []*backend.Backend, attempts int, timeout time.Duration) *httptest.Server { + t.Helper() + cm := circuit.NewManager(circuit.Policy{MinRequests: 100}) + reg := backend.NewRegistry(bs) + sel := selector.NewRangeSelector(reg, health.NewRegistry(), cm, 0) + p := pool.NewHTTPPool() + t.Cleanup(p.CloseIdle) + s := New("ignored", forwarder.NewHTTP(sel, p, cm, forwarder.Policy{MaxAttempts: attempts, PerAttemptTimeout: timeout})) + s.SetResponseCache(cache.NewResponseCache(cache.ResponseCacheOpts{Capacity: 10, MaxBytes: 16 << 20}), func() int64 { return 200 }, 0, time.Minute) + front := httptest.NewServer(s.Handler()) + t.Cleanup(front.Close) + return front +} + +func getTruncationResponse(t *testing.T, front *httptest.Server) (*http.Response, []byte) { + t.Helper() + resp, err := front.Client().Get(front.URL + blockResultsPath) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("buffered response was not complete: %v", err) + } + if resp.Header.Get("x-request-id") == "" { + t.Fatal("caller request ID was lost") + } + return resp, body +} + +func TestBufferedBodyFailureReturns502WithoutCaching(t *testing.T) { + for _, mode := range []string{"content-length", "chunked", "deadline", "valid-json-prefix"} { + t.Run(mode, func(t *testing.T) { + good := completeBlockResults() + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) > 1 { + _, _ = w.Write(good) + return + } + w.Header().Set("ETag", "failed-attempt") + w.Header().Set("Cache-Control", "public,max-age=86400") + prefix := good[:128<<10] // Exercise the historical-error replay boundary. + switch mode { + case "content-length": + w.Header().Set("Content-Length", strconv.Itoa(len(good))) + case "valid-json-prefix": + // Even syntactically valid JSON must not be cached when the + // HTTP body failed before its promised end. + prefix = []byte(`{"jsonrpc":"2.0","id":-1,"result":{"height":"75","poison":true}}`) + w.Header().Set("Content-Length", strconv.Itoa(len(prefix)+100)) + } + _, _ = w.Write(prefix) + w.(http.Flusher).Flush() + switch mode { + case "chunked": + panic(http.ErrAbortHandler) // Omit the final HTTP chunk. + case "deadline": + <-r.Context().Done() + } + })) + defer upstream.Close() + front := newTruncationServer(t, []*backend.Backend{truncationBackend("bad-"+mode, upstream.URL, 1, 100)}, 1, 300*time.Millisecond) + resp, body := getTruncationResponse(t, front) + if resp.StatusCode != http.StatusBadGateway || !bytes.Contains(body, []byte(`"error":`)) || bytes.Contains(body, []byte(`"result":`)) { + t.Fatalf("failed attempt leaked success: status=%d body=%q", resp.StatusCode, body) + } + if resp.Header.Get("ETag") != "" || resp.Header.Get("Cache-Control") != "" { + t.Fatalf("failed attempt headers survived: %v", resp.Header) + } + if resp.Header.Get("x-stitch-cache") != "miss" { + t.Fatal("cache miss header lost during reset") + } + for _, wantCache := range []string{"miss", "hit"} { + resp, body = getTruncationResponse(t, front) + if resp.StatusCode != http.StatusOK || !bytes.Equal(body, good) || resp.Header.Get("x-stitch-cache") != wantCache { + t.Fatalf("recovery/cache: status=%d bytes=%d cache=%q, want %q", resp.StatusCode, len(body), resp.Header.Get("x-stitch-cache"), wantCache) + } + } + if calls.Load() != 2 { + t.Fatalf("upstream calls=%d, want one failure and one successful fill", calls.Load()) + } + }) + } +} + +func TestBufferedBodyFailureRetriesOnlyEligibleCandidatesWithinBudget(t *testing.T) { + for _, maxAttempts := range []int{1, 2} { + t.Run(fmt.Sprint(maxAttempts), func(t *testing.T) { + goodBody := completeBlockResults() + var badCalls, goodCalls, ineligibleCalls atomic.Int32 + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + badCalls.Add(1) + w.Header().Set("Content-Length", "9000000") + w.Header().Set("ETag", "bad") + _, _ = w.Write(goodBody[:128<<10]) + })) + defer bad.Close() + good := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + goodCalls.Add(1) + w.Header().Set("ETag", "good") + _, _ = w.Write(goodBody) + })) + defer good.Close() + ineligible := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + ineligibleCalls.Add(1) + _, _ = w.Write(goodBody) + })) + defer ineligible.Close() + front := newTruncationServer(t, []*backend.Backend{ + truncationBackend("ineligible", ineligible.URL, 100, 1000), + truncationBackend("bad", bad.URL, 1, 200), + truncationBackend("good", good.URL, 1, 100), + }, maxAttempts, time.Second) + resp, body := getTruncationResponse(t, front) + if maxAttempts == 1 { + if resp.StatusCode != http.StatusBadGateway || goodCalls.Load() != 0 { + t.Fatalf("attempt budget exceeded: status=%d good calls=%d", resp.StatusCode, goodCalls.Load()) + } + } else if resp.StatusCode != http.StatusOK || !bytes.Equal(body, goodBody) || resp.Header.Get("ETag") != "good" || goodCalls.Load() != 1 { + t.Fatalf("eligible retry failed: status=%d bytes=%d headers=%v good calls=%d", resp.StatusCode, len(body), resp.Header, goodCalls.Load()) + } + if maxAttempts == 2 { + resp, body = getTruncationResponse(t, front) + if !bytes.Equal(body, goodBody) || resp.Header.Get("x-stitch-cache") != "hit" || goodCalls.Load() != 1 { + t.Fatal("successful retry was not cached intact") + } + } + if badCalls.Load() != 1 || ineligibleCalls.Load() != 0 { + t.Fatalf("bad calls=%d ineligible calls=%d", badCalls.Load(), ineligibleCalls.Load()) + } + }) + } +} + +func TestLargeSlowBufferedResponseCompletesAndCaches(t *testing.T) { + want := completeBlockResults() + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + calls.Add(1) + for offset := 0; offset < len(want); offset += 128 << 10 { + end := min(offset+128<<10, len(want)) + if _, err := w.Write(want[offset:end]); err != nil { + return + } + w.(http.Flusher).Flush() + time.Sleep(5 * time.Millisecond) + } + })) + defer upstream.Close() + front := newTruncationServer(t, []*backend.Backend{truncationBackend("slow", upstream.URL, 1, 100)}, 1, 2*time.Second) + for _, wantCache := range []string{"miss", "hit"} { + resp, body := getTruncationResponse(t, front) + if resp.StatusCode != http.StatusOK || !bytes.Equal(body, want) || resp.Header.Get("x-stitch-cache") != wantCache { + t.Fatalf("large response: status=%d bytes=%d cache=%q", resp.StatusCode, len(body), resp.Header.Get("x-stitch-cache")) + } + } + if calls.Load() != 1 { + t.Fatalf("upstream calls=%d, want one complete cache fill", calls.Load()) + } +} + +func TestWebSocketQueryTruncationReturnsErrorAndStaysUsable(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if bytes.Contains(body, []byte(`"id":1`)) { + w.Header().Set("Content-Length", "1000") + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":`)) + return + } + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":2,"result":{}}`)) + })) + defer upstream.Close() + front := newTruncationServer(t, []*backend.Backend{truncationBackend("ws-http", upstream.URL, 1, 100)}, 1, time.Second) + conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(front.URL, "http")+"/websocket", nil) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + for id := 1; id <= 2; id++ { + if err := conn.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"jsonrpc":"2.0","id":%d,"method":"status","params":{}}`, id))); err != nil { + t.Fatal(err) + } + if err := conn.SetReadDeadline(time.Now().Add(3 * time.Second)); err != nil { + t.Fatal(err) + } + _, body, err := conn.ReadMessage() + if err != nil { + t.Fatal(err) + } + want := `"error":` + if id == 2 { + want = `"result":` + } + if !bytes.Contains(body, []byte(want)) { + t.Fatalf("query %d: %s", id, body) + } + } +} diff --git a/internal/server/cmt_rpc/websocket.go b/internal/server/cmt_rpc/websocket.go index b077dc4..60bfef3 100644 --- a/internal/server/cmt_rpc/websocket.go +++ b/internal/server/cmt_rpc/websocket.go @@ -345,6 +345,15 @@ type wsResponse struct { func (w *wsResponse) Header() http.Header { return w.header } func (w *wsResponse) WriteHeader(int) {} + +// ResetResponse keeps HTTP query failures inside the WebSocket response buffer; +// the forwarder must not try to abort a hijacked HTTP connection. +func (w *wsResponse) ResetResponse(headers http.Header) { + w.header = headers.Clone() + w.body.Reset() + w.overflow = false +} + func (w *wsResponse) Write(b []byte) (int, error) { if len(b) > wsReadLimit-w.body.Len() { w.overflow = true diff --git a/internal/server/eth_rpc/server.go b/internal/server/eth_rpc/server.go index b142022..6668f7e 100644 --- a/internal/server/eth_rpc/server.go +++ b/internal/server/eth_rpc/server.go @@ -317,6 +317,24 @@ func (s *Server) handleBatch(w http.ResponseWriter, r *http.Request, body []byte rec := server.NewCapture(parentHeader) s.handleSingle(rec, r, raw[i]) results[i] = rec.BodyBytes() + if rec.Status() < 200 || rec.Status() >= 300 { + // HTTP forwarding errors are plain JSON. Inside a batch they + // need a JSON-RPC envelope and the original item's request ID. + // Preserve errors already framed by the JSON-RPC handler. + var response struct { + JSONRPC string `json:"jsonrpc"` + Error json.RawMessage `json:"error"` + } + if json.Unmarshal(results[i], &response) == nil && response.JSONRPC == "2.0" && + len(response.Error) > 0 && response.Error[0] == '{' { + continue + } + var request jsonRPCRequest + if json.Unmarshal(raw[i], &request) != nil || !cache.IsJSONRPCID(request.ID) { + request.ID = nil + } + results[i] = jsonRPCErrorBytes(request.ID, -32000, "upstream request failed") + } } w.WriteHeader(http.StatusOK) diff --git a/internal/server/eth_rpc/truncation_test.go b/internal/server/eth_rpc/truncation_test.go new file mode 100644 index 0000000..a378e80 --- /dev/null +++ b/internal/server/eth_rpc/truncation_test.go @@ -0,0 +1,199 @@ +package eth_rpc + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/InjectiveLabs/stitch/internal/backend" + "github.com/InjectiveLabs/stitch/internal/cache" + "github.com/InjectiveLabs/stitch/internal/circuit" + "github.com/InjectiveLabs/stitch/internal/forwarder" + "github.com/InjectiveLabs/stitch/internal/pool" + "github.com/InjectiveLabs/stitch/internal/types" +) + +type truncationSelector []*backend.Backend + +func (s truncationSelector) Candidates(types.RouteKey) []*backend.Backend { return s } + +func newTruncationEVM(t *testing.T, endpoints ...string) *Server { + t.Helper() + var sel truncationSelector + for i, endpoint := range endpoints { + sel = append(sel, &backend.Backend{Name: fmt.Sprintf("evm-truncation-%d", i), Endpoints: map[types.Protocol]string{types.ProtoEthRPC: endpoint}}) + } + p := pool.NewHTTPPool() + t.Cleanup(p.CloseIdle) + cm := circuit.NewManager(circuit.Policy{MinRequests: 100}) + return New("ignored", forwarder.NewHTTP(sel, p, cm, forwarder.Policy{MaxAttempts: 3, PerAttemptTimeout: time.Second})) +} + +func TestTruncatedEVMBatchItemPreservesIDAndOtherResults(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req jsonRPCRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Error(err) + return + } + switch string(req.ID) { + case `"failed:string"`, "9007199254740993": + w.Header().Set("Content-Length", "1000") + _, _ = fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":"poison"}`, req.ID) + case "4": + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":4,"error":{"code":-32042,"message":"upstream error","data":{"detail":true}}}`)) + default: + _, _ = fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":"0x59f"}`, req.ID) + } + })) + defer upstream.Close() + s := newTruncationEVM(t, upstream.URL) + front := httptest.NewServer(s.Handler()) + defer front.Close() + resp := post(t, front.URL, `[ + {"jsonrpc":"2.0","id":"failed:string","method":"eth_chainId","params":[]}, + {"jsonrpc":"2.0","id":9007199254740993,"method":"eth_chainId","params":[]}, + {"jsonrpc":"2.0","id":3,"method":"eth_chainId","params":[]}, + {"jsonrpc":"2.0","id":4,"method":"eth_chainId","params":[]} + ]`) + defer resp.Body.Close() + var items []struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id"` + Result json.RawMessage `json:"result"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data"` + } `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&items); err != nil { + t.Fatal(err) + } + if resp.StatusCode != http.StatusOK || len(items) != 4 { + t.Fatalf("batch status=%d items=%d", resp.StatusCode, len(items)) + } + for i, wantID := range []string{`"failed:string"`, "9007199254740993", "3", "4"} { + if items[i].JSONRPC != "2.0" || string(items[i].ID) != wantID { + t.Fatalf("item %d lost framing/ID: %+v", i, items[i]) + } + } + for i := range 2 { + if items[i].Error == nil || items[i].Error.Code != -32000 || len(items[i].Result) != 0 { + t.Fatalf("truncated item %d became success: %+v", i, items[i]) + } + } + if items[2].Error != nil || string(items[2].Result) != `"0x59f"` { + t.Fatalf("unrelated successful item changed: %+v", items[2]) + } + if items[3].Error == nil || items[3].Error.Code != -32042 || string(items[3].Error.Data) != `{"detail":true}` { + t.Fatalf("existing JSON-RPC error changed: %+v", items[3]) + } +} + +func TestTruncatedFilterMintDoesNotBindOrRetry(t *testing.T) { + var badCalls, nextCalls atomic.Int32 + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + badCalls.Add(1) + w.Header().Set("Content-Length", "1000") + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":"0xdead"}`)) + })) + defer bad.Close() + next := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + nextCalls.Add(1) + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":"0xbeef"}`)) + })) + defer next.Close() + s := newTruncationEVM(t, bad.URL, next.URL) + front := httptest.NewServer(s.Handler()) + defer front.Close() + resp := post(t, front.URL, `{"jsonrpc":"2.0","id":1,"method":"eth_newFilter","params":[{}]}`) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil || resp.StatusCode != http.StatusBadGateway || !json.Valid(body) || bytes.Contains(body, []byte(`"result"`)) { + t.Fatalf("filter response: status=%d body=%q err=%v", resp.StatusCode, body, err) + } + if s.FilterStore().Size() != 0 || badCalls.Load() != 1 || nextCalls.Load() != 0 { + t.Fatalf("failed mint bound/retried: bindings=%d bad=%d next=%d", s.FilterStore().Size(), badCalls.Load(), nextCalls.Load()) + } +} + +func TestTruncatedEVMResponseDoesNotPoisonResponseOrHashCache(t *testing.T) { + for _, responseCacheEnabled := range []bool{false, true} { + t.Run(fmt.Sprint(responseCacheEnabled), func(t *testing.T) { + badHash, goodHash := "0x"+strings.Repeat("ab", 32), "0x"+strings.Repeat("cd", 32) + var calls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req jsonRPCRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Error(err) + return + } + hash := goodHash + first := calls.Add(1) == 1 + if first { + hash = badHash + } + body := fmt.Sprintf(`{"jsonrpc":"2.0","id":%s,"result":{"number":"0x4b","hash":%q}}`, req.ID, hash) + if first { + // The JSON alone looks complete and cacheable. Transport + // completion must be checked before either cache sees it. + w.Header().Set("Content-Length", strconv.Itoa(len(body)+100)) + } + _, _ = w.Write([]byte(body)) + })) + defer upstream.Close() + s := newTruncationEVM(t, upstream.URL) + idx := cache.New(10) + s.SetHashCache(idx) + rc := cache.NewResponseCache(cache.ResponseCacheOpts{Capacity: 10}) + if responseCacheEnabled { + s.SetResponseCache(rc, func() int64 { return 200 }, 0, time.Minute) + } + front := httptest.NewServer(s.Handler()) + defer front.Close() + for id := 1; id <= 3; id++ { + resp := post(t, front.URL, fmt.Sprintf(`{"jsonrpc":"2.0","id":%d,"method":"eth_getBlockByNumber","params":["0x4b",false]}`, id)) + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil || !json.Valid(body) { + t.Fatalf("invalid response: body=%q err=%v", body, err) + } + if _, found := idx.Get(cache.EthBlockKey(badHash)); found { + t.Fatal("failed response poisoned hash index") + } + if id == 1 { + if resp.StatusCode != http.StatusBadGateway || rc.Size() != 0 { + t.Fatalf("failed response returned/cached success: status=%d entries=%d", resp.StatusCode, rc.Size()) + } + continue + } + var envelope struct { + ID json.RawMessage `json:"id"` + } + if json.Unmarshal(body, &envelope) != nil || string(envelope.ID) != strconv.Itoa(id) || !bytes.Contains(body, []byte(goodHash)) || resp.StatusCode != http.StatusOK { + t.Fatalf("valid response/cache replay failed: status=%d body=%q", resp.StatusCode, body) + } + if height, found := idx.Get(cache.EthBlockKey(goodHash)); !found || height != 75 { + t.Fatal("valid response failed to populate hash index") + } + } + wantCalls := int32(3) + if responseCacheEnabled { + wantCalls = 2 + } + if calls.Load() != wantCalls { + t.Fatalf("upstream calls=%d want=%d", calls.Load(), wantCalls) + } + }) + } +}