Merge branch 'main' into feat/ulimit

Author: ntk148v

CHANGELOG.md

@@ -23,6 +23,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 ### Fixed
 
 - [#8211](https://github.com/thanos-io/thanos/pull/8211) Query: fix panic on nested partial response in distributed instant query
+- [#8216](https://github.com/thanos-io/thanos/pull/8216) Query/Receive: fix iter race between `next()` and `stop()` introduced in https://github.com/thanos-io/thanos/pull/7821.
 
 ## [v0.38.0](https://github.com/thanos-io/thanos/tree/release-0.38) - 03.04.2025
 

pkg/store/storepb/inprocess.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"iter"
 	"runtime/debug"
+	"sync"
 
 	"google.golang.org/grpc"
 )
@@ -40,17 +41,21 @@ type inProcessClient struct {
 	ctx  context.Context
 	next func() (*SeriesResponse, error, bool)
 	stop func()
+	mu   sync.Mutex // protects next and stop
 }
 
 func newInProcessClient(ctx context.Context, next func() (*SeriesResponse, error, bool), stop func()) *inProcessClient {
 	return &inProcessClient{
 		ctx:  ctx,
 		next: next,
 		stop: stop,
+		mu:   sync.Mutex{},
 	}
 }
 
 func (c *inProcessClient) Recv() (*SeriesResponse, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	resp, err, ok := c.next()
 	if err != nil {
 		c.stop()
@@ -70,6 +75,8 @@ func (c *inProcessClient) Context() context.Context {
 }
 
 func (c *inProcessClient) CloseSend() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	c.stop()
 	return nil
 }

pkg/store/storepb/inprocess_test.go

@@ -6,6 +6,7 @@ package storepb
 import (
 	"context"
 	"io"
+	"sync"
 	"testing"
 
 	"github.com/thanos-io/thanos/pkg/testutil/custom"
@@ -152,6 +153,33 @@ func TestServerAsClient(t *testing.T) {
 				s.seriesLastReq = nil
 			}
 		})
+		t.Run("race", func(t *testing.T) {
+			s.err = nil
+			for i := 0; i < 20; i++ {
+				r := &SeriesRequest{
+					MinTime:                 -214,
+					MaxTime:                 213,
+					Matchers:                []LabelMatcher{{Value: "wfsdfs", Name: "__name__", Type: LabelMatcher_EQ}},
+					PartialResponseStrategy: PartialResponseStrategy_ABORT,
+				}
+				client, err := ServerAsClient(s).Series(ctx, r)
+				testutil.Ok(t, err)
+				var wg sync.WaitGroup
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					for {
+						_, err := client.Recv()
+						if err != nil {
+							break
+						}
+					}
+				}()
+				testutil.Ok(t, client.CloseSend())
+				wg.Wait()
+				s.seriesLastReq = nil
+			}
+		})
 	})
 	t.Run("LabelNames", func(t *testing.T) {
 		s := &testStoreServer{}