fix deadlock in filters.go (#5734)

When the filter is in the onNewTxs function, the subscription function
exits and no longer receives information from the msg channel. In this
case, the unsubscribe_xxx function is triggered, which will cause
filter.mu read and write locks to enter a deadlock state

Co-authored-by: dc <dctrlbox@gmail.com>
Co-authored-by: Alexey Sharp <alexeysharp@Alexeys-iMac.local>

turbo/rpchelper/filters.go

@@ -312,17 +312,34 @@ func (ff *Filters) SubscribeNewHeads(out chan *types.Header) HeadsSubID {
 }
 
 func (ff *Filters) UnsubscribeHeads(id HeadsSubID) bool {
-	ff.mu.Lock()
-	defer ff.mu.Unlock()
-	if ch, ok := ff.headsSubs[id]; ok {
-		close(ch)
-		delete(ff.headsSubs, id)
-		ff.storeMu.Lock()
-		defer ff.storeMu.Unlock()
-		delete(ff.pendingHeadsStores, id)
-		return true
+	ff.mu.RLock()
+	ch, ok := ff.headsSubs[id]
+	ff.mu.RUnlock()
+	if !ok {
+		return false
+	}
+	// Drain the channel to avoid the deadlock in the OnNewEvent function
+	// Draining of the channel is safe without a lock because it does not panic
+	// when the channel is closed
+	for {
+		select {
+		case <-ch:
+		default:
+			ff.mu.Lock()
+			defer ff.mu.Unlock()
+			// Need to re-check the channel because it might have been closed and removed from the map
+			// which we were not holding the lock
+			if ch, ok = ff.headsSubs[id]; !ok {
+				return false
+			}
+			close(ch)
+			delete(ff.headsSubs, id)
+			ff.storeMu.Lock()
+			delete(ff.pendingHeadsStores, id)
+			ff.storeMu.Unlock()
+			return true
+		}
 	}
-	return false
 }
 
 func (ff *Filters) SubscribePendingLogs(c chan types.Logs) PendingLogsSubID {
@@ -362,17 +379,34 @@ func (ff *Filters) SubscribePendingTxs(out chan []types.Transaction) PendingTxsS
 }
 
 func (ff *Filters) UnsubscribePendingTxs(id PendingTxsSubID) bool {
-	ff.mu.Lock()
-	defer ff.mu.Unlock()
-	if ch, ok := ff.pendingTxsSubs[id]; ok {
-		close(ch)
-		delete(ff.pendingTxsSubs, id)
-		ff.storeMu.Lock()
-		defer ff.storeMu.Unlock()
-		delete(ff.pendingTxsStores, id)
-		return true
+	ff.mu.RLock()
+	ch, ok := ff.pendingTxsSubs[id]
+	ff.mu.RUnlock()
+	if !ok {
+		return false
+	}
+	// Drain the channel to avoid the deadlock in the OnNewTxs function
+	// Draining of the channel is safe without a lock because it does not panic
+	// when the channel is closed
+	for {
+		select {
+		case <-ch:
+		default:
+			ff.mu.Lock()
+			defer ff.mu.Unlock()
+			// Need to re-check the channel because it might have been closed and removed from the map
+			// which we were not holding the lock
+			if ch, ok = ff.pendingTxsSubs[id]; !ok {
+				return false
+			}
+			close(ch)
+			delete(ff.pendingTxsSubs, id)
+			ff.storeMu.Lock()
+			delete(ff.pendingTxsStores, id)
+			ff.storeMu.Unlock()
+			return true
+		}
 	}
-	return false
 }
 
 func (ff *Filters) SubscribeLogs(out chan *types.Log, crit filters.FilterCriteria) LogsSubID {