rpc: validate log query limit when creating filters

ChangeLog.md

@@ -2,6 +2,7 @@
 
 ### Breaking Changes
 
+- rpc: `eth_newFilter` now rejects criteria exceeding `rpc.logs.querylimit` with `-32602`. Such filters can no longer be created for live-only `eth_getFilterChanges` use; use `eth_subscribe("logs")` or set the limit to `0` instead (#23598) — by @yperbasis
 - rpc: `eth_getFilterLogs` now runs a historical query from the criteria stored by `eth_newFilter`; it neither drains the live-log queue read by `eth_getFilterChanges` nor refreshes the filter expiry deadline. Historical queries enforce `rpc.blockrange.limit`, `rpc.logs.maxresults`, and `rpc.logs.querylimit`; a filter can return `-32602` when one of these limits is exceeded, and queries may surface initialization, pruned-history, or not-yet-executed errors. This can be a breaking change (#23296) — by @taratorio
 - rpc: `eth_newFilter` and `eth_subscribe("logs")` now reject criteria exceeding the per-filter `rpc.subscription.filters.maxaddresses` or `rpc.subscription.filters.maxtopics` limit with `-32602` instead of silently capping them. This can be a breaking change for nodes that configure either limit above `0`; both limits default to `0` (unlimited) (#23296) — by @taratorio
 - rpc: `eth_newFilter` and `eth_subscribe("logs")` now reject criteria with more than four topic positions with `-32602` instead of accepting a filter that cannot match any Ethereum log. This can be a breaking change (#23296) — by @taratorio

cmd/utils/flags.go

@@ -482,7 +482,7 @@ var (
 	}
 	RpcLogQueryLimit = cli.IntFlag{
 		Name:  "rpc.logs.querylimit",
-		Usage: "Maximum number of alternative addresses or topics allowed per search position in eth_getLogs and eth_getFilterLogs filter criteria (<=0 = unlimited)",
+		Usage: "Maximum number of alternative addresses or topics allowed per search position in eth_getLogs, eth_newFilter, and eth_getFilterLogs filter criteria (<=0 = unlimited)",
 		Value: 1_000,
 	}
 	RpcTraceCompatFlag = cli.BoolFlag{

docs/site/docs/fundamentals/configuring-erigon.mdx

@@ -299,9 +299,9 @@ Flags for configuring various RPC servers and their behavior. See [Interacting w
   * Applies to: `eth_getLogs`, `eth_getFilterLogs`, `erigon_getLogs`, `erigon_getLatestLogs`
   * Set to `0` to remove the limit entirely (use with caution on large ranges).
   * Works in tandem with `--rpc.blockrange.limit`: both constraints apply independently — a query can be blocked by either limit.
-* `--rpc.logs.querylimit value`: Maximum number of alternative addresses or topics allowed per search position in the `eth_getLogs` and `eth_getFilterLogs` filter criteria.
+* `--rpc.logs.querylimit value`: Maximum number of alternative addresses or topics allowed per search position in the `eth_getLogs`, `eth_newFilter`, and `eth_getFilterLogs` filter criteria.
   * Default: `1000`
-  * Applies to: `eth_getLogs`, `eth_getFilterLogs`
+  * Applies to: `eth_getLogs`, `eth_newFilter`, `eth_getFilterLogs`
   * Set to `0` or a negative value for no limit.
 * `--rpc.returndata.limit value`: Sets the maximum return data size for `eth_call`.
   * Default: `100000`

docs/site/static/llms-full.txt

@@ -2328,9 +2328,9 @@ Flags for configuring various RPC servers and their behavior. See [Interacting w
   * Applies to: `eth_getLogs`, `eth_getFilterLogs`, `erigon_getLogs`, `erigon_getLatestLogs`
   * Set to `0` to remove the limit entirely (use with caution on large ranges).
   * Works in tandem with `--rpc.blockrange.limit`: both constraints apply independently — a query can be blocked by either limit.
-* `--rpc.logs.querylimit value`: Maximum number of alternative addresses or topics allowed per search position in the `eth_getLogs` and `eth_getFilterLogs` filter criteria.
+* `--rpc.logs.querylimit value`: Maximum number of alternative addresses or topics allowed per search position in the `eth_getLogs`, `eth_newFilter`, and `eth_getFilterLogs` filter criteria.
   * Default: `1000`
-  * Applies to: `eth_getLogs`, `eth_getFilterLogs`
+  * Applies to: `eth_getLogs`, `eth_newFilter`, `eth_getFilterLogs`
   * Set to `0` or a negative value for no limit.
 * `--rpc.returndata.limit value`: Sets the maximum return data size for `eth_call`.
   * Default: `100000`

llms-full.txt

@@ -2328,9 +2328,9 @@ Flags for configuring various RPC servers and their behavior. See [Interacting w
   * Applies to: `eth_getLogs`, `eth_getFilterLogs`, `erigon_getLogs`, `erigon_getLatestLogs`
   * Set to `0` to remove the limit entirely (use with caution on large ranges).
   * Works in tandem with `--rpc.blockrange.limit`: both constraints apply independently — a query can be blocked by either limit.
-* `--rpc.logs.querylimit value`: Maximum number of alternative addresses or topics allowed per search position in the `eth_getLogs` and `eth_getFilterLogs` filter criteria.
+* `--rpc.logs.querylimit value`: Maximum number of alternative addresses or topics allowed per search position in the `eth_getLogs`, `eth_newFilter`, and `eth_getFilterLogs` filter criteria.
   * Default: `1000`
-  * Applies to: `eth_getLogs`, `eth_getFilterLogs`
+  * Applies to: `eth_getLogs`, `eth_newFilter`, `eth_getFilterLogs`
   * Set to `0` or a negative value for no limit.
 * `--rpc.returndata.limit value`: Sets the maximum return data size for `eth_call`.
   * Default: `100000`

rpc/jsonrpc/eth_filters.go

@@ -63,6 +63,12 @@ func (api *APIImpl) NewFilter(_ context.Context, crit filters.FilterCriteria) (s
 	if api.filters == nil {
 		return "", rpc.ErrNotificationsUnsupported
 	}
+	if err := crit.ValidateTopicPositions(); err != nil {
+		return "", err
+	}
+	if err := validateLogQueryLimit(crit, api.logQueryLimit); err != nil {
+		return "", err
+	}
 	logs, id, err := api.filters.SubscribeLogs(256, crit, rpchelper.ProtocolHTTP)
 	if err != nil {
 		return "", err

rpc/jsonrpc/eth_filters_test.go

@@ -97,6 +97,16 @@ func TestLogFilterEndpointsRejectTooManyTopicPositions(t *testing.T) {
 	}
 }
 
+func TestGetLogsAppliesLogQueryLimitBeforeOpeningTransaction(t *testing.T) {
+	api := &APIImpl{BaseAPI: &BaseAPI{logQueryLimit: 1}}
+	_, err := api.GetLogs(t.Context(), filters.FilterCriteria{Addresses: make(common.Addresses, 2)})
+
+	var rpcErr rpc.Error
+	require.ErrorAs(t, err, &rpcErr)
+	require.Equal(t, rpc.ErrCodeInvalidParams, rpcErr.ErrorCode())
+	require.EqualError(t, err, "query exceeds the maximum of 1 addresses or topics per search position")
+}
+
 func TestSubscriptionsRequireFiltersAndNotifier(t *testing.T) {
 	m := execmoduletester.New(t)
 	ctx, conn := rpcdaemontest.CreateTestGrpcConn(t, m)
@@ -249,62 +259,38 @@ func TestGetFilterLogsReturnsInvalidParamsWhenStoredRangeExceedsLimit(t *testing
 	require.Equal(t, errExceedBlockRange+": 1", rpcErr.Error())
 }
 
-func TestGetFilterLogsAppliesLogQueryLimitAtPollTime(t *testing.T) {
-	if testing.Short() {
-		t.Skip("slow test")
-	}
-	m, _, _ := rpcdaemontest.CreateTestExecModule(t)
-	ctx, conn := rpcdaemontest.CreateTestGrpcConn(t, m)
-	mining := txpoolproto.NewMiningClient(conn)
-	stateCache := kvcache.New(kvcache.DefaultCoherentConfig)
+func TestNewFilterAppliesLogQueryLimitAtCreation(t *testing.T) {
+	filterManager := rpchelper.New(t.Context(), rpchelper.DefaultFiltersConfig, nil, nil, nil, func() {}, log.New(), nil)
+	api := &APIImpl{BaseAPI: &BaseAPI{filters: filterManager, logQueryLimit: 1}}
 
 	tests := []struct {
-		name       string
-		criteria   filters.FilterCriteria
-		filterConf rpchelper.FiltersConfig
+		name     string
+		criteria filters.FilterCriteria
 	}{
 		{
 			name: "addresses",
 			criteria: filters.FilterCriteria{
 				Addresses: common.Addresses{{1}, {2}},
 			},
-			filterConf: rpchelper.FiltersConfig{
-				RpcSubscriptionFiltersMaxAddresses: 2,
-			},
 		},
 		{
 			name: "topic alternatives",
 			criteria: filters.FilterCriteria{
 				Topics: [][]common.Hash{{{1}, {2}}},
 			},
-			filterConf: rpchelper.FiltersConfig{
-				RpcSubscriptionFiltersMaxTopics: 2,
-			},
 		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			test.filterConf.RpcSubscriptionFiltersTimeout = rpchelper.DefaultFilterTimeout
-			ff := rpchelper.New(ctx, test.filterConf, nil, nil, mining, func() {}, m.Log, nil)
-			base := NewBaseApi(ff, stateCache, m.BlockReader, m.Engine, &rpccfg.BaseApiConfig{
-				Dirs:          m.Dirs,
-				LogQueryLimit: 1,
-			})
-			api := newEthApiForTest(base, m.DB, nil, nil)
-			test.criteria.FromBlock = big.NewInt(10)
-			test.criteria.ToBlock = big.NewInt(10)
-
-			filterID, err := api.NewFilter(ctx, test.criteria)
-			require.NoError(t, err)
-			t.Cleanup(func() {
-				_, _ = api.UninstallFilter(ctx, filterID)
-			})
-
-			_, err = api.GetLogs(ctx, test.criteria)
-			require.ErrorContains(t, err, "query exceeds the maximum of 1 addresses or topics per search position")
-
-			_, err = api.GetFilterLogs(ctx, filterID)
+			filterID, err := api.NewFilter(t.Context(), test.criteria)
+			if err == nil {
+				_, _ = api.UninstallFilter(t.Context(), filterID)
+			}
 			require.ErrorContains(t, err, "query exceeds the maximum of 1 addresses or topics per search position")
+			require.Empty(t, filterID)
+			var rpcErr rpc.Error
+			require.ErrorAs(t, err, &rpcErr)
+			require.Equal(t, rpc.ErrCodeInvalidParams, rpcErr.ErrorCode())
 		})
 	}
 }
@@ -343,22 +329,19 @@ func TestGetFilterLogsDoesNotKeepFilterAlive(t *testing.T) {
 }
 
 func TestLogsSubscribeAndUnsubscribe_WithoutConcurrentMapIssue(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)
+	ff := rpchelper.New(t.Context(), rpchelper.DefaultFiltersConfig, nil, nil, nil, func() {}, log.New(), nil)
 
 	// generate some random topics
-	topics := make([][]common.Hash, 0)
-	for range 10 {
+	topics := make([][]common.Hash, 0, filters.MaxTopicPositions)
+	for range filters.MaxTopicPositions {
 		bytes := make([]byte, length.Hash)
 		rand.Read(bytes)
 		toAdd := []common.Hash{common.BytesToHash(bytes)}
 		topics = append(topics, toAdd)
 	}
 
 	// generate some addresses
-	addresses := make([]common.Address, 0)
+	addresses := make([]common.Address, 0, 10)
 	for range 10 {
 		bytes := make([]byte, length.Addr)
 		rand.Read(bytes)
@@ -371,21 +354,30 @@ func TestLogsSubscribeAndUnsubscribe_WithoutConcurrentMapIssue(t *testing.T) {
 	}
 
 	ids := make([]rpchelper.LogsSubID, 1000)
+	errs := make([]error, len(ids))
+	unsubscribed := make([]bool, len(ids))
 
 	// make a lot of subscriptions
 	wg := sync.WaitGroup{}
 	for i := range 1000 {
 		idx := i
 		wg.Go(func() {
-			_, id, _ := ff.SubscribeLogs(32, crit, "")
-			defer func() {
-				time.Sleep(100 * time.Nanosecond)
-				ff.UnsubscribeLogs(id)
-			}()
+			_, id, err := ff.SubscribeLogs(32, crit, rpchelper.ProtocolWS)
 			ids[idx] = id
+			errs[idx] = err
+			if err != nil {
+				return
+			}
+			time.Sleep(100 * time.Nanosecond)
+			unsubscribed[idx] = ff.UnsubscribeLogs(id)
 		})
 	}
 	wg.Wait()
+	for i := range ids {
+		require.NoError(t, errs[i])
+		require.NotEmpty(t, ids[i])
+		require.True(t, unsubscribed[i])
+	}
 }
 
 func TestBlockFilterGetFilterChangesInitiallyEmpty(t *testing.T) {

rpc/jsonrpc/eth_receipts.go

@@ -83,21 +83,19 @@ func (api *BaseAPI) getCachedReceipts(ctx context.Context, hash common.Hash) (ty
 	return api.receiptsGenerator.GetCachedReceipts(ctx, hash)
 }
 
-// exceedsLogQueryLimit reports whether the filter has more addresses, or more
-// alternatives in one topic position, than limit allows (<=0 = unlimited).
-func exceedsLogQueryLimit(crit filters.FilterCriteria, limit int) bool {
+func validateLogQueryLimit(crit filters.FilterCriteria, limit int) error {
 	if limit <= 0 {
-		return false
+		return nil
 	}
 	if len(crit.Addresses) > limit {
-		return true
+		return &rpc.CustomError{Message: fmt.Sprintf(errExceedLogQueryLimit, limit), Code: rpc.ErrCodeInvalidParams}
 	}
 	for _, topics := range crit.Topics {
 		if len(topics) > limit {
-			return true
+			return &rpc.CustomError{Message: fmt.Sprintf(errExceedLogQueryLimit, limit), Code: rpc.ErrCodeInvalidParams}
 		}
 	}
-	return false
+	return nil
 }
 
 // resolveLogsRange resolves a filter's block range. A BlockHash pins the range to that
@@ -181,6 +179,9 @@ func (api *APIImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (t
 	if err := crit.ValidateTopicPositions(); err != nil {
 		return nil, err
 	}
+	if err := validateLogQueryLimit(crit, api.logQueryLimit); err != nil {
+		return nil, err
+	}
 
 	logs := types.RPCLogs{}
 
@@ -190,10 +191,6 @@ func (api *APIImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (t
 	}
 	defer tx.Rollback()
 
-	if exceedsLogQueryLimit(crit, api.logQueryLimit) {
-		return nil, &rpc.CustomError{Message: fmt.Sprintf(errExceedLogQueryLimit, api.logQueryLimit), Code: rpc.ErrCodeInvalidParams}
-	}
-
 	if crit.BlockHash != nil && (crit.FromBlock != nil || crit.ToBlock != nil) {
 		return nil, &rpc.CustomError{Message: errBlockHashWithRange, Code: rpc.ErrCodeInvalidParams}
 	}