eth/filters: decouple client notification delivery from event fan-out

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
 			}

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 <http://www.gnu.org/licenses/>.
+
+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)
+	}
+}

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 <http://www.gnu.org/licenses/>.
+
+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):
+		}
+	}
+}