feat(rpc): gate WSS requests and bound active subscriptions - #3883
feat(rpc): gate WSS requests and bound active subscriptions#3883infrmtcs wants to merge 1 commit into
Conversation
|
Claude finished @infrmtcs's task in 7m 12s —— View job Review complete
SummaryThe direction is right and the two mechanisms are wired in cleanly. I traced the acquire/release pairing carefully and it is correct — no leaks, no double-releases:
No blockers. Findings by severity: Important
Nit
Not verified: |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| handlerv8 := rpcv8.New(bcReader, syncReader, virtualMachine, logger) | ||
| handlerv9 := rpcv9.New(bcReader, syncReader, virtualMachine, logger) | ||
| handlerv10 := rpcv10.New(bcReader, syncReader, virtualMachine, logger) | ||
| subscriptionLimiter := semaphore.NewWeighted(rpccore.DefaultMaxSubscriptions) |
There was a problem hiding this comment.
important — the limit is process-wide, not per-connection
The PR description says "Bound the number of active subscriptions per connection", but this is a single semaphore.Weighted(2048) shared by all three handlers and, transitively, by every websocket connection in the process. There is no per-connection accounting anywhere in subscribe().
Combined with jsonrpc/websocket.go:21 (maxConns = 2048), a single client that opens one connection and issues 2048 starknet_subscribe* calls exhausts the global pool, and every other client — including already-connected ones — gets ErrTooManySubscriptions on their next subscribe. That's cheaper to trigger than the resource exhaustion this is meant to prevent.
Suggest tracking a per-connection counter (keyed on the jsonrpc.Conn, which subscribe already receives, and which has Equal for identity) and keeping the global cap as a second, larger backstop.
| func (ws *Websocket) handleMessage(wsc *websocketConn) error { | ||
| if ws.gate != nil { | ||
| acquireCtx := wsc.ctx | ||
| if ws.requestTimeout > 0 { | ||
| var cancel context.CancelFunc | ||
| acquireCtx, cancel = context.WithTimeout(acquireCtx, ws.requestTimeout) | ||
| defer cancel() | ||
| } | ||
| if err := ws.gate.Acquire(acquireCtx); err != nil { | ||
| if errors.Is(err, context.Canceled) { | ||
| return err | ||
| } | ||
| _, writeErr := wsc.Write(serverBusyResponse) | ||
| return writeErr | ||
| } | ||
| defer ws.gate.Release() | ||
| } | ||
|
|
||
| return ws.rpc.HandleReadWriter(wsc.ctx, ws.requestTimeout, wsc) | ||
| } |
There was a problem hiding this comment.
important — a websocket request can now consume 2 × rpcRequestTimeout
acquireCtx gets its own requestTimeout deadline (line 172), and then HandleReadWriter derives a fresh requestTimeout from wsc.ctx (jsonrpc/server.go:322-323), not from acquireCtx. So a request that waits requestTimeout-ε in the gate queue still gets a full requestTimeout to execute — worst case double the configured budget, during which the connection's read loop (ServeHTTP line 132) is blocked and no other message on that connection makes progress.
The HTTP path doesn't have this: jsonrpc/http.go:80-105 builds one ctx with the timeout and uses it for both the gate acquire and the handler, so queue time is charged against the request budget.
Suggest building the deadline once here and threading it into HandleReadWriter, e.g. give HandleReadWriter the already-deadlined ctx and pass 0 for requestTimeout:
func (ws *Websocket) handleMessage(wsc *websocketConn) error {
ctx := wsc.ctx
if ws.requestTimeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, ws.requestTimeout)
defer cancel()
}
if ws.gate != nil {
if err := ws.gate.Acquire(ctx); err != nil { ... }
defer ws.gate.Release()
}
return ws.rpc.HandleReadWriter(ctx, 0, wsc)
}Careful with that shape though: HandleReadWriter passes connCtx into connection.ctx, which subscription writes use after the initial response, so handing it a deadlined ctx would break long-lived subscriptions. Either keep the two ctxs separate but subtract the queue wait, or pass the deadline explicitly. Worth a comment either way, since the current asymmetry with HTTP looks unintentional.
| if err := ws.gate.Acquire(acquireCtx); err != nil { | ||
| if errors.Is(err, context.Canceled) { | ||
| return err | ||
| } | ||
| _, writeErr := wsc.Write(serverBusyResponse) | ||
| return writeErr | ||
| } |
There was a problem hiding this comment.
important — rejections here are silent, and DeadlineExceeded is reported as "server busy"
Two things:
-
No log on rejection. The HTTP path calls
h.logServerBusy()(jsonrpc/http.go:60-66), which emits a sampled warning with running/queued/rejected. The websocket path drops the request with no log at all, so an operator debugging "clients randomly get -32603" has nothing to go on. Consider givingWebsocketthe samelog.Sampledlogger and calling it here. -
The two failure modes are conflated.
Gate.Acquirereturns eitherErrServerBusy(queue full — a genuine overload) orcontext.DeadlineExceeded(this request queued longer thanrequestTimeout— the server is busy but this was a timeout). Both fall through toserverBusyResponse. HTTP distinguishes them (http.go:94-102). Worth at least an expliciterrors.Is(err, ErrServerBusy)branch so the two are separable in logs/metrics rather than relying on "notcontext.Canceled" as the catch-all.
Also note the metrics side: makeHTTPGateMetrics (node/metrics.go:110-135) registers these under subsystemHTTP with help text "Number of HTTP RPC requests…". Now that the same gate serves websockets, rpc_http_active_requests / rpc_http_rejected_requests silently include websocket traffic. Either rename to a version-neutral subsystem or update the help strings, otherwise existing dashboards/alerts become wrong after this lands.
| var serverBusyResponse = func() []byte { | ||
| b, err := json.Marshal(&response{ | ||
| Version: "2.0", | ||
| Error: &Error{Code: InternalError, Message: ErrServerBusy.Error()}, | ||
| }) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| return b | ||
| }() |
There was a problem hiding this comment.
important — the busy response is always "id":null, so pipelining clients can't correlate it
Because the gate is checked before the frame is parsed (HandleReadWriter is what reads wsc.r), this canned response carries no request id. Websocket clients routinely have several requests in flight on one connection and match responses by id; an id:null error is unattributable, so a client typically either drops it or fails the wrong request. The HTTP path doesn't have this problem — it answers with a 503 at the transport level, which is unambiguous.
Related: a client that sent a notification (no id) now receives an unsolicited response frame, which is a JSON-RPC violation.
Options: read the frame into a bounded buffer and decode just the id before gating (cost: one extra copy bounded by ReadLimit), or close the connection with a websocket close code instead of a fake response. If the current behaviour is a deliberate trade-off, please note it in a comment on serverBusyResponse.
Separately, InternalError (-32603) is a poor fit semantically — nothing failed internally, the server is shedding load. A dedicated code (or ErrServerBusy mapped to one) would let clients implement backoff instead of treating it as a bug.
|
|
||
| DefaultMaxSubscriptions int64 = 2048 |
There was a problem hiding this comment.
nit — hardcoded, unlike the sibling request gate
rpc-max-concurrent-requests / rpc-max-request-queue are both configurable flags (cmd/juno/juno.go:110,176), but the subscription cap is a compile-time constant with no way to tune or disable it. Node operators with different memory budgets can't adjust it, and there's no way to turn it off if it turns out to be too tight. Worth a rpc-max-subscriptions flag for symmetry — the plumbing already supports it since WithSubscriptionLimiter takes the semaphore.
Also 2048 deserves the same "arbitrary default, revisit after monitoring" note that maxConns carries in jsonrpc/websocket.go:21, and ErrTooManySubscriptions (code 101) is a new Juno-specific error that clients will hit — please add it to the RPC error docs.
There was a problem hiding this comment.
In fact, if we make them configurable, it should be a pair of flags: 1 for connections and 1 for subscriptions.
| if h.subscriptionLimiter != nil && !h.subscriptionLimiter.TryAcquire(1) { | ||
| return "", rpccore.ErrTooManySubscriptions | ||
| } |
There was a problem hiding this comment.
nit — the limit is checked after the expensive work, and the pair is triplicated
The acquire/release pairing itself is correct: there is no early return between TryAcquire (line 57) and sub.wg.Go (line 80), the release is deferred first inside the goroutine so it runs after unsubscribe, and Unsubscribe does subs.wg.Wait() (line 230), so the slot is provably free before starknet_unsubscribe returns true. No leak, no double-release.
Two smaller things:
-
Load is shed too late. All five
Subscribe*entrypoints resolve the block range before callingsubscribe, so a rejected subscription has already paidbcReader.Height()and — for historical subscriptions — the header/commitment/state-update reads. Your own test makes this visible:Height()is expected.Times(3)for two successful subscriptions. Checking the limiter at the top of theSubscribe*methods (or hoisting it into a tiny helper called first) would reject before touching the DB, which is the point of a limiter. -
Identical 6 lines now live in
v8/subscriptions.go:121,138,v9/subscriptions.go:109,131and here. The threesubscribebodies were already near-duplicates, so this follows precedent, but it means a future fix (e.g. the per-connection accounting discussed onrpc/handlers.go) has to be applied three times and can silently diverge. Worth a shared helper inrpccoreif that's cheap here.
| var rpcGate *jsonrpc.Gate | ||
| if cfg.RPCMaxConcurrentRequests > 0 { | ||
| rpcGate = jsonrpc.NewGate(cfg.RPCMaxConcurrentRequests, uint64(cfg.RPCMaxRequestQueue)) | ||
| if cfg.Metrics { | ||
| makeHTTPGateMetrics(rpcGate) | ||
| } | ||
| } |
There was a problem hiding this comment.
nit — hoisting the gate here so HTTP and WS share it is the right call, but it now runs unconditionally: with --http=false --ws=false the node still builds a Gate and (with metrics on) registers three prometheus collectors for a gate nobody acquires. Harmless, but if cfg.HTTP || cfg.Websocket around it would keep /metrics honest.
| func TestSubscribeNewHeadsRespectsLimit(t *testing.T) { | ||
| logger := log.NewNopZapLogger() | ||
| client := feeder.NewTestClient(t, &networks.Sepolia) | ||
| block1, commitments1, stateUpdate1 := GetTestBlockWithCommitments(t, client, 56377) | ||
| adaptedHeader := AdaptBlockHeader(block1.Header, commitments1, stateUpdate1.StateDiff) | ||
|
|
||
| mockCtrl := gomock.NewController(t) | ||
| t.Cleanup(mockCtrl.Finish) | ||
| mockChain := mocks.NewMockReader(mockCtrl) | ||
|
|
||
| handler := New(mockChain, nil, nil, logger). | ||
| WithSubscriptionLimiter(semaphore.NewWeighted(1)) | ||
|
|
||
| mockChain.EXPECT().Height().Return(block1.Number, nil).Times(3) | ||
| mockChain.EXPECT().BlockHeaderByNumber(block1.Number).Return(block1.Header, nil).Times(2) | ||
| mockChain.EXPECT().BlockCommitmentsByNumber(block1.Number).Return(commitments1, nil).Times(2) | ||
| mockChain.EXPECT().StateUpdateByNumber(block1.Number).Return(stateUpdate1, nil).Times(2) | ||
|
|
||
| blockIDLatest := BlockIDLatest() | ||
|
|
||
| subID1, conn1 := createTestNewHeadsWebsocket(t, handler, (*SubscriptionBlockID)(&blockIDLatest)) | ||
| assertNextHead(t, conn1, subID1, &adaptedHeader) | ||
|
|
||
| serverConn, clientConn := net.Pipe() | ||
| t.Cleanup(func() { | ||
| require.NoError(t, serverConn.Close()) | ||
| require.NoError(t, clientConn.Close()) | ||
| }) | ||
| rejConn := &fakeConn{Conn: clientConn, w: serverConn} | ||
| rejCtx := context.WithValue(t.Context(), jsonrpc.ConnKey{}, rejConn) | ||
| id, rpcErr := handler.SubscribeNewHeads(rejCtx, nil) | ||
| assert.Zero(t, id) | ||
| assert.Equal(t, rpccore.ErrTooManySubscriptions, rpcErr) | ||
|
|
||
| unsubCtx := context.WithValue(t.Context(), jsonrpc.ConnKey{}, conn1) | ||
| ok, rpcErr := handler.Unsubscribe(unsubCtx, string(subID1)) | ||
| require.Nil(t, rpcErr) | ||
| require.True(t, ok) | ||
|
|
||
| subID3, conn3 := createTestNewHeadsWebsocket(t, handler, (*SubscriptionBlockID)(&blockIDLatest)) | ||
| assertNextHead(t, conn3, subID3, &adaptedHeader) | ||
| } |
There was a problem hiding this comment.
nit — good test, two gaps
Solid: it exercises reject-at-limit and slot-reclaimed-after-unsubscribe, and it correctly relies on Unsubscribe's wg.Wait() rather than a sleep, so it isn't flaky.
- The identical limiter code in
rpc/v8/subscriptions.go:121andrpc/v9/subscriptions.go:109has no coverage. Since it's copy-pasted three times, that's exactly where a divergence would go unnoticed. - Only
SubscribeNewHeadsis covered.SubscribeEvents/SubscribeTransactionStatus/SubscribeNewTransactions/SubscribeNewTransactionReceiptsall funnel through the samesubscribe, so a small table-driven test over the entrypoints would be cheap insurance that none of them grows an early return betweenTryAcquireandsub.wg.Go.
Also subID3/conn3 skips 2 — presumably left over from an earlier draft.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #3883 +/- ##
==========================================
- Coverage 75.29% 75.12% -0.18%
==========================================
Files 446 446
Lines 40136 40187 +51
==========================================
- Hits 30220 30190 -30
- Misses 7825 7900 +75
- Partials 2091 2097 +6 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
e25b0e8 to
650ead8
Compare
d266b82 to
0401efe
Compare
|
Claude finished @infrmtcs's task in 5m 20s —— View job Review complete
This push was a rebase, not a content change. HEAD moved New findingsNit — Nit — After this PR both flags govern websocket traffic too. Worth fixing in the same pass as the Re-confirmed from the previous passAlso re-verified against the flag defaults this time (
Acquire/release pairing re-traced and still correct — no leaks, no double-releases, and the Not verified: |
| close(release) | ||
| _, _, err = connA.Read(t.Context()) | ||
| require.NoError(t, err) | ||
| require.Eventually(t, func() bool { return gate.Running() == 0 }, time.Second, 5*time.Millisecond) |
There was a problem hiding this comment.
nit — Running() == 0 is not sufficient to know the slot is fully reclaimed
Gate.Release() (jsonrpc/gate.go) does:
func (g *Gate) Release() {
<-g.sem // Running() drops to 0 here
g.decreaseActiveReq() // activeRequests drops to 0 only here
}Running() is len(g.sem), but admission is decided by activeRequests vs maxRequests. So there is a window where Running() == 0 while activeRequests is still 1. If Eventually samples inside that window, the follow-up id:3 request on connB hits increaseActiveReq() == 2 > maxRequests == 1 and gets serverBusyResponse instead of {"result":"hi"} — the test fails.
It's a nanosecond-wide window and there's a websocket round trip after it, so it will almost never fire in CI, but it's a real ordering gap rather than a theoretical one. Cheap fix — assert both counters are drained:
| require.Eventually(t, func() bool { return gate.Running() == 0 }, time.Second, 5*time.Millisecond) | |
| require.Eventually(t, func() bool { return gate.Running() == 0 && gate.Queued() == 0 }, | |
| time.Second, 5*time.Millisecond) |
650ead8 to
ab4786a
Compare
0401efe to
4b82a5c
Compare
|
Claude finished @infrmtcs's task in 6m 31s —— View job Review complete
What changed in this pushThe base branch The new testsThey're good, and deterministic — I traced the ordering rather than assuming it:
One nit posted inline on the new v9 test: the Standing findings — all still openI re-verified each against the current tree rather than trusting the earlier pass. None have been addressed; I have not re-posted them inline.
No blockers. The acquire/release pairing on both mechanisms is correct — no leaks, no double-releases — and the Verification: |
| // Every subscribe resolves the range (Height); only the two that acquire a | ||
| // slot go on to send one historical header (BlockHeaderByNumber). | ||
| mockChain.EXPECT().Height().Return(uint64(0), nil).Times(3) |
There was a problem hiding this comment.
nit — these expectations now pin the "shed load after the DB read" behaviour in a third place
Height().Times(3) for two successful subscribes is the assertion that the rejected subscribe still paid resolveBlockRange before the limiter looked at it — same thing Height().Times(3) records in rpc/v10/subscriptions_test.go:1233 and HeadsHeader().Times(3) in rpc/v8/subscriptions_test.go:562. That's the late-shedding issue from my earlier note on rpc/v10/subscriptions.go. It's fine as a description of today's behaviour, but if the limiter check moves ahead of resolveBlockRange these three exact counts all have to be edited in lockstep, and the comments above them go stale. A short // note: limiter runs after resolveBlockRange, hence 3 not 2 would make that intentional rather than incidental.
Two smaller things while here:
- The
Times(2)onBlockHeaderByNumberand the singleclient1.Readare coupled tolatestBlock == startBlock == 0. If the mocked height ever becomes non-zero,sendHistoricalHeaderswrites more than one frame andUnsubscribe'swg.Wait()blocks on the unbuffered pipe — the test then hangs to the package timeout rather than failing. Your comment on line 1058 already spots the mechanism; draining in a loop until the goroutine parks (or using a buffered writer instead ofnet.Pipe) would make it fail loudly instead. - This body is now duplicated verbatim in v8 and v9 (and near-verbatim in v10). Same concern as the triplicated
subscribebodies: a divergence in one of the three won't be noticed.
Not a blocker — I could not execute go test in this sandbox, so the hang scenario is reasoned from the code, not observed.
|
Stopped, it looks like it requires a bigger refactor all around with http. |
2 participants
Summary
jsonrpc.Gate, same as HTTPHandleReadWriternow takesrequestTimeout)Conn.Context()so subscriptions key off the connection, not the timed-out request contextrpccore.DefaultMaxSubscriptions), returningErrTooManySubscriptionswhen exceeded (v8/v9/v10)node.go/http.gointo the HTTP and WebSocket servers