eth/filters: decouple client notification delivery from event fan-out - #2335
eth/filters: decouple client notification delivery from event fan-out#2335mukul3097 wants to merge 2 commits into
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).
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
|
codegenie review |
🧞 Codegenie ReviewReviewed all 9 hunks (4 deep, 5 normal), no skipped or failed hunks. Two verified issues remain after deduplication.
Open follow-ups for the author (not filed as findings): whether the CoverageReviewed 9/9 hunks.
|
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. |
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.