diff --git a/cmd/stellar-rpc/internal/jsonrpc/bridge.go b/cmd/stellar-rpc/internal/jsonrpc/bridge.go
new file mode 100644
index 000000000..af1cfc888
--- /dev/null
+++ b/cmd/stellar-rpc/internal/jsonrpc/bridge.go
@@ -0,0 +1,336 @@
+package jsonrpc
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "mime"
+ "net"
+ "net/http"
+ "runtime"
+ "slices"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/creachadair/jrpc2"
+ "golang.org/x/sync/semaphore"
+
+ "github.com/stellar/go-stellar-sdk/support/log"
+
+ "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/network"
+ "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/util"
+)
+
+const contentTypeJSON = "application/json"
+
+var (
+ errBridgeClosed = errors.New("the bridge is closed")
+ errEmptyMethod = &jrpc2.Error{Code: jrpc2.InvalidRequest, Message: "empty method name"}
+ errNoSuchMethod = &jrpc2.Error{Code: jrpc2.MethodNotFound, Message: jrpc2.MethodNotFound.String()}
+)
+
+// bridge serves JSON-RPC 2.0 over HTTP POST by calling the handlers of an
+// Assigner directly, so a result is encoded once and a json.RawMessage result
+// is written as is. It stands in for jhttp.Bridge, which ran every request
+// through an in-memory client and server and re-encoded each response on the
+// way, and keeps its wire behavior; TestBridge_ParityWithJHTTP checks that.
+// It enforces the global duration limit itself, so a response is never
+// buffered to be swapped for a 504.
+type bridge struct {
+ methods jrpc2.Assigner
+ sem *semaphore.Weighted // bounds the handlers running at once, as jrpc2.Server does
+ limits durationLimits
+
+ mu sync.RWMutex // read-held while a request's handlers run
+ closed bool
+ ctx context.Context //nolint:containedctx // ends at Close, canceling the requests in flight
+ cancel context.CancelFunc
+}
+
+// durationLimits are the global request-duration thresholds, with the counters
+// and logger for requests that pass them, as network's HTTP limiter had them.
+type durationLimits struct {
+ warning, limit time.Duration
+ warnings, timeouts interface{ Inc() }
+ logger *log.Entry
+}
+
+func newBridge(methods jrpc2.Assigner, limits durationLimits) *bridge {
+ ctx, cancel := context.WithCancel(context.Background())
+ return &bridge{
+ methods: methods,
+ sem: semaphore.NewWeighted(int64(runtime.NumCPU())),
+ limits: limits,
+ ctx: ctx,
+ cancel: cancel,
+ }
+}
+
+// Close cancels the requests in flight, waits for their handlers to return,
+// and fails later requests.
+func (b *bridge) Close() {
+ b.cancel()
+ b.mu.Lock()
+ b.closed = true
+ b.mu.Unlock()
+}
+
+// ServeHTTP implements http.Handler.
+func (b *bridge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ defer b.recoverPanic(w)
+ w.Header().Set("Accept-Post", contentTypeJSON)
+ if r.Method != http.MethodPost {
+ w.WriteHeader(http.StatusMethodNotAllowed)
+ return
+ }
+ mt, params, _ := mime.ParseMediaType(r.Header.Get("Content-Type"))
+ if mt != contentTypeJSON {
+ http.Error(w, "content-type must be application/json", http.StatusUnsupportedMediaType)
+ return
+ }
+ if cs, ok := params["charset"]; ok && cs != "utf-8" && cs != "utf8" {
+ http.Error(w, "invalid content-type charset", http.StatusUnsupportedMediaType)
+ return
+ }
+ start := time.Now()
+ ctx, cancel := b.limits.bound(r.Context())
+ defer cancel()
+ batch, rsps, err := b.serve(ctx, r)
+ if b.limits.timedOut(r, time.Since(start)) {
+ if r.Context().Err() == nil {
+ w.WriteHeader(http.StatusGatewayTimeout)
+ }
+ return
+ }
+ if err == nil && len(rsps) == 0 {
+ w.WriteHeader(http.StatusNoContent) // only notifications, or an empty batch
+ return
+ }
+ if err == nil {
+ err = write(w, batch, rsps)
+ }
+ if err != nil {
+ w.WriteHeader(http.StatusInternalServerError)
+ _, _ = fmt.Fprintln(w, err.Error()) //nolint:gosec // a read or parse error, as text; jhttp.Bridge did the same
+ }
+}
+
+// serve runs the requests in r's body on ctx and returns their responses.
+func (b *bridge) serve(ctx context.Context, r *http.Request) (bool, []*response, error) {
+ body, err := io.ReadAll(r.Body)
+ if err != nil {
+ return false, nil, err
+ }
+ reqs, err := jrpc2.ParseRequests(body)
+ if err != nil {
+ return false, nil, err
+ }
+ batch := len(reqs) > 0 && reqs[0].Batch
+ // Invalid requests are answered first, as jhttp.Bridge did.
+ slices.SortStableFunc(reqs, func(x, y *jrpc2.ParsedRequest) int { return rank(x) - rank(y) })
+ rsps, err := b.run(ctx, reqs)
+ return batch, rsps, err
+}
+
+// bound returns ctx ending at the limit, as the handlers see it.
+func (l durationLimits) bound(ctx context.Context) (context.Context, context.CancelFunc) {
+ if l.limit == network.RequestDurationLimiterNoLimit {
+ return ctx, func() {}
+ }
+ return context.WithTimeout(ctx, l.limit)
+}
+
+// timedOut reports whether a request that took elapsed passed the limit, and
+// counts and logs it if it passed the limit or, failing that, the warning.
+func (l durationLimits) timedOut(r *http.Request, elapsed time.Duration) bool {
+ switch {
+ case l.limit == network.RequestDurationLimiterNoLimit:
+ case l.limit != 0 && elapsed >= l.limit:
+ l.timeouts.Inc()
+ l.logger.Infof("Request processing for %s exceed limiting threshold of %v", r.URL.Path, l.limit)
+ return true
+ case l.warning != 0 && l.warning < l.limit && elapsed >= l.warning:
+ l.warnings.Inc()
+ l.logger.Infof("Request processing for %s exceed warning threshold of %v", r.URL.Path, l.warning)
+ }
+ return false
+}
+
+// recoverPanic answers 500 to a request whose serving panicked outside a
+// handler call and logs the stack, as network's HTTP limiter did.
+func (b *bridge) recoverPanic(w http.ResponseWriter) {
+ p := recover()
+ if p == nil {
+ return
+ }
+ w.WriteHeader(http.StatusInternalServerError)
+ for _, line := range util.CallStack(p, "", "(*bridge).ServeHTTP", 8) {
+ b.limits.logger.Warn(line)
+ }
+}
+
+// rank orders invalid requests before valid ones.
+func rank(pr *jrpc2.ParsedRequest) int {
+ if pr.Error != nil {
+ return 0
+ }
+ return 1
+}
+
+// run calls the handlers for reqs, on a context that also ends at Close, and
+// returns the responses in request order. Notifications get none.
+func (b *bridge) run(ctx context.Context, reqs []*jrpc2.ParsedRequest) ([]*response, error) {
+ b.mu.RLock()
+ defer b.mu.RUnlock()
+ if b.closed {
+ return nil, errBridgeClosed
+ }
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+ defer context.AfterFunc(b.ctx, cancel)() //nolint:contextcheck // b.ctx ends at Close, not with the request
+
+ rsps := make([]*response, len(reqs))
+ var wg sync.WaitGroup
+ for i, pr := range reqs {
+ method, rsp := b.route(ctx, pr)
+ if method == nil {
+ rsps[i] = rsp
+ continue
+ }
+ if err := b.sem.Acquire(ctx, 1); err != nil {
+ rsps[i] = respond(pr, nil, err)
+ continue
+ }
+ call := func() {
+ defer b.sem.Release(1)
+ defer func() {
+ if p := recover(); p != nil { // answer just this call, as the duration limiter does
+ for _, line := range util.CallStack(p, pr.Method, "(*bridge).run.func", 8) {
+ b.limits.logger.Warn(line)
+ }
+ rsps[i] = respond(pr, nil, network.ErrFailToProcessDueToInternalIssue)
+ }
+ }()
+ v, err := method(ctx, pr.ToRequest())
+ rsps[i] = respond(pr, v, err)
+ }
+ if i == len(reqs)-1 {
+ call() // the last request runs here, as in jrpc2.Server
+ } else {
+ wg.Go(call)
+ }
+ }
+ wg.Wait()
+ return slices.DeleteFunc(rsps, func(r *response) bool { return r == nil }), nil
+}
+
+// route returns pr's handler, or else the response pr gets without one:
+// invalid requests are always answered, notifications to unknown methods never.
+func (b *bridge) route(ctx context.Context, pr *jrpc2.ParsedRequest) (jrpc2.Handler, *response) {
+ if pr.Error != nil {
+ return nil, &response{id: pr.ID, err: pr.Error}
+ }
+ if pr.Method != "" {
+ if method := b.methods.Assign(ctx, pr.Method); method != nil {
+ return method, nil
+ }
+ }
+ switch {
+ case pr.ID == "":
+ return nil, nil
+ case pr.Method == "":
+ return nil, &response{id: pr.ID, err: errEmptyMethod}
+ }
+ return nil, &response{id: pr.ID, err: errNoSuchMethod.WithData(pr.Method)}
+}
+
+// respond builds the response to pr from a handler's result, passing a
+// non-empty json.RawMessage through as is. A notification gets none.
+func respond(pr *jrpc2.ParsedRequest, v any, err error) *response {
+ if pr.ID == "" {
+ return nil
+ }
+ rsp := &response{id: pr.ID}
+ if err == nil {
+ raw, ok := v.(json.RawMessage)
+ if !ok || len(raw) == 0 {
+ raw, err = json.Marshal(v)
+ }
+ if err == nil {
+ rsp.result = raw
+ return rsp
+ }
+ }
+ if e, ok := err.(*jrpc2.Error); ok { //nolint:errorlint // as jrpc2.Server does: no unwrapping
+ rsp.err = e
+ } else if c := jrpc2.ErrorCode(err); c != jrpc2.NoError {
+ rsp.err = &jrpc2.Error{Code: c, Message: err.Error()}
+ } else {
+ rsp.err = &jrpc2.Error{Code: jrpc2.InternalError, Message: err.Error()}
+ }
+ return rsp
+}
+
+// A response is one JSON-RPC response, with a result or an error.
+type response struct {
+ id string // the request ID as sent, "" for null
+ result json.RawMessage
+ err *jrpc2.Error
+}
+
+// write writes rsps as the body of a 200 response, as an array if batch is
+// set. Results are written without copying.
+func write(w http.ResponseWriter, batch bool, rsps []*response) error {
+ parts := make(net.Buffers, 0, 3*len(rsps)+2)
+ if batch {
+ parts = append(parts, []byte("["))
+ }
+ for i, r := range rsps {
+ head := `{"jsonrpc":"2.0","id":` + escapeID(r.id)
+ if i > 0 {
+ head = "," + head
+ }
+ body := r.result
+ if r.err != nil {
+ e, err := json.Marshal(r.err)
+ if err != nil {
+ return err
+ }
+ head, body = head+`,"error":`, e
+ } else {
+ head += `,"result":`
+ }
+ parts = append(parts, []byte(head), body, []byte("}"))
+ }
+ if batch {
+ parts = append(parts, []byte("]"))
+ }
+ n := 0
+ for _, p := range parts {
+ n += len(p)
+ }
+ w.Header().Set("Content-Type", contentTypeJSON)
+ w.Header().Set("Content-Length", strconv.Itoa(n))
+ w.WriteHeader(http.StatusOK)
+ _, _ = parts.WriteTo(w) // a write error means the client is gone
+ return nil
+}
+
+// escapeID HTML-escapes a request ID before it is echoed, as re-encoding the
+// response through jhttp.Bridge did. An absent ID is null.
+func escapeID(id string) string {
+ if id == "" {
+ return "null"
+ }
+ if !strings.ContainsAny(id, "<>&\u2028\u2029") {
+ return id
+ }
+ var buf bytes.Buffer
+ json.HTMLEscape(&buf, []byte(id))
+ return buf.String()
+}
diff --git a/cmd/stellar-rpc/internal/jsonrpc/bridge_test.go b/cmd/stellar-rpc/internal/jsonrpc/bridge_test.go
new file mode 100644
index 000000000..97450effa
--- /dev/null
+++ b/cmd/stellar-rpc/internal/jsonrpc/bridge_test.go
@@ -0,0 +1,305 @@
+package jsonrpc
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/creachadair/jrpc2"
+ "github.com/creachadair/jrpc2/handler"
+ "github.com/creachadair/jrpc2/jhttp"
+ "github.com/sirupsen/logrus"
+ logtest "github.com/sirupsen/logrus/hooks/test"
+ "github.com/stretchr/testify/require"
+
+ "github.com/stellar/go-stellar-sdk/support/log"
+
+ "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/network"
+)
+
+// unlimited lets requests run as long as their handlers take.
+var unlimited = durationLimits{limit: network.RequestDurationLimiterNoLimit, logger: log.DefaultLogger}
+
+type counter struct{ n atomic.Int64 }
+
+func (c *counter) Inc() { c.n.Add(1) }
+
+// testLogger returns a quiet logger whose entries hook records.
+func testLogger() (*log.Entry, *logtest.Hook) {
+ hook := new(logtest.Hook)
+ logger := log.New()
+ logger.SetOutput(io.Discard)
+ logger.SetLevel(logrus.DebugLevel)
+ logger.AddHook(hook)
+ return logger, hook
+}
+
+// bridgeMethods is the method table the bridge tests serve. Its raw result is
+// compact and free of HTML characters, so jhttp.Bridge re-encodes it to the
+// bytes the bridge passes through.
+func bridgeMethods() handler.Map {
+ return handler.Map{
+ "echo": handler.New(func(_ context.Context, r *jrpc2.Request) (any, error) {
+ var v any
+ if err := r.UnmarshalParams(&v); err != nil {
+ return nil, err
+ }
+ return v, nil
+ }),
+ "html": handler.New(func(context.Context) (any, error) {
+ return map[string]string{"s": "&"}, nil
+ }),
+ "raw": handler.New(func(context.Context) (any, error) {
+ return json.RawMessage(`{"pre":"rendered"}`), nil
+ }),
+ "rawEmpty": handler.New(func(context.Context) (any, error) {
+ return json.RawMessage(nil), nil
+ }),
+ "ptrErr": handler.New(func(context.Context) (any, error) {
+ return nil, &jrpc2.Error{Code: jrpc2.InvalidParams, Message: "", Data: json.RawMessage(`{"k":1}`)}
+ }),
+ "valErr": handler.New(func(context.Context) (any, error) {
+ return nil, jrpc2.Error{Code: jrpc2.InternalError, Message: "val"}
+ }),
+ "plainErr": handler.New(func(context.Context) (any, error) {
+ return nil, errors.New("boom")
+ }),
+ "canceled": handler.New(func(context.Context) (any, error) {
+ return nil, context.Canceled
+ }),
+ "unmarshalable": handler.New(func(context.Context) (any, error) {
+ return make(chan int), nil
+ }),
+ }
+}
+
+type bridgeCase struct {
+ name, method, contentType, body string
+}
+
+func bridgeCases() []bridgeCase {
+ call := func(name, body string) bridgeCase {
+ return bridgeCase{name: name, method: http.MethodPost, contentType: contentTypeJSON, body: body}
+ }
+ const echo = `{"jsonrpc":"2.0","id":1,"method":"echo","params":[1]}`
+ return []bridgeCase{
+ call("call", `{"jsonrpc":"2.0","id":1,"method":"echo","params":[1,"a",{"b":null}]}`),
+ call("null result", `{"jsonrpc":"2.0","id":1,"method":"echo"}`),
+ call("html in result", `{"jsonrpc":"2.0","id":1,"method":"html"}`),
+ call("raw result", `{"jsonrpc":"2.0","id":"r","method":"raw"}`),
+ call("empty raw result", `{"jsonrpc":"2.0","id":"r","method":"rawEmpty"}`),
+ call("pointer error", `{"jsonrpc":"2.0","id":1,"method":"ptrErr"}`),
+ call("value error", `{"jsonrpc":"2.0","id":1,"method":"valErr"}`),
+ call("plain error", `{"jsonrpc":"2.0","id":1,"method":"plainErr"}`),
+ call("canceled error", `{"jsonrpc":"2.0","id":1,"method":"canceled"}`),
+ call("unmarshalable result", `{"jsonrpc":"2.0","id":1,"method":"unmarshalable"}`),
+ call("notification", `{"jsonrpc":"2.0","method":"echo","params":[1]}`),
+ call("notification to unknown method", `{"jsonrpc":"2.0","method":"nope"}`),
+ call("notification with empty method", `{"jsonrpc":"2.0","method":""}`),
+ call("notification with bad params", `{"jsonrpc":"2.0","method":"echo","params":"s"}`),
+ call("unknown method", `{"jsonrpc":"2.0","id":1,"method":"nope"}`),
+ call("empty method", `{"jsonrpc":"2.0","id":1,"method":""}`),
+ call("bad params", `{"jsonrpc":"2.0","id":1,"method":"echo","params":"s"}`),
+ call("bad version", `{"jsonrpc":"1.0","id":1,"method":"echo"}`),
+ call("html id", `{"jsonrpc":"2.0","id":"&`+"\u2028"+`","method":"echo","params":[]}`),
+ call("html id on invalid request", `{"jsonrpc":"2.0","id":"&","method":"echo","params":"s"}`),
+ call("null id", `{"jsonrpc":"2.0","id":null,"method":"echo","params":[]}`),
+ call("number id", `{"jsonrpc":"2.0","id":1.5e3,"method":"echo","params":[]}`),
+ call("batch", `[`+echo+`,`+
+ `{"jsonrpc":"2.0","method":"echo","params":[2]},`+
+ `{"jsonrpc":"2.0","id":2,"method":"nope"},`+
+ `{"jsonrpc":"1.0","id":3,"method":"echo"},`+
+ `{"jsonrpc":"2.0","id":"","method":"ptrErr"},`+
+ `{"jsonrpc":"2.0","id":4,"method":"valErr"},`+
+ `{"jsonrpc":"2.0","id":5,"method":"raw"}]`),
+ call("batch of one", `[`+echo+`]`),
+ call("batch of notifications", `[{"jsonrpc":"2.0","method":"echo"},{"jsonrpc":"2.0","method":"nope"}]`),
+ call("batch with invalid entries", `[1,`+echo+`,"x"]`),
+ call("empty batch", `[]`),
+ call("empty object", `{}`),
+ call("invalid json", `{"jsonrpc":`),
+ call("empty body", ``),
+ call("not json", `hello`),
+ {name: "get", method: http.MethodGet, contentType: contentTypeJSON, body: ``},
+ {name: "put", method: http.MethodPut, contentType: contentTypeJSON, body: echo},
+ {name: "text content type", method: http.MethodPost, contentType: "text/plain", body: echo},
+ {name: "no content type", method: http.MethodPost, contentType: "", body: echo},
+ {name: "json-rpc content type", method: http.MethodPost, contentType: "application/json-rpc", body: echo},
+ {name: "charset utf-8", method: http.MethodPost, contentType: contentTypeJSON + "; charset=utf-8", body: echo},
+ {name: "charset UTF8", method: http.MethodPost, contentType: contentTypeJSON + "; charset=UTF8", body: echo},
+ {name: "charset latin1", method: http.MethodPost, contentType: contentTypeJSON + "; charset=latin1", body: echo},
+ }
+}
+
+func serveBridge(t *testing.T, h http.Handler, tc bridgeCase) *httptest.ResponseRecorder {
+ t.Helper()
+ req := httptest.NewRequestWithContext(t.Context(), tc.method, "/", strings.NewReader(tc.body))
+ if tc.contentType != "" {
+ req.Header.Set("Content-Type", tc.contentType)
+ }
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ return rec
+}
+
+func postBridge(t *testing.T, h http.Handler, body string) *httptest.ResponseRecorder {
+ t.Helper()
+ return serveBridge(t, h, bridgeCase{method: http.MethodPost, contentType: contentTypeJSON, body: body})
+}
+
+// The bridge answers every request with the status, headers and body that
+// jhttp.Bridge, the transport it replaces, gives on upstream jrpc2.
+func TestBridge_ParityWithJHTTP(t *testing.T) {
+ methods := bridgeMethods()
+ old := jhttp.NewBridge(methods, &jhttp.BridgeOptions{Server: &jrpc2.ServerOptions{DisableBuiltin: true}})
+ t.Cleanup(func() { _ = old.Close() })
+ b := newBridge(methods, unlimited)
+ t.Cleanup(b.Close)
+ for _, tc := range bridgeCases() {
+ t.Run(tc.name, func(t *testing.T) {
+ want, got := serveBridge(t, old, tc), serveBridge(t, b, tc)
+ require.Equal(t, want.Code, got.Code)
+ require.Equal(t, want.Header(), got.Header())
+ require.Equal(t, want.Body.String(), got.Body.String())
+ })
+ }
+}
+
+// A non-empty json.RawMessage result is written as is, with the Content-Length
+// to match.
+func TestBridge_RawResultPassthrough(t *testing.T) {
+ raw := json.RawMessage("{\n \"pretty\": \"\"\n}")
+ b := newBridge(handler.Map{"raw": handler.New(func(context.Context) (any, error) { return raw, nil })}, unlimited)
+ t.Cleanup(b.Close)
+ rec := postBridge(t, b, `{"jsonrpc":"2.0","id":1,"method":"raw"}`)
+ require.Equal(t, http.StatusOK, rec.Code)
+ require.Equal(t, `{"jsonrpc":"2.0","id":1,"result":`+string(raw)+`}`, rec.Body.String())
+ require.Equal(t, strconv.Itoa(rec.Body.Len()), rec.Header().Get("Content-Length"))
+}
+
+// A handler runs on the HTTP request's context, so a client disconnect
+// cancels it, and its error reaches the wire as a cancellation.
+func TestBridge_RequestContext(t *testing.T) {
+ ctx, disconnect := context.WithCancel(t.Context())
+ b := newBridge(handler.Map{"m": handler.New(func(hctx context.Context) (any, error) {
+ disconnect()
+ return nil, hctx.Err()
+ })}, unlimited)
+ t.Cleanup(b.Close)
+ req := httptest.NewRequestWithContext(ctx, http.MethodPost, "/",
+ strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"m"}`))
+ req.Header.Set("Content-Type", contentTypeJSON)
+ rec := httptest.NewRecorder()
+ b.ServeHTTP(rec, req)
+ require.Equal(t, http.StatusOK, rec.Code)
+ require.JSONEq(t,
+ fmt.Sprintf(`{"jsonrpc":"2.0","id":1,"error":{"code":%d,"message":"context canceled"}}`,
+ jrpc2.ErrorCode(context.Canceled)),
+ rec.Body.String())
+}
+
+// Close cancels the requests in flight, waits for their handlers to return,
+// and fails the requests after it, as jhttp.Bridge.Close did.
+func TestBridge_CloseDrains(t *testing.T) {
+ started, unwound := make(chan struct{}), make(chan struct{})
+ b := newBridge(handler.Map{"slow": handler.New(func(hctx context.Context) (any, error) {
+ close(started)
+ <-hctx.Done()
+ time.Sleep(50 * time.Millisecond) // unwinding takes a while
+ close(unwound)
+ return nil, hctx.Err()
+ })}, unlimited)
+ served := make(chan *httptest.ResponseRecorder, 1)
+ go func() { served <- postBridge(t, b, `{"jsonrpc":"2.0","id":1,"method":"slow"}`) }()
+ <-started
+ b.Close()
+ select {
+ case <-unwound:
+ default:
+ t.Fatal("Close returned before the handler did")
+ }
+ require.Equal(t, http.StatusOK, (<-served).Code)
+
+ rec := postBridge(t, b, `{"jsonrpc":"2.0","id":1,"method":"slow"}`)
+ require.Equal(t, http.StatusInternalServerError, rec.Code)
+ require.Equal(t, errBridgeClosed.Error()+"\n", rec.Body.String())
+}
+
+// A request that outlasts the global limit is answered 504 with no body, and
+// one past the limit or the warning is counted and logged.
+func TestBridge_DurationLimits(t *testing.T) {
+ const ok = `{"jsonrpc":"2.0","id":1,"result":1}`
+ for _, tc := range []struct {
+ name string
+ warning, limit, run time.Duration
+ wantCode int
+ wantBody string
+ wantWarnings, wantLimits int64
+ }{
+ {"limited", time.Second / 20, time.Second / 10, 10 * time.Second, http.StatusGatewayTimeout, "", 0, 1},
+ {"not limited", 5 * time.Second, 10 * time.Second, time.Second / 10, http.StatusOK, ok, 0, 0},
+ {"warned", time.Second / 10, 10 * time.Second, time.Second / 5, http.StatusOK, ok, 1, 0},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ var warnings, limits counter
+ logger, hook := testLogger()
+ b := newBridge(handler.Map{"slow": handler.New(func(ctx context.Context) (any, error) {
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(tc.run):
+ return 1, nil
+ }
+ })}, durationLimits{tc.warning, tc.limit, &warnings, &limits, logger})
+ t.Cleanup(b.Close)
+ rec := postBridge(t, b, `{"jsonrpc":"2.0","id":1,"method":"slow"}`)
+ require.Equal(t, tc.wantCode, rec.Code)
+ require.Equal(t, tc.wantBody, rec.Body.String())
+ require.Equal(t, tc.wantWarnings, warnings.n.Load())
+ require.Equal(t, tc.wantLimits, limits.n.Load())
+ require.Len(t, hook.AllEntries(), int(tc.wantWarnings+tc.wantLimits))
+ })
+ }
+}
+
+// A handler panic answers just that call with -32003 and its stack is logged,
+// whether the handler ran on the request's goroutine or on its own.
+func TestBridge_Panic(t *testing.T) {
+ methods := handler.Map{
+ "panic": handler.New(func(context.Context) (any, error) { panic("test panic") }),
+ "ok": handler.New(func(context.Context) (any, error) { return 1, nil }),
+ }
+ failed := `{"jsonrpc":"2.0","id":1,"error":` +
+ `{"code":-32003,"message":"[-32003] request failed to process due to internal issue"}}`
+ for name, tc := range map[string]struct{ body, want string }{
+ "single": {`{"jsonrpc":"2.0","id":1,"method":"panic"}`, failed},
+ "batch element": {
+ `[{"jsonrpc":"2.0","id":1,"method":"panic"},{"jsonrpc":"2.0","id":2,"method":"ok"}]`,
+ `[` + failed + `,{"jsonrpc":"2.0","id":2,"result":1}]`,
+ },
+ } {
+ t.Run(name, func(t *testing.T) {
+ logger, hook := testLogger()
+ b := newBridge(methods, durationLimits{limit: network.RequestDurationLimiterNoLimit, logger: logger})
+ t.Cleanup(b.Close)
+ rec := postBridge(t, b, tc.body)
+ require.Equal(t, http.StatusOK, rec.Code)
+ require.Equal(t, tc.want, rec.Body.String())
+ entries := hook.AllEntries()
+ require.NotEmpty(t, entries)
+ require.Equal(t, "test panic when calling panic", entries[0].Message)
+ for _, e := range entries {
+ require.Equal(t, logrus.WarnLevel, e.Level)
+ }
+ })
+ }
+}
diff --git a/cmd/stellar-rpc/internal/jsonrpc/jsonrpc.go b/cmd/stellar-rpc/internal/jsonrpc/jsonrpc.go
index 1f691e488..6273e9365 100644
--- a/cmd/stellar-rpc/internal/jsonrpc/jsonrpc.go
+++ b/cmd/stellar-rpc/internal/jsonrpc/jsonrpc.go
@@ -12,12 +12,12 @@ import (
"net/http"
"strconv"
"strings"
+ "sync"
"time"
"unicode"
"github.com/creachadair/jrpc2"
"github.com/creachadair/jrpc2/handler"
- "github.com/creachadair/jrpc2/jhttp"
"github.com/go-chi/chi/middleware"
"github.com/prometheus/client_golang/prometheus"
"github.com/rs/cors"
@@ -52,16 +52,15 @@ const (
type Handler struct {
http.Handler
- bridge jhttp.Bridge
- logger *log.Entry
+ bridge *bridge
+ inflight *sync.WaitGroup
}
-// Close closes all the resources held by the Handler instances.
-// After Close is called the Handler instance will stop accepting JSON RPC requests.
+// Close stops accepting JSON-RPC requests, cancels those in flight and waits
+// for their handlers to return, so none outlives the stores closed after it.
func (h Handler) Close() {
- if err := h.bridge.Close(); err != nil {
- h.logger.WithError(err).Warn("could not close bridge")
- }
+ h.bridge.Close()
+ h.inflight.Wait() // handlers a duration limiter stopped waiting for
}
// HandlerSpec describes one JSON-RPC method: its handler plus the per-method
@@ -120,7 +119,8 @@ func decorateHandlers(daemon host.Daemon, logger *log.Entry, m handler.Map) hand
duration := time.Since(startTime)
label := prometheus.Labels{"endpoint": r.Method(), "status": "ok"}
simulateTransactionResponse, ok := result.(protocol.SimulateTransactionResponse)
- if ok && simulateTransactionResponse.Error != "" {
+ simulateFailed := ok && simulateTransactionResponse.Error != ""
+ if simulateFailed {
label[labelStatus] = "error"
} else if err != nil {
var jsonRPCErr *jrpc2.Error
@@ -130,8 +130,12 @@ func decorateHandlers(daemon host.Daemon, logger *log.Entry, m handler.Map) hand
label[labelStatus] = status
}
}
+ if ctx.Err() != nil && (err != nil || simulateFailed) {
+ // Failed after its context ended: the client left or timed out, not a server fault.
+ label[labelStatus] = "canceled"
+ }
requestMetric.With(label).Observe(duration.Seconds())
- logResponse(logger, reqID, duration, label[labelStatus])
+ logResponse(logger, reqID, r.ID(), duration, label[labelStatus])
return result, err
})
}
@@ -152,12 +156,12 @@ func logRequest(logger *log.Entry, reqID string, req *jrpc2.Request) {
logger.Debug("starting JSONRPC request params")
}
-func logResponse(logger *log.Entry, reqID string, duration time.Duration, status string) {
+func logResponse(logger *log.Entry, reqID, jsonReq string, duration time.Duration, status string) {
logger = logger.WithFields(log.F{
"subsys": "jsonrpc",
"req": reqID,
"duration": duration.String(),
- "json_req": reqID,
+ "json_req": jsonReq,
"status": status,
})
logger.Info("finished JSONRPC request")
@@ -177,7 +181,7 @@ func toSnakeCase(s string) string {
// wrapWithLimiters applies the per-method backlog-queue and request-duration
// limiters (and their metrics) around a single method handler.
-func wrapWithLimiters(spec HandlerSpec, daemon host.Daemon, logger *log.Entry) jrpc2.Handler {
+func wrapWithLimiters(spec HandlerSpec, daemon host.Daemon, logger *log.Entry, inflight *sync.WaitGroup) jrpc2.Handler {
longName := toSnakeCase(spec.MethodName)
queueLimiterGaugeName := longName + "_inflight_requests"
queueLimiterGaugeHelp := "Number of concurrenty in-flight " + spec.MethodName + " requests"
@@ -214,46 +218,30 @@ func wrapWithLimiters(spec HandlerSpec, daemon host.Daemon, logger *log.Entry) j
requestDurationWarn := spec.RequestDurationLimit / warningThresholdDenominator
durationLimiter := network.MakeJrpcRequestDurationLimiter(
queueLimiter.Handle,
+ inflight,
requestDurationWarn,
spec.RequestDurationLimit,
requestDurationWarnCounter,
requestDurationLimitCounter,
logger)
+ if spec.MethodName == protocol.SendTransactionMethodName {
+ // The bridge cancels a handler's context when its client disconnects.
+ // Submission has side effects, so let it finish; the duration limit
+ // still applies.
+ return func(ctx context.Context, req *jrpc2.Request) (any, error) {
+ return durationLimiter.Handle(context.WithoutCancel(ctx), req)
+ }
+ }
return durationLimiter.Handle
}
// NewHandler constructs a Handler instance from the given method specs
func NewHandler(params Params) Handler {
- bridgeOptions := jhttp.BridgeOptions{
- Server: &jrpc2.ServerOptions{
- Logger: func(text string) { params.Logger.Debug(text) },
- // Disable built-in rpc.* methods (e.g. rpc.serverInfo) that
- // bypass the handler allowlist and request limiters.
- DisableBuiltin: true,
- },
- }
-
handlersMap := handler.Map{}
+ inflight := new(sync.WaitGroup)
for _, spec := range params.Specs {
- handlersMap[spec.MethodName] = wrapWithLimiters(spec, params.Daemon, params.Logger)
+ handlersMap[spec.MethodName] = wrapWithLimiters(spec, params.Daemon, params.Logger, inflight)
}
- bridge := jhttp.NewBridge(decorateHandlers(
- params.Daemon,
- params.Logger,
- handlersMap),
- &bridgeOptions)
-
- // globalQueueRequestBacklogLimiter is a metric for measuring the total concurrent inflight requests
- globalQueueRequestBacklogLimiter := prometheus.NewGauge(prometheus.GaugeOpts{
- Namespace: params.Daemon.MetricsNamespace(), Subsystem: subsystemNetwork, Name: "global_inflight_requests",
- Help: "Number of concurrenty in-flight http requests",
- })
-
- queueLimitedBridge := network.MakeHTTPBacklogQueueLimiter(
- bridge,
- globalQueueRequestBacklogLimiter,
- uint64(params.GlobalQueueLimit),
- params.Logger)
globalQueueRequestExecutionDurationWarningCounter := prometheus.NewCounter(prometheus.CounterOpts{
Namespace: params.Daemon.MetricsNamespace(),
@@ -267,15 +255,27 @@ func NewHandler(params Params) Handler {
Name: "global_request_execution_duration_threshold_limit",
Help: "The metric measures the count of requests that surpassed the limit threshold for execution time",
})
- handler := network.MakeHTTPRequestDurationLimiter(
- queueLimitedBridge,
- params.GlobalDurationWarning,
- params.GlobalDurationLimit,
- globalQueueRequestExecutionDurationWarningCounter,
- globalQueueRequestExecutionDurationLimitCounter,
+ rpc := newBridge(decorateHandlers(params.Daemon, params.Logger, handlersMap), durationLimits{
+ warning: params.GlobalDurationWarning,
+ limit: params.GlobalDurationLimit,
+ warnings: globalQueueRequestExecutionDurationWarningCounter,
+ timeouts: globalQueueRequestExecutionDurationLimitCounter,
+ logger: params.Logger,
+ })
+
+ // globalQueueRequestBacklogLimiter is a metric for measuring the total concurrent inflight requests
+ globalQueueRequestBacklogLimiter := prometheus.NewGauge(prometheus.GaugeOpts{
+ Namespace: params.Daemon.MetricsNamespace(), Subsystem: subsystemNetwork, Name: "global_inflight_requests",
+ Help: "Number of concurrenty in-flight http requests",
+ })
+
+ queueLimitedBridge := network.MakeHTTPBacklogQueueLimiter(
+ rpc,
+ globalQueueRequestBacklogLimiter,
+ uint64(params.GlobalQueueLimit),
params.Logger)
- handler = http.MaxBytesHandler(handler, maxHTTPRequestSize)
+ handler := http.MaxBytesHandler(queueLimitedBridge, maxHTTPRequestSize)
corsMiddleware := cors.New(cors.Options{
AllowedOrigins: []string{},
@@ -285,8 +285,8 @@ func NewHandler(params Params) Handler {
})
return Handler{
- bridge: bridge,
- logger: params.Logger,
- Handler: corsMiddleware.Handler(handler),
+ bridge: rpc,
+ inflight: inflight,
+ Handler: corsMiddleware.Handler(handler),
}
}
diff --git a/cmd/stellar-rpc/internal/jsonrpc/jsonrpc_test.go b/cmd/stellar-rpc/internal/jsonrpc/jsonrpc_test.go
new file mode 100644
index 000000000..50d91e10d
--- /dev/null
+++ b/cmd/stellar-rpc/internal/jsonrpc/jsonrpc_test.go
@@ -0,0 +1,53 @@
+package jsonrpc
+
+import (
+ "context"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/creachadair/jrpc2"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ protocol "github.com/stellar/go-stellar-sdk/protocols/rpc"
+ "github.com/stellar/go-stellar-sdk/support/log"
+
+ "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/host"
+)
+
+// A client disconnect cancels the handler context, except for sendTransaction,
+// which keeps its duration limit.
+func TestWrapWithLimiters_ClientDisconnect(t *testing.T) {
+ for _, test := range []struct {
+ method string
+ wantCanceled bool
+ }{
+ {protocol.SendTransactionMethodName, false},
+ {protocol.GetHealthMethodName, true},
+ } {
+ t.Run(test.method, func(t *testing.T) {
+ ctx, disconnect := context.WithCancel(t.Context())
+ var deadline bool
+ h := wrapWithLimiters(HandlerSpec{
+ MethodName: test.method,
+ Handler: func(hctx context.Context, _ *jrpc2.Request) (any, error) {
+ disconnect()
+ _, deadline = hctx.Deadline()
+ return nil, hctx.Err()
+ },
+ QueueLimit: 1,
+ RequestDurationLimit: time.Minute,
+ }, host.MakeNoOpDaemon(), log.DefaultLogger, new(sync.WaitGroup))
+
+ req := (&jrpc2.ParsedRequest{ID: "1", Method: test.method}).ToRequest()
+ _, err := h(ctx, req)
+ if test.wantCanceled {
+ require.ErrorIs(t, err, context.Canceled)
+ } else {
+ require.NoError(t, err)
+ }
+ assert.True(t, deadline, "duration limit not applied")
+ })
+ }
+}
diff --git a/cmd/stellar-rpc/internal/methods/get_ledgers.go b/cmd/stellar-rpc/internal/methods/get_ledgers.go
index 961134a09..5cdce2b92 100644
--- a/cmd/stellar-rpc/internal/methods/get_ledgers.go
+++ b/cmd/stellar-rpc/internal/methods/get_ledgers.go
@@ -40,6 +40,8 @@ func NewGetLedgersHandler(ledgerReader store.LedgerReader, maxLimit, defaultLimi
// getLedgers fetch ledgers and relevant metadata from DB and falling back to
// the remote rpcdatastore if necessary.
+//
+//nolint:cyclop // the context check on the datastore log adds one branch
func (h ledgersHandler) getLedgers(
ctx context.Context, request protocol.GetLedgersRequest,
) (protocol.GetLedgersResponse, error) {
@@ -70,8 +72,10 @@ func (h ledgersHandler) getLedgers(
if h.datastoreLedgerReader != nil {
dsRange, err := h.datastoreLedgerReader.GetAvailableLedgerRange(ctx)
if err != nil {
- // log error but continue using local ledger range
- h.logger.WithError(err).Error("failed to get available ledger range from datastore")
+ // log error but continue using local ledger range; a canceled request is not a fault
+ if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+ h.logger.WithError(err).Error("failed to get available ledger range from datastore")
+ }
} else {
// extend available range to include datastore
availableLedgerRange.FirstLedger = min(dsRange.FirstLedger, availableLedgerRange.FirstLedger)
diff --git a/cmd/stellar-rpc/internal/methods/get_version_info.go b/cmd/stellar-rpc/internal/methods/get_version_info.go
index 54d7c1266..22388ebf7 100644
--- a/cmd/stellar-rpc/internal/methods/get_version_info.go
+++ b/cmd/stellar-rpc/internal/methods/get_version_info.go
@@ -2,6 +2,7 @@ package methods
import (
"context"
+ "errors"
"github.com/creachadair/jrpc2"
@@ -24,7 +25,8 @@ func NewGetVersionInfoHandler(
// version once it starts core, which is after handlers are built.
captiveCoreVersion := daemon.CoreVersion()
protocolVersion, err := getProtocolVersion(ctx, ledgerReader)
- if err != nil {
+ // a canceled request is not a fault
+ if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.WithError(err).Error("failed to fetch protocol version")
}
diff --git a/cmd/stellar-rpc/internal/network/requestdurationlimiter.go b/cmd/stellar-rpc/internal/network/requestdurationlimiter.go
index 6cc3ee170..b820003db 100644
--- a/cmd/stellar-rpc/internal/network/requestdurationlimiter.go
+++ b/cmd/stellar-rpc/internal/network/requestdurationlimiter.go
@@ -2,18 +2,13 @@ package network
import (
"context"
- "maps"
"math"
- "net/http"
- "reflect"
- "runtime"
+ "sync"
"time"
"github.com/creachadair/jrpc2"
"github.com/stellar/go-stellar-sdk/support/log"
-
- "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/util"
)
const maxDuration = time.Duration(math.MaxInt64)
@@ -37,174 +32,16 @@ type requestDurationLimiter struct {
limitCounter increasingCounter
}
-type httpRequestDurationLimiter struct {
- requestDurationLimiter
-
- httpDownstreamHandler http.Handler
-}
-
-func MakeHTTPRequestDurationLimiter(
- downstream http.Handler,
- warningThreshold time.Duration,
- limitThreshold time.Duration,
- warningCounter increasingCounter,
- limitCounter increasingCounter,
- logger *log.Entry,
-) http.Handler {
- // make sure the warning threshold is less then the limit threshold; otherwise, just set it to the limit threshold.
- if warningThreshold > limitThreshold {
- warningThreshold = limitThreshold
- }
- return &httpRequestDurationLimiter{
- httpDownstreamHandler: downstream,
- requestDurationLimiter: requestDurationLimiter{
- warningThreshold: warningThreshold,
- limitThreshold: limitThreshold,
- logger: logger,
- warningCounter: warningCounter,
- limitCounter: limitCounter,
- },
- }
-}
-
-type bufferedResponseWriter struct {
- header http.Header
- buffer []byte
- statusCode int
-}
-
-func makeBufferedResponseWriter(rw http.ResponseWriter) *bufferedResponseWriter {
- header := rw.Header()
- bw := &bufferedResponseWriter{
- header: make(http.Header, 0),
- }
- maps.Copy(bw.header, header)
- return bw
-}
-
-func (w *bufferedResponseWriter) Header() http.Header {
- return w.header
-}
-
-func (w *bufferedResponseWriter) Write(buf []byte) (int, error) {
- w.buffer = append(w.buffer, buf...)
- return len(buf), nil
-}
-
-func (w *bufferedResponseWriter) WriteHeader(statusCode int) {
- w.statusCode = statusCode
-}
-
-func (w *bufferedResponseWriter) WriteOut(ctx context.Context, rw http.ResponseWriter) {
- // update the headers map.
- headers := rw.Header()
- for k := range headers {
- delete(headers, k)
- }
- maps.Copy(headers, w.header)
-
- if len(w.buffer) == 0 {
- if w.statusCode != 0 {
- rw.WriteHeader(w.statusCode)
- }
- return
- }
- if w.statusCode != 0 {
- rw.WriteHeader(w.statusCode)
- }
-
- if ctx.Err() == nil {
- // the following return size/error won't help us much at this point. The request is already finalized.
- rw.Write(w.buffer) //nolint:errcheck
- }
-}
-
-// TODO: refactor and simplify this function
-//
-//nolint:gocognit,cyclop
-func (q *httpRequestDurationLimiter) ServeHTTP(res http.ResponseWriter, req *http.Request) {
- if q.limitThreshold == RequestDurationLimiterNoLimit {
- // if specified max duration, pass-through
- q.httpDownstreamHandler.ServeHTTP(res, req)
- return
- }
- var warningCh <-chan time.Time
- if q.warningThreshold != time.Duration(0) && q.warningThreshold < q.limitThreshold {
- warningCh = time.NewTimer(q.warningThreshold).C
- }
- var limitCh <-chan time.Time
- if q.limitThreshold != time.Duration(0) {
- limitCh = time.NewTimer(q.limitThreshold).C
- }
- requestCompleted := make(chan []string, 1)
- requestCtx, requestCtxCancel := context.WithTimeout(req.Context(), q.limitThreshold)
- defer requestCtxCancel()
- timeLimitedRequest := req.WithContext(requestCtx)
- responseBuffer := makeBufferedResponseWriter(res)
- go func() {
- defer func() {
- if err := recover(); err != nil {
- functionName := runtime.FuncForPC(reflect.ValueOf(q.httpDownstreamHandler.ServeHTTP).Pointer()).Name()
- callStack := util.CallStack(err, functionName, "(*httpRequestDurationLimiter).ServeHTTP.func1()", 8)
- requestCompleted <- callStack
- } else {
- close(requestCompleted)
- }
- }()
- q.httpDownstreamHandler.ServeHTTP(responseBuffer, timeLimitedRequest)
- }()
-
- warn := false
- for {
- select {
- case <-warningCh:
- // warn
- warn = true
- case <-limitCh:
- // limit
- requestCtxCancel()
- if q.limitCounter != nil {
- q.limitCounter.Inc()
- }
- if q.logger != nil {
- q.logger.Infof("Request processing for %s exceed limiting threshold of %v", req.URL.Path, q.limitThreshold)
- }
- if req.Context().Err() == nil {
- res.WriteHeader(http.StatusGatewayTimeout)
- }
- return
- case errStrings := <-requestCompleted:
- if warn {
- if q.warningCounter != nil {
- q.warningCounter.Inc()
- }
- if q.logger != nil {
- q.logger.Infof("Request processing for %s exceed warning threshold of %v", req.URL.Path, q.warningThreshold)
- }
- }
- if len(errStrings) == 0 {
- responseBuffer.WriteOut(req.Context(), res)
- } else {
- res.WriteHeader(http.StatusInternalServerError)
- for _, errStr := range errStrings {
- if q.logger != nil {
- q.logger.Warn(errStr)
- }
- }
- }
- return
- }
- }
-}
-
type RPCRequestDurationLimiter struct {
requestDurationLimiter
jrpcDownstreamHandler jrpc2.Handler
+ inflight *sync.WaitGroup // downstream calls, which can outlive Handle past the limit
}
func MakeJrpcRequestDurationLimiter(
downstream jrpc2.Handler,
+ inflight *sync.WaitGroup,
warningThreshold time.Duration,
limitThreshold time.Duration,
warningCounter increasingCounter,
@@ -218,6 +55,7 @@ func MakeJrpcRequestDurationLimiter(
return &RPCRequestDurationLimiter{
jrpcDownstreamHandler: downstream,
+ inflight: inflight,
requestDurationLimiter: requestDurationLimiter{
warningThreshold: warningThreshold,
limitThreshold: limitThreshold,
@@ -253,7 +91,7 @@ func (q *RPCRequestDurationLimiter) Handle(ctx context.Context, req *jrpc2.Reque
requestCtx, requestCtxCancel := context.WithTimeout(ctx, q.limitThreshold)
defer requestCtxCancel()
- go func() {
+ q.inflight.Go(func() {
defer func() {
if err := recover(); err != nil {
q.logger.Errorf("Request for method %s resulted in an error : %v", req.Method(), err)
@@ -263,7 +101,7 @@ func (q *RPCRequestDurationLimiter) Handle(ctx context.Context, req *jrpc2.Reque
var res requestResultOutput
res.data, res.err = q.jrpcDownstreamHandler(requestCtx, req)
requestCompleted <- res
- }()
+ })
warn := false
for {
diff --git a/cmd/stellar-rpc/internal/network/requestdurationlimiter_test.go b/cmd/stellar-rpc/internal/network/requestdurationlimiter_test.go
index b22970f84..66768c7ec 100644
--- a/cmd/stellar-rpc/internal/network/requestdurationlimiter_test.go
+++ b/cmd/stellar-rpc/internal/network/requestdurationlimiter_test.go
@@ -2,9 +2,10 @@ package network
import (
"context"
- "io"
"net"
"net/http"
+ "sync"
+ "sync/atomic"
"testing"
"time"
@@ -42,132 +43,6 @@ func createTestServer(ctx context.Context) (string, *TestServerHandlerWrapper, c
}
}
-func TestHTTPRequestDurationLimiter_Limiting(t *testing.T) {
- ctx := t.Context()
- addr, redirector, shutdown := createTestServer(ctx)
- longExecutingHandler := &TestServerHandlerWrapper{
- f: func(res http.ResponseWriter, req *http.Request) {
- select {
- case <-req.Context().Done():
- return
- case <-time.After(time.Second * 10):
- }
- n, err := res.Write([]byte{1, 2, 3})
- require.Equal(t, 3, n)
- require.NoError(t, err)
- },
- }
- warningCounter := TestingCounter{}
- limitCounter := TestingCounter{}
- logCounter := makeTestLogCounter()
- redirector.f = MakeHTTPRequestDurationLimiter(
- longExecutingHandler,
- time.Second/20,
- time.Second/10,
- &warningCounter,
- &limitCounter,
- logCounter.Entry()).ServeHTTP
-
- client := http.Client{}
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+addr+"/", nil)
- require.NoError(t, err)
- resp, err := client.Do(req)
- require.NoError(t, err)
- bytes, err := io.ReadAll(resp.Body)
- require.NoError(t, resp.Body.Close())
- require.NoError(t, err)
- require.Equal(t, []byte{}, bytes)
- require.Equal(t, http.StatusGatewayTimeout, resp.StatusCode)
- require.Zero(t, warningCounter.count)
- require.Equal(t, int64(1), limitCounter.count)
- require.Equal(t, [7]int{0, 0, 0, 0, 1, 0, 0}, logCounter.writtenLogEntries)
- shutdown()
-}
-
-func TestHTTPRequestDurationLimiter_NoLimiting(t *testing.T) {
- ctx := t.Context()
- addr, redirector, shutdown := createTestServer(ctx)
- longExecutingHandler := &TestServerHandlerWrapper{
- f: func(res http.ResponseWriter, req *http.Request) {
- select {
- case <-req.Context().Done():
- return
- case <-time.After(time.Second / 10):
- }
- n, err := res.Write([]byte{1, 2, 3})
- require.Equal(t, 3, n)
- require.NoError(t, err)
- },
- }
- warningCounter := TestingCounter{}
- limitCounter := TestingCounter{}
- logCounter := makeTestLogCounter()
- redirector.f = MakeHTTPRequestDurationLimiter(
- longExecutingHandler,
- time.Second*5,
- time.Second*10,
- &warningCounter,
- &limitCounter,
- logCounter.Entry()).ServeHTTP
-
- client := http.Client{}
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+addr+"/", nil)
- require.NoError(t, err)
- resp, err := client.Do(req)
- require.NoError(t, err)
- bytes, err := io.ReadAll(resp.Body)
- require.NoError(t, resp.Body.Close())
- require.NoError(t, err)
- require.Equal(t, []byte{1, 2, 3}, bytes)
- require.Equal(t, http.StatusOK, resp.StatusCode)
- require.Zero(t, warningCounter.count)
- require.Zero(t, limitCounter.count)
- require.Equal(t, [7]int{0, 0, 0, 0, 0, 0, 0}, logCounter.writtenLogEntries)
- shutdown()
-}
-
-func TestHTTPRequestDurationLimiter_NoLimiting_Warn(t *testing.T) {
- ctx := t.Context()
- addr, redirector, shutdown := createTestServer(ctx)
- longExecutingHandler := &TestServerHandlerWrapper{
- f: func(res http.ResponseWriter, req *http.Request) {
- select {
- case <-req.Context().Done():
- return
- case <-time.After(time.Second / 5):
- }
- n, err := res.Write([]byte{1, 2, 3})
- require.Equal(t, 3, n)
- require.NoError(t, err)
- },
- }
- warningCounter := TestingCounter{}
- limitCounter := TestingCounter{}
- logCounter := makeTestLogCounter()
- redirector.f = MakeHTTPRequestDurationLimiter(
- longExecutingHandler,
- time.Second/10,
- time.Second*10,
- &warningCounter,
- &limitCounter,
- logCounter.Entry()).ServeHTTP
-
- client := http.Client{}
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+addr+"/", nil)
- require.NoError(t, err)
- resp, err := client.Do(req)
- require.NoError(t, err)
- bytes, err := io.ReadAll(resp.Body)
- require.NoError(t, resp.Body.Close())
- require.NoError(t, err)
- require.Equal(t, []byte{1, 2, 3}, bytes)
- require.Equal(t, http.StatusOK, resp.StatusCode)
- require.Equal(t, int64(1), warningCounter.count)
- require.Zero(t, limitCounter.count)
- require.Equal(t, [7]int{0, 0, 0, 0, 1, 0, 0}, logCounter.writtenLogEntries)
- shutdown()
-}
-
type JRPCHandlerFunc func(ctx context.Context, r *jrpc2.Request) (any, error)
func bindRPCHoist(redirector *TestServerHandlerWrapper) *JRPCHandlerFunc {
@@ -188,9 +63,12 @@ func TestJRPCRequestDurationLimiter_Limiting(t *testing.T) {
addr, redirector, shutdown := createTestServer(ctx)
hoistFunction := bindRPCHoist(redirector)
+ var returned atomic.Bool
longExecutingHandler := handler.New(func(ctx context.Context, _ *jrpc2.Request) (any, error) {
select {
case <-ctx.Done():
+ time.Sleep(time.Second / 20) // still unwinding after the limiter has answered
+ returned.Store(true)
return nil, ctx.Err()
case <-time.After(time.Second * 10):
}
@@ -200,8 +78,10 @@ func TestJRPCRequestDurationLimiter_Limiting(t *testing.T) {
warningCounter := TestingCounter{}
limitCounter := TestingCounter{}
logCounter := makeTestLogCounter()
+ inflight := new(sync.WaitGroup)
*hoistFunction = MakeJrpcRequestDurationLimiter(
longExecutingHandler,
+ inflight,
time.Second/20,
time.Second/10,
&warningCounter,
@@ -225,6 +105,8 @@ func TestJRPCRequestDurationLimiter_Limiting(t *testing.T) {
require.Zero(t, warningCounter.count)
require.Equal(t, int64(1), limitCounter.count)
require.Equal(t, [7]int{0, 0, 0, 0, 1, 0, 0}, logCounter.writtenLogEntries)
+ inflight.Wait()
+ require.True(t, returned.Load(), "inflight did not wait for the handler")
shutdown()
}
@@ -248,6 +130,7 @@ func TestJRPCRequestDurationLimiter_NoLimiting(t *testing.T) {
logCounter := makeTestLogCounter()
*hoistFunction = MakeJrpcRequestDurationLimiter(
longExecutingHandler,
+ new(sync.WaitGroup),
time.Second*5,
time.Second*10,
&warningCounter,
@@ -291,6 +174,7 @@ func TestJRPCRequestDurationLimiter_NoLimiting_Warn(t *testing.T) {
logCounter := makeTestLogCounter()
*hoistFunction = MakeJrpcRequestDurationLimiter(
longExecutingHandler,
+ new(sync.WaitGroup),
time.Second/10,
time.Second*10,
&warningCounter,
@@ -314,34 +198,3 @@ func TestJRPCRequestDurationLimiter_NoLimiting_Warn(t *testing.T) {
shutdown()
}
-func TestHTTPRequestDurationLimiter_Panicing(t *testing.T) {
- ctx := t.Context()
- addr, redirector, shutdown := createTestServer(ctx)
- longExecutingHandler := &TestServerHandlerWrapper{
- f: func(_ http.ResponseWriter, _ *http.Request) {
- panic("test panic")
- },
- }
-
- logCounter := makeTestLogCounter()
- redirector.f = MakeHTTPRequestDurationLimiter(
- longExecutingHandler,
- time.Second*10,
- time.Second*10,
- nil,
- nil,
- logCounter.Entry()).ServeHTTP
-
- client := http.Client{}
- req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+addr+"/", nil)
- require.NoError(t, err)
- resp, err := client.Do(req)
- require.NoError(t, err)
- bytes, err := io.ReadAll(resp.Body)
- require.NoError(t, err)
- require.NoError(t, resp.Body.Close())
- require.Equal(t, http.StatusInternalServerError, resp.StatusCode)
- require.Equal(t, []byte{}, bytes)
- require.Equal(t, [7]int{0, 0, 0, 7, 0, 0, 0}, logCounter.writtenLogEntries)
- shutdown()
-}
diff --git a/cmd/stellar-rpc/internal/preflight/preflight.go b/cmd/stellar-rpc/internal/preflight/preflight.go
index 6f808e21f..5401548ec 100644
--- a/cmd/stellar-rpc/internal/preflight/preflight.go
+++ b/cmd/stellar-rpc/internal/preflight/preflight.go
@@ -19,6 +19,7 @@ import "C"
import (
"context"
+ "errors"
"fmt"
"runtime/cgo"
"unsafe"
@@ -56,7 +57,10 @@ func SnapshotSourceGet(handle C.uintptr_t, cLedgerKey C.xdr_t) C.ledger_entry_an
}
entries, _, err := h.ledgerEntryGetter.GetLedgerEntries(h.ctx, []xdr.LedgerKey{ledgerKey})
if err != nil {
- h.logger.WithError(err).Error("SnapshotSourceGet(): GetLedgerEntries() failed")
+ // a canceled request is not a fault
+ if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+ h.logger.WithError(err).Error("SnapshotSourceGet(): GetLedgerEntries() failed")
+ }
return C.ledger_entry_and_ttl_t{}
}
if len(entries) > 1 {