diff --git a/ChangeLog.md b/ChangeLog.md
index 77018d7012f..4bfeff3bda9 100644
--- a/ChangeLog.md
+++ b/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
---
diff --git a/cmd/rpcdaemon/cli/config.go b/cmd/rpcdaemon/cli/config.go
index 20a8ebaa3db..5118aa21733 100644
--- a/cmd/rpcdaemon/cli/config.go
+++ b/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)
diff --git a/diagnostics/metrics/countervec.go b/diagnostics/metrics/countervec.go
new file mode 100644
index 00000000000..b3467be0322
--- /dev/null
+++ b/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 .
+
+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...)}
+}
diff --git a/diagnostics/metrics/register.go b/diagnostics/metrics/register.go
index 51464c32f3d..cf65fe22e16 100644
--- a/diagnostics/metrics/register.go
+++ b/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.
diff --git a/diagnostics/metrics/set.go b/diagnostics/metrics/set.go
index fbd9ac8d739..09c98a85d48 100644
--- a/diagnostics/metrics/set.go
+++ b/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.
diff --git a/node/cli/default_flags.go b/node/cli/default_flags.go
index 2843533fc1e..61c934544ac 100644
--- a/node/cli/default_flags.go
+++ b/node/cli/default_flags.go
@@ -123,6 +123,7 @@ var DefaultFlags = []cli.Flag{
&RpcSubscriptionFiltersMaxTxsFlag,
&RpcSubscriptionFiltersMaxAddressesFlag,
&RpcSubscriptionFiltersMaxTopicsFlag,
+ &RpcSubscriptionFiltersTimeoutFlag,
&utils.SnapKeepBlocksFlag,
&utils.SnapStopFlag,
diff --git a/node/cli/flags.go b/node/cli/flags.go
index 60b670ef2f3..66a5f706495 100644
--- a/node/cli/flags.go
+++ b/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),
diff --git a/rpc/filters/api.go b/rpc/filters/api.go
index f15d54cf02e..5ce708dd59c 100644
--- a/rpc/filters/api.go
+++ b/rpc/filters/api.go
@@ -48,444 +48,13 @@ func DefaultLogFilterOptions() LogFilterOptions {
}
}
-// TransactionReceiptsFilter defines criteria for transaction receipts subscription.
+// ReceiptsFilterCriteria defines criteria for transaction receipts subscription.
// If TransactionHashes is nil or empty, receipts for all transactions included in new blocks will be delivered.
// Otherwise, only receipts for the specified transactions will be delivered.
type ReceiptsFilterCriteria struct {
TransactionHashes []common.Hash `json:"transactionHashes"`
}
-/*
-// filter is a helper struct that holds meta information over the filter type
-// and associated subscription in the event system.
-type filter struct {
- typ Type
- deadline *time.Timer // filter is inactiv when deadline triggers
- hashes []common.Hash
- crit FilterCriteria
- logs []*types.Log
- s *Subscription // associated subscription in event system
-}
-
-// PublicFilterAPI offers support to create and manage filters. This will allow external clients to retrieve various
-// information related to the Ethereum protocol such as blocks, transactions and logs.
-type PublicFilterAPI struct {
- backend Backend
- quit chan struct{}
- chainDb ethdb.Database
- events *EventSystem
- filtersMu sync.Mutex
- filters map[rpc.ID]*filter
- timeout time.Duration
-}
-
-// NewPublicFilterAPI returns a new PublicFilterAPI instance.
-func NewPublicFilterAPI(backend Backend, timeout time.Duration) *PublicFilterAPI {
- api := &PublicFilterAPI{
- backend: backend,
- quit: make(chan struct{}, 1),
- chainDb: backend.ChainDb(),
- events: NewEventSystem(backend),
- filters: make(map[rpc.ID]*filter),
- timeout: timeout,
- }
- go api.timeoutLoop(timeout)
-
- return api
-}
-
-// timeoutLoop runs every 5 minutes and deletes filters that have not been recently used.
-// Tt is started when the api is created.
-func (api *PublicFilterAPI) timeoutLoop(timeout time.Duration) {
- var toUninstall []*Subscription
- ticker := time.NewTicker(timeout)
- defer ticker.Stop()
- for {
- select {
- case <-ticker.C:
- //nothing to do
- case <-api.quit:
- return
- }
-
- api.filtersMu.Lock()
- for id, f := range api.filters {
- select {
- case <-f.deadline.C:
- toUninstall = append(toUninstall, f.s)
- delete(api.filters, id)
- default:
- continue
- }
- }
- api.filtersMu.Unlock()
-
- // Unsubscribes are processed outside the lock to avoid the following scenario:
- // event loop attempts broadcasting events to still active filters while
- // Unsubscribe is waiting for it to process the uninstall request.
- for _, s := range toUninstall {
- s.Unsubscribe()
- }
- toUninstall = nil
- }
-}
-
-func (api *PublicFilterAPI) Close() {
- close(api.quit)
-}
-
-// NewPendingTransactionFilter creates a filter that fetches pending transaction hashes
-// as transactions enter the pending state.
-//
-// It is part of the filter package because this filter can be used through the
-// `eth_getFilterChanges` polling method that is also used for log filters.
-//
-// https://eth.wiki/json-rpc/API#eth_newpendingtransactionfilter
-func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
- var (
- pendingTxs = make(chan []common.Hash)
- pendingTxSub = api.events.SubscribePendingTxs(pendingTxs)
- )
-
- api.filtersMu.Lock()
- api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, deadline: time.NewTimer(api.timeout), hashes: make([]common.Hash, 0), s: pendingTxSub}
- api.filtersMu.Unlock()
-
- go func() {
- for {
- select {
- case ph := <-pendingTxs:
- api.filtersMu.Lock()
- if f, found := api.filters[pendingTxSub.ID]; found {
- f.hashes = append(f.hashes, ph...)
- }
- api.filtersMu.Unlock()
- case <-pendingTxSub.Err():
- api.filtersMu.Lock()
- delete(api.filters, pendingTxSub.ID)
- api.filtersMu.Unlock()
- return
- case <-api.quit:
- return
- }
- }
- }()
-
- return pendingTxSub.ID
-}
-
-// NewPendingTransactions creates a subscription that is triggered each time a transaction
-// enters the transaction pool and was signed from one of the transactions this nodes manages.
-func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Subscription, error) {
- notifier, supported := rpc.NotifierFromContext(ctx)
- if !supported {
- return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
- }
-
- rpcSub := notifier.CreateSubscription()
-
- go func() {
- txHashes := make(chan []common.Hash, 128)
- pendingTxSub := api.events.SubscribePendingTxs(txHashes)
-
- for {
- select {
- case hashes := <-txHashes:
- // To keep the original behaviour, send a single txn hash in one notification.
- // TODO(rjl493456442) Send a batch of txn hashes in one notification
- for _, h := range hashes {
- notifier.Notify(rpcSub.ID, h)
- }
- case <-rpcSub.Err():
- pendingTxSub.Unsubscribe()
- return
- case <-notifier.Closed():
- pendingTxSub.Unsubscribe()
- return
- case <-api.quit:
- return
- }
- }
- }()
-
- return rpcSub, nil
-}
-
-// NewBlockFilter creates a filter that fetches blocks that are imported into the chain.
-// It is part of the filter package since polling goes with eth_getFilterChanges.
-//
-// https://eth.wiki/json-rpc/API#eth_newblockfilter
-func (api *PublicFilterAPI) NewBlockFilter() rpc.ID {
- var (
- headers = make(chan *types.Header)
- headerSub = api.events.SubscribeNewHeads(headers)
- )
-
- api.filtersMu.Lock()
- api.filters[headerSub.ID] = &filter{typ: BlocksSubscription, deadline: time.NewTimer(api.timeout), hashes: make([]common.Hash, 0), s: headerSub}
- api.filtersMu.Unlock()
-
- go func() {
- for {
- select {
- case h := <-headers:
- api.filtersMu.Lock()
- if f, found := api.filters[headerSub.ID]; found {
- f.hashes = append(f.hashes, h.Hash())
- }
- api.filtersMu.Unlock()
- case <-headerSub.Err():
- api.filtersMu.Lock()
- delete(api.filters, headerSub.ID)
- api.filtersMu.Unlock()
- return
- case <-api.quit:
- return
- }
- }
- }()
-
- return headerSub.ID
-}
-
-// NewHeads send a notification each time a new (header) block is appended to the chain.
-func (api *PublicFilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
- notifier, supported := rpc.NotifierFromContext(ctx)
- if !supported {
- return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
- }
-
- rpcSub := notifier.CreateSubscription()
-
- go func() {
- headers := make(chan *types.Header)
- headersSub := api.events.SubscribeNewHeads(headers)
-
- for {
- select {
- case h := <-headers:
- notifier.Notify(rpcSub.ID, h)
- case <-rpcSub.Err():
- headersSub.Unsubscribe()
- return
- case <-notifier.Closed():
- headersSub.Unsubscribe()
- return
- case <-api.quit:
- return
- }
- }
- }()
-
- return rpcSub, nil
-}
-
-// Logs creates a subscription that fires for all new log that match the given filter criteria.
-func (api *PublicFilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
- notifier, supported := rpc.NotifierFromContext(ctx)
- if !supported {
- return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
- }
-
- var (
- rpcSub = notifier.CreateSubscription()
- matchedLogs = make(chan []*types.Log)
- )
-
- logsSub, err := api.events.SubscribeLogs(bind.FilterQuery(crit), matchedLogs)
- if err != nil {
- return nil, err
- }
-
- go func() {
-
- for {
- select {
- case logs := <-matchedLogs:
- for _, log := range logs {
- log := log
- notifier.Notify(rpcSub.ID, &log)
- }
- case <-rpcSub.Err(): // client send an unsubscribe request
- logsSub.Unsubscribe()
- return
- case <-notifier.Closed(): // connection dropped
- logsSub.Unsubscribe()
- return
- case <-api.quit:
- return
- }
- }
- }()
-
- return rpcSub, nil
-}
-// NewFilter creates a new filter and returns the filter id. It can be
-// used to retrieve logs when the state changes. This method cannot be
-// used to fetch logs that are already stored in the state.
-//
-// Default criteria for the from and to block are "latest".
-// Using "latest" as block number will return logs for mined blocks.
-// Using "pending" as block number returns logs for not yet mined (pending) blocks.
-// In case logs are removed (chain reorg) previously returned logs are returned
-// again but with the removed property set to true.
-//
-// In case "fromBlock" > "toBlock" an error is returned.
-//
-// https://eth.wiki/json-rpc/API#eth_newfilter
-func (api *PublicFilterAPI) NewFilter(crit FilterCriteria) (rpc.ID, error) {
- logs := make(chan []*types.Log)
- logsSub, err := api.events.SubscribeLogs(bind.FilterQuery(crit), logs)
- if err != nil {
- return rpc.ID(""), err
- }
-
- api.filtersMu.Lock()
- api.filters[logsSub.ID] = &filter{typ: LogsSubscription, crit: crit, deadline: time.NewTimer(api.timeout), logs: make([]*types.Log, 0), s: logsSub}
- api.filtersMu.Unlock()
-
- go func() {
- for {
- select {
- case l := <-logs:
- api.filtersMu.Lock()
- if f, found := api.filters[logsSub.ID]; found {
- f.logs = append(f.logs, l...)
- }
- api.filtersMu.Unlock()
- case <-logsSub.Err():
- api.filtersMu.Lock()
- delete(api.filters, logsSub.ID)
- api.filtersMu.Unlock()
- return
- case <-api.quit:
- return
- }
- }
- }()
-
- return logsSub.ID, nil
-}
-
-// GetLogs returns logs matching the given argument that are stored within the state.
-//
-// https://eth.wiki/json-rpc/API#eth_getlogs
-func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) {
- var filter *Filter
- if crit.BlockHash != nil {
- // Block filter requested, construct a single-shot filter
- filter = NewBlockFilter(api.backend, *crit.BlockHash, crit.Addresses, crit.Topics)
- } else {
- // Convert the RPC block numbers into internal representations
- begin := rpc.LatestBlockNumber.Int64()
- if crit.FromBlock != nil {
- begin = crit.FromBlock.Int64()
- }
- end := rpc.LatestBlockNumber.Int64()
- if crit.ToBlock != nil {
- end = crit.ToBlock.Int64()
- }
- // Construct the range filter
- filter = NewRangeFilter(api.backend, begin, end, crit.Addresses, crit.Topics)
- }
- // Run the filter and return all the logs
- logs, err := filter.Logs(ctx)
- if err != nil {
- return nil, err
- }
- return returnLogs(logs), err
-}
-
-// UninstallFilter removes the filter with the given filter id.
-//
-// https://eth.wiki/json-rpc/API#eth_uninstallfilter
-func (api *PublicFilterAPI) UninstallFilter(id rpc.ID) bool {
- api.filtersMu.Lock()
- f, found := api.filters[id]
- if found {
- delete(api.filters, id)
- }
- api.filtersMu.Unlock()
- if found {
- f.s.Unsubscribe()
- }
-
- return found
-}
-
-// GetFilterLogs returns the logs for the filter with the given id.
-// If the filter could not be found an empty array of logs is returned.
-//
-// https://eth.wiki/json-rpc/API#eth_getfilterlogs
-func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*types.Log, error) {
- api.filtersMu.Lock()
- f, found := api.filters[id]
- api.filtersMu.Unlock()
-
- if !found || f.typ != LogsSubscription {
- return nil, errors.New("filter not found")
- }
-
- var filter *Filter
- if f.crit.BlockHash != nil {
- // Block filter requested, construct a single-shot filter
- filter = NewBlockFilter(api.backend, *f.crit.BlockHash, f.crit.Addresses, f.crit.Topics)
- } else {
- // Convert the RPC block numbers into internal representations
- begin := rpc.LatestBlockNumber.Int64()
- if f.crit.FromBlock != nil {
- begin = f.crit.FromBlock.Int64()
- }
- end := rpc.LatestBlockNumber.Int64()
- if f.crit.ToBlock != nil {
- end = f.crit.ToBlock.Int64()
- }
- // Construct the range filter
- filter = NewRangeFilter(api.backend, begin, end, f.crit.Addresses, f.crit.Topics)
- }
- // Run the filter and return all the logs
- logs, err := filter.Logs(ctx)
- if err != nil {
- return nil, err
- }
- return returnLogs(logs), nil
-}
-
-// GetFilterChanges returns the logs for the filter with the given id since
-// last time it was called. This can be used for polling.
-//
-// For pending transaction and block filters the result is []common.Hash.
-// (pending)Log filters return []Log.
-//
-// https://eth.wiki/json-rpc/API#eth_getfilterchanges
-func (api *PublicFilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
- api.filtersMu.Lock()
- defer api.filtersMu.Unlock()
-
- if f, found := api.filters[id]; found {
- if !f.deadline.Stop() {
- // timer expired but filter is not yet removed in timeout loop
- // receive timer value and reset timer
- <-f.deadline.C
- }
- f.deadline.Reset(api.timeout)
-
- switch f.typ {
- case PendingTransactionsSubscription, BlocksSubscription:
- hashes := f.hashes
- f.hashes = nil
- return returnHashes(hashes), nil
- case LogsSubscription, MinedAndPendingLogsSubscription:
- logs := f.logs
- f.logs = nil
- return returnLogs(logs), nil
- }
- }
-
- return []interface{}{}, errors.New("filter not found")
-}
-*/
-
// UnmarshalJSON sets *args fields with given data.
func (args *FilterCriteria) UnmarshalJSON(data []byte) error {
type input struct {
diff --git a/rpc/jsonrpc/eth_filters.go b/rpc/jsonrpc/eth_filters.go
index 3501d4edc60..2e6854bb69b 100644
--- a/rpc/jsonrpc/eth_filters.go
+++ b/rpc/jsonrpc/eth_filters.go
@@ -35,7 +35,7 @@ func (api *APIImpl) NewPendingTransactionFilter(_ context.Context) (string, erro
if api.filters == nil {
return "", rpc.ErrNotificationsUnsupported
}
- txsCh, id := api.filters.SubscribePendingTxs(32)
+ txsCh, id := api.filters.SubscribePendingTxs(32, rpchelper.ProtocolHTTP)
go func() {
for txs := range txsCh {
api.filters.AddPendingTxs(id, txs)
@@ -49,7 +49,7 @@ func (api *APIImpl) NewBlockFilter(_ context.Context) (string, error) {
if api.filters == nil {
return "", rpc.ErrNotificationsUnsupported
}
- ch, id := api.filters.SubscribeNewHeads(32)
+ ch, id := api.filters.SubscribeNewHeads(32, rpchelper.ProtocolHTTP)
go func() {
for block := range ch {
api.filters.AddPendingBlock(id, block)
@@ -63,7 +63,10 @@ func (api *APIImpl) NewFilter(_ context.Context, crit filters.FilterCriteria) (s
if api.filters == nil {
return "", rpc.ErrNotificationsUnsupported
}
- logs, id := api.filters.SubscribeLogs(256, crit)
+ logs, id, err := api.filters.SubscribeLogs(256, crit, rpchelper.ProtocolHTTP)
+ if err != nil {
+ return "", err
+ }
go func() {
for lg := range logs {
api.filters.AddLogs(id, lg)
@@ -101,37 +104,33 @@ func (api *APIImpl) GetFilterChanges(_ context.Context, index string) ([]any, er
stub := make([]any, 0)
// remove 0x
cutIndex := strings.TrimPrefix(index, "0x")
- // Validate the id exists for any subscription type first to distinguish "never created" from "no changes yet".
- exists := api.filters.HasHeadsSubscription(rpchelper.HeadsSubID(cutIndex)) ||
- api.filters.HasPendingTxsSubscription(rpchelper.PendingTxsSubID(cutIndex)) ||
- api.filters.HasSubscription(rpchelper.LogsSubID(cutIndex))
- if !exists {
+ ft, ok := api.filters.TouchSubscription(rpchelper.SubscriptionID(cutIndex))
+ if !ok {
return nil, rpc.ErrFilterNotFound
}
-
- // Identify the subscription type by probing each store; if none have data yet, return empty slice
- if blocks, ok := api.filters.ReadPendingBlocks(rpchelper.HeadsSubID(cutIndex)); ok {
- for _, v := range blocks {
- stub = append(stub, v.Hash())
+ switch ft {
+ case rpchelper.FilterTypeHeads:
+ if blocks, ok := api.filters.ReadPendingBlocks(rpchelper.HeadsSubID(cutIndex)); ok {
+ for _, v := range blocks {
+ stub = append(stub, v.Hash())
+ }
}
- return stub, nil
- }
- if txs, ok := api.filters.ReadPendingTxs(rpchelper.PendingTxsSubID(cutIndex)); ok {
- if len(txs) > 0 {
- for _, txn := range txs[0] {
- stub = append(stub, txn.Hash())
+ case rpchelper.FilterTypePendingTxs:
+ if txs, ok := api.filters.ReadPendingTxs(rpchelper.PendingTxsSubID(cutIndex)); ok {
+ for _, batch := range txs {
+ for _, txn := range batch {
+ stub = append(stub, txn.Hash())
+ }
}
- return stub, nil
}
- return stub, nil
- }
- if logs, ok := api.filters.ReadLogs(rpchelper.LogsSubID(cutIndex)); ok {
- for _, v := range logs {
- stub = append(stub, v)
+ case rpchelper.FilterTypeLogs:
+ if logs, ok := api.filters.ReadLogs(rpchelper.LogsSubID(cutIndex)); ok {
+ for _, v := range logs {
+ stub = append(stub, v)
+ }
}
- return stub, nil
}
- return []any{}, nil
+ return stub, nil
}
// GetFilterLogs implements eth_getFilterLogs.
@@ -142,7 +141,7 @@ func (api *APIImpl) GetFilterLogs(_ context.Context, index string) ([]*types.Log
return nil, rpc.ErrNotificationsUnsupported
}
cutIndex := strings.TrimPrefix(index, "0x")
- if found := api.filters.HasSubscription(rpchelper.LogsSubID(cutIndex)); !found {
+ if ft, ok := api.filters.TouchSubscription(rpchelper.SubscriptionID(cutIndex)); !ok || ft != rpchelper.FilterTypeLogs {
return nil, rpc.ErrFilterNotFound
}
if logs, ok := api.filters.ReadLogs(rpchelper.LogsSubID(cutIndex)); ok {
@@ -156,7 +155,7 @@ func (api *APIImpl) GetFilterLogs(_ context.Context, index string) ([]*types.Log
// the channel closes or the client goes away. subscribe is called inside the goroutine
// and must return the item channel plus an unsubscribe func. notify receives an emit
// func that sends a payload to the client, logging on failure.
-func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, subscribe func() (<-chan T, func()), notify func(emit func(payload any), item T), closedWarn string) (*rpc.Subscription, error) {
+func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, subscribe func() (<-chan T, func(), error), notify func(emit func(payload any), item T), closedWarn string) (*rpc.Subscription, error) {
if apiFilters == nil {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
}
@@ -165,11 +164,14 @@ func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, sub
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
}
+ ch, unsubscribe, err := subscribe()
+ if err != nil {
+ return nil, err
+ }
rpcSub := notifier.CreateSubscription()
go func() {
defer dbg.LogPanic()
- ch, unsubscribe := subscribe()
defer unsubscribe()
emit := func(payload any) {
@@ -197,9 +199,9 @@ func subscribeRPC[T any](ctx context.Context, apiFilters *rpchelper.Filters, sub
// NewHeads send a notification each time a new (header) block is appended to the chain.
func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
return subscribeRPC(ctx, api.filters,
- func() (<-chan *types.Header, func()) {
- headers, id := api.filters.SubscribeNewHeads(32)
- return headers, func() { api.filters.UnsubscribeHeads(id) }
+ func() (<-chan *types.Header, func(), error) {
+ headers, id := api.filters.SubscribeNewHeads(32, rpchelper.ProtocolWS)
+ return headers, func() { api.filters.UnsubscribeHeads(id) }, nil
},
func(emit func(payload any), h *types.Header) {
if h != nil {
@@ -221,9 +223,9 @@ func (api *APIImpl) NewPendingTransactionsWithBody(ctx context.Context) (*rpc.Su
func (api *APIImpl) subscribePendingTransactions(ctx context.Context, chanSize int, fullTx bool) (*rpc.Subscription, error) {
return subscribeRPC(ctx, api.filters,
- func() (<-chan []types.Transaction, func()) {
- txsCh, id := api.filters.SubscribePendingTxs(chanSize)
- return txsCh, func() { api.filters.UnsubscribePendingTxs(id) }
+ func() (<-chan []types.Transaction, func(), error) {
+ txsCh, id := api.filters.SubscribePendingTxs(chanSize, rpchelper.ProtocolWS)
+ return txsCh, func() { api.filters.UnsubscribePendingTxs(id) }, nil
},
func(emit func(payload any), txs []types.Transaction) {
for _, t := range txs {
@@ -242,9 +244,12 @@ func (api *APIImpl) subscribePendingTransactions(ctx context.Context, chanSize i
// Logs send a notification each time a new log appears.
func (api *APIImpl) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc.Subscription, error) {
return subscribeRPC(ctx, api.filters,
- func() (<-chan *types.Log, func()) {
- logs, id := api.filters.SubscribeLogs(api.SubscribeLogsChannelSize, crit)
- return logs, func() { api.filters.UnsubscribeLogs(id) }
+ func() (<-chan *types.Log, func(), error) {
+ logs, id, err := api.filters.SubscribeLogs(api.SubscribeLogsChannelSize, crit, rpchelper.ProtocolWS)
+ if err != nil {
+ return nil, nil, err
+ }
+ return logs, func() { api.filters.UnsubscribeLogs(id) }, nil
},
func(emit func(payload any), h *types.Log) {
if h != nil {
@@ -257,9 +262,12 @@ func (api *APIImpl) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc
// TransactionReceipts send a notification each time a new receipt appears.
func (api *APIImpl) TransactionReceipts(ctx context.Context, crit filters.ReceiptsFilterCriteria) (*rpc.Subscription, error) {
return subscribeRPC(ctx, api.filters,
- func() (<-chan *remoteproto.SubscribeReceiptsReply, func()) {
- receipts, id := api.filters.SubscribeReceipts(api.SubscribeLogsChannelSize, crit)
- return receipts, func() { api.filters.UnsubscribeReceipts(id) }
+ func() (<-chan *remoteproto.SubscribeReceiptsReply, func(), error) {
+ receipts, id, err := api.filters.SubscribeReceipts(api.SubscribeLogsChannelSize, crit)
+ if err != nil {
+ return nil, nil, err
+ }
+ return receipts, func() { api.filters.UnsubscribeReceipts(id) }, nil
},
func(emit func(payload any), protoReceipt *remoteproto.SubscribeReceiptsReply) {
if protoReceipt != nil {
diff --git a/rpc/jsonrpc/eth_filters_test.go b/rpc/jsonrpc/eth_filters_test.go
index 950b14c6376..00457bf426a 100644
--- a/rpc/jsonrpc/eth_filters_test.go
+++ b/rpc/jsonrpc/eth_filters_test.go
@@ -18,6 +18,7 @@ package jsonrpc
import (
"math/rand"
+ "strings"
"sync"
"testing"
"time"
@@ -143,7 +144,7 @@ func TestLogsSubscribeAndUnsubscribe_WithoutConcurrentMapIssue(t *testing.T) {
for i := 0; i < 1000; i++ {
wg.Add(1)
go func(idx int) {
- _, id := ff.SubscribeLogs(32, crit)
+ _, id, _ := ff.SubscribeLogs(32, crit, "")
defer func() {
time.Sleep(100 * time.Nanosecond)
ff.UnsubscribeLogs(id)
@@ -243,6 +244,38 @@ func TestNewPendingTransactionIncludesFrom(t *testing.T) {
require.Equal(t, m.Address, rpcTx.From)
}
+func TestPendingTxsFilterChangesReturnsAllBatches(t *testing.T) {
+ m := execmoduletester.New(t)
+ ctx, conn := rpcdaemontest.CreateTestGrpcConn(t, m)
+ mining := txpoolproto.NewMiningClient(conn)
+ ff := rpchelper.New(ctx, rpchelper.DefaultFiltersConfig, nil, nil, mining, func() {}, m.Log, nil)
+ stateCache := kvcache.New(kvcache.DefaultCoherentConfig)
+ api := newEthApiForTest(newBaseApiWithFiltersForTest(ff, stateCache, m), m.DB, nil, nil)
+
+ ptf, err := api.NewPendingTransactionFilter(ctx)
+ require.NoError(t, err)
+ id := rpchelper.PendingTxsSubID(strings.TrimPrefix(ptf, "0x"))
+
+ signer := types.LatestSignerForChainID(m.ChainConfig.ChainID)
+ makeTx := func(nonce uint64) types.Transaction {
+ tx, err := types.SignTx(
+ types.NewTransaction(nonce, m.Address, uint256.NewInt(1), params.TxGas, uint256.NewInt(1), nil),
+ *signer,
+ m.Key,
+ )
+ require.NoError(t, err)
+ return tx
+ }
+ tx0, tx1, tx2 := makeTx(0), makeTx(1), makeTx(2)
+
+ ff.AddPendingTxs(id, []types.Transaction{tx0, tx1})
+ ff.AddPendingTxs(id, []types.Transaction{tx2})
+
+ changes, err := api.GetFilterChanges(ctx, ptf)
+ require.NoError(t, err)
+ require.Equal(t, []any{tx0.Hash(), tx1.Hash(), tx2.Hash()}, changes)
+}
+
func TestGetFilterChangesReturnsFilterNotFoundForUnknownID(t *testing.T) {
if testing.Short() {
t.Skip("slow test")
diff --git a/rpc/jsonrpc/eth_subscribe_test.go b/rpc/jsonrpc/eth_subscribe_test.go
index 45f38bf8ffb..dd4c1943596 100644
--- a/rpc/jsonrpc/eth_subscribe_test.go
+++ b/rpc/jsonrpc/eth_subscribe_test.go
@@ -62,7 +62,7 @@ func TestEthSubscribe(t *testing.T) {
}
ff := rpchelper.New(ctx, rpchelper.DefaultFiltersConfig, backend, nil, nil, onNewSnapshot, m.Log, nil)
subscriptionReadyWg.Wait() // This is needed *before* inserting the blocks, which sends NEW_HEADER events
- newHeads, id := ff.SubscribeNewHeads(16)
+ newHeads, id := ff.SubscribeNewHeads(16, "")
defer ff.UnsubscribeHeads(id)
highestSeenHeader := chain.TopBlock.NumberU64()
err = m.InsertChain(chain)
@@ -94,7 +94,7 @@ func TestEthSubscribeReceipts(t *testing.T) {
}
ff := rpchelper.New(ctx, rpchelper.DefaultFiltersConfig, backend, nil, nil, onNewSnapshot, m.Log, nil)
subscriptionReadyWg.Wait()
- newReceipts, id := ff.SubscribeReceipts(16, filters.ReceiptsFilterCriteria{
+ newReceipts, id, _ := ff.SubscribeReceipts(16, filters.ReceiptsFilterCriteria{
TransactionHashes: []common.Hash{},
})
defer ff.UnsubscribeReceipts(id)
diff --git a/rpc/jsonrpc/send_transaction.go b/rpc/jsonrpc/send_transaction.go
index fa8c6b74257..3948a58b8fb 100644
--- a/rpc/jsonrpc/send_transaction.go
+++ b/rpc/jsonrpc/send_transaction.go
@@ -103,7 +103,10 @@ func (api *APIImpl) SendRawTransactionSync(ctx context.Context, encodedTx hexuti
criteria := filters.ReceiptsFilterCriteria{
TransactionHashes: []common.Hash{hash},
}
- receiptsCh, id := api.filters.SubscribeReceipts(128, criteria)
+ receiptsCh, id, err := api.filters.SubscribeReceipts(128, criteria)
+ if err != nil {
+ return nil, err
+ }
defer api.filters.UnsubscribeReceipts(id)
// Theoretically, we should subscribe *before* submitting the transaction, but then we couldn't filter by hash.
diff --git a/rpc/jsonrpc/send_transaction_test.go b/rpc/jsonrpc/send_transaction_test.go
index 140798b3a1b..eb88075f5a7 100644
--- a/rpc/jsonrpc/send_transaction_test.go
+++ b/rpc/jsonrpc/send_transaction_test.go
@@ -59,7 +59,7 @@ func TestSendRawTransaction(t *testing.T) {
buf := bytes.NewBuffer(nil)
err = txn.MarshalBinary(buf)
require.NoError(err)
- txsCh, id := ff.SubscribePendingTxs(1)
+ txsCh, id := ff.SubscribePendingTxs(1, "")
defer ff.UnsubscribePendingTxs(id)
txHash, err := api.SendRawTransaction(ctx, buf.Bytes())
require.NoError(err)
@@ -102,7 +102,7 @@ func TestSendRawTransactionUnprotected(t *testing.T) {
buf := bytes.NewBuffer(nil)
err = txn.MarshalBinary(buf)
require.NoError(err)
- txsCh, id := ff.SubscribePendingTxs(1)
+ txsCh, id := ff.SubscribePendingTxs(1, "")
defer ff.UnsubscribePendingTxs(id)
txHash, err := api.SendRawTransaction(ctx, buf.Bytes())
require.NoError(err)
diff --git a/rpc/rpchelper/config.go b/rpc/rpchelper/config.go
index ff85a580874..eacf35fe107 100644
--- a/rpc/rpchelper/config.go
+++ b/rpc/rpchelper/config.go
@@ -16,15 +16,21 @@
package rpchelper
+import "time"
+
+// DefaultFilterTimeout matches geth's deadline for evicting idle filters; 0 disables eviction.
+const DefaultFilterTimeout = 5 * time.Minute
+
// FiltersConfig defines the configuration settings for RPC subscription filters.
// Each field represents a limit on the number of respective items that can be stored per subscription.
// A value of 0 disables the limit (no cap). Oldest items are evicted first (FIFO) when the limit is reached.
type FiltersConfig struct {
- RpcSubscriptionFiltersMaxLogs int // Maximum number of logs to store per subscription. Default: 10000
- RpcSubscriptionFiltersMaxHeaders int // Maximum number of block headers to store per subscription. Default: 10000
- RpcSubscriptionFiltersMaxTxs int // Maximum number of transactions to store per subscription. Default: 10000
- RpcSubscriptionFiltersMaxAddresses int // Maximum number of addresses per subscription to filter logs by. Default: 0 (no limit)
- RpcSubscriptionFiltersMaxTopics int // Maximum number of topics per subscription to filter logs by. Default: 0 (no limit)
+ RpcSubscriptionFiltersMaxLogs int // Maximum number of logs to store per subscription. Default: 10000
+ RpcSubscriptionFiltersMaxHeaders int // Maximum number of block headers to store per subscription. Default: 10000
+ RpcSubscriptionFiltersMaxTxs int // Maximum number of transactions to store per subscription. Default: 10000
+ RpcSubscriptionFiltersMaxAddresses int // Maximum number of addresses per subscription to filter logs by. Default: 0 (no limit)
+ RpcSubscriptionFiltersMaxTopics int // Maximum number of topics per subscription to filter logs by. Default: 0 (no limit)
+ RpcSubscriptionFiltersTimeout time.Duration // Timeout before idle filters are evicted. Default: 5m; 0 disables eviction
}
// DefaultFiltersConfig defines the default settings for filter configurations.
@@ -37,4 +43,5 @@ var DefaultFiltersConfig = FiltersConfig{
RpcSubscriptionFiltersMaxTxs: 10000,
RpcSubscriptionFiltersMaxAddresses: 0, // no limit
RpcSubscriptionFiltersMaxTopics: 0, // no limit
+ RpcSubscriptionFiltersTimeout: DefaultFilterTimeout,
}
diff --git a/rpc/rpchelper/filters.go b/rpc/rpchelper/filters.go
index 101d2e6277a..e2ed32fd5bb 100644
--- a/rpc/rpchelper/filters.go
+++ b/rpc/rpchelper/filters.go
@@ -47,6 +47,23 @@ import (
"github.com/erigontech/erigon/txnprovider/txpool"
)
+// FilterType represents the type of a filter subscription for metrics tracking.
+type FilterType string
+
+const (
+ FilterTypeLogs FilterType = "logs"
+ FilterTypeHeads FilterType = "heads"
+ FilterTypePendingTxs FilterType = "pendingTxs"
+)
+
+// trackedSub indexes a pollable subscription for O(1) id lookup: polls resolve the
+// filter type and reset the eviction deadline in one step, and the eviction loop
+// iterates all pollable subscriptions regardless of element type.
+type trackedSub struct {
+ ft FilterType
+ tracker SubTracker
+}
+
// Filters holds the state for managing subscriptions to various Ethereum events.
// It allows for the subscription and management of events such as new blocks, pending transactions,
// logs, and other Ethereum-related activities.
@@ -69,6 +86,7 @@ type Filters struct {
logsStores *concurrent.SyncMap[LogsSubID, []*types.Log]
pendingHeadsStores *concurrent.SyncMap[HeadsSubID, []*types.Header]
pendingTxsStores *concurrent.SyncMap[PendingTxsSubID, [][]types.Transaction]
+ trackedSubs *concurrent.SyncMap[SubscriptionID, trackedSub]
logger log.Logger
// latestSD is the local fallback for the most recent SharedDomains.
@@ -99,11 +117,29 @@ func New(ctx context.Context, config FiltersConfig, ethBackend ApiBackend, txPoo
logsStores: concurrent.NewSyncMap[LogsSubID, []*types.Log](),
pendingHeadsStores: concurrent.NewSyncMap[HeadsSubID, []*types.Header](),
pendingTxsStores: concurrent.NewSyncMap[PendingTxsSubID, [][]types.Transaction](),
+ trackedSubs: concurrent.NewSyncMap[SubscriptionID, trackedSub](),
logger: logger,
config: config,
events: events,
}
+ // Start the timeout loop for filter eviction if timeout is configured
+ if config.RpcSubscriptionFiltersTimeout > 0 {
+ // Check more frequently than the timeout to ensure timely eviction, with a
+ // 1s floor: time.NewTicker panics on non-positive intervals and a
+ // sub-second sweep would burn CPU for no practical benefit.
+ checkInterval := config.RpcSubscriptionFiltersTimeout / 2
+ if checkInterval < time.Second {
+ checkInterval = time.Second
+ }
+ logger.Info("[rpc] [filters] starting timeout loop for idle filter eviction",
+ "timeout", config.RpcSubscriptionFiltersTimeout,
+ "checkInterval", checkInterval)
+ go ff.timeoutLoop(ctx, config.RpcSubscriptionFiltersTimeout, checkInterval)
+ } else {
+ logger.Info("[rpc] [filters] timeout-based filter eviction disabled")
+ }
+
go func() {
if ethBackend == nil {
return
@@ -286,6 +322,95 @@ func (ff *Filters) LastPendingBlock() *types.Block {
return ff.pendingBlock
}
+// timeoutLoop runs periodically and evicts subscriptions that have not been polled within the timeout duration.
+// This prevents unbounded accumulation of idle subscriptions and matches geth's behavior.
+func (ff *Filters) timeoutLoop(ctx context.Context, timeout, checkInterval time.Duration) {
+ ticker := time.NewTicker(checkInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ ff.logger.Debug("[rpc] [filters] timeout loop stopping due to context cancellation")
+ return
+ case <-ticker.C:
+ ff.logger.Trace("[rpc] [filters] running eviction check")
+ ff.evictStaleSubscriptions(timeout)
+ }
+ }
+}
+
+// evictStaleSubscriptions removes filters that haven't been accessed within the timeout
+// duration. CloseIfIdle decides staleness and closes in one atomic step, so a concurrent
+// Touch either lands before it (the subscription survives) or finds the subscription
+// already closed; teardown and metrics follow after the Range to avoid mutating the
+// maps during iteration. Evicted logs filters share one remote-filter update per cycle.
+func (ff *Filters) evictStaleSubscriptions(timeout time.Duration) {
+ type victim struct {
+ id SubscriptionID
+ ft FilterType
+ protocol SubProtocol
+ }
+ checked := 0
+ var victims []victim
+ ff.trackedSubs.Range(func(id SubscriptionID, sub trackedSub) error {
+ checked++
+ if sub.tracker.CloseIfIdle(timeout) {
+ victims = append(victims, victim{id, sub.ft, sub.tracker.Protocol()})
+ }
+ return nil
+ })
+
+ evicted := 0
+ logsEvicted := false
+ for _, v := range victims {
+ var removed bool
+ switch v.ft {
+ case FilterTypeHeads:
+ removed = ff.unsubscribeHeadsInternal(HeadsSubID(v.id))
+ case FilterTypePendingTxs:
+ removed = ff.unsubscribePendingTxsInternal(PendingTxsSubID(v.id))
+ case FilterTypeLogs:
+ removed = ff.removeLogsSubscription(LogsSubID(v.id), false)
+ logsEvicted = logsEvicted || removed
+ }
+ if !removed {
+ continue
+ }
+ evicted++
+ ff.logger.Info("[rpc] [filters] evicted idle filter", "type", v.ft, "id", v.id, "protocol", v.protocol, "timeout", timeout)
+ ff.decrementMetrics(v.ft, v.protocol)
+ reapedSubscriptionsCounter.WithLabelValues(string(v.ft)).Inc()
+ }
+ if logsEvicted {
+ ff.updateRemoteLogsFilter()
+ }
+
+ if evicted > 0 {
+ ff.logger.Debug("[rpc] [filters] eviction cycle complete", "checked", checked, "evicted", evicted)
+ } else {
+ ff.logger.Trace("[rpc] [filters] eviction cycle complete, no stale filters", "checked", checked)
+ }
+}
+
+// TouchSubscription resets the eviction deadline of the subscription and reports its
+// filter type, so a poll on a quiet chain keeps the filter alive even when nothing is
+// buffered. Called on every poll: a single map lookup, no logging.
+func (ff *Filters) TouchSubscription(id SubscriptionID) (FilterType, bool) {
+ sub, ok := ff.trackedSubs.Get(id)
+ if !ok {
+ return "", false
+ }
+ sub.tracker.Touch()
+ return sub.ft, true
+}
+
+func (ff *Filters) registerSubscription(id SubscriptionID, ft FilterType, tracker SubTracker) {
+ ff.trackedSubs.Put(id, trackedSub{ft: ft, tracker: tracker})
+ ff.logger.Debug("[rpc] [filters] registered subscription", "type", ft, "id", id, "protocol", tracker.Protocol())
+ ff.incrementMetrics(ft, tracker.Protocol())
+}
+
// subscribeToPendingTransactions subscribes to pending transactions using the given transaction pool client.
// It listens for new transactions and processes them as they arrive.
func (ff *Filters) subscribeToPendingTransactions(ctx context.Context, txPool txpoolproto.TxpoolClient) error {
@@ -402,25 +527,52 @@ func (ff *Filters) HandlePendingLogs(reply *txpoolproto.OnPendingLogsReply) {
// SubscribeNewHeads subscribes to new block headers and returns a channel to receive the headers
// and a subscription ID to manage the subscription.
-func (ff *Filters) SubscribeNewHeads(size int) (<-chan *types.Header, HeadsSubID) {
+func (ff *Filters) SubscribeNewHeads(size int, protocol SubProtocol) (<-chan *types.Header, HeadsSubID) {
id := HeadsSubID(generateSubscriptionID())
- sub := newChanSub[*types.Header](size)
+ sub := newChanSub[*types.Header](size, protocol)
ff.headsSubs.Put(id, sub)
+ ff.registerSubscription(SubscriptionID(id), FilterTypeHeads, sub)
return sub.ch, id
}
// UnsubscribeHeads unsubscribes from new block headers using the given subscription ID.
// It returns true if the unsubscription was successful, otherwise false.
func (ff *Filters) UnsubscribeHeads(id HeadsSubID) bool {
- ch, ok := ff.headsSubs.Get(id)
+ sub, ok := ff.headsSubs.Get(id)
if !ok {
+ ff.logger.Debug("[rpc] [filters] unsubscribe heads filter not found", "id", id)
return false
}
- ch.Close()
- if _, ok = ff.headsSubs.Delete(id); !ok {
+ protocol := sub.Protocol()
+ if !ff.unsubscribeHeadsInternal(id) {
+ return false
+ }
+ ff.logger.Debug("[rpc] [filters] unsubscribed heads filter", "id", id, "protocol", protocol)
+ ff.decrementMetrics(FilterTypeHeads, protocol)
+ return true
+}
+
+func (ff *Filters) unsubscribeHeadsInternal(id HeadsSubID) bool {
+ if !unsubscribeSubInternal(ff.headsSubs, ff.pendingHeadsStores, id) {
return false
}
- ff.pendingHeadsStores.Delete(id)
+ ff.trackedSubs.Delete(SubscriptionID(id))
+ return true
+}
+
+// unsubscribeSubInternal tears down a subscription and its buffered store without
+// updating metrics: eviction and client unsubscribe share it, and the caller owns
+// the single decrement, gated on the map delete succeeding.
+func unsubscribeSubInternal[K comparable, T, S any](subs *concurrent.SyncMap[K, Sub[T]], stores *concurrent.SyncMap[K, S], id K) bool {
+ sub, ok := subs.Get(id)
+ if !ok {
+ return false
+ }
+ sub.Close()
+ if _, ok = subs.Delete(id); !ok {
+ return false
+ }
+ stores.Delete(id)
return true
}
@@ -428,7 +580,7 @@ func (ff *Filters) UnsubscribeHeads(id HeadsSubID) bool {
// and a subscription ID to manage the subscription. It uses the specified filter criteria.
func (ff *Filters) SubscribePendingLogs(size int) (<-chan types.Logs, PendingLogsSubID) {
id := PendingLogsSubID(generateSubscriptionID())
- sub := newChanSub[types.Logs](size)
+ sub := newChanSub[types.Logs](size, "")
ff.pendingLogsSubs.Put(id, sub)
return sub.ch, id
}
@@ -449,7 +601,7 @@ func (ff *Filters) UnsubscribePendingLogs(id PendingLogsSubID) bool {
// and a subscription ID to manage the subscription.
func (ff *Filters) SubscribePendingBlock(size int) (<-chan *types.Block, PendingBlockSubID) {
id := PendingBlockSubID(generateSubscriptionID())
- sub := newChanSub[*types.Block](size)
+ sub := newChanSub[*types.Block](size, "")
ff.pendingBlockSubs.Put(id, sub)
return sub.ch, id
}
@@ -468,38 +620,50 @@ func (ff *Filters) UnsubscribePendingBlock(id PendingBlockSubID) bool {
// SubscribePendingTxs subscribes to pending transactions and returns a channel to receive the transactions
// and a subscription ID to manage the subscription.
-func (ff *Filters) SubscribePendingTxs(size int) (<-chan []types.Transaction, PendingTxsSubID) {
+func (ff *Filters) SubscribePendingTxs(size int, protocol SubProtocol) (<-chan []types.Transaction, PendingTxsSubID) {
id := PendingTxsSubID(generateSubscriptionID())
- sub := newChanSub[[]types.Transaction](size)
+ sub := newChanSub[[]types.Transaction](size, protocol)
ff.pendingTxsSubs.Put(id, sub)
+ ff.registerSubscription(SubscriptionID(id), FilterTypePendingTxs, sub)
return sub.ch, id
}
// UnsubscribePendingTxs unsubscribes from pending transactions using the given subscription ID.
// It returns true if the unsubscription was successful, otherwise false.
func (ff *Filters) UnsubscribePendingTxs(id PendingTxsSubID) bool {
- ch, ok := ff.pendingTxsSubs.Get(id)
+ sub, ok := ff.pendingTxsSubs.Get(id)
if !ok {
+ ff.logger.Debug("[rpc] [filters] unsubscribe pending txs filter not found", "id", id)
return false
}
- ch.Close()
- if _, ok = ff.pendingTxsSubs.Delete(id); !ok {
+ protocol := sub.Protocol()
+ if !ff.unsubscribePendingTxsInternal(id) {
return false
}
- ff.pendingTxsStores.Delete(id)
+ ff.logger.Debug("[rpc] [filters] unsubscribed pending txs filter", "id", id, "protocol", protocol)
+ ff.decrementMetrics(FilterTypePendingTxs, protocol)
+ return true
+}
+
+func (ff *Filters) unsubscribePendingTxsInternal(id PendingTxsSubID) bool {
+ if !unsubscribeSubInternal(ff.pendingTxsSubs, ff.pendingTxsStores, id) {
+ return false
+ }
+ ff.trackedSubs.Delete(SubscriptionID(id))
return true
}
// SubscribeReceipts subscribes to transaction receipts and returns a channel to receive the receipts
-// and a subscription ID to manage the subscription.
-func (ff *Filters) SubscribeReceipts(size int, criteria filters.ReceiptsFilterCriteria) (<-chan *remoteproto.SubscribeReceiptsReply, ReceiptsSubID) {
- sub := newChanSub[*remoteproto.SubscribeReceiptsReply](size)
+// and a subscription ID to manage the subscription. When the remote filter update fails, no subscription
+// is installed and the error is returned.
+func (ff *Filters) SubscribeReceipts(size int, criteria filters.ReceiptsFilterCriteria) (<-chan *remoteproto.SubscribeReceiptsReply, ReceiptsSubID, error) {
+ sub := newChanSub[*remoteproto.SubscribeReceiptsReply](size, "")
id := ff.receiptsSubs.insertReceiptsFilter(sub, criteria.TransactionHashes, ff.config.RpcSubscriptionFiltersMaxLogs)
if err := ff.sendReceiptsFilterUpdate(); err != nil {
- ff.logger.Warn("Could not update remote receipts filter", "err", err)
ff.receiptsSubs.removeReceiptsFilter(id)
+ return nil, "", fmt.Errorf("could not update remote receipts filter: %w", err)
}
- return sub.ch, id
+ return sub.ch, id, nil
}
// UnsubscribeReceipts unsubscribes from transaction receipts using the given subscription ID.
@@ -534,9 +698,10 @@ func (ff *Filters) sendReceiptsFilterUpdate() error {
}
// SubscribeLogs subscribes to logs using the specified filter criteria and returns a channel to receive the logs
-// and a subscription ID to manage the subscription.
-func (ff *Filters) SubscribeLogs(size int, criteria filters.FilterCriteria) (<-chan *types.Log, LogsSubID) {
- sub := newChanSub[*types.Log](size)
+// and a subscription ID to manage the subscription. When the remote filter update fails, no subscription is
+// installed and the error is returned.
+func (ff *Filters) SubscribeLogs(size int, criteria filters.FilterCriteria, protocol SubProtocol) (<-chan *types.Log, LogsSubID, error) {
+ sub := newChanSub[*types.Log](size, protocol)
id, f := ff.logsSubs.insertLogsFilter(sub)
// Initialize address and topic maps
@@ -601,12 +766,13 @@ func (ff *Filters) SubscribeLogs(size int, criteria filters.FilterCriteria) (<-c
loaded := ff.loadLogsRequester()
if loaded != nil {
if err := loaded.(func(*remoteproto.LogsFilterRequest) error)(lfr); err != nil {
- ff.logger.Warn("Could not update remote logs filter", "err", err)
ff.logsSubs.removeLogsFilter(id)
+ return nil, "", fmt.Errorf("could not update remote logs filter: %w", err)
}
}
- return sub.ch, id
+ ff.registerSubscription(SubscriptionID(id), FilterTypeLogs, sub)
+ return sub.ch, id, nil
}
// loadLogsRequester loads the current logs requester and returns it.
@@ -616,49 +782,58 @@ func (ff *Filters) loadLogsRequester() any {
return ff.logsRequestor.Load()
}
-func (ff *Filters) HasSubscription(id LogsSubID) bool {
- return ff.logsSubs.hasLogsFilter(id)
-}
-
-// HasHeadsSubscription returns true if a heads (new block headers) subscription exists for the given ID.
-func (ff *Filters) HasHeadsSubscription(id HeadsSubID) bool {
- _, ok := ff.headsSubs.Get(id)
- return ok
+// UnsubscribeLogs unsubscribes from logs using the given subscription ID.
+// It returns true if the unsubscription was successful, otherwise false.
+func (ff *Filters) UnsubscribeLogs(id LogsSubID) bool {
+ filter, ok := ff.logsSubs.logsFilters.Get(id)
+ if !ok {
+ ff.logger.Debug("[rpc] [filters] unsubscribe logs filter not found", "id", id)
+ return false
+ }
+ protocol := filter.sender.Protocol()
+ if !ff.removeLogsSubscription(id, true) {
+ return false
+ }
+ ff.logger.Debug("[rpc] [filters] unsubscribed logs filter", "id", id, "protocol", protocol)
+ ff.decrementMetrics(FilterTypeLogs, protocol)
+ return true
}
-// HasPendingTxsSubscription returns true if a pending transactions subscription exists for the given ID.
-func (ff *Filters) HasPendingTxsSubscription(id PendingTxsSubID) bool {
- _, ok := ff.pendingTxsSubs.Get(id)
- return ok
+// removeLogsSubscription is the logs counterpart of unsubscribeSubInternal (metrics
+// stay with the caller); logs additionally push the shrunken aggregate to the remote,
+// which eviction batches into one update per cycle via pushRemote=false. The store
+// and tracking entry are released regardless of whether the remote update succeeds —
+// a failed update only means the upstream keeps sending logs we'll now discard, it
+// does not reanimate the local subscription.
+func (ff *Filters) removeLogsSubscription(id LogsSubID, pushRemote bool) bool {
+ if !ff.logsSubs.removeLogsFilter(id) {
+ return false
+ }
+ if pushRemote {
+ ff.updateRemoteLogsFilter()
+ }
+ ff.deleteLogStore(id)
+ ff.trackedSubs.Delete(SubscriptionID(id))
+ return true
}
-// UnsubscribeLogs unsubscribes from logs using the given subscription ID.
-// It returns true if the unsubscription was successful, otherwise false.
-func (ff *Filters) UnsubscribeLogs(id LogsSubID) bool {
- isDeleted := ff.logsSubs.removeLogsFilter(id)
- // if any filters in the aggregate need all addresses or all topics then the request to the central
- // log subscription needs to honour this
+// updateRemoteLogsFilter pushes the aggregated filter state to the remote log source.
+// If any filters in the aggregate need all addresses or all topics then the request to
+// the central log subscription needs to honour this.
+func (ff *Filters) updateRemoteLogsFilter() {
lfr := ff.logsSubs.createFilterRequest()
-
addresses, topics := ff.logsSubs.getAggMaps()
-
for addr := range addresses {
lfr.Addresses = append(lfr.Addresses, gointerfaces.ConvertAddressToH160(addr))
}
for topic := range topics {
lfr.Topics = append(lfr.Topics, gointerfaces.ConvertHashToH256(topic))
}
- loaded := ff.loadLogsRequester()
- if loaded != nil {
+ if loaded := ff.loadLogsRequester(); loaded != nil {
if err := loaded.(func(*remoteproto.LogsFilterRequest) error)(lfr); err != nil {
ff.logger.Warn("Could not update remote logs filter", "err", err)
- return isDeleted || ff.logsSubs.removeLogsFilter(id)
}
}
-
- ff.deleteLogStore(id)
-
- return isDeleted
}
// deleteLogStore deletes the log store associated with the given subscription ID.
@@ -776,7 +951,15 @@ func (ff *Filters) OnNewLogs(reply *remoteproto.SubscribeLogsReply) {
// AddLogs adds logs to the store associated with the given subscription ID.
func (ff *Filters) AddLogs(id LogsSubID, log *types.Log) {
- ff.logsStores.DoAndStore(id, func(st []*types.Log, ok bool) []*types.Log {
+ ff.logsStores.Do(id, func(st []*types.Log, ok bool) ([]*types.Log, bool) {
+ // Drop (and clear) the entry when the subscription is gone: reads are gated
+ // on the subscription's existence, so a late write from the forwarding
+ // goroutine draining a closed channel would orphan the entry forever.
+ // Checking under the store lock makes this race-free against the
+ // remove-subscription-then-delete-store teardown ordering.
+ if !ff.logsSubs.hasLogsFilter(id) {
+ return nil, false
+ }
if !ok {
st = make([]*types.Log, 0)
}
@@ -793,7 +976,7 @@ func (ff *Filters) AddLogs(id LogsSubID, log *types.Log) {
// Append the new log
st = append(st, log)
- return st
+ return st, true
})
}
@@ -805,7 +988,11 @@ func (ff *Filters) ReadLogs(id LogsSubID) ([]*types.Log, bool) {
// AddPendingBlock adds a pending block header to the store associated with the given subscription ID.
func (ff *Filters) AddPendingBlock(id HeadsSubID, block *types.Header) {
- ff.pendingHeadsStores.DoAndStore(id, func(st []*types.Header, ok bool) []*types.Header {
+ ff.pendingHeadsStores.Do(id, func(st []*types.Header, ok bool) ([]*types.Header, bool) {
+ // Same orphaned-store guard as AddLogs.
+ if _, exists := ff.headsSubs.Get(id); !exists {
+ return nil, false
+ }
if !ok {
st = make([]*types.Header, 0)
}
@@ -822,7 +1009,7 @@ func (ff *Filters) AddPendingBlock(id HeadsSubID, block *types.Header) {
// Append the new header
st = append(st, block)
- return st
+ return st, true
})
}
@@ -834,7 +1021,11 @@ func (ff *Filters) ReadPendingBlocks(id HeadsSubID) ([]*types.Header, bool) {
// AddPendingTxs adds pending transactions to the store associated with the given subscription ID.
func (ff *Filters) AddPendingTxs(id PendingTxsSubID, txs []types.Transaction) {
- ff.pendingTxsStores.DoAndStore(id, func(st [][]types.Transaction, ok bool) [][]types.Transaction {
+ ff.pendingTxsStores.Do(id, func(st [][]types.Transaction, ok bool) ([][]types.Transaction, bool) {
+ // Same orphaned-store guard as AddLogs.
+ if _, exists := ff.pendingTxsSubs.Get(id); !exists {
+ return nil, false
+ }
if !ok {
st = make([][]types.Transaction, 0)
}
@@ -867,7 +1058,7 @@ func (ff *Filters) AddPendingTxs(id PendingTxsSubID, txs []types.Transaction) {
// Append the new transactions as a new batch
st = append(st, txs)
- return st
+ return st, true
})
}
@@ -923,3 +1114,19 @@ func (ff *Filters) WithTemporalOverlay(tx kv.TemporalTx) kv.TemporalTx {
}
return tx
}
+
+func (ff *Filters) incrementMetrics(ft FilterType, protocol SubProtocol) {
+ if protocol == "" {
+ return // internal subscription, not tracked
+ }
+ activeSubscriptionsGauge.WithLabelValues(string(ft), string(protocol)).Inc()
+ createdSubscriptionsCounter.WithLabelValues(string(ft), string(protocol)).Inc()
+}
+
+func (ff *Filters) decrementMetrics(ft FilterType, protocol SubProtocol) {
+ if protocol == "" {
+ return // internal subscription, not tracked
+ }
+ activeSubscriptionsGauge.WithLabelValues(string(ft), string(protocol)).Dec()
+ unsubscribedSubscriptionsCounter.WithLabelValues(string(ft), string(protocol)).Inc()
+}
diff --git a/rpc/rpchelper/filters_deadlock_test.go b/rpc/rpchelper/filters_deadlock_test.go
index 637023b8b9d..ff827e712f0 100644
--- a/rpc/rpchelper/filters_deadlock_test.go
+++ b/rpc/rpchelper/filters_deadlock_test.go
@@ -44,7 +44,7 @@ func TestFiltersDeadlock(t *testing.T) {
ctx, cancel := context.WithCancel(context.TODO())
for i := 0; i < subCount; i++ {
n := &sub{}
- n.ch, n.id = f.SubscribeLogs(128, crit)
+ n.ch, n.id, _ = f.SubscribeLogs(128, crit, "")
// start a loop similar to an rpcdaemon subscription, that calls unsubscribe on return
go func() {
defer f.UnsubscribeLogs(n.id)
diff --git a/rpc/rpchelper/filters_eviction_test.go b/rpc/rpchelper/filters_eviction_test.go
new file mode 100644
index 00000000000..c608701a6a2
--- /dev/null
+++ b/rpc/rpchelper/filters_eviction_test.go
@@ -0,0 +1,217 @@
+// 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 .
+
+package rpchelper
+
+import (
+ "runtime"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/execution/types"
+ "github.com/erigontech/erigon/node/gointerfaces/remoteproto"
+ "github.com/erigontech/erigon/rpc/filters"
+)
+
+func newTestFilters(t *testing.T) *Filters {
+ return New(t.Context(), DefaultFiltersConfig, nil, nil, nil, func() {}, log.New(), nil)
+}
+
+func backdateSub[T any](t *testing.T, sub Sub[T], age time.Duration) {
+ t.Helper()
+ cs, ok := sub.(*chan_sub[T])
+ require.True(t, ok)
+ cs.lock.Lock()
+ cs.lastAccess = time.Now().Add(-age)
+ cs.lock.Unlock()
+}
+
+func (ff *Filters) hasTrackedSub(id SubscriptionID) bool {
+ _, ok := ff.trackedSubs.Get(id)
+ return ok
+}
+
+func TestEvictStaleSubscriptionsRemovesIdleFilters(t *testing.T) {
+ f := newTestFilters(t)
+
+ headsCh, headsID := f.SubscribeNewHeads(8, ProtocolHTTP)
+ txsCh, txsID := f.SubscribePendingTxs(8, ProtocolHTTP)
+ logsCh, logsID, err := f.SubscribeLogs(8, filters.FilterCriteria{}, ProtocolHTTP)
+ require.NoError(t, err)
+
+ headsSub, ok := f.headsSubs.Get(headsID)
+ require.True(t, ok)
+ backdateSub(t, headsSub, 2*time.Hour)
+ txsSub, ok := f.pendingTxsSubs.Get(txsID)
+ require.True(t, ok)
+ backdateSub(t, txsSub, 2*time.Hour)
+ logsFilter, ok := f.logsSubs.logsFilters.Get(logsID)
+ require.True(t, ok)
+ backdateSub(t, logsFilter.sender, 2*time.Hour)
+
+ f.evictStaleSubscriptions(time.Hour)
+
+ _, ok = f.headsSubs.Get(headsID)
+ require.False(t, ok)
+ _, ok = f.pendingTxsSubs.Get(txsID)
+ require.False(t, ok)
+ _, ok = f.logsSubs.logsFilters.Get(logsID)
+ require.False(t, ok)
+ require.False(t, f.hasTrackedSub(SubscriptionID(headsID)))
+ require.False(t, f.hasTrackedSub(SubscriptionID(txsID)))
+ require.False(t, f.hasTrackedSub(SubscriptionID(logsID)))
+
+ _, open := <-headsCh
+ require.False(t, open)
+ _, open = <-txsCh
+ require.False(t, open)
+ _, open = <-logsCh
+ require.False(t, open)
+}
+
+func TestTouchSubscriptionPreventsEviction(t *testing.T) {
+ f := newTestFilters(t)
+
+ _, touchedID := f.SubscribePendingTxs(8, ProtocolHTTP)
+ _, idleID := f.SubscribePendingTxs(8, ProtocolHTTP)
+
+ for _, id := range []PendingTxsSubID{touchedID, idleID} {
+ sub, ok := f.pendingTxsSubs.Get(id)
+ require.True(t, ok)
+ backdateSub(t, sub, 2*time.Hour)
+ }
+
+ ft, ok := f.TouchSubscription(SubscriptionID(touchedID))
+ require.True(t, ok)
+ require.Equal(t, FilterTypePendingTxs, ft)
+ f.evictStaleSubscriptions(time.Hour)
+
+ _, ok = f.pendingTxsSubs.Get(touchedID)
+ require.True(t, ok)
+ _, ok = f.pendingTxsSubs.Get(idleID)
+ require.False(t, ok)
+}
+
+func TestEvictStaleSubscriptionsSkipsWebSocketSubscriptions(t *testing.T) {
+ f := newTestFilters(t)
+
+ _, id := f.SubscribeNewHeads(8, ProtocolWS)
+
+ f.evictStaleSubscriptions(time.Nanosecond)
+
+ _, ok := f.headsSubs.Get(id)
+ require.True(t, ok)
+}
+
+// A forwarding goroutine can drain channel-buffered items after its subscription is
+// unsubscribed or evicted; such late writes must not recreate the per-filter store,
+// which would be unreachable (reads are gated on the subscription) and leak forever.
+func TestAddAfterUnsubscribeDoesNotOrphanStore(t *testing.T) {
+ f := newTestFilters(t)
+
+ t.Run("heads", func(t *testing.T) {
+ _, id := f.SubscribeNewHeads(8, ProtocolHTTP)
+ require.True(t, f.UnsubscribeHeads(id))
+ require.False(t, f.hasTrackedSub(SubscriptionID(id)))
+ f.AddPendingBlock(id, &types.Header{})
+ _, ok := f.pendingHeadsStores.Get(id)
+ require.False(t, ok)
+ })
+ t.Run("pendingTxs", func(t *testing.T) {
+ _, id := f.SubscribePendingTxs(8, ProtocolHTTP)
+ require.True(t, f.UnsubscribePendingTxs(id))
+ require.False(t, f.hasTrackedSub(SubscriptionID(id)))
+ f.AddPendingTxs(id, []types.Transaction{})
+ _, ok := f.pendingTxsStores.Get(id)
+ require.False(t, ok)
+ })
+ t.Run("logs", func(t *testing.T) {
+ _, id, err := f.SubscribeLogs(8, filters.FilterCriteria{}, ProtocolHTTP)
+ require.NoError(t, err)
+ require.True(t, f.UnsubscribeLogs(id))
+ require.False(t, f.hasTrackedSub(SubscriptionID(id)))
+ f.AddLogs(id, &types.Log{})
+ _, ok := f.logsStores.Get(id)
+ require.False(t, ok)
+ })
+}
+
+func TestLogsEvictionBatchesRemoteFilterUpdate(t *testing.T) {
+ f := newTestFilters(t)
+
+ var updates atomic.Int32
+ f.logsRequestor.Store(func(*remoteproto.LogsFilterRequest) error {
+ updates.Add(1)
+ return nil
+ })
+
+ ids := make([]LogsSubID, 0, 3)
+ for range 3 {
+ _, id, err := f.SubscribeLogs(8, filters.FilterCriteria{}, ProtocolHTTP)
+ require.NoError(t, err)
+ lf, ok := f.logsSubs.logsFilters.Get(id)
+ require.True(t, ok)
+ backdateSub(t, lf.sender, 2*time.Hour)
+ ids = append(ids, id)
+ }
+
+ updates.Store(0) // SubscribeLogs issues its own updates; count only eviction's
+ f.evictStaleSubscriptions(time.Hour)
+
+ for _, id := range ids {
+ _, ok := f.logsSubs.logsFilters.Get(id)
+ require.False(t, ok)
+ }
+ require.EqualValues(t, 1, updates.Load())
+}
+
+func TestConcurrentTouchAndEviction(t *testing.T) {
+ f := newTestFilters(t)
+
+ _, id := f.SubscribePendingTxs(8, ProtocolHTTP)
+
+ stop := make(chan struct{})
+ var wg sync.WaitGroup
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ f.TouchSubscription(SubscriptionID(id))
+ runtime.Gosched()
+ }
+ }
+ }()
+
+ deadline := time.Now().Add(200 * time.Millisecond)
+ for time.Now().Before(deadline) {
+ f.evictStaleSubscriptions(50 * time.Millisecond)
+ runtime.Gosched()
+ }
+ close(stop)
+ wg.Wait()
+
+ _, ok := f.pendingTxsSubs.Get(id)
+ require.True(t, ok)
+}
diff --git a/rpc/rpchelper/filters_subscribe_test.go b/rpc/rpchelper/filters_subscribe_test.go
new file mode 100644
index 00000000000..de153d1c849
--- /dev/null
+++ b/rpc/rpchelper/filters_subscribe_test.go
@@ -0,0 +1,53 @@
+// 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 .
+
+package rpchelper
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/node/gointerfaces/remoteproto"
+ "github.com/erigontech/erigon/rpc/filters"
+)
+
+// A filter id must never be handed out for a subscription that is not installed:
+// when the remote filter update fails, Subscribe* reports the error instead.
+func TestSubscribeLogsRemoteUpdateFailureReturnsError(t *testing.T) {
+ f := newTestFilters(t)
+ f.logsRequestor.Store(func(*remoteproto.LogsFilterRequest) error {
+ return errors.New("remote logs source unavailable")
+ })
+
+ _, id, err := f.SubscribeLogs(8, filters.FilterCriteria{}, ProtocolHTTP)
+ require.Error(t, err)
+ _, ok := f.logsSubs.logsFilters.Get(id)
+ require.False(t, ok)
+ require.False(t, f.hasTrackedSub(SubscriptionID(id)))
+}
+
+func TestSubscribeReceiptsRemoteUpdateFailureReturnsError(t *testing.T) {
+ f := newTestFilters(t)
+ f.receiptsRequestor.Store(func(*remoteproto.ReceiptsFilterRequest) error {
+ return errors.New("remote receipts source unavailable")
+ })
+
+ _, id, err := f.SubscribeReceipts(8, filters.ReceiptsFilterCriteria{})
+ require.Error(t, err)
+ require.False(t, f.receiptsSubs.removeReceiptsFilter(id))
+}
diff --git a/rpc/rpchelper/filters_test.go b/rpc/rpchelper/filters_test.go
index cf1ae0f8f98..9dc3f89537e 100644
--- a/rpc/rpchelper/filters_test.go
+++ b/rpc/rpchelper/filters_test.go
@@ -84,7 +84,7 @@ func TestFilters_SingleSubscription_OnlyTopicsSubscribedAreBroadcast(t *testing.
Topics: [][]common.Hash{{subbedTopic}},
}
- outChan, _ := f.SubscribeLogs(10, criteria)
+ outChan, _, _ := f.SubscribeLogs(10, criteria, "")
// now create a log for some other topic and distribute it
log := createLog()
@@ -118,7 +118,7 @@ func TestFilters_SingleSubscription_EmptyTopicsInCriteria_OnlyTopicsSubscribedAr
Topics: [][]common.Hash{{nilTopic, subbedTopic, nilTopic}},
}
- outChan, _ := f.SubscribeLogs(10, criteria)
+ outChan, _, _ := f.SubscribeLogs(10, criteria, "")
// now create a log for some other topic and distribute it
log := createLog()
@@ -151,7 +151,7 @@ func TestFilters_SingleSubscription_TopicPositionWildcardIsPreserved(t *testing.
Topics: [][]common.Hash{{topic0}, nil, {topic2}},
}
- outChan, _ := f.SubscribeLogs(10, criteria)
+ outChan, _, _ := f.SubscribeLogs(10, criteria, "")
matchingLog := createLog()
matchingLog.Topics = []*typesproto.H256{
@@ -188,7 +188,7 @@ func TestFilters_SingleSubscription_WildcardOnlyTopicRowMatches(t *testing.T) {
Topics: [][]common.Hash{nil},
}
- outChan, _ := f.SubscribeLogs(10, criteria)
+ outChan, _, _ := f.SubscribeLogs(10, criteria, "")
log1 := createLog()
f.OnNewLogs(log1)
@@ -218,8 +218,8 @@ func TestFilters_TwoSubscriptionsWithDifferentCriteria(t *testing.T) {
Topics: [][]common.Hash{{topic1}},
}
- chan1, _ := f.SubscribeLogs(256, criteria1)
- chan2, _ := f.SubscribeLogs(256, criteria2)
+ chan1, _, _ := f.SubscribeLogs(256, criteria1, "")
+ chan2, _, _ := f.SubscribeLogs(256, criteria2, "")
// now create a log for some other topic and distribute it
log := createLog()
@@ -263,9 +263,9 @@ func TestFilters_ThreeSubscriptionsWithDifferentCriteria(t *testing.T) {
Topics: [][]common.Hash{},
}
- chan1, _ := f.SubscribeLogs(256, criteria1)
- chan2, _ := f.SubscribeLogs(256, criteria2)
- chan3, _ := f.SubscribeLogs(256, criteria3)
+ chan1, _, _ := f.SubscribeLogs(256, criteria1, "")
+ chan2, _, _ := f.SubscribeLogs(256, criteria2, "")
+ chan3, _, _ := f.SubscribeLogs(256, criteria3, "")
// now create a log for some other topic and distribute it
log := createLog()
@@ -332,7 +332,7 @@ func TestFilters_SubscribeLogsGeneratesCorrectLogFilterRequest(t *testing.T) {
Addresses: []common.Address{},
Topics: [][]common.Hash{},
}
- _, id1 := f.SubscribeLogs(1, criteria1)
+ _, id1, _ := f.SubscribeLogs(1, criteria1, "")
// request should have all addresses and topics enabled
if lastFilterRequest.AllAddresses == false {
@@ -347,7 +347,7 @@ func TestFilters_SubscribeLogsGeneratesCorrectLogFilterRequest(t *testing.T) {
Addresses: []common.Address{address1},
Topics: [][]common.Hash{},
}
- _, id2 := f.SubscribeLogs(1, criteria2)
+ _, id2, _ := f.SubscribeLogs(1, criteria2, "")
// request should have all addresses and all topics still and the new address
if lastFilterRequest.AllAddresses == false {
@@ -365,7 +365,7 @@ func TestFilters_SubscribeLogsGeneratesCorrectLogFilterRequest(t *testing.T) {
Addresses: []common.Address{},
Topics: [][]common.Hash{{topic1}},
}
- _, id3 := f.SubscribeLogs(1, criteria3)
+ _, id3, _ := f.SubscribeLogs(1, criteria3, "")
// request should have all addresses and all topics as well as the previous address and new topic
if lastFilterRequest.AllAddresses == false {
@@ -450,7 +450,7 @@ func TestFilters_AddLogs(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
config := FiltersConfig{RpcSubscriptionFiltersMaxLogs: tt.maxLogs}
f := New(t.Context(), config, nil, nil, nil, func() {}, log.New(), nil)
- logID := LogsSubID("test-log")
+ _, logID, _ := f.SubscribeLogs(8, filters.FilterCriteria{}, "")
logEntry := &types.Log{Address: common.HexToAddress("095e7baea6a6c7c4c2dfeb977efac326af552d87")}
for i := 0; i < tt.numToAdd; i++ {
@@ -484,7 +484,7 @@ func TestFilters_AddPendingBlocks(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
config := FiltersConfig{RpcSubscriptionFiltersMaxHeaders: tt.maxHeaders}
f := New(t.Context(), config, nil, nil, nil, func() {}, log.New(), nil)
- blockID := HeadsSubID("test-block")
+ _, blockID := f.SubscribeNewHeads(8, "")
header := &types.Header{}
for i := 0; i < tt.numToAdd; i++ {
@@ -519,7 +519,7 @@ func TestFilters_AddPendingTxs(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
config := FiltersConfig{RpcSubscriptionFiltersMaxTxs: tt.maxTxs}
f := New(t.Context(), config, nil, nil, nil, func() {}, log.New(), nil)
- txID := PendingTxsSubID("test-tx")
+ _, txID := f.SubscribePendingTxs(8, "")
var txn types.Transaction = types.NewTransaction(0, common.HexToAddress("095e7baea6a6c7c4c2dfeb977efac326af552d87"), uint256.NewInt(10), 50000, uint256.NewInt(10), nil)
txn, _ = txn.WithSignature(*types.LatestSignerForChainID(nil), common.Hex2Bytes("9bea4c4daac7c7c52e093e6a4c35dbbcf8856f1af7b059ba20253e70848d094f8a8fae537ce25ed8cb5af9adac3f141af69bd515bd2ba031522df09b97dd72b100"))
@@ -602,7 +602,7 @@ func TestFilters_SingleReceiptsSubscription_OnlyTransactionHashesSubscribedAreBr
TransactionHashes: []common.Hash{txHash1},
}
- outChan, _ := f.SubscribeReceipts(10, criteria)
+ outChan, _, _ := f.SubscribeReceipts(10, criteria)
// Create a receipt for a different transaction hash
receipt := createReceipt(txHash2)
@@ -633,7 +633,7 @@ func TestFilters_ReceiptsSubscription_EmptyFilterSubscribesToAll(t *testing.T) {
TransactionHashes: []common.Hash{},
}
- outChan, _ := f.SubscribeReceipts(10, criteria)
+ outChan, _, _ := f.SubscribeReceipts(10, criteria)
// Any receipt should be received
receipt1 := createReceipt(txHash1)
@@ -665,8 +665,8 @@ func TestFilters_TwoReceiptsSubscriptionsWithDifferentCriteria(t *testing.T) {
TransactionHashes: []common.Hash{txHash1},
}
- chan1, _ := f.SubscribeReceipts(256, criteria1)
- chan2, _ := f.SubscribeReceipts(256, criteria2)
+ chan1, _, _ := f.SubscribeReceipts(256, criteria1)
+ chan2, _, _ := f.SubscribeReceipts(256, criteria2)
// Create a receipt for txHash2
receipt := createReceipt(txHash2)
@@ -708,9 +708,9 @@ func TestFilters_ThreeReceiptsSubscriptionsWithDifferentCriteria(t *testing.T) {
TransactionHashes: []common.Hash{txHash1, txHash2},
}
- chan1, _ := f.SubscribeReceipts(256, criteria1)
- chan2, _ := f.SubscribeReceipts(256, criteria2)
- chan3, _ := f.SubscribeReceipts(256, criteria3)
+ chan1, _, _ := f.SubscribeReceipts(256, criteria1)
+ chan2, _, _ := f.SubscribeReceipts(256, criteria2)
+ chan3, _, _ := f.SubscribeReceipts(256, criteria3)
// Receipt for txHash3 (not subscribed by chan2 or chan3)
receipt := createReceipt(txHash3)
@@ -774,7 +774,7 @@ func TestFilters_SubscribeReceiptsGeneratesCorrectReceiptsFilterRequest(t *testi
criteria1 := filters.ReceiptsFilterCriteria{
TransactionHashes: []common.Hash{},
}
- _, id1 := f.SubscribeReceipts(1, criteria1)
+ _, id1, _ := f.SubscribeReceipts(1, criteria1)
// Request should have AllTransactions=true and empty TransactionHashes
if !lastFilterRequest.AllTransactions {
@@ -788,7 +788,7 @@ func TestFilters_SubscribeReceiptsGeneratesCorrectReceiptsFilterRequest(t *testi
criteria2 := filters.ReceiptsFilterCriteria{
TransactionHashes: []common.Hash{txHash1},
}
- _, id2 := f.SubscribeReceipts(1, criteria2)
+ _, id2, _ := f.SubscribeReceipts(1, criteria2)
// Request should have AllTransactions=true and include txHash1
// Backend uses OR logic: send if (AllTransactions OR hash matches)
@@ -817,7 +817,7 @@ func TestFilters_SubscribeReceiptsGeneratesCorrectReceiptsFilterRequest(t *testi
criteria3 := filters.ReceiptsFilterCriteria{
TransactionHashes: []common.Hash{txHash2, txHash3},
}
- _, id3 := f.SubscribeReceipts(1, criteria3)
+ _, id3, _ := f.SubscribeReceipts(1, criteria3)
// Request should have all three transaction hashes
if len(lastFilterRequest.TransactionHashes) != 3 {
diff --git a/rpc/rpchelper/logsfilter.go b/rpc/rpchelper/logsfilter.go
index 32d097d6eaf..7b63b42ea79 100644
--- a/rpc/rpchelper/logsfilter.go
+++ b/rpc/rpchelper/logsfilter.go
@@ -102,8 +102,8 @@ func (a *LogsFilterAggregator) removeLogsFilter(filterId LogsSubID) bool {
// hasLogsFilter checks if a log filter identified by filterId is present in the LogsFilterAggregator.
func (a *LogsFilterAggregator) hasLogsFilter(filterId LogsSubID) bool {
- a.logsFilterLock.Lock()
- defer a.logsFilterLock.Unlock()
+ a.logsFilterLock.RLock()
+ defer a.logsFilterLock.RUnlock()
_, ok := a.logsFilters.Get(filterId)
return ok
diff --git a/rpc/rpchelper/metrics.go b/rpc/rpchelper/metrics.go
index a91dd46609a..0afd12368a6 100644
--- a/rpc/rpchelper/metrics.go
+++ b/rpc/rpchelper/metrics.go
@@ -21,12 +21,17 @@ import (
)
const (
- filterLabelName = "filter"
- clientLabelName = "client"
+ filterLabelName = "filter"
+ clientLabelName = "client"
+ protocolLabelName = "protocol"
)
var (
- activeSubscriptionsGauge = metrics.GetOrCreateGaugeVec("subscriptions", []string{filterLabelName}, "Current number of subscriptions")
+ activeSubscriptionsGauge = metrics.GetOrCreateGaugeVec("subscriptions_active", []string{filterLabelName, protocolLabelName}, "Current number of active subscriptions")
+ createdSubscriptionsCounter = metrics.GetOrCreateCounterVec("subscriptions_created_total", []string{filterLabelName, protocolLabelName}, "Total number of subscriptions created")
+ unsubscribedSubscriptionsCounter = metrics.GetOrCreateCounterVec("subscriptions_unsubscribed_total", []string{filterLabelName, protocolLabelName}, "Total number of subscriptions removed by client unsubscribe or timeout eviction")
+ reapedSubscriptionsCounter = metrics.GetOrCreateCounterVec("subscriptions_reaped_total", []string{filterLabelName}, "Total number of idle subscriptions evicted by timeout")
+
activeSubscriptionsLogsAllAddressesGauge = metrics.GetOrCreateGauge("subscriptions_logs_all_addresses")
activeSubscriptionsLogsAllTopicsGauge = metrics.GetOrCreateGauge("subscriptions_logs_all_topics")
activeSubscriptionsLogsAddressesGauge = metrics.GetOrCreateGauge("subscriptions_logs_addresses")
diff --git a/rpc/rpchelper/subscription.go b/rpc/rpchelper/subscription.go
index 25ec40722d1..b9e33bc2e03 100644
--- a/rpc/rpchelper/subscription.go
+++ b/rpc/rpchelper/subscription.go
@@ -18,28 +18,58 @@ package rpchelper
import (
"sync"
+ "time"
+)
+
+// SubProtocol says how a subscription is consumed and thereby its lifecycle: HTTP
+// polling filters are subject to timeout eviction, WS subscriptions end with their
+// connection, and the empty value marks internal subscriptions that are neither
+// evicted nor counted in metrics.
+type SubProtocol string
+
+const (
+ ProtocolHTTP SubProtocol = "http"
+ ProtocolWS SubProtocol = "ws"
)
// a simple interface for subscriptions for rpc helper
type Sub[T any] interface {
Send(T)
Close()
+ SubTracker
+}
+
+// SubTracker is the element-type-independent surface of a subscription,
+// used for timeout-based eviction and metrics.
+type SubTracker interface {
+ Touch() // reset the eviction deadline (HTTP polling filters)
+ CloseIfIdle(timeout time.Duration) bool // close and report true when idle longer than timeout
+ Protocol() SubProtocol
}
type chan_sub[T any] struct {
- lock sync.Mutex // protects all fileds of this struct
- ch chan T
- closed bool
+ lock sync.Mutex // protects all mutable fields of this struct
+ ch chan T
+ closed bool
+ protocol SubProtocol // immutable after construction
+ // lastAccess is the last poll time for timeout eviction; zero means no
+ // timeout tracking (push subscriptions die with their connection instead).
+ lastAccess time.Time
}
// newChanSub - buffered channel
-func newChanSub[T any](size int) *chan_sub[T] {
+func newChanSub[T any](size int, protocol SubProtocol) *chan_sub[T] {
if size < 8 { // set min size to 8
size = 8
}
- o := &chan_sub[T]{}
- o.ch = make(chan T, size)
- return o
+ s := &chan_sub[T]{
+ ch: make(chan T, size),
+ protocol: protocol,
+ }
+ if protocol == ProtocolHTTP {
+ s.lastAccess = time.Now()
+ }
+ return s
}
func (s *chan_sub[T]) Send(x T) {
s.lock.Lock()
@@ -52,12 +82,46 @@ func (s *chan_sub[T]) Send(x T) {
default: // the sub is overloaded, dispose message
}
}
+
+// Close is idempotent and safe to call concurrently with Send and CloseIfIdle.
func (s *chan_sub[T]) Close() {
s.lock.Lock()
defer s.lock.Unlock()
+ s.closeLocked()
+}
+
+func (s *chan_sub[T]) closeLocked() {
if s.closed {
return
}
s.closed = true
close(s.ch)
}
+
+// Touch resets the last access time, preventing timeout eviction.
+// A no-op unless the subscription is tracked for eviction.
+func (s *chan_sub[T]) Touch() {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ if !s.lastAccess.IsZero() && !s.closed {
+ s.lastAccess = time.Now()
+ }
+}
+
+// CloseIfIdle closes the subscription and reports true when timeout tracking is
+// enabled and the last access is older than timeout. Deciding and closing under
+// the same lock Touch uses guarantees a subscription touched within the timeout
+// is never evicted.
+func (s *chan_sub[T]) CloseIfIdle(timeout time.Duration) bool {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ if s.closed || s.lastAccess.IsZero() || time.Since(s.lastAccess) <= timeout {
+ return false
+ }
+ s.closeLocked()
+ return true
+}
+
+func (s *chan_sub[T]) Protocol() SubProtocol {
+ return s.protocol
+}