Function TestQueueFilledDeadlock

storage/remote/queue_manager_test.go:654–695

TestQueueFilledDeadlock makes sure the code does not deadlock in the case where a large scrape (> capacity + max samples per send) is appended at the same time as a batch times out according to the batch send deadline.

(t *testing.T)

// where a large scrape (> capacity + max samples per send) is appended at the
// same time as a batch times out according to the batch send deadline.
func TestQueueFilledDeadlock(t *testing.T) {
	for _, protoMsg := range []remoteapi.WriteMessageType{remoteapi.WriteV1MessageType, remoteapi.WriteV2MessageType} {
		t.Run(fmt.Sprint(protoMsg), func(t *testing.T) {
			recs := testwal.GenerateRecords(recCase{
				NoST:   protoMsg == remoteapi.WriteV1MessageType, // RW1 does not support ST.
				Series: 50, SamplesPerSeries: 1,
			})

			c := NewNopWriteClient()

			cfg := testDefaultQueueConfig()
			mcfg := config.DefaultMetadataConfig
			cfg.MaxShards = 1
			cfg.MaxSamplesPerSend = 10
			cfg.Capacity = 20
			flushDeadline := time.Second
			batchSendDeadline := time.Millisecond
			cfg.BatchSendDeadline = model.Duration(batchSendDeadline)

			m := newTestQueueManager(t, cfg, mcfg, flushDeadline, c, protoMsg)
			m.StoreSeries(recs.Series, 0)
			m.Start()
			defer m.Stop()

			for range 100 {
				done := make(chan struct{})
				go func() {
					time.Sleep(batchSendDeadline)
					m.Append(recs.Samples)
					done <- struct{}{}
				}()
				select {
				case <-done:
				case <-time.After(2 * time.Second):
					t.Error("Deadlock between sending and appending detected")
					pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
					t.FailNow()
				}
			}
		})
	}
}

func TestReleaseNoninternedString(t *testing.T) {
	for _, protoMsg := range []remoteapi.WriteMessageType{remoteapi.WriteV1MessageType, remoteapi.WriteV2MessageType} {