Skip to content

[rpc] Add Tracking to Subscriptions Made via HTTP for Unused Filter Eviction #18591

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
onelapahead wants to merge 1 commit into from
Closed

[rpc] Add Tracking to Subscriptions Made via HTTP for Unused Filter Eviction #18591

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cmd/rpcdaemon/cli/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
1 change: 1 addition & 0 deletions node/cli/default_flags.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,7 @@ var DefaultFlags = []cli.Flag{
&RpcSubscriptionFiltersMaxTxsFlag,
&RpcSubscriptionFiltersMaxAddressesFlag,
&RpcSubscriptionFiltersMaxTopicsFlag,
&RpcSubscriptionFiltersTimeoutFlag,

&utils.SnapKeepBlocksFlag,
&utils.SnapStopFlag,
Expand Down
6 changes: 6 additions & 0 deletions node/cli/flags.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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),
Expand Down
52 changes: 31 additions & 21 deletions rpc/jsonrpc/eth_filters.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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.
Expand All @@ -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
}
Expand All @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
17 changes: 12 additions & 5 deletions rpc/rpchelper/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -37,4 +43,5 @@ var DefaultFiltersConfig = FiltersConfig{
RpcSubscriptionFiltersMaxTxs: 10000,
RpcSubscriptionFiltersMaxAddresses: 0, // no limit
RpcSubscriptionFiltersMaxTopics: 0, // no limit
RpcSubscriptionFiltersTimeout: DefaultFilterTimeout,
}
Loading
Loading