eth/filters: fix flaky test TestPendingTxFilterDeadlock (#28376)

eth/filters/filter_system_test.go

@@ -915,10 +915,14 @@ func TestPendingTxFilterDeadlock(t *testing.T) {
 
 	// Create a bunch of filters that will
 	// timeout either in 100ms or 200ms
-	fids := make([]rpc.ID, 20)
-	for i := 0; i < len(fids); i++ {
+	subs := make([]*Subscription, 20)
+	for i := 0; i < len(subs); i++ {
 		fid := api.NewPendingTransactionFilter(nil)
-		fids[i] = fid
+		f, ok := api.filters[fid]
+		if !ok {
+			t.Fatalf("Filter %s should exist", fid)
+		}
+		subs[i] = f.s
 		// Wait for at least one tx to arrive in filter
 		for {
 			hashes, err := api.GetFilterChanges(fid)
@@ -932,21 +936,13 @@ func TestPendingTxFilterDeadlock(t *testing.T) {
 		}
 	}
 
-	// Wait until filters have timed out
-	time.Sleep(3 * timeout)
-
-	// If tx loop doesn't consume `done` after a second
-	// it's hanging.
-	select {
-	case done <- struct{}{}:
-		// Check that all filters have been uninstalled
-		for _, fid := range fids {
-			if _, err := api.GetFilterChanges(fid); err == nil {
-				t.Errorf("Filter %s should have been uninstalled\n", fid)
-			}
+	// Wait until filters have timed out and have been uninstalled.
+	for _, sub := range subs {
+		select {
+		case <-sub.Err():
+		case <-time.After(1 * time.Second):
+			t.Fatalf("Filter timeout is hanging")
 		}
-	case <-time.After(1 * time.Second):
-		t.Error("Tx sending loop hangs")
 	}
 }