diff --git a/eth/filters/api.go b/eth/filters/api.go
index 3ca35276fa..9d5d8b61d3 100644
--- a/eth/filters/api.go
+++ b/eth/filters/api.go
@@ -31,6 +31,8 @@ import (
"github.com/ethereum/go-ethereum/core/history"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/internal/ethapi"
+ "github.com/ethereum/go-ethereum/log"
+ "github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rpc"
)
@@ -206,6 +208,81 @@ func (api *FilterAPI) NewPendingTransactionFilter(fullTx *bool) rpc.ID {
return pendingTxSub.ID
}
+// clientNotificationBuffer bounds how many notifications may sit queued for a
+// single subscription client. The filter EventSystem fans events out to every
+// subscription from one shared goroutine with blocking sends, so client
+// delivery must never back-pressure into it: a WebSocket client that stops
+// reading (or reads very slowly) would otherwise stall the shared fan-out
+// loop and starve every other subscription on the node.
+const clientNotificationBuffer = 512
+
+// subscriptionsDroppedCounter counts subscriptions dropped because the client
+// 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
+// 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.
+type clientNotifier struct {
+ id rpc.ID
+ queue chan any
+ failed chan struct{}
+ failOnce sync.Once
+}
+
+// notifyAsync returns a clientNotifier whose queue is drained into
+// notifier.Notify by a background goroutine. That goroutine exits when stop is
+// closed or when delivery fails, so it never outlives the subscription.
+func notifyAsync(notifier *rpc.Notifier, id rpc.ID, stop <-chan struct{}) *clientNotifier {
+ c := &clientNotifier{
+ id: id,
+ queue: make(chan any, clientNotificationBuffer),
+ failed: make(chan struct{}),
+ }
+
+ go func() {
+ for {
+ select {
+ case v := <-c.queue:
+ if err := notifier.Notify(id, v); err != nil {
+ c.fail("notification write failed", err)
+ return
+ }
+ case <-stop:
+ return
+ }
+ }
+ }()
+
+ return c
+}
+
+// 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) {
+ select {
+ case c.queue <- v:
+ default:
+ c.fail("client fell behind", nil)
+ }
+}
+
+// fail closes c.failed exactly once, recording why the subscription is going
+// away. Both callers may race: send runs on the subscription goroutine while
+// the drain goroutine reports write errors.
+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)
+ })
+}
+
// NewPendingTransactions creates a subscription that is triggered each time a
// transaction enters the transaction pool. If fullTx is true the full tx is
// sent to the client, otherwise the hash is sent.
@@ -222,6 +299,10 @@ func (api *FilterAPI) NewPendingTransactions(ctx context.Context, fullTx *bool)
pendingTxSub := api.events.SubscribePendingTxs(txs)
defer pendingTxSub.Unsubscribe()
+ stop := make(chan struct{})
+ defer close(stop)
+ client := notifyAsync(notifier, rpcSub.ID, stop)
+
chainConfig := api.sys.backend.ChainConfig()
for {
@@ -234,11 +315,13 @@ func (api *FilterAPI) NewPendingTransactions(ctx context.Context, fullTx *bool)
for _, tx := range txs {
if fullTx != nil && *fullTx {
rpcTx := ethapi.NewRPCPendingTransaction(tx, latest, chainConfig)
- _ = notifier.Notify(rpcSub.ID, rpcTx)
+ client.send(rpcTx)
} else {
- _ = notifier.Notify(rpcSub.ID, tx.Hash())
+ client.send(tx.Hash())
}
}
+ case <-client.failed:
+ return
case <-rpcSub.Err():
return
}
@@ -297,10 +380,16 @@ func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
headersSub := api.events.SubscribeNewHeads(headers)
defer headersSub.Unsubscribe()
+ stop := make(chan struct{})
+ defer close(stop)
+ client := notifyAsync(notifier, rpcSub.ID, stop)
+
for {
select {
case h := <-headers:
- notifier.Notify(rpcSub.ID, h)
+ client.send(h)
+ case <-client.failed:
+ return
case <-rpcSub.Err():
return
}
@@ -329,12 +418,19 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
go func() {
defer logsSub.Unsubscribe()
+
+ stop := make(chan struct{})
+ defer close(stop)
+ client := notifyAsync(notifier, rpcSub.ID, stop)
+
for {
select {
case logs := <-matchedLogs:
for _, log := range logs {
- notifier.Notify(rpcSub.ID, &log)
+ client.send(&log)
}
+ case <-client.failed:
+ return
case <-rpcSub.Err(): // client send an unsubscribe request
return
}
@@ -390,6 +486,10 @@ func (api *FilterAPI) TransactionReceipts(ctx context.Context, filter *Transacti
go func() {
defer receiptsSub.Unsubscribe()
+ stop := make(chan struct{})
+ defer close(stop)
+ client := notifyAsync(notifier, rpcSub.ID, stop)
+
signer := types.LatestSigner(api.sys.backend.ChainConfig())
for {
@@ -410,8 +510,10 @@ func (api *FilterAPI) TransactionReceipts(ctx context.Context, filter *Transacti
}
// Send a batch of tx receipts in one notification
- notifier.Notify(rpcSub.ID, marshaledReceipts)
+ client.send(marshaledReceipts)
}
+ case <-client.failed:
+ return
case <-rpcSub.Err():
return
}
diff --git a/eth/filters/api_notifier_test.go b/eth/filters/api_notifier_test.go
new file mode 100644
index 0000000000..86974bf719
--- /dev/null
+++ b/eth/filters/api_notifier_test.go
@@ -0,0 +1,60 @@
+// Copyright 2015 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package filters
+
+import "testing"
+
+// TestClientNotifierDropsSubscriptionOnOverflow asserts the overflow policy: a
+// client that falls further than clientNotificationBuffer behind has its
+// subscription dropped and counted, rather than being served a stream with a
+// silent gap it cannot detect.
+func TestClientNotifierDropsSubscriptionOnOverflow(t *testing.T) {
+ // Constructed directly rather than via notifyAsync: with no drain goroutine
+ // nothing leaves the queue, so overflow is reached deterministically.
+ c := &clientNotifier{
+ id: "test-subscription",
+ queue: make(chan any, clientNotificationBuffer),
+ failed: make(chan struct{}),
+ }
+
+ before := subscriptionsDroppedCounter.Snapshot().Count()
+
+ for i := 0; i < clientNotificationBuffer; i++ {
+ c.send(i)
+
+ select {
+ case <-c.failed:
+ t.Fatalf("subscription dropped after %d of %d buffered notifications", i+1, clientNotificationBuffer)
+ default:
+ }
+ }
+
+ c.send("overflow")
+
+ select {
+ case <-c.failed:
+ default:
+ t.Fatal("subscription not dropped after the client fell past the buffer")
+ }
+
+ // Later sends must not re-close the channel or re-count the drop.
+ c.send("after overflow")
+
+ if dropped := subscriptionsDroppedCounter.Snapshot().Count() - before; dropped != 1 {
+ t.Fatalf("counted %d drops, want 1", dropped)
+ }
+}
diff --git a/eth/filters/api_slow_client_test.go b/eth/filters/api_slow_client_test.go
new file mode 100644
index 0000000000..db939cf0ca
--- /dev/null
+++ b/eth/filters/api_slow_client_test.go
@@ -0,0 +1,149 @@
+// Copyright 2026 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package filters
+
+import (
+ "context"
+ "math/big"
+ "net"
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core"
+ "github.com/ethereum/go-ethereum/core/rawdb"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/rpc"
+)
+
+// TestSlowClientDoesNotStarveOtherSubscribers reproduces a production failure:
+// a WebSocket client that subscribes to newPendingTransactions and then stops
+// reading its connection must not affect delivery to other subscribers. The
+// EventSystem fans events out to all subscriptions from one shared goroutine
+// with blocking sends, so if client delivery back-pressures into the
+// subscription channel, one stalled client freezes every subscription on the
+// node until the write deadline fires.
+func TestSlowClientDoesNotStarveOtherSubscribers(t *testing.T) {
+ t.Parallel()
+
+ var (
+ db = rawdb.NewMemoryDatabase()
+ backend, sys = newTestFilterSystem(db, Config{})
+ api = NewFilterAPI(sys, false)
+ )
+
+ server := rpc.NewServer("", 0, 0)
+ defer server.Stop()
+
+ if err := server.RegisterName("eth", api); err != nil {
+ t.Fatal(err)
+ }
+
+ // Stalled client: subscribe over a raw pipe, read the subscription reply,
+ // then never read again. Notification writes to this connection block
+ // until the server's write deadline.
+ srvConn, cliConn := net.Pipe()
+ defer srvConn.Close()
+ defer cliConn.Close()
+
+ go server.ServeCodec(rpc.NewCodec(srvConn), 0)
+
+ if err := cliConn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil {
+ t.Fatal(err)
+ }
+
+ if _, err := cliConn.Write([]byte(`{"jsonrpc":"2.0","id":1,"method":"eth_subscribe","params":["newPendingTransactions"]}` + "\n")); err != nil {
+ t.Fatal(err)
+ }
+
+ reply := make([]byte, 512)
+ if _, err := cliConn.Read(reply); err != nil {
+ t.Fatalf("stalled client never got subscription reply: %v", err)
+ }
+
+ // Healthy client on its own connection.
+ client := rpc.DialInProc(server)
+ defer client.Close()
+
+ const events = 200
+
+ healthy := make(chan common.Hash, events)
+
+ sub, err := client.EthSubscribe(context.Background(), healthy, "newPendingTransactions")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer sub.Unsubscribe()
+
+ // EthSubscribe returns once the server has assigned a subscription ID, which
+ // happens before the handler goroutine installs the subscription in the
+ // EventSystem. Events sent in that window are dropped by the feed and would
+ // make the exact-count assertion below fail for the wrong reason, so send
+ // warm-up events until one is delivered, then discard what they produced.
+ awaitInstalled(t, backend, healthy)
+
+ 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}})
+ }
+
+ // The healthy subscriber must receive every event promptly even though the
+ // stalled client stopped reading. Keep the deadline well under the RPC
+ // write timeout so recovery-by-disconnect cannot mask starvation.
+ received := 0
+ timeout := time.After(5 * time.Second)
+
+ for received < events {
+ select {
+ case <-healthy:
+ received++
+ case err := <-sub.Err():
+ t.Fatalf("healthy subscription failed after %d events: %v", received, err)
+ case <-timeout:
+ t.Fatalf("healthy subscriber starved by stalled client: got %d of %d events", received, events)
+ }
+ }
+}
+
+// awaitInstalled blocks until the subscription feeding delivered is installed in
+// the EventSystem, then drains the events the probing produced.
+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):
+ }
+ }
+}
diff --git a/eth/filters/bor_api.go b/eth/filters/bor_api.go
index fca26d99ad..6a4a39eda6 100644
--- a/eth/filters/bor_api.go
+++ b/eth/filters/bor_api.go
@@ -72,14 +72,21 @@ func (api *FilterAPI) NewDeposits(ctx context.Context, crit ethereum.StateSyncFi
stateSyncData := make(chan *types.StateSyncData, 10)
stateSyncSub := api.events.SubscribeNewDeposits(stateSyncData)
+ stop := make(chan struct{})
+ defer close(stop)
+ client := notifyAsync(notifier, rpcSub.ID, stop)
+
//nolint:staticcheck
for {
select {
case h := <-stateSyncData:
if h != nil && (crit.ID == h.ID || crit.Contract == h.Contract ||
(crit.ID == 0 && crit.Contract == common.Address{})) {
- notifier.Notify(rpcSub.ID, h)
+ client.send(h)
}
+ case <-client.failed:
+ stateSyncSub.Unsubscribe()
+ return
case <-rpcSub.Err():
stateSyncSub.Unsubscribe()
return