Fix data race on chunks shared by DedupQueue

chunk.go

@@ -23,6 +23,16 @@ func NewChunk(b []byte) *Chunk {
 	return &Chunk{data: b}
 }
 
+// clone returns a shallow copy of the chunk. The copy shares the underlying
+// data and storage slices, which are never modified after construction, but
+// gets its own lazily-calculated fields (plain data, ID). Data() and ID()
+// mutate the chunk without synchronization, so a chunk handed to multiple
+// goroutines must be cloned for each of them.
+func (c *Chunk) clone() *Chunk {
+	cp := *c
+	return &cp
+}
+
 // NewChunkWithID creates a new chunk from either compressed or uncompressed data
 // (or both if available). It also expects an ID and validates that it matches
 // the uncompressed data unless skipVerify is true. If called with just compressed

dedupqueue.go

@@ -35,7 +35,11 @@ func (q *DedupQueue) GetChunk(id ChunkID) (*Chunk, error) {
 		case nil:
 			return nil, err
 		case *Chunk:
-			return b, err
+			// The chunk in the request is shared with all waiters. Chunk
+			// materializes its plain data and ID lazily without locking,
+			// so hand every caller its own copy. The copies still share
+			// the underlying (read-only) chunk data.
+			return b.clone(), err
 		default:
 			return nil, fmt.Errorf("internal error: unexpected type %T", data)
 		}
@@ -52,7 +56,12 @@ func (q *DedupQueue) GetChunk(id ChunkID) (*Chunk, error) {
 	// in memory after the request is done
 	q.getChunkQueue.delete(id)
 
-	return b, err
+	if b == nil {
+		return nil, err
+	}
+	// Return a copy for the same reason the waiters get one: the chunk stored
+	// in the request must stay untouched while waiters are still reading it.
+	return b.clone(), err
 }
 
 func (q *DedupQueue) HasChunk(id ChunkID) (bool, error) {

dedupqueue_test.go

@@ -7,6 +7,7 @@ import (
 	"testing/synctest"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -73,3 +74,48 @@ func TestDedupQueueParallel(t *testing.T) {
 		require.EqualValues(t, 1, requests.Load(), "requests to the store")
 	})
 }
+
+func TestDedupQueueChunkNotShared(t *testing.T) {
+	data := []byte("some data")
+	id := NewChunk(data).ID()
+	compressed, err := Compress(data)
+	require.NoError(t, err)
+
+	store := &TestStore{
+		GetChunkFunc: func(id ChunkID) (*Chunk, error) {
+			// Give the other goroutines time to pile up on this request
+			time.Sleep(10 * time.Millisecond)
+			// Return the chunk in storage form, the plain data is only
+			// materialized lazily when the caller asks for it
+			return NewChunkFromStorage(id, compressed, Converters{Compressor{}}, true)
+		},
+	}
+	q := NewDedupQueue(store)
+
+	// Request the same chunk from many goroutines at once. Each caller must
+	// receive its own copy since accessing the plain data modifies the chunk.
+	// Run with -race to confirm the callers don't share state.
+	chunks := make(chan *Chunk, 10)
+	var wg sync.WaitGroup
+	for range 10 {
+		wg.Go(func() {
+			c, err := q.GetChunk(id)
+			if !assert.NoError(t, err) {
+				return
+			}
+			b, err := c.Data()
+			assert.NoError(t, err)
+			assert.Equal(t, data, b)
+			chunks <- c
+		})
+	}
+	wg.Wait()
+	close(chunks)
+
+	seen := make(map[*Chunk]struct{})
+	for c := range chunks {
+		_, shared := seen[c]
+		assert.False(t, shared, "the same *Chunk was handed to multiple callers")
+		seen[c] = struct{}{}
+	}
+}