Query: fix streamSeriesSet defer in for loop (#4830)

* Query: fix streamSeriesSet defer in for loop
which lead too many timer in heap.

Signed-off-by: Jimmiehan <[email protected]>

* Query: Add comment to handleRecvResponse

Signed-off-by: Jimmiehan <[email protected]>

pkg/store/proxy.go

@@ -418,28 +418,30 @@ func startStreamSeriesSet(
 				}
 			}
 		}()
-		for {
+		// The `defer` only executed when function return, we do `defer cancel` in for loop,
+		// so make the loop body as a function, release timers created by context as early.
+		handleRecvResponse := func() (next bool) {
 			frameTimeoutCtx, cancel := frameCtx(s.responseTimeout)
 			defer cancel()
 			var rr *recvResponse
 			select {
 			case <-ctx.Done():
 				s.handleErr(errors.Wrapf(ctx.Err(), "failed to receive any data from %s", s.name), done)
-				return
+				return false
 			case <-frameTimeoutCtx.Done():
 				s.handleErr(errors.Wrapf(frameTimeoutCtx.Err(), "failed to receive any data in %s from %s", s.responseTimeout.String(), s.name), done)
-				return
+				return false
 			case rr = <-rCh:
 			}
 
 			if rr.err == io.EOF {
 				close(done)
-				return
+				return false
 			}
 
 			if rr.err != nil {
 				s.handleErr(errors.Wrapf(rr.err, "receive series from %s", s.name), done)
-				return
+				return false
 			}
 			numResponses++
 			bytesProcessed += rr.r.Size()
@@ -455,9 +457,15 @@ func startStreamSeriesSet(
 				case s.recvCh <- series:
 				case <-ctx.Done():
 					s.handleErr(errors.Wrapf(ctx.Err(), "failed to receive any data from %s", s.name), done)
-					return
+					return false
 				}
 			}
+			return true
+		}
+		for {
+			if !handleRecvResponse() {
+				return
+			}
 		}
 	}()
 	return s