rpc: evict idle polling filters after a timeout, add subscription metrics

ChangeLog.md

@@ -36,11 +36,27 @@ Aligns Erigon with the `eth_simulateV1` error code specification ([NethermindEth
 
 - If your tooling matches on error code `-32602` to detect base-fee-too-low conditions in `eth_simulateV1` responses, update it to match `-38012` instead.
 
+---
+
+#### JSON-RPC: idle polling filters are evicted after 5 minutes
+
+Filters created with `eth_newFilter`, `eth_newBlockFilter`, and `eth_newPendingTransactionFilter` are now evicted when not polled for 5 minutes, matching geth's stale-filter deadline. Previously they lived — and kept buffering data — until `eth_uninstallFilter` or a restart.
+
+**What changed:**
+
+| Aspect | Before | After |
+|---|---|---|
+| Idle polling filter | kept until uninstalled or restart | evicted after 5 minutes without a poll |
+| `eth_getFilterChanges` / `eth_getFilterLogs` on an evicted id | — | `filter not found` |
+
+**Migration:** poll more often than the timeout, or recreate the filter when `filter not found` is returned (as with geth). Tune with `--rpc.subscription.filters.timeout`; set it to 0 to restore the previous keep-forever behavior. (#22261 by @onelapahead)
+
 ### Added
 
 #### CLI & Operations
 
 - `--prune.distance.blocks` now accepts readable policy names — `keep-post-merge` and `keep-all` — instead of the raw `MaxUint64`-based magic numbers (`18446744073709551615` / `18446744073709551614`); `--prune.distance` likewise accepts `keep-all`. Numeric values still work (#22119) — by @yperbasis
+- `--rpc.subscription.filters.timeout` — deadline for evicting idle RPC polling filters (default 5m; 0 disables). New `subscriptions_active` gauge and `subscriptions_created_total` / `subscriptions_unsubscribed_total` / `subscriptions_reaped_total` counters track the filter lifecycle (#22261) — by @onelapahead
 
 ---
 

cmd/rpcdaemon/cli/config.go

@@ -181,6 +181,7 @@ func RootCommand() (*cobra.Command, *httpcfg.HttpCfg) {
 	rootCmd.PersistentFlags().IntVar(&cfg.RpcFiltersConfig.RpcSubscriptionFiltersMaxTxs, "rpc.subscription.filters.maxtxs", rpchelper.DefaultFiltersConfig.RpcSubscriptionFiltersMaxTxs, "Maximum number of transactions to store per subscription.")
 	rootCmd.PersistentFlags().IntVar(&cfg.RpcFiltersConfig.RpcSubscriptionFiltersMaxAddresses, "rpc.subscription.filters.maxaddresses", rpchelper.DefaultFiltersConfig.RpcSubscriptionFiltersMaxAddresses, "Maximum number of addresses per subscription to filter logs by.")
 	rootCmd.PersistentFlags().IntVar(&cfg.RpcFiltersConfig.RpcSubscriptionFiltersMaxTopics, "rpc.subscription.filters.maxtopics", rpchelper.DefaultFiltersConfig.RpcSubscriptionFiltersMaxTopics, "Maximum number of topics per subscription to filter logs by.")
+	rootCmd.PersistentFlags().DurationVar(&cfg.RpcFiltersConfig.RpcSubscriptionFiltersTimeout, "rpc.subscription.filters.timeout", rpchelper.DefaultFiltersConfig.RpcSubscriptionFiltersTimeout, "Timeout before idle filters are evicted. Defaults to 5m; set to 0 to disable eviction.")
 	rootCmd.PersistentFlags().IntVar(&cfg.BlockRangeLimit, utils.RpcBlockRangeLimit.Name, utils.RpcBlockRangeLimit.Value, utils.RpcBlockRangeLimit.Usage)
 	rootCmd.PersistentFlags().IntVar(&cfg.GetLogsMaxResults, utils.RpcGetLogsMaxResults.Name, utils.RpcGetLogsMaxResults.Value, utils.RpcGetLogsMaxResults.Usage)
 	rootCmd.PersistentFlags().IntVar(&cfg.BatchLimit, utils.RpcBatchLimit.Name, utils.RpcBatchLimit.Value, utils.RpcBatchLimit.Usage)

diagnostics/metrics/countervec.go

@@ -0,0 +1,75 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon 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.
+//
+// Erigon 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 Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+// Since we wrapped prometheus.Counter into a custom interface, we need to wrap prometheus.CounterVec into a custom struct
+type CounterVec struct {
+	*prometheus.CounterVec
+}
+
+func (cv *CounterVec) Collect(ch chan<- prometheus.Metric) {
+	cv.MetricVec.Collect(ch)
+}
+
+func (cv *CounterVec) CurryWith(labels prometheus.Labels) (*CounterVec, error) {
+	cv2, err := cv.CounterVec.CurryWith(labels)
+	return &CounterVec{cv2}, err
+}
+
+func (cv *CounterVec) Delete(labels prometheus.Labels) bool {
+	return cv.CounterVec.MetricVec.Delete(labels)
+}
+
+func (cv *CounterVec) DeleteLabelValues(lvs ...string) bool {
+	return cv.CounterVec.MetricVec.DeleteLabelValues(lvs...)
+}
+
+func (cv *CounterVec) DeletePartialMatch(labels prometheus.Labels) int {
+	return cv.CounterVec.MetricVec.DeletePartialMatch(labels)
+}
+
+func (cv *CounterVec) Describe(ch chan<- *prometheus.Desc) {
+	cv.CounterVec.MetricVec.Describe(ch)
+}
+
+func (cv *CounterVec) GetMetricWith(labels prometheus.Labels) (Counter, error) {
+	c, err := cv.CounterVec.GetMetricWith(labels)
+	return &counter{c}, err
+}
+
+func (cv *CounterVec) GetMetricWithLabelValues(lvs ...string) (Counter, error) {
+	c, err := cv.CounterVec.GetMetricWithLabelValues(lvs...)
+	return &counter{c}, err
+}
+
+func (cv *CounterVec) MustCurryWith(labels prometheus.Labels) *CounterVec {
+	return &CounterVec{cv.CounterVec.MustCurryWith(labels)}
+}
+
+func (cv *CounterVec) Reset() {
+	cv.CounterVec.MetricVec.Reset()
+}
+
+func (cv *CounterVec) With(labels prometheus.Labels) Counter {
+	return &counter{cv.CounterVec.With(labels)}
+}
+
+func (cv *CounterVec) WithLabelValues(lvs ...string) Counter {
+	return &counter{cv.CounterVec.WithLabelValues(lvs...)}
+}

diagnostics/metrics/register.go

@@ -63,6 +63,26 @@ func GetOrCreateCounter(name string) Counter {
 	return &counter{c}
 }
 
+// GetOrCreateCounterVec returns registered CounterVec with the given name
+// or creates a new CounterVec if the registry doesn't contain a CounterVec with
+// the given name and labels.
+//
+// name must be a valid Prometheus-compatible metric with possible labels.
+// labels are the names of the dimensions associated with the counter vector.
+// For instance,
+//
+//   - foo, with labels []string{"bar", "baz"}
+//
+// The returned CounterVec is safe to use from concurrent goroutines.
+func GetOrCreateCounterVec(name string, labels []string, help ...string) *CounterVec {
+	cv, err := defaultSet.GetOrCreateCounterVec(name, labels, help...)
+	if err != nil {
+		panic(fmt.Errorf("could not get or create new countervec: %w", err))
+	}
+
+	return &CounterVec{cv}
+}
+
 // NewGauge registers and returns gauge with the given name.
 //
 // name must be valid Prometheus-compatible metric with possible labels.

diagnostics/metrics/set.go

@@ -211,6 +211,46 @@ func (s *Set) GetOrCreateCounter(name string, help ...string) (prometheus.Counte
 	})
 }
 
+// GetOrCreateCounterVec returns registered CounterVec in s with the given name
+// or creates new CounterVec if s doesn't contain CounterVec with the given name.
+//
+// name must be valid Prometheus-compatible metric with possible labels.
+// For instance,
+//
+//   - foo
+//   - foo{bar="baz"}
+//   - foo{bar="baz",aaa="b"}
+//
+// labels are the labels associated with the CounterVec.
+//
+// The returned CounterVec is safe to use from concurrent goroutines.
+func (s *Set) GetOrCreateCounterVec(name string, labels []string, help ...string) (*prometheus.CounterVec, error) {
+	return getOrCreateVec(s, name, func() (*prometheus.CounterVec, error) {
+		return newCounterVec(name, labels, help...)
+	})
+}
+
+// newCounterVec creates a new Prometheus CounterVec.
+func newCounterVec(name string, labels []string, help ...string) (*prometheus.CounterVec, error) {
+	name, constLabels, err := parseMetric(name)
+	if err != nil {
+		return nil, err
+	}
+
+	helpStr := "counter metric"
+	if len(help) > 0 {
+		helpStr = strings.Join(help, ", ")
+	}
+
+	cv := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name:        name,
+		Help:        helpStr,
+		ConstLabels: constLabels,
+	}, labels)
+
+	return cv, nil
+}
+
 // NewGauge registers and returns gauge with the given name.
 //
 // name must be valid Prometheus-compatible metric with possible labels.

node/cli/default_flags.go

@@ -123,6 +123,7 @@ var DefaultFlags = []cli.Flag{
 	&RpcSubscriptionFiltersMaxTxsFlag,
 	&RpcSubscriptionFiltersMaxAddressesFlag,
 	&RpcSubscriptionFiltersMaxTopicsFlag,
+	&RpcSubscriptionFiltersTimeoutFlag,
 
 	&utils.SnapKeepBlocksFlag,
 	&utils.SnapStopFlag,

node/cli/flags.go

@@ -207,6 +207,11 @@ var (
 		Usage: "Maximum number of topics per subscription to filter logs by.",
 		Value: rpchelper.DefaultFiltersConfig.RpcSubscriptionFiltersMaxTopics,
 	}
+	RpcSubscriptionFiltersTimeoutFlag = cli.DurationFlag{
+		Name:  "rpc.subscription.filters.timeout",
+		Usage: "Timeout before idle filters are evicted. Defaults to 5m; set to 0 to disable eviction.",
+		Value: rpchelper.DefaultFiltersConfig.RpcSubscriptionFiltersTimeout,
+	}
 )
 
 // BuildEthConfig applies all CLI flags to the ethconfig.Config. This is the single
@@ -490,6 +495,7 @@ func setEmbeddedRpcDaemon(ctx *cli.Command, cfg *nodecfg.Config, logger log.Logg
 			RpcSubscriptionFiltersMaxTxs:       ctx.Int(RpcSubscriptionFiltersMaxTxsFlag.Name),
 			RpcSubscriptionFiltersMaxAddresses: ctx.Int(RpcSubscriptionFiltersMaxAddressesFlag.Name),
 			RpcSubscriptionFiltersMaxTopics:    ctx.Int(RpcSubscriptionFiltersMaxTopicsFlag.Name),
+			RpcSubscriptionFiltersTimeout:      ctx.Duration(RpcSubscriptionFiltersTimeoutFlag.Name),
 		},
 		Gascap:              utils.RpcGasCap(ctx),
 		BlockRangeLimit:     ctx.Int(utils.RpcBlockRangeLimit.Name),