Fix ChunkStream buffer-reuse race causing corrupted stores (folbricht#314)

Author: folbricht

chunker.go

@@ -202,7 +202,9 @@ func (c *Chunker) fillBuffer() (n int, err error) {
 }
 
 // Next returns the starting position as well as the chunk data. Returns
-// an empty byte slice when complete
+// an empty byte slice when complete. The returned byte slice is only valid
+// until the next call to Next; callers that pass the slice to other
+// goroutines must copy it first.
 func (c *Chunker) Next() (uint64, []byte, error) {
 	if len(c.buf) < int(c.max) {
 		n, err := c.fillBuffer()

index.go

@@ -6,6 +6,7 @@ import (
 	"crypto"
 	"fmt"
 	"math"
+	"slices"
 	"sync"
 
 	"golang.org/x/sync/errgroup"
@@ -192,6 +193,12 @@ loop:
 			break
 		}
 
+		// Copy the buffer before sending it to workers. The chunker
+		// reuses its internal backing buffer across fillBuffer() calls,
+		// so the slice returned by Next() may be overwritten by the
+		// next call.
+		b = slices.Clone(b)
+
 		// Send it off for compression and storage
 		select {
 		case <-ctx.Done():

index_test.go

@@ -3,6 +3,7 @@ package desync
 import (
 	"bytes"
 	"context"
+	"encoding/binary"
 	"os"
 	"reflect"
 	"testing"
@@ -144,6 +145,52 @@ func TestIndexChunking(t *testing.T) {
 	}
 }
 
+// TestChunkStreamIntegrity verifies that chunks stored by ChunkStream match
+// their IDs. This catches buffer-reuse races where the chunker's internal
+// backing buffer is overwritten by fillBuffer() before workers finish
+// processing the slice sent through the channel. The input must be large
+// enough to trigger multiple fillBuffer() calls (> 10*ChunkSizeMaxDefault).
+func TestChunkStreamIntegrity(t *testing.T) {
+	// 8MB of pseudo-random data ensures multiple fillBuffer() cycles.
+	// The chunker's internal buffer is 10*max = 2.5MB, so this triggers
+	// at least 3 refills and produces ~125 chunks at 64KB average.
+	data := make([]byte, 8*1024*1024)
+	for i := range len(data) / 8 {
+		binary.LittleEndian.PutUint64(data[i*8:], uint64(i)*0x9E3779B97F4A7C15)
+	}
+
+	c, err := NewChunker(bytes.NewReader(data), ChunkSizeMinDefault, ChunkSizeAvgDefault, ChunkSizeMaxDefault)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	dir := t.TempDir()
+	s, err := NewLocalStore(dir, StoreOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	index, err := ChunkStream(context.Background(), c, s, 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify every stored chunk: read it back, decompress, and check
+	// that its content hashes to the expected ChunkID. GetChunk with
+	// SkipVerify=false (the default) returns ChunkInvalid on mismatch.
+	var corrupted int
+	for i, chunk := range index.Chunks {
+		_, err := s.GetChunk(chunk.ID)
+		if err != nil {
+			corrupted++
+			t.Errorf("chunk %d (%s): %v", i, chunk.ID, err)
+		}
+	}
+	if corrupted > 0 {
+		t.Fatalf("%d of %d chunks are corrupted", corrupted, len(index.Chunks))
+	}
+}
+
 // Global var to store benchmark output
 var idx Index