Bug: Distributor panic with gRPC push stream

[SungJin1212]

Problem Statement

When I enabled a push stream with the Cortex v1.21.0, I noticed the Distributors get a panic.
The reason is:

1. PushStreamConnection put a job in the channel and waited on <-reqCtx.Done().
2. The worker called stream.Send(), which pre-computes Size() then calls MarshalToSizedBuffer()
3. When the ctx deadline fired, <-reqCtx.Done() unblocked and PushStreamConnection returned before Send() finished.
4. The distributor reused the TimeSeries with more labels.
5. MarshalToSizedBuffer() used the old (smaller) buffer → panic.

Related log

panic: runtime error: slice bounds out of range [-40:]

goroutine 1242 [running]:
github.com/cortexproject/cortex/pkg/cortexpb.(*TimeSeries).MarshalToSizedBuffer(0xc0c143afc0, {0xc059ba328f, 0x2e6, 0xaed71})
	/__w/cortex/cortex/pkg/cortexpb/cortex.pb.go:3091 +0x238
github.com/cortexproject/cortex/pkg/cortexpb.(*TimeSeries).MarshalTo(0xc0c143afc0, {0xc059ba328f, 0xc0bde49b00?, 0xaed71})
	/__w/cortex/cortex/pkg/cortexpb/cortex.pb.go:3036 +0x45
github.com/cortexproject/cortex/pkg/cortexpb.(*WriteRequest).MarshalToSizedBuffer(0xc0c15ba240, {0xc059b52000, 0x55276, 0x100000})
	/__w/cortex/cortex/pkg/cortexpb/cortex.pb.go:2645 +0x212
github.com/cortexproject/cortex/pkg/cortexpb.(*StreamWriteRequest).MarshalToSizedBuffer(0xc0bde49b00, {0xc059b52000, 0x55279, 0x100000})
	/__w/cortex/cortex/pkg/cortexpb/cortex.pb.go:2954 +0x7e
github.com/cortexproject/cortex/pkg/cortexpb.(*cortexCodec).Marshal(0xc000524180, {0x3a0d660, 0xc0bde49b00})
	/__w/cortex/cortex/pkg/cortexpb/codec.go:78 +0x19d
google.golang.org/grpc.encode({0x7fc150ab1198?, 0xc000524180?}, {0x3a0d660?, 0xc0bde49b00?})
	/__w/cortex/cortex/vendor/google.golang.org/grpc/rpc_util.go:691 +0x4a
google.golang.org/grpc.prepareMsg({0x3a0d660?, 0xc0bde49b00?}, {0x7fc150ab1198?, 0xc000524180?}, {0x0, 0x0}, {0x415da60, 0xc0004749b0}, {0x414e6c8, 0xc000453140})
	/__w/cortex/cortex/vendor/google.golang.org/grpc/stream.go:1814 +0xe5
google.golang.org/grpc.(*clientStream).SendMsg(0xc001034b40, {0x3a0d660, 0xc0bde49b00})
	/__w/cortex/cortex/vendor/google.golang.org/grpc/stream.go:907 +0xf1
github.com/cortexproject/cortex/pkg/util/middleware.(*instrumentedReusableClientStream).SendMsg(0xc000f45b60, {0x3a0d660, 0xc0bde49b00})
	/__w/cortex/cortex/pkg/util/middleware/grpc.go:109 +0x67
github.com/opentracing-contrib/go-grpc.(*openTracingClientStream).SendMsg(0xc000d04840, {0x3a0d660?, 0xc0bde49b00?})
	/__w/cortex/cortex/vendor/github.com/opentracing-contrib/go-grpc/client.go:203 +0x2e
github.com/cortexproject/cortex/pkg/ingester/client.(*ingesterPushStreamClient).Send(0xc0047f1f90?, 0xc0047f1f40?)
	/__w/cortex/cortex/pkg/ingester/client/ingester.pb.go:2880 +0x25
github.com/cortexproject/cortex/pkg/ingester/client.(*closableHealthAndIngesterClient).worker.func1()
	/__w/cortex/cortex/pkg/ingester/client/client.go:236 +0xfe
created by github.com/cortexproject/cortex/pkg/ingester/client.(*closableHealthAndIngesterClient).worker in goroutine 1110
	/__w/cortex/cortex/pkg/ingester/client/client.go:227 +0x117