From 6d67fb1c8bb203dfc0c926f72f1fde4eef349f24 Mon Sep 17 00:00:00 2001 From: hfuss Date: Thu, 8 Jan 2026 09:39:13 -0500 Subject: [PATCH] [rpc] Add Tracking to Subscriptions Made via HTTP for Unused Filter Eviction Signed-off-by: hfuss self-review fixes Signed-off-by: hfuss --- cmd/rpcdaemon/cli/config.go | 1 + node/cli/default_flags.go | 1 + node/cli/flags.go | 6 + rpc/jsonrpc/eth_filters.go | 52 +++--- rpc/rpchelper/config.go | 17 +- rpc/rpchelper/filters.go | 319 ++++++++++++++++++++++++++++++++-- rpc/rpchelper/metrics.go | 34 +++- rpc/rpchelper/subscription.go | 67 ++++++- 8 files changed, 454 insertions(+), 43 deletions(-) diff --git a/cmd/rpcdaemon/cli/config.go b/cmd/rpcdaemon/cli/config.go index eafe14097e7..e692f42574f 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 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/node/cli/default_flags.go b/node/cli/default_flags.go index 46c914537f2..25af40f859e 100644 --- a/node/cli/default_flags.go +++ b/node/cli/default_flags.go @@ -121,6 +121,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 1302e6c545c..a174a29d769 100644 --- a/node/cli/flags.go +++ b/node/cli/flags.go @@ -206,6 +206,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. Set to 0 to disable eviction.", + Value: rpchelper.DefaultFiltersConfig.RpcSubscriptionFiltersTimeout, + } ) // BuildEthConfig applies all CLI flags to the ethconfig.Config. This is the single @@ -466,6 +471,7 @@ func setEmbeddedRpcDaemon(ctx *cli.Context, 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: ctx.Uint64(utils.RpcGasCapFlag.Name), BlockRangeLimit: ctx.Int(utils.RpcBlockRangeLimit.Name), diff --git a/rpc/jsonrpc/eth_filters.go b/rpc/jsonrpc/eth_filters.go index eb3c926357e..cd5a63e234a 100644 --- a/rpc/jsonrpc/eth_filters.go +++ b/rpc/jsonrpc/eth_filters.go @@ -35,6 +35,7 @@ func (api *APIImpl) NewPendingTransactionFilter(_ context.Context) (string, erro return "", rpc.ErrNotificationsUnsupported } txsCh, id := api.filters.SubscribePendingTxs(32) + api.filters.TrackSubscription(rpchelper.SubscriptionID(id), rpchelper.FilterTypePendingTxs, rpchelper.ProtocolHTTP) go func() { for txs := range txsCh { api.filters.AddPendingTxs(id, txs) @@ -49,6 +50,7 @@ func (api *APIImpl) NewBlockFilter(_ context.Context) (string, error) { return "", rpc.ErrNotificationsUnsupported } ch, id := api.filters.SubscribeNewHeads(32) + api.filters.TrackSubscription(rpchelper.SubscriptionID(id), rpchelper.FilterTypeHeads, rpchelper.ProtocolHTTP) go func() { for block := range ch { api.filters.AddPendingBlock(id, block) @@ -63,6 +65,7 @@ func (api *APIImpl) NewFilter(_ context.Context, crit filters.FilterCriteria) (s return "", rpc.ErrNotificationsUnsupported } logs, id := api.filters.SubscribeLogs(256, crit) + api.filters.TrackSubscription(rpchelper.SubscriptionID(id), rpchelper.FilterTypeLogs, rpchelper.ProtocolHTTP) go func() { for lg := range logs { api.filters.AddLogs(id, lg) @@ -100,37 +103,38 @@ 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 { - 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()) + // Identify which subscription type the id belongs to, touch it to reset the + // eviction deadline, then probe the store. Touch must happen on every poll + // — not only when there is buffered data — otherwise an actively-polling + // client whose chain is quiet would be evicted. + switch { + case api.filters.HasHeadsSubscription(rpchelper.HeadsSubID(cutIndex)): + api.filters.TouchSubscription(rpchelper.SubscriptionID(cutIndex), 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 { + case api.filters.HasPendingTxsSubscription(rpchelper.PendingTxsSubID(cutIndex)): + api.filters.TouchSubscription(rpchelper.SubscriptionID(cutIndex), rpchelper.FilterTypePendingTxs) + if txs, ok := api.filters.ReadPendingTxs(rpchelper.PendingTxsSubID(cutIndex)); ok && len(txs) > 0 { for _, txn := range txs[0] { 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 api.filters.HasSubscription(rpchelper.LogsSubID(cutIndex)): + api.filters.TouchSubscription(rpchelper.SubscriptionID(cutIndex), rpchelper.FilterTypeLogs) + if logs, ok := api.filters.ReadLogs(rpchelper.LogsSubID(cutIndex)); ok { + for _, v := range logs { + stub = append(stub, v) + } } return stub, nil + default: + return nil, rpc.ErrFilterNotFound } - return []any{}, nil } // GetFilterLogs implements eth_getFilterLogs. @@ -144,6 +148,8 @@ func (api *APIImpl) GetFilterLogs(_ context.Context, index string) ([]*types.Log if found := api.filters.HasSubscription(rpchelper.LogsSubID(cutIndex)); !found { return nil, rpc.ErrFilterNotFound } + // Reset the filter deadline since it was just accessed + api.filters.TouchSubscription(rpchelper.SubscriptionID(cutIndex), rpchelper.FilterTypeLogs) if logs, ok := api.filters.ReadLogs(rpchelper.LogsSubID(cutIndex)); ok { return logs, nil } @@ -165,6 +171,7 @@ func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) { go func() { defer dbg.LogPanic() headers, id := api.filters.SubscribeNewHeads(32) + api.filters.SetSubscriptionProtocol(rpchelper.SubscriptionID(id), rpchelper.FilterTypeHeads, rpchelper.ProtocolWS) defer api.filters.UnsubscribeHeads(id) for { select { @@ -203,6 +210,7 @@ func (api *APIImpl) NewPendingTransactions(ctx context.Context, fullTx *bool) (* go func() { defer dbg.LogPanic() txsCh, id := api.filters.SubscribePendingTxs(256) + api.filters.SetSubscriptionProtocol(rpchelper.SubscriptionID(id), rpchelper.FilterTypePendingTxs, rpchelper.ProtocolWS) defer api.filters.UnsubscribePendingTxs(id) for { @@ -250,6 +258,7 @@ func (api *APIImpl) NewPendingTransactionsWithBody(ctx context.Context) (*rpc.Su go func() { defer dbg.LogPanic() txsCh, id := api.filters.SubscribePendingTxs(512) + api.filters.SetSubscriptionProtocol(rpchelper.SubscriptionID(id), rpchelper.FilterTypePendingTxs, rpchelper.ProtocolWS) defer api.filters.UnsubscribePendingTxs(id) for { @@ -291,6 +300,7 @@ func (api *APIImpl) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc go func() { defer dbg.LogPanic() logs, id := api.filters.SubscribeLogs(api.SubscribeLogsChannelSize, crit) + api.filters.SetSubscriptionProtocol(rpchelper.SubscriptionID(id), rpchelper.FilterTypeLogs, rpchelper.ProtocolWS) defer api.filters.UnsubscribeLogs(id) for { diff --git a/rpc/rpchelper/config.go b/rpc/rpchelper/config.go index ff85a580874..ea078a30a26 100644 --- a/rpc/rpchelper/config.go +++ b/rpc/rpchelper/config.go @@ -16,15 +16,21 @@ package rpchelper +import "time" + +// DefaultFilterTimeout is the default eviction timeout; 0 disables eviction. +const DefaultFilterTimeout time.Duration = 0 + // 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: 0 (no 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..7a8b461d3b6 100644 --- a/rpc/rpchelper/filters.go +++ b/rpc/rpchelper/filters.go @@ -47,6 +47,15 @@ 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" +) + // 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. @@ -104,6 +113,16 @@ func New(ctx context.Context, config FiltersConfig, ethBackend ApiBackend, txPoo events: events, } + // Start the timeout loop for filter eviction if timeout is configured + if config.RpcSubscriptionFiltersTimeout > 0 { + logger.Info("[rpc] [filters] starting timeout loop for idle filter eviction", + "timeout", config.RpcSubscriptionFiltersTimeout, + "checkInterval", config.RpcSubscriptionFiltersTimeout/2) + go ff.timeoutLoop(ctx, config.RpcSubscriptionFiltersTimeout) + } else { + logger.Info("[rpc] [filters] timeout-based filter eviction disabled") + } + go func() { if ethBackend == nil { return @@ -286,6 +305,210 @@ 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 time.Duration) { + // Check more frequently than the timeout to ensure timely eviction. + // Floor at 1s so a sub-2s timeout doesn't produce a zero or sub-second + // interval (time.NewTicker panics on non-positive durations and a tight + // loop would burn CPU for no practical benefit). + checkInterval := timeout / 2 + if checkInterval < time.Second { + checkInterval = time.Second + } + ticker := time.NewTicker(checkInterval) + defer ticker.Stop() + + ff.logger.Debug("[rpc] [filters] timeout loop started", "timeout", timeout, "checkInterval", checkInterval) + + 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) + } + } +} + +// evictStaleFilters removes filters that haven't been accessed within the timeout duration. +// It iterates the existing subscription maps and checks each subscription's embedded tracking. +func (ff *Filters) evictStaleSubscriptions(timeout time.Duration) { + var headsChecked, txsChecked, logsChecked int + var headsEvicted, txsEvicted, logsEvicted int + + // Collect expired heads subscriptions + var headsToEvict []HeadsSubID + ff.headsSubs.Range(func(id HeadsSubID, sub Sub[*types.Header]) error { + headsChecked++ + if sub.ShouldEvict(timeout) { + ff.logger.Trace("[rpc] [filters] heads filter marked for eviction", "id", id, "protocol", sub.Protocol()) + headsToEvict = append(headsToEvict, id) + } + return nil + }) + for _, id := range headsToEvict { + if sub, ok := ff.headsSubs.Get(id); ok { + // Re-check ShouldEvict: a client may have called TouchSubscription between phase 1 and phase 2. + // This shrinks the TOCTOU window but does not eliminate it; an unavoidable race remains + // between this check and unsubscribeHeadsInternal, but the timeout is on the order of minutes + // so the practical impact is bounded. + if !sub.ShouldEvict(timeout) { + continue + } + protocol := sub.Protocol() + if ff.unsubscribeHeadsInternal(id) { + headsEvicted++ + ff.logger.Info("[rpc] [filters] evicted idle heads filter", "id", id, "protocol", protocol, "timeout", timeout) + ff.decrementMetrics(FilterTypeHeads, protocol) + getReapedCounter(string(FilterTypeHeads)).Inc() + } + } + } + + // Collect expired pending txs subscriptions + var txsToEvict []PendingTxsSubID + ff.pendingTxsSubs.Range(func(id PendingTxsSubID, sub Sub[[]types.Transaction]) error { + txsChecked++ + if sub.ShouldEvict(timeout) { + ff.logger.Trace("[rpc] [filters] pending txs filter marked for eviction", "id", id, "protocol", sub.Protocol()) + txsToEvict = append(txsToEvict, id) + } + return nil + }) + for _, id := range txsToEvict { + if sub, ok := ff.pendingTxsSubs.Get(id); ok { + if !sub.ShouldEvict(timeout) { + continue + } + protocol := sub.Protocol() + if ff.unsubscribePendingTxsInternal(id) { + txsEvicted++ + ff.logger.Info("[rpc] [filters] evicted idle pending txs filter", "id", id, "protocol", protocol, "timeout", timeout) + ff.decrementMetrics(FilterTypePendingTxs, protocol) + getReapedCounter(string(FilterTypePendingTxs)).Inc() + } + } + } + + // Collect expired logs filter subscriptions + var logsToEvict []LogsSubID + ff.logsSubs.logsFilters.Range(func(id LogsSubID, filter *LogsFilter) error { + logsChecked++ + if filter.sender != nil && filter.sender.ShouldEvict(timeout) { + ff.logger.Trace("[rpc] [filters] logs filter marked for eviction", "id", id, "protocol", filter.sender.Protocol()) + logsToEvict = append(logsToEvict, id) + } + return nil + }) + for _, id := range logsToEvict { + if filter, ok := ff.logsSubs.logsFilters.Get(id); ok && filter.sender != nil { + if !filter.sender.ShouldEvict(timeout) { + continue + } + protocol := filter.sender.Protocol() + if ff.unsubscribeLogsInternal(id) { + logsEvicted++ + ff.logger.Info("[rpc] [filters] evicted idle logs filter", "id", id, "protocol", protocol, "timeout", timeout) + ff.decrementMetrics(FilterTypeLogs, protocol) + getReapedCounter(string(FilterTypeLogs)).Inc() + } + } + } + + // Log summary at DEBUG level + totalEvicted := headsEvicted + txsEvicted + logsEvicted + if totalEvicted > 0 { + ff.logger.Debug("[rpc] [filters] eviction cycle complete", + "headsChecked", headsChecked, "headsEvicted", headsEvicted, + "txsChecked", txsChecked, "txsEvicted", txsEvicted, + "logsChecked", logsChecked, "logsEvicted", logsEvicted) + } else { + ff.logger.Trace("[rpc] [filters] eviction cycle complete, no stale filters", + "headsChecked", headsChecked, "txsChecked", txsChecked, "logsChecked", logsChecked) + } +} + +// TouchSubscription resets the deadline for the given filter, preventing it from being evicted. +// This should be called whenever a filter is accessed (e.g., via GetFilterChanges). +// Called on every poll, so it must stay cheap — no logging on the hot path. +func (ff *Filters) TouchSubscription(id SubscriptionID, ft FilterType) { + switch ft { + case FilterTypeHeads: + if sub, ok := ff.headsSubs.Get(HeadsSubID(id)); ok { + sub.Touch() + } + case FilterTypePendingTxs: + if sub, ok := ff.pendingTxsSubs.Get(PendingTxsSubID(id)); ok { + sub.Touch() + } + case FilterTypeLogs: + if filter, ok := ff.logsSubs.logsFilters.Get(LogsSubID(id)); ok && filter.sender != nil { + filter.sender.Touch() + } + } +} + +// TrackSubscription enables timeout tracking and metrics for a subscription. +// This should be called for HTTP polling filters. For WebSocket subscriptions, call +// SetSubscriptionProtocol directly without enabling tracking as they are "evicted" once +// connections are closed. +func (ff *Filters) TrackSubscription(id SubscriptionID, ft FilterType, protocol string) { + // Only increment metrics when the subscription is actually found and tagged. + // Incrementing on a missing lookup would leak the gauge — nothing later + // decrements it because there is no subscription to unsubscribe. + switch ft { + case FilterTypeHeads: + if sub, ok := ff.headsSubs.Get(HeadsSubID(id)); ok { + sub.SetProtocol(protocol) + sub.EnableTimeout() + ff.logger.Debug("[rpc] [filters] registered heads subscription with timeout tracking", "id", id, "protocol", protocol) + ff.incrementMetrics(ft, protocol) + } + case FilterTypePendingTxs: + if sub, ok := ff.pendingTxsSubs.Get(PendingTxsSubID(id)); ok { + sub.SetProtocol(protocol) + sub.EnableTimeout() + ff.logger.Debug("[rpc] [filters] registered pending txs subscription with timeout tracking", "id", id, "protocol", protocol) + ff.incrementMetrics(ft, protocol) + } + case FilterTypeLogs: + if filter, ok := ff.logsSubs.logsFilters.Get(LogsSubID(id)); ok && filter.sender != nil { + filter.sender.SetProtocol(protocol) + filter.sender.EnableTimeout() + ff.logger.Debug("[rpc] [filters] registered logs subscription with timeout tracking", "id", id, "protocol", protocol) + ff.incrementMetrics(ft, protocol) + } + } +} + +// SetSubscriptionProtocol sets the protocol for metrics tracking without enabling timeout. +// Use this for WebSocket subscriptions that don't need eviction but need metrics. +func (ff *Filters) SetSubscriptionProtocol(id SubscriptionID, ft FilterType, protocol string) { + switch ft { + case FilterTypeHeads: + if sub, ok := ff.headsSubs.Get(HeadsSubID(id)); ok { + sub.SetProtocol(protocol) + ff.logger.Debug("[rpc] [filters] registered heads subscription (no timeout)", "id", id, "protocol", protocol) + ff.incrementMetrics(ft, protocol) + } + case FilterTypePendingTxs: + if sub, ok := ff.pendingTxsSubs.Get(PendingTxsSubID(id)); ok { + sub.SetProtocol(protocol) + ff.logger.Debug("[rpc] [filters] registered pending txs subscription (no timeout)", "id", id, "protocol", protocol) + ff.incrementMetrics(ft, protocol) + } + case FilterTypeLogs: + if filter, ok := ff.logsSubs.logsFilters.Get(LogsSubID(id)); ok && filter.sender != nil { + filter.sender.SetProtocol(protocol) + ff.logger.Debug("[rpc] [filters] registered logs subscription (no timeout)", "id", id, "protocol", protocol) + ff.incrementMetrics(ft, 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 { @@ -412,11 +635,28 @@ func (ff *Filters) SubscribeNewHeads(size int) (<-chan *types.Header, HeadsSubID // 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() + 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 +} + +// unsubscribeHeadsInternal performs the actual unsubscription without updating metrics. +// Used by eviction to avoid double-counting. +func (ff *Filters) unsubscribeHeadsInternal(id HeadsSubID) bool { + sub, ok := ff.headsSubs.Get(id) + if !ok { + return false + } + sub.Close() if _, ok = ff.headsSubs.Delete(id); !ok { return false } @@ -478,11 +718,28 @@ func (ff *Filters) SubscribePendingTxs(size int) (<-chan []types.Transaction, Pe // 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() + protocol := sub.Protocol() + if !ff.unsubscribePendingTxsInternal(id) { + return false + } + ff.logger.Debug("[rpc] [filters] unsubscribed pending txs filter", "id", id, "protocol", protocol) + ff.decrementMetrics(FilterTypePendingTxs, protocol) + return true +} + +// unsubscribePendingTxsInternal performs the actual unsubscription without updating metrics. +// Used by eviction to avoid double-counting. +func (ff *Filters) unsubscribePendingTxsInternal(id PendingTxsSubID) bool { + sub, ok := ff.pendingTxsSubs.Get(id) + if !ok { + return false + } + sub.Close() if _, ok = ff.pendingTxsSubs.Delete(id); !ok { return false } @@ -635,7 +892,30 @@ func (ff *Filters) HasPendingTxsSubscription(id PendingTxsSubID) bool { // 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 { + // Get protocol before unsubscribing + var protocol string + filter, ok := ff.logsSubs.logsFilters.Get(id) + if ok && filter.sender != nil { + protocol = filter.sender.Protocol() + } + if !ok { + ff.logger.Debug("[rpc] [filters] unsubscribe logs filter not found", "id", id) + } + if !ff.unsubscribeLogsInternal(id) { + return false + } + ff.logger.Debug("[rpc] [filters] unsubscribed logs filter", "id", id, "protocol", protocol) + ff.decrementMetrics(FilterTypeLogs, protocol) + return true +} + +// unsubscribeLogsInternal performs the actual unsubscription without updating metrics. +// Used by eviction to avoid double-counting. +func (ff *Filters) unsubscribeLogsInternal(id LogsSubID) bool { isDeleted := ff.logsSubs.removeLogsFilter(id) + if !isDeleted { + return false + } // if any filters in the aggregate need all addresses or all topics then the request to the central // log subscription needs to honour this lfr := ff.logsSubs.createFilterRequest() @@ -648,17 +928,17 @@ func (ff *Filters) UnsubscribeLogs(id LogsSubID) bool { for topic := range topics { lfr.Topics = append(lfr.Topics, gointerfaces.ConvertHashToH256(topic)) } - loaded := ff.loadLogsRequester() - if loaded != nil { + // The local filter has been removed; the log store and metric must be released + // regardless of whether the remote-filter update succeeds. A failed remote update + // only means the upstream will keep sending logs we'll now discard — it does not + // reanimate the local subscription. + 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 + return true } // deleteLogStore deletes the log store associated with the given subscription ID. @@ -923,3 +1203,22 @@ func (ff *Filters) WithTemporalOverlay(tx kv.TemporalTx) kv.TemporalTx { } return tx } + +func (ff *Filters) incrementMetrics(ft FilterType, protocol string) { + activeSubscriptionsGauge.With(prometheus.Labels{ + filterLabelName: string(ft), + protocolLabelName: protocol, + }).Inc() + getSubscriptionCounter("created", string(ft), protocol).Inc() +} + +func (ff *Filters) decrementMetrics(ft FilterType, protocol string) { + if protocol == "" { + return // Not tracked + } + activeSubscriptionsGauge.With(prometheus.Labels{ + filterLabelName: string(ft), + protocolLabelName: protocol, + }).Dec() + getSubscriptionCounter("unsubscribed", string(ft), protocol).Inc() +} diff --git a/rpc/rpchelper/metrics.go b/rpc/rpchelper/metrics.go index a91dd46609a..523821a2ae4 100644 --- a/rpc/rpchelper/metrics.go +++ b/rpc/rpchelper/metrics.go @@ -21,15 +21,43 @@ import ( ) const ( - filterLabelName = "filter" - clientLabelName = "client" + filterLabelName = "filter" + clientLabelName = "client" + protocolLabelName = "protocol" +) + +// Protocol types for metrics labeling +const ( + ProtocolHTTP = "http" + ProtocolWS = "ws" ) var ( - activeSubscriptionsGauge = metrics.GetOrCreateGaugeVec("subscriptions", []string{filterLabelName}, "Current number of subscriptions") + // activeSubscriptionsGauge tracks current active subscriptions by filter type and protocol + activeSubscriptionsGauge = metrics.GetOrCreateGaugeVec( + "subscriptions_active", + []string{filterLabelName, protocolLabelName}, + "Current number of active subscriptions", + ) + activeSubscriptionsLogsAllAddressesGauge = metrics.GetOrCreateGauge("subscriptions_logs_all_addresses") activeSubscriptionsLogsAllTopicsGauge = metrics.GetOrCreateGauge("subscriptions_logs_all_topics") activeSubscriptionsLogsAddressesGauge = metrics.GetOrCreateGauge("subscriptions_logs_addresses") activeSubscriptionsLogsTopicsGauge = metrics.GetOrCreateGauge("subscriptions_logs_topics") activeSubscriptionsLogsClientGauge = metrics.GetOrCreateGaugeVec("subscriptions_logs_client", []string{clientLabelName}, "Current number of subscriptions by client") ) + +// getSubscriptionCounter returns a counter for subscription lifecycle events. +// pattern: subscriptions_{event}_total{{filter="{filterType}",protocol="{protocol}"}} +func getSubscriptionCounter(event, filterType, protocol string) metrics.Counter { + return metrics.GetOrCreateCounter( + `subscriptions_` + event + `_total{filter="` + filterType + `",protocol="` + protocol + `"}`, + ) +} + +// getReapedCounter returns a counter for reaped (timeout-evicted) subscriptions. +func getReapedCounter(filterType string) metrics.Counter { + return metrics.GetOrCreateCounter( + `subscriptions_reaped_total{filter="` + filterType + `"}`, + ) +} diff --git a/rpc/rpchelper/subscription.go b/rpc/rpchelper/subscription.go index 25ec40722d1..c2a4f1f5f4b 100644 --- a/rpc/rpchelper/subscription.go +++ b/rpc/rpchelper/subscription.go @@ -18,18 +18,33 @@ package rpchelper import ( "sync" + "sync/atomic" + "time" ) // a simple interface for subscriptions for rpc helper type Sub[T any] interface { Send(T) Close() + // Tracking methods for timeout-based eviction and metrics + Touch() // Reset last access time (for HTTP polling filters) + SetProtocol(protocol string) // Set protocol label for metrics + EnableTimeout() // Enable timeout tracking (HTTP only) + ShouldEvict(timeout time.Duration) bool // Check if subscription should be evicted + Protocol() string // Get protocol label for metrics } type chan_sub[T any] struct { - lock sync.Mutex // protects all fileds of this struct + lock sync.Mutex // protects ch and closed fields ch chan T closed bool + + // protocol is published once via SetProtocol and read concurrently by the eviction loop. + // atomic.Pointer keeps the write/read ordered without a lock. + protocol atomic.Pointer[string] + // lastAccess tracks last poll time for timeout eviction (atomic for concurrent Touch/ShouldEvict) + // lastAccess == 0 means no timeout tracking (WebSocket subscriptions) + lastAccess atomic.Int64 } // newChanSub - buffered channel @@ -37,9 +52,9 @@ func newChanSub[T any](size int) *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 + return &chan_sub[T]{ + ch: make(chan T, size), + } } func (s *chan_sub[T]) Send(x T) { s.lock.Lock() @@ -52,6 +67,12 @@ func (s *chan_sub[T]) Send(x T) { default: // the sub is overloaded, dispose message } } + +// Close is safe to call concurrently and idempotent — the eviction loop and a +// client-initiated unsubscribe can both reach Close for the same subscription. +// The s.closed guard prevents the double channel close panic; the +// unsubscribe*Internal funcs guard the matching metrics decrement via the +// SyncMap.Delete result so the gauge is decremented at most once. func (s *chan_sub[T]) Close() { s.lock.Lock() defer s.lock.Unlock() @@ -61,3 +82,41 @@ func (s *chan_sub[T]) Close() { s.closed = true close(s.ch) } + +// SetProtocol sets the protocol label for metrics tracking. +// Safe to call concurrently with Protocol(); intended to be called once at subscription setup. +func (s *chan_sub[T]) SetProtocol(protocol string) { + s.protocol.Store(&protocol) +} + +// EnableTimeout enables timeout tracking for this subscription (HTTP polling filters). +// For WebSocket subscriptions, don't call this - they will never be evicted. +func (s *chan_sub[T]) EnableTimeout() { + s.lastAccess.Store(time.Now().UnixNano()) +} + +// Touch resets the last access time, preventing timeout eviction. +// Only effective if EnableTimeout was called. +func (s *chan_sub[T]) Touch() { + if s.lastAccess.Load() != 0 { + s.lastAccess.Store(time.Now().UnixNano()) + } +} + +// ShouldEvict returns true if the subscription has not been accessed within the timeout. +// Returns false if tracking is not enabled (lastAccess == 0). +func (s *chan_sub[T]) ShouldEvict(timeout time.Duration) bool { + last := s.lastAccess.Load() + if last == 0 { + return false // No tracking, never evict (WebSocket) + } + return time.Since(time.Unix(0, last)) > timeout +} + +// Protocol returns the protocol label for metrics. Returns "" if SetProtocol was never called. +func (s *chan_sub[T]) Protocol() string { + if p := s.protocol.Load(); p != nil { + return *p + } + return "" +}