Conversation
The filter EventSystem fans events out to every installed subscription from a single eventLoop goroutine using blocking channel sends, and the per-subscription goroutines in the RPC API deliver to clients with a synchronous notifier.Notify. A WebSocket client that stops reading (or reads very slowly, never tripping the write deadline) therefore back-pressures through its subscription channel into the shared loop: one stalled client freezes newPendingTransactions, newHeads, logs, receipts and state-sync delivery for every other subscriber on the node, while eth_subscribe keeps returning valid IDs because installs interleave with the blocked sends. Observed in production on Polygon mainnet: a single stalled subscriber reduced newPendingTransactions delivery for all other clients from ~900 to ~5 notifications per 15s for days; node restarts did not help because the client reconnected immediately. Insert a bounded queue between each subscription's event feed and the client write: enqueueing never blocks, and a per-subscription goroutine drains the queue into notifier.Notify. A client that falls more than clientNotificationBuffer notifications behind loses subsequent notifications for itself only; in-process EventSystem delivery semantics are unchanged. The regression test stalls a raw-pipe client after subscribing and asserts a healthy client still receives all events promptly; without this change it stalls after exactly buffer-size events (129 of 200).
|
codegenie review |
🧞 Codegenie ReviewWarning Review incomplete. Some review or verification work did not complete. Treat the results below as partial and consider rerunning. Two distinct verified issues in the new
Coverage: 11/11 hunks reviewed (7 deep, 3 normal, 1 light), 0 skipped, 0 failed. Partial: 1 candidate verification was incomplete, and open follow-up questions remain about the new tests' timing assumptions ( CoverageReview completed with incomplete verification for 1 candidate.
|
There was a problem hiding this comment.
Pull request overview
This PR addresses an RPC-client-triggerable head-of-line blocking issue in eth/filters subscriptions by decoupling client notification delivery from the shared EventSystem fan-out loop, ensuring one slow/stalled subscription client cannot starve other subscribers.
Changes:
- Add a per-subscription bounded async notification queue (
notifyAsync/queueNotification) to prevent client back-pressure from blocking the shared event fan-out loop. - Update RPC subscription endpoints (
NewPendingTransactions,NewHeads,Logs,TransactionReceipts, and BorNewDeposits) to enqueue notifications instead of callingnotifier.Notifyinline. - Add a regression test that reproduces the slow-client starvation scenario and asserts other subscribers remain unaffected.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| eth/filters/api.go | Introduces async notification queueing and switches multiple subscription paths to non-blocking enqueue. |
| eth/filters/bor_api.go | Applies async notification queueing to Bor deposits subscription delivery. |
| eth/filters/api_slow_client_test.go | Adds regression test covering slow/stalled client behavior vs. healthy subscribers. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| case logs := <-matchedLogs: | ||
| for _, log := range logs { | ||
| notifier.Notify(rpcSub.ID, &log) | ||
| queueNotification(queue, &log) | ||
| } |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧞 Codegenie Review
Reviewed all 9 hunks (4 deep, 5 normal), no skipped or failed hunks. Two verified issues remain after deduplication.
-
Notification drop policy is unobservable to clients (
eth/filters/api.go,eth/filters/bor_api.go). Four packets independently converged on the same delta:queueNotificationuses a non-blocking send with an emptydefault, andnotifyAsyncdiscardsnotifier.Notify's error, so a subscriber more thanclientNotificationBuffer(512) notifications behind gets a gappedlogs/newHeads/newPendingTransactions/transactionReceipts/newDepositsstream with no error, no gap marker, no teardown, and no log/metric. The decoupling itself matches declared intent; the open question is the client-facing contract for overflow. Merged into one inline finding. -
New regression test in
eth/filters/api_slow_client_test.gois both flaky and unable to fail pre-fix. Two verified sub-findings merged: (a) no barrier betweenEthSubscribereturning andEventSysteminstallation, so earlytxFeed.Sendcalls can be dropped and the exact-count loop then fails with a misleading starvation message; (b) the 5s deadline is created after the blocking send loop andhealthyis buffered for all 200 events, so with the fix reverted the ~10srpcwrite timeout tears down the stalled client and the test still passes.
Open follow-ups for the author (not filed as findings): whether the NewPendingTransactions doc comment was intentionally detached by inserting clientNotificationBuffer/notifyAsync between comment and function; whether any existing eth/filters test encodes a full-delivery/ordering contract for newHeads/logs that the drop policy would violate.
Reviewed 9/9 hunks.
Coverage levels: deep 4, normal 5, light 0, skip 0.
— codegenie v0.5.5 (58f82a9b2c) · View Workflow Job
| if h != nil && (crit.ID == h.ID || crit.Contract == h.Contract || | ||
| (crit.ID == 0 && crit.Contract == common.Address{})) { | ||
| notifier.Notify(rpcSub.ID, h) | ||
| queueNotification(queue, h) |
There was a problem hiding this comment.
Every eth_subscribe stream in this package now discards payloads once the client falls more than clientNotificationBuffer (512) notifications behind, and nothing observable happens when it does — no error, no gap marker, no subscription teardown, no log or metric.
// eth/filters/api.go
const clientNotificationBuffer = 512
func notifyAsync(notifier *rpc.Notifier, id rpc.ID, stop <-chan struct{}) chan<- any {
queue := make(chan any, clientNotificationBuffer)
go func() {
for {
select {
case v := <-queue:
_ = notifier.Notify(id, v) // error discarded
case <-stop:
return
}
}
}()
return queue
}
func queueNotification(queue chan<- any, v any) {
select {
case queue <- v:
default: // payload dropped, caller cannot tell
}
}All five subscription paths were converted to this lossy send: NewPendingTransactions (eth/filters/api.go:280, :282), NewHeads (:350), Logs (:388), TransactionReceipts (:469), and NewDeposits (eth/filters/bor_api.go:85).
// before
notifier.Notify(rpcSub.ID, h)
// after
queueNotification(queue, h)Impact: the delivery contract changes. Previously rpc.Notifier.Notify wrote synchronously to the connection, so a subscriber either received every matching event or the write failed and rpcSub.Err() fired, which the client could observe and act on. Now a slow client keeps an apparently healthy subscription while silently receiving an incomplete stream — it sees head N then head N+k with no indication that N+1..N+k-1 were skipped. Consumers that assume completeness (log indexers, reorg/head tracking, confirmation tracking, bor bridge/state-sync consumers of newDeposits) will derive wrong state with no trigger to resubscribe or backfill. Because notifyAsync also drops Notify's error, a genuinely failed write is equally invisible.
The code comments state the intent (client delivery must never back-pressure into it; Once a client falls this far behind, further notifications are dropped for that client only), and decoupling the shared eventLoop from client back-pressure is clearly desirable. What is not settled by the PR description is the client-facing overflow contract: whether callers should be disconnected so they can resubscribe, or should keep a truncated stream. Please confirm the intended semantics with the RPC spec/callers.
Suggested fix: make the overflow observable. Either terminate the affected subscription so rpcSub.Err() fires, or at minimum emit a log/metric so gapped streams are diagnosable.
func queueNotification(queue chan<- any, v any) bool {
select {
case queue <- v:
return true
default:
return false
}
}
// at each call site:
if !queueNotification(queue, h) {
droppedNotificationsMeter.Mark(1)
return // client must resubscribe rather than silently miss data
}If the lossy behavior is deliberate and should stay, document it in the eth_subscribe RPC docs so clients know the stream is not gap-free.
Suggested test: install a logs/newHeads/newDeposits subscription with a notifier whose writes are stalled, push more than clientNotificationBuffer events, then assert the intended overflow contract (subscription torn down with an error, or drop surfaced via metric/log) rather than payloads vanishing unobserved. No existing test in eth/filters covers this boundary:
grep -rn 'queueNotification\|notifyAsync\|clientNotificationBuffer' eth/filters/*_test.go
# no matches| sub, err := client.EthSubscribe(context.Background(), healthy, "newPendingTransactions") | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| defer sub.Unsubscribe() | ||
|
|
||
| for i := 0; i < events; i++ { |
There was a problem hiding this comment.
TestSlowClientDoesNotStarveOtherSubscribers has two defects that together make it both flaky and unable to detect a regression of the fix it guards.
1. No barrier between EthSubscribe returning and the EventSystem subscription being installed.
sub, err := client.EthSubscribe(context.Background(), healthy, "newPendingTransactions")
if err != nil {
t.Fatal(err)
}
defer sub.Unsubscribe()
for i := 0; i < events; i++ {
tx := types.NewTransaction(uint64(i), common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil)
backend.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{tx}})
}FilterAPI.NewPendingTransactions installs the subscription inside a background goroutine after returning rpcSub, and installation only completes when es.subscribe round-trips through the eventLoop (eth/filters/api.go:251-292, eth/filters/filter_system.go:428-441). EthSubscribe returning therefore establishes no happens-before with installation. Any of the 200 txFeed.Send calls processed before installation is never fanned out, and because the receive loop demands exactly events deliveries, one missed event blocks for the full 5s and fails with healthy subscriber starved by stalled client — a misleading, non-deterministic failure that falsely accuses the code under test. Pre-existing tests in the same package (eth/filters/filter_system_test.go:303, :335, :395, :740) insert time.Sleep for exactly this reason, and this test crosses an additional RPC boundary with no barrier at all.
2. The timeout window starts after the stall has already resolved, so the test passes pre-fix.
for i := 0; i < events; i++ {
// ... backend.txFeed.Send(...) <- this is where pre-fix back-pressure manifests; untimed
}
received := 0
timeout := time.After(5 * time.Second)
for received < events {
select {
case <-healthy:
received++
case <-timeout:
t.Fatalf("healthy subscriber starved by stalled client: got %d of %d events", received, events)
}
}With queueNotification(queue, tx.Hash()) reverted to _ = notifier.Notify(rpcSub.ID, tx.Hash()), the stalled net.Pipe subscriber blocks the shared eventLoop and thus blocks txFeed.Send inside the send loop — which carries no time assertion. That block is bounded at ~10s by rpc's write deadline, after which the stalled connection is closed and its subscription removed:
// rpc/json.go
defaultWriteTimeout = 10 * time.Second // used if context has no deadline
func (c *jsonCodec) writeJSON(ctx context.Context, v interface{}, isErrorResponse bool) error {
deadline, ok := ctx.Deadline()
if !ok {
deadline = time.Now().Add(defaultWriteTimeout)
}
c.conn.SetWriteDeadline(deadline)Notifier.send passes context.Background(), so the default applies. The remaining events then flow into healthy, which is created as make(chan common.Hash, events) and buffers all 200. Only afterwards is timeout created, and the loop drains the buffer instantly. The assertion passes with and without the fix.
Impact: no production behavior is affected, but this is the only regression test accompanying a behavior-changing fix. Today it can fail spuriously on loaded CI (burning ~5s); tomorrow a revert or refactor that reintroduces blocking delivery would ship green, and the starvation bug described in the PR body could return undetected.
Suggested fix: add an installation barrier, then assert latency during fan-out rather than after it.
// Barrier: publish warm-up txs until one is observed, proving installation.
warm := time.NewTicker(20 * time.Millisecond)
defer warm.Stop()
installed := time.After(5 * time.Second)
for ready := false; !ready; {
select {
case <-warm.C:
backend.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{warmupTx}})
case <-healthy:
ready = true
case <-installed:
t.Fatal("subscription never installed")
}
}
done := make(chan struct{})
go func() {
defer close(done)
for i := 0; i < events; i++ {
backend.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{newTx(i)}})
}
}()
received := 0
deadline := time.After(3 * time.Second) // well under rpc defaultWriteTimeout (10s)
for received < events {
select {
case <-healthy:
received++
case err := <-sub.Err():
t.Fatalf("healthy subscription failed: %v", err)
case <-deadline:
t.Fatalf("healthy subscriber starved by stalled client: got %d of %d events", received, events)
}
}
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("txFeed.Send blocked by stalled client")
}Suggested test: validate the guard by temporarily reverting queueNotification(queue, tx.Hash()) to _ = notifier.Notify(rpcSub.ID, tx.Hash()) in FilterAPI.NewPendingTransactions and confirming the test fails. If it still passes, it does not protect the fix.
go test ./eth/filters/ -run TestSlowClientDoesNotStarveOtherSubscribers -race -count=20The first version of this change bounded per-client notification delivery but dropped payloads silently once a client fell clientNotificationBuffer behind: no error, no gap marker, no teardown, no log or metric. That changed the delivery contract, since a subscriber previously either received every matching event or saw its write fail through rpcSub.Err(). A consumer that assumes completeness (log indexers, head and reorg tracking, state-sync consumers) would derive wrong state with nothing to trigger a resubscribe or backfill. Overflow now drops the subscription rather than the payload. The queue and its drain goroutine move into a clientNotifier that closes a failed channel on either a full queue or a Notify error, which each subscription loop selects on and returns from, unsubscribing from the EventSystem. Notify errors are no longer discarded, and drops increment rpc/subscription/dropped and log the reason, so a slow consumer is distinguishable from a node that stopped producing events. Also harden the regression test: EthSubscribe returns before the handler installs the subscription in the EventSystem, so events sent in that window were dropped by the feed and could fail the exact-count assertion for the wrong reason. It now waits for installation first. Adds a unit test for the overflow policy itself.
|
Pushed a commit addressing the overflow contract. Unobservable drops. Fixed. Overflow now drops the subscription rather than the payload: the queue and its drain goroutine live in a Note the current Test install race. Real, fixed. The test now waits for the subscription to be installed in the "Test cannot fail pre-fix". This one does not reproduce. With the fix reverted and the test otherwise unchanged, it fails deterministically 3/3: The healthy subscriber starves well inside the 5s deadline, before the ~10s RPC write timeout can tear down the stalled client. With the fix applied it passes 20/20 under
One request: the workflow runs on this fork PR need approval, so only the Socket Security checks have executed. The unit test, lint, Diffguard and Sonar gates have not run against these commits yet. |
|
@claude review |
|
codegenie review |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (3)
eth/filters/api.go:431
- In Logs subscription delivery,
client.send(&log)queues the address of the loop variable. Because delivery is now asynchronous, all queued pointers can end up referencing the same variable and encode the wrong log entry (and it also sends **types.Log sincelogsis already []*types.Log). Send the log pointer itself instead.
case logs := <-matchedLogs:
for _, log := range logs {
client.send(&log)
}
eth/filters/api.go:267
- This comment says a full queue "drops the subscription". In practice it only signals failure via
failedand relies on the caller to tear down the EventSystem subscription; the RPC subscription remains until the client unsubscribes or disconnects. Reword to avoid implying RPC-side unsubscribe.
// send enqueues v for asynchronous delivery to the client. It never blocks,
// isolating the caller (and transitively the shared event fan-out loop) from
// slow clients. A full queue drops the subscription rather than the payload.
func (c *clientNotifier) send(v any) {
eth/filters/api.go:229
- The doc comment claims the subscription is "dropped" and that the client can observe a closed subscription. This code only signals
failedand relies on the caller to stop producing/unsubscribe from the EventSystem; it does not close the RPC subscription. Please adjust the comment to match actual behavior.
This issue also appears on line 264 of the same file.
// ever blocking the caller. Delivery is bounded rather than lossy: a client that
// falls clientNotificationBuffer behind, or whose connection write fails, has
// its subscription dropped instead of being served a stream with a silent gap it
// cannot detect. Callers must select on failed and return, which unsubscribes
// from the EventSystem and lets the client observe the closed subscription.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| // could not keep up, so operators can tell a slow consumer apart from a node | ||
| // that stopped producing events. | ||
| var subscriptionsDroppedCounter = metrics.GetOrRegisterCounter("rpc/subscription/dropped", nil) | ||
|
|
||
| // clientNotifier delivers notifications to a single subscription client without | ||
| // ever blocking the caller. Delivery is bounded rather than lossy: a client that | ||
| // falls clientNotificationBuffer behind, or whose connection write fails, has |
There was a problem hiding this comment.
🟡 subscriptionsDroppedCounter (rpc/subscription/dropped, eth/filters/api.go:222) is a single process-wide counter incremented identically by clientNotifier.fail() for all five subscription types (NewPendingTransactions, NewHeads, Logs, TransactionReceipts, NewDeposits), so an operator watching it rise has no way to tell which RPC stream is overloaded — the accompanying log.Warn carries the subscription id but not the method name either. Consider a per-method label (e.g. metrics.NewRegisteredLabelCounter) or separate counters per subscription type, and including the method name in the log line, so operators can see which surface is under load directly from metrics.
Extended reasoning...
The bug: subscriptionsDroppedCounter (eth/filters/api.go:222) is declared once as metrics.GetOrRegisterCounter("rpc/subscription/dropped", nil) and incremented only inside clientNotifier.fail() (api.go:278-284). Every one of the five subscription constructors in this PR — NewPendingTransactions, NewHeads, Logs, TransactionReceipts (all in api.go), and NewDeposits (bor_api.go) — builds its clientNotifier via the same notifyAsync helper and routes both overflow (send() hitting a full queue) and write failures through the same fail() method. clientNotifier itself only stores the opaque id rpc.ID, never the subscription's method name, so there is no way to recover which RPC surface a given drop came from once it reaches the counter.\n\nWhy this matters for this PR specifically: the whole point of this change is to give operators a way to detect and reason about a slow-consumer scenario that previously manifested as silent, hard-to-diagnose mempool starvation (per the PR description, it took the author's team days to even characterize the symptom in production). The one new signal this PR adds for that purpose — rpc/subscription/dropped — collapses five semantically distinct failure modes (a flooded newPendingTransactions client, a slow logs consumer, a stalled newDeposits/state-sync bridge consumer, etc.) into a single scalar. An operator's dashboard can show the counter climbing with zero indication of which subscription type is responsible, forcing them back to grepping log.Warn lines by id — and even those log lines don't carry the method name, only the opaque subscription id, so correlating still requires extra bookkeeping the operator doesn't have.\n\nConcrete walk-through: suppose a logs subscriber with a heavy filter falls behind at the same time a newDeposits (state-sync bridge) consumer on a different connection also falls behind. Both call paths hit clientNotifier.send() → queue full → fail("client fell behind", nil) → subscriptionsDroppedCounter.Inc(1). The metric now reads 2, with no label distinguishing the RPC-log drop from the state-sync drop — despite these having very different operational implications (a lagging log indexer vs. a bridge consumer missing deposits). The operator's only recourse is to scan log.Warn output for every "Dropping RPC subscription" line and manually correlate ids to subscriptions they've handed out, which for a busy public RPC endpoint with many concurrent subscriptions is impractical.\n\nWhy existing code doesn't prevent it: nothing in the design routes the subscription's method name into clientNotifier or into the counter; it's a straightforward oversight of using one global counter for all call sites rather than parameterizing it per caller.\n\nSuggested fix: pass the subscription type into notifyAsync/clientNotifier and use metrics.NewRegisteredLabelCounter (or five separate counters, one per method) keyed by that type, and include the same label in the log.Warn call. Cardinality is bounded at 5, so this is cheap.\n\nOn the refutation: one verifier argued this isn't a "cleanup" in the reuse/simplification/efficiency/altitude sense and is instead a subjective enhancement request, noting the single counter was a deliberate, minimal addition and the log already carries the id. That's a fair characterization of why it's low priority, but it doesn't make the finding factually wrong or harmful to raise — it's a genuine, low-cost observability gap in the one new signal this PR introduces for exactly the failure mode the PR is about, and per this repo's grading rubric a quality/observability finding like this is scored as a nit (does not block merge) rather than omitted.
| defer pendingTxSub.Unsubscribe() | ||
|
|
||
| stop := make(chan struct{}) | ||
| defer close(stop) | ||
| client := notifyAsync(notifier, rpcSub.ID, stop) | ||
|
|
There was a problem hiding this comment.
🟡 sweep:stop := make\(chan struct\{\}\)\s*\n\s*defer close\(stop\)\s*\n\s*client := notifyAsync
This three-line setup (create stop channel, defer close, construct notifier) is copy-pasted verbatim in all five subscription handlers, e.g. NewPendingTransactions (eth/filters/api.go:302-304) and NewDeposits (eth/filters/bor_api.go:76-78). notifyAsync could own the channel's lifecycle internally and return a clientNotifier with a close()/stop() method, collapsing each site to client := notifyAsync(notifier, rpcSub.ID); defer client.close().
Extended reasoning...
The exact three statements — stop := make(chan struct{}), defer close(stop), client := notifyAsync(notifier, rpcSub.ID, stop) — are duplicated verbatim across all five subscription handlers this PR touches: NewPendingTransactions, NewHeads, Logs, and TransactionReceipts in eth/filters/api.go, plus NewDeposits in eth/filters/bor_api.go. In every single case the stop channel is created immediately before the notifyAsync call and closed only by the immediately-following defer, and it has no other reader or writer anywhere in the goroutine. Its entire lifecycle — birth and death — is scoped 1:1 to the clientNotifier it feeds into.
That means the channel does not need to be a caller-managed value at all. notifyAsync already owns and starts the drain goroutine that consumes stop; there is no reason the caller has to also own the channel that stops it. The channel could instead live inside clientNotifier itself (which already exists as a struct with id, queue, failed, and failOnce fields), with notifyAsync allocating it internally and exposing a close()/Stop() method that closes it. Each call site then collapses from three lines to two: client := notifyAsync(notifier, rpcSub.ID); defer client.close().
Step-by-step proof of the duplication:
eth/filters/api.go:301-304(NewPendingTransactions):stop := make(chan struct{}); defer close(stop); client := notifyAsync(notifier, rpcSub.ID, stop).eth/filters/api.go(NewHeads, right afterdefer headersSub.Unsubscribe()): identical triple.eth/filters/api.go(Logs, right afterdefer logsSub.Unsubscribe()): identical triple.eth/filters/api.go(TransactionReceipts, right afterdefer receiptsSub.Unsubscribe()): identical triple.eth/filters/bor_api.go:75-78(NewDeposits): identical triple.
In every one of these five sites, grepping for any other use of the local stop variable inside the enclosing goroutine turns up nothing beyond the make, the defer close, and the pass into notifyAsync — confirming the channel is genuinely internal plumbing that leaked into the callers scope rather than a value the caller needs to inspect or reuse.
This does not affect correctness of the current PR — the behavior is identical either way — so it does not block merging. But it is exactly the kind of boilerplate that a sixth subscription type (or a future refactor of one of these five) will copy-paste again rather than notice and factor out, since nothing about the current API signature signals that the channel is purely internal. Moving channel ownership into notifyAsync/clientNotifier removes the raw channel from every call site and makes it structurally impossible to forget the paired create/defer-close, since the constructor and destructor become one API. This is a code-quality/reuse observation with no test or behavior implications either way.
| func awaitInstalled(t *testing.T, backend *testBackend, delivered <-chan common.Hash) { | ||
| t.Helper() | ||
|
|
||
| tx := types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil) | ||
| deadline := time.After(5 * time.Second) | ||
|
|
||
| for { | ||
| backend.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{tx}}) | ||
|
|
||
| select { | ||
| case <-delivered: | ||
| // Drain anything the earlier probes delivered so the caller starts | ||
| // from an empty channel. | ||
| for { | ||
| select { | ||
| case <-delivered: | ||
| default: | ||
| return | ||
| } | ||
| } | ||
| case <-deadline: | ||
| t.Fatal("subscription was never installed in the EventSystem") | ||
| case <-time.After(10 * time.Millisecond): | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🟡 In awaitInstalled (eth/filters/api_slow_client_test.go:124-149), the drain at lines 137-143 is non-blocking, so if two probes are ever in flight at once, the still-in-flight one isn't caught by the drain and lands later during the exact-count receive loop. Since that loop just does received++ on any arrival without checking payload identity, a stray probe can let received == events be satisfied one real event short, contradicting the barrier's own comment that the caller "starts from an empty channel." Low-probability and test-only (can mask at most ~1 dropped event); fix by looping the drain until quiescent or tagging probes with an identifiable payload.
Extended reasoning...
awaitInstalled sends a probe transaction every 10ms until one is observed on delivered, then performs a single non-blocking drain (the inner select/default at lines 138-144) before returning. The comment directly above it states the intent plainly: "Drain anything the earlier probes delivered so the caller starts from an empty channel." That guarantee does not actually hold, because delivery past the initial backend.txFeed.Send is asynchronous with variable latency: feed.Send only synchronously hands off to the EventSystem's eventLoop channel; everything downstream of that (subscription queue → notifyAsync drain goroutine → notifier.Notify → RPC write → in-proc client dispatch → the healthy/delivered channel) runs on separate goroutines with no synchronization back to the sender.
Walking through the failure mode concretely: on some iteration of the loop, probe A is sent and takes longer than 10ms to traverse the pipeline. Before A lands, the 10ms timeout fires and probe B is sent. Then A arrives on delivered, satisfying the <-delivered case in the outer select — at that instant B is still somewhere in the pipeline (not yet enqueued on the channel delivered reads from). The subsequent drain loop only sees "what's currently sitting in the channel," finds nothing (since only A had arrived and was already consumed by the outer select), hits default, and returns. awaitInstalled reports success while probe B is still in flight. B eventually lands — either during the 200-event fan-out loop (lines 99-102) or, worse, during the exact-count receive loop (lines 110-119). That loop is for received < events { case <-healthy: received++ ... } — it has no way to distinguish a stray probe hash from a real event hash (both use the same synthetic 0xb794... recipient address construction), so the stray simply counts as one of the events needed, and the loop can exit with received == events while one genuine event was actually dropped by the pipeline.
This is a distinct, narrower defect from the already-known "no install barrier" issue that awaitInstalled itself was written to fix: this is a residual race inside the barrier's own cleanup step, not the absence of a barrier. It's also much lower-impact than the starvation bug the test exists to catch — the pre-fix regression drops on the order of 70 of 200 events (the PR reports "got 128 of 200"), which a single stray probe obviously cannot mask. But the exact-count assertion itself is not actually exact: under load (this test runs with t.Parallel()), a latency spike past 10ms on the async delivery path is plausible enough that this is a real, if rare, test-hygiene gap, not a theoretical one.
The fix is straightforward: either loop the drain until it observes a period of quiescence (e.g., drain, then wait one full probe interval and drain again, repeating until a pass finds nothing), or give probes an identifiable payload (e.g., a reserved nonce/sentinel value) so the exact-count loop can skip them instead of blindly incrementing received on any arrival.
All three independent verifiers reproduced the same causal chain and agreed the race is real, while noting (correctly) that the "silently masks the exact regression this test guards against" framing in the original bug description overstates impact — the pre-fix failure mode is wholesale starvation, not a single dropped event, so this residual race only threatens a hypothetical future one-event regression, not the one this PR is actually about. That caveat doesn't change the conclusion that the barrier doesn't deliver the guarantee its own comment claims.
| func (c *clientNotifier) fail(reason string, err error) { | ||
| c.failOnce.Do(func() { | ||
| subscriptionsDroppedCounter.Inc(1) | ||
| log.Warn("Dropping RPC subscription", "id", c.id, "reason", reason, "buffer", clientNotificationBuffer, "err", err) | ||
| close(c.failed) | ||
| }) | ||
| } |
There was a problem hiding this comment.
🟡 clientNotifier.fail() (eth/filters/api.go:278-284, mirrored in NewHeads/Logs/TransactionReceipts and bor_api.go's NewDeposits) tears down the EventSystem subscription and drain goroutine but never removes the entry from rpc.Handler's serverSubs map or closes rpcSub.err, since rpc.Subscription exposes no server-side teardown hook. On a long-lived connection that repeatedly triggers the 512-notification overflow, each drop leaves a dangling serverSubs entry (cleaned only on full disconnect) and the client's ClientSubscription.Err() never fires, so eth_unsubscribe on an already-dropped ID still returns true and the client is left holding a subscription that looks healthy but silently delivers nothing.
Extended reasoning...
The bug
clientNotifier.fail() is the new teardown path this PR introduces for a client that falls behind or hits a write error. It closes c.failed, which the API goroutine selects on via the new case <-client.failed: return branch. That return runs the deferred pendingTxSub.Unsubscribe() / headersSub.Unsubscribe() / logsSub.Unsubscribe() / receiptsSub.Unsubscribe() (or the explicit stateSyncSub.Unsubscribe() in bor_api.go), which tears down the internal EventSystem subscription, and it lets stop close, which kills the notifyAsync drain goroutine. That is the entire cleanup this callback performs.
Why it leaks
rpc.Handler tracks a second, independent piece of server-side state per subscription: handler.serverSubs[id] (rpc/handler.go:71), populated by addSubscriptions right after the subscription method returns (rpc/handler.go:378-386). I verified there are only two places that ever remove an entry from that map: handler.unsubscribe(), triggered by the client sending eth_unsubscribe (rpc/handler.go:625-638, closes s.err and deletes the map entry), and handler.cancelServerSubscriptions(), triggered when the whole connection tears down (rpc/handler.go:391-400, iterates the map, sends the error into s.err, closes it, deletes it). rpc.Subscription itself (rpc/subscription.go:203-214) exposes only ID, Err(), and MarshalJSON — there is no method the filters package can call to reach into the handler and delete its own entry or close its own err channel.
Pre-PR, the API goroutines only had two exits: rpcSub.Err() (which fires because one of the two paths above already ran, so serverSubs was already consistent) or the deferred Unsubscribe() on connection close (cleaned via cancelServerSubscriptions). This PR adds a third exit — <-client.failed — that goes around both cleanup mechanisms entirely. Nothing in the new code calls back into rpc.Handler.
Concrete walk-through
- A WS client opens one subscription (say
newHeads) and it happens to fall behind —queueNotification/sendfills the 512-entry buffer,clientNotifier.fail("client fell behind", nil)runs, closingc.failed. - The
NewHeadsgoroutine hitscase <-client.failed: return, which unsubscribes from theEventSystemand stops the drain goroutine.rpcSub.err(therpc.Subscriptioncreated for this call) is untouched, andhandler.serverSubs[rpcSub.ID]still holds the entry. - The client, still on the same connection, calls
eth_subscribe("newHeads")again to get a fresh stream, and later triggers overflow again. Repeat. - Each cycle leaves one more stale entry in
serverSubsthat nothing frees until the whole connection eventually disconnects (cancelServerSubscriptions) — exactly the multi-day-connection scenario the PR's own description says these production nodes run under. - Separately, if the client (not knowing delivery already stopped) calls
eth_unsubscribe(id)on the dropped ID,handler.unsubscribe()finds the entry still present inserverSubs, happily deletes it, and returnstrue— indistinguishable from unsubscribing a live, healthy stream. And becauserpcSub.errwas never closed by the callback, the client-sideClientSubscription.Err()also never fires, so a well-behaved client has no signal at all that its subscription died; it just silently stops receiving updates.
Why existing code/tests don't catch this
The new TestClientNotifierDropsSubscriptionOnOverflow test only exercises clientNotifier in isolation and asserts c.failed closes — it never goes through the full FilterAPI subscription path or touches rpc.Handler, so it can't observe the serverSubs leak or the missing rpcSub.err closure. TestSlowClientDoesNotStarveOtherSubscribers only checks that a healthy subscriber isn't starved; it never asserts anything about the stalled subscriber's own cleanup.
Fix
The callback needs a way to close rpcSub.err and remove the entry from handler.serverSubs when a subscription is dropped — either by exposing a server-side teardown method on rpc.Subscription/rpc.Notifier that internally calls the equivalent of handler.unsubscribe(ctx, id), or by having Notify's error path and the overflow path route through that same call. The author already flagged in the PR timeline that "the current rpc.Notifier API has no way to hand a server-side error to a subscriber" — this is the concrete gap that needs to be closed, not just the client-observability question raised there.
Why this is a nit rather than blocking
Each leaked *Subscription is small and inert — the underlying EventSystem subscription and drain goroutine are already torn down by the time the entry goes stale, so there's no ongoing resource consumption or goroutine leak, just a dangling map entry that is fully reclaimed the moment the connection closes. Reaching a meaningfully large leak requires one connection to repeatedly resubscribe-and-overflow many times, which is a slow, low-severity accumulation rather than a crash, data-loss, or consensus-affecting bug. It is real and worth fixing, and it does undercut the PR's stated observability goal for the client side, but it should not block merge.
There was a problem hiding this comment.
🧞 Codegenie Review
Warning
Review incomplete. Some review or verification work did not complete. Treat the results below as partial and consider rerunning.
Two distinct verified issues in the new clientNotifier async delivery path in eth/filters/api.go, both bounded to per-subscription impact and both marked intentional_needs_confirmation (mixed refactor/behavior-change intent):
-
Queue-overflow teardown is not observable by the client (3 findings merged, high/medium confidence).
fail()only increments a counter, logs, and closesc.failed; the subscription goroutine returns and unsubscribes from theEventSystem, but the RPC-level subscription stays registered on a healthy open connection with nothing written to the client. This contradicts the new doc comment ateth/filters/api.go:224-229which claims the client observes the closed subscription. Needs either a client-visible teardown or a corrected comment. -
A single oversized event batch can drop healthy subscribers (2 findings merged, medium confidence).
LogsandNewPendingTransactionsenqueue one queue slot per item from an in-memory batch while the drain goroutine performs one synchronous socket write per item, so a block with more thanclientNotificationBuffer(512) matching logs can overflow and terminate the subscription of a normally-reading client.TransactionReceiptsis unaffected (one batch = one send).
Coverage: 11/11 hunks reviewed (7 deep, 3 normal, 1 light), 0 skipped, 0 failed. Partial: 1 candidate verification was incomplete, and open follow-up questions remain about the new tests' timing assumptions (awaitInstalled drain semantics and subscription-installation ordering in eth/filters/api_slow_client_test.go) — those were not converted into findings because they were not verified.
Review completed with incomplete verification for 1 candidate.
Reviewed 11/11 hunks before stopping.
Coverage disclosure:
- Verification incomplete for 1 candidate.
— codegenie v0.5.5 (58f82a9b2c) · View Workflow Job
| // clientNotifier delivers notifications to a single subscription client without | ||
| // ever blocking the caller. Delivery is bounded rather than lossy: a client that | ||
| // falls clientNotificationBuffer behind, or whose connection write fails, has | ||
| // its subscription dropped instead of being served a stream with a silent gap it |
There was a problem hiding this comment.
The queue-overflow teardown path emits no signal to the client, so the new doc comment's claim that the client "observe[s] the closed subscription" does not hold.
When send finds the queue full it takes the default arm and calls fail, which only increments a counter, logs, and closes c.failed:
func (c *clientNotifier) send(v any) {
select {
case c.queue <- v:
default:
c.fail("client fell behind", nil)
}
}
func (c *clientNotifier) fail(reason string, err error) {
c.failOnce.Do(func() {
subscriptionsDroppedCounter.Inc(1)
log.Warn("Dropping RPC subscription", "id", c.id, "reason", reason, "buffer", clientNotificationBuffer, "err", err)
close(c.failed)
})
}The only caller reaction is:
case <-client.failed:
returnReturning runs defer close(stop) and the Unsubscribe() defer, which detaches the subscription from the filter EventSystem only. Nothing touches the RPC transport. In rpc/handler.go, a server *Subscription's err channel is closed exclusively by cancelServerSubscriptions (connection teardown) or handler.unsubscribe (client-initiated *_unsubscribe), and rpc/subscription.go exposes no server-initiated close/error API on Notifier:
type Subscription struct {
ID ID
namespace string
err chan error // closed on unsubscribe
}Impact: on an otherwise healthy connection the subscription ID remains registered in handler.serverSubs, the client's ClientSubscription.Err() never fires, and no further notifications ever arrive. The client cannot distinguish this from an idle chain — the exact undetectable gap the new comment says is prevented. The pre-change blocking _ = notifier.Notify(...) either delivered every payload or failed the write (which closes the connection and is therefore visible). Note the write-failure path (fail("notification write failed", err)) remains client-visible because Notify closes the connection on error; only the fell-behind path is silent. Blast radius is per-subscription: applications on eth_subscribe (newPendingTransactions, newHeads, logs, receipts) stall indefinitely on a dead-but-open subscription and will not resubscribe.
This is a contract change with mixed intent signals. Please confirm the intended client-facing semantics.
Suggested fix: either make the drop observable — emit a terminal error/gap notification via notifier.Notify before returning, or add an RPC-layer API to close the server subscription so handler.serverSubs is cleared and err is closed — or correct the comment at eth/filters/api.go:224-229 to state that the client only observes the absence of further notifications and must rely on its own liveness detection.
Suggested test: install a subscription over an in-process WebSocket/RPC client, stall the client read loop until more than clientNotificationBuffer events are queued, then assert what the client observes. Today it receives nothing further and ClientSubscription.Err() never fires; eth/filters/api_notifier_test.go currently only asserts the server-side drop counter.
go test ./eth/filters/ -run 'TestClientNotifier|TestSlowClient' -race -count=5| // send enqueues v for asynchronous delivery to the client. It never blocks, | ||
| // isolating the caller (and transitively the shared event fan-out loop) from | ||
| // slow clients. A full queue drops the subscription rather than the payload. | ||
| func (c *clientNotifier) send(v any) { |
There was a problem hiding this comment.
Logs and NewPendingTransactions enqueue one queue slot per item from an in-memory batch, so a single oversized event batch can exceed clientNotificationBuffer (512) and terminate the subscription of a client that is reading normally.
stop := make(chan struct{})
defer close(stop)
client := notifyAsync(notifier, rpcSub.ID, stop)
for {
select {
case logs := <-matchedLogs:
for _, log := range logs {
client.send(&log)
}
case <-client.failed:
returnIn eth/filters/filter_system.go, one channel send carries every log matching a whole block:
func (es *EventSystem) handleLogs(filters filterIndex, ev []*types.Log) {
for _, f := range filters[LogsSubscription] {
matchedLogs := filterLogs(ev, ...)
if len(matchedLogs) > 0 {
f.logs <- matchedLogs
}So the batch size is bounded by the block's log count, not by client speed. The subscription goroutine fills the 512-slot queue in a tight in-memory loop while the drain goroutine performs one JSON marshal plus one synchronous notifier.Notify connection write per item. The first overflow hits the default branch in send, fail("client fell behind", nil) closes failed, and the goroutine returns and unsubscribes permanently. NewPendingTransactions has the same per-item loop:
for _, tx := range txs {
if fullTx != nil && *fullTx {
rpcTx := ethapi.NewRPCPendingTransaction(tx, latest, chainConfig)
client.send(rpcTx)
} else {
client.send(tx.Hash())
}
}TransactionReceipts is not affected: it performs a single client.send(marshaledReceipts) per batch.
Impact: a broad eth_subscribe("logs") filter on a busy chain — commonly well over 512 matching logs per block — can have its stream permanently torn down on the first busy block even though the client never stopped reading. Base code delivered the whole batch via blocking Notify (bounded by the RPC write deadline). The dropped counter and warn log will attribute this to a slow consumer, masking a capacity mismatch between the fixed 512-item bound and per-block batch sizes. The comment at eth/filters/api.go:211-217 justifies the buffer in terms of "a WebSocket client that stops reading (or reads very slowly)", so the intended trigger appears to be a non-reading client rather than one large batch — please confirm whether tearing down healthy subscribers on burst is the intended contract.
Suggested fix: count the bound in event batches rather than individual notifications — enqueue the []*types.Log slice as one queue item, as TransactionReceipts already does — or tolerate transient producer bursts by selecting on queue, stop/failed and a bounded timer before failing, so only a genuinely non-reading client is dropped.
Suggested test: drive Logs (or clientNotifier via notifyAsync with a slow-but-progressing notifier stub) with one matchedLogs batch of clientNotificationBuffer + N logs and assert the subscription is not dropped and every log is delivered. The existing eth/filters/api_notifier_test.go exercises the overflow policy with no drain goroutine and does not cover a single batch larger than the buffer.
go test ./eth/filters/ -run TestLogs -race -count=5
pratikspatil024
left a comment
pratikspatil024
left a comment
There was a problem hiding this comment.
Requesting changes for three additional blockers not covered by the existing bot threads. Codegenie/Claude already raised the client-invisible RPC teardown and the >512 healthy-batch overflow, so I have not duplicated those comments. These findings are in addition to those open threads. I also confirmed the current head remains bbe7009; the eth/filters race suite passes locally, while the required diffguard check is currently failing.
| func notifyAsync(notifier *rpc.Notifier, id rpc.ID, stop <-chan struct{}) *clientNotifier { | ||
| c := &clientNotifier{ | ||
| id: id, | ||
| queue: make(chan any, clientNotificationBuffer), |
There was a problem hiding this comment.
This bounds each queue but not the number of queues. Every accepted public eth_subscribe call now eagerly allocates a 512-entry chan any (about 8 KiB of interface slots before payload retention) and another goroutine, while the RPC layer has no per-connection or global subscription budget. A client can therefore create unbounded idle subscriptions and turn this fix into a deterministic memory/scheduler amplification path; fullTx streams can retain substantially more through queued payload objects. Please enforce a subscription budget before allocating this state (and ideally bound queued bytes, not only item count), or use a shared/lazy dispatcher, with an exhaustion test.
| for { | ||
| select { | ||
| case v := <-c.queue: | ||
| if err := notifier.Notify(id, v); err != nil { |
There was a problem hiding this comment.
The 512-item bound is bypassed before the RPC subscription is activated. This drain goroutine starts immediately, but rpc.Notifier.Notify appends to the notifier’s unbounded buffer while activated == false (rpc/subscription.go:150-154). If the client delays reading the subscribe response while events arrive, notifications are continuously moved out of this bounded queue into that unbounded buffer. Please either delay draining until activation succeeds or enforce the same bound inside rpc.Notifier, and add a codec test that blocks the initial subscription response while publishing more than 512 events.
| client.send(h) | ||
| } | ||
| case <-client.failed: | ||
| stateSyncSub.Unsubscribe() |
There was a problem hiding this comment.
This overflow teardown can deadlock the shared EventSystem. After client.failed, this goroutine stops reading stateSyncData and calls stateSyncSub.Unsubscribe() synchronously. Generic Subscription.Unsubscribe drains logs, txs, headers, and receipts while waiting to submit the uninstall request, but it does not drain stateSyncData (filter_system.go:306-313). If handleStateSyncEvent is blocked sending the next deposit (bor_filter_system.go:11-14), the event loop cannot receive the uninstall request and this call cannot unblock it, freezing every subscription type again. Please drain stateSyncData in generic unsubscribe (and centralize teardown with a defer), then add an overflow/unsubscribe race test proving other subscribers continue to receive events.
|
This PR is stale because it has been open 21 days with no activity. Remove stale label or comment or this will be closed in 14 days. |
Problem
The filter
EventSystemfans events out to every installed subscription from a singleeventLoopgoroutine using blocking channel sends (eth/filters/filter_system.go,handleTxsEventet al.), and the per-subscription goroutines ineth/filters/api.godeliver to clients with a synchronousnotifier.Notify. The two together create a back-pressure chain from an untrusted RPC client into shared node state:A WebSocket client that stops reading its connection — or reads just slowly enough never to trip the RPC write deadline — blocks its subscription goroutine in
Notify, its subscription channel fills, and the sharedeventLoopthen blocks on that one subscriber. From that moment every subscription on the node starves:newPendingTransactions,newHeads,logs, transaction receipts, and state-sync deposits (they all share the loop, so the starvation crosses subscription types). The failure is deceptive becauseeth_subscribekeeps returning valid subscription IDs — the install channel interleaves between blocked sends — while delivery trickles at the pace of the slowest client.Per the repo's own threat-model framing this is an RPC-user-triggerable DoS on a public endpoint: any single WS client can, accidentally or deliberately, suppress subscription delivery for all other clients of the node.
Production impact
We operate large Polygon PoS RPC infrastructure. On a mainnet full node (bor v2.9.0, 200 peers, at chain tip, txpool ingesting ~64 tx/s throughout),
newPendingTransactionssubscribers received 2–13 hashes per 15s instead of ~900 for several days. Restarting bor did not help — the offending client auto-reconnected through the load balancer and re-wedged the fresh process; the node recovered only when an LB restart severed all client sessions. We reported the symptom ("progressive mempool starvation on subscribe") through the operator channel in April without a reproduction; this PR includes the reproduction that was missing.Fix
Insert a bounded queue between each subscription's event feed and the client write (
notifyAsync/queueNotificationinapi.go): enqueueing never blocks, and a per-subscription goroutine drains the queue intonotifier.Notify. A client that falls more thanclientNotificationBuffer(512) notifications behind loses subsequent notifications for itself only.Deliberate properties of this approach:
EventSystemsemantics are untouched. In-process subscribers keep guaranteed, ordered, blocking delivery — all existingeth/filterstests pass unmodified. The isolation boundary sits exactly where the untrusted party (the RPC client) attaches.rpc/rpchelper'schan_sub.Senddrops on overflow), so the two clients become consistent under a slow consumer.Alternatives considered: per-send timeouts in the fan-out loop (retains head-of-line blocking for the timeout duration, multiplied across subscribers); dropping at the
EventSystemlayer (breaks the guaranteed-delivery contract thatTestBlockSubscriptionandTestTransactionReceiptsSubscriptioncorrectly encode for in-process consumers — rejected after trying it); relying on the RPC write deadline (already insufficient in practice — a trickling client never trips it).Testing
TestSlowClientDoesNotStarveOtherSubscribers: a raw-pipe client subscribes and then stops reading; a healthy in-proc client must still receive all 200 events promptly. On currentdevelopit fails withgot 129 of 200 events— exactly the stalled subscriber's channel buffer (128) plus one in-flight before the shared loop froze. With this change it passes in ~1s.eth/filterssuite passes, including with-race(29/29).Happy to adjust details (queue size, a metrics counter for dropped notifications, drop-oldest vs drop-newest) if maintainers prefer — the property we need is that one slow client cannot affect other subscribers.