rpc: preserve topic wildcard positions in log subscriptions (#21067)

This fixes log subscription topic filtering to preserve wildcard
positions, and adds regression tests for wildcard-only and positional
matching.

Co-authored-by: lupin012 <58134934+lupin012@users.noreply.github.com>

Author: serikssonn

rpc/rpchelper/filters.go

@@ -567,7 +567,7 @@ func (ff *Filters) SubscribeLogs(size int, criteria filters.FilterCriteria) (<-c
 	} else {
 		// Limit the number of topics
 		topicCount := 0
-		allowedTopics := [][]common.Hash{}
+		allowedTopics := make([][]common.Hash, 0, len(criteria.Topics))
 		for _, topics := range criteria.Topics {
 			allowedTopicsRow := []common.Hash{}
 			for _, topic := range topics {
@@ -579,9 +579,8 @@ func (ff *Filters) SubscribeLogs(size int, criteria filters.FilterCriteria) (<-c
 					break
 				}
 			}
-			if len(allowedTopicsRow) > 0 {
-				allowedTopics = append(allowedTopics, allowedTopicsRow)
-			}
+			// Preserve per-position wildcard slots (empty rows) for correct positional matching.
+			allowedTopics = append(allowedTopics, allowedTopicsRow)
 		}
 		f.topicsOriginal = allowedTopics
 	}

rpc/rpchelper/filters_test.go

@@ -139,6 +139,71 @@ func TestFilters_SingleSubscription_EmptyTopicsInCriteria_OnlyTopicsSubscribedAr
 	}
 }
 
+func TestFilters_SingleSubscription_TopicPositionWildcardIsPreserved(t *testing.T) {
+	t.Parallel()
+	config := FiltersConfig{}
+	f := New(t.Context(), config, nil, nil, nil, func() {}, log.New(), nil)
+
+	topic0 := common.BytesToHash([]byte{10, 20})
+	topic2 := common.BytesToHash([]byte{30, 40})
+	criteria := filters.FilterCriteria{
+		Addresses: nil,
+		Topics:    [][]common.Hash{{topic0}, nil, {topic2}},
+	}
+
+	outChan, _ := f.SubscribeLogs(10, criteria)
+
+	matchingLog := createLog()
+	matchingLog.Topics = []*typesproto.H256{
+		gointerfaces.ConvertHashToH256(topic0),
+		gointerfaces.ConvertHashToH256(common.BytesToHash([]byte{50, 60})),
+		gointerfaces.ConvertHashToH256(topic2),
+	}
+	f.OnNewLogs(matchingLog)
+
+	if len(outChan) != 1 {
+		t.Error("expected wildcard middle topic to preserve positional matching")
+	}
+
+	nonMatchingLog := createLog()
+	nonMatchingLog.Topics = []*typesproto.H256{
+		gointerfaces.ConvertHashToH256(topic0),
+		gointerfaces.ConvertHashToH256(common.BytesToHash([]byte{50, 60})),
+		gointerfaces.ConvertHashToH256(common.BytesToHash([]byte{70, 80})),
+	}
+	f.OnNewLogs(nonMatchingLog)
+
+	if len(outChan) != 1 {
+		t.Error("expected non-matching topic at constrained position to be filtered out")
+	}
+}
+
+func TestFilters_SingleSubscription_WildcardOnlyTopicRowMatches(t *testing.T) {
+	t.Parallel()
+	config := FiltersConfig{}
+	f := New(t.Context(), config, nil, nil, nil, func() {}, log.New(), nil)
+
+	criteria := filters.FilterCriteria{
+		Addresses: nil,
+		Topics:    [][]common.Hash{nil},
+	}
+
+	outChan, _ := f.SubscribeLogs(10, criteria)
+
+	log1 := createLog()
+	f.OnNewLogs(log1)
+	if len(outChan) != 1 {
+		t.Error("expected wildcard-only topic row to match arbitrary topics")
+	}
+
+	log2 := createLog()
+	log2.Topics = []*typesproto.H256{gointerfaces.ConvertHashToH256(common.BytesToHash([]byte{1, 2, 3}))}
+	f.OnNewLogs(log2)
+	if len(outChan) != 2 {
+		t.Error("expected wildcard-only topic row to continue matching subsequent logs")
+	}
+}
+
 func TestFilters_TwoSubscriptionsWithDifferentCriteria(t *testing.T) {
 	t.Parallel()
 	config := FiltersConfig{}

rpc/rpchelper/logsfilter.go

@@ -270,16 +270,6 @@ func (a *LogsFilterAggregator) distributeLog(eventLog *remoteproto.SubscribeLogs
 // chooseTopics checks if the log topics match the filter's topics.
 // It returns true if the log topics match the filter's topics, otherwise false.
 func (a *LogsFilterAggregator) chooseTopics(filter *LogsFilter, logTopics []common.Hash) bool {
-	var found bool
-	for _, logTopic := range logTopics {
-		if _, ok := filter.topics.Get(logTopic); ok {
-			found = true
-			break
-		}
-	}
-	if !found {
-		return false
-	}
 	if len(filter.topicsOriginal) > len(logTopics) {
 		return false
 	}