Fix data race on ChunkingStats in IndexFromFile

make.go

@@ -27,6 +27,7 @@ func IndexFromFile(ctx context.Context,
 ) (Index, ChunkingStats, error) {
 
 	stats := ChunkingStats{}
+	var wg sync.WaitGroup
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
@@ -121,14 +122,15 @@ func IndexFromFile(ctx context.Context,
 
 	// Start the workers
 	for _, w := range worker {
-		go w.start(ctx)
-		defer w.stop() // shouldn't be necessary, but better be safe
+		wg.Go(func() { w.start(ctx) })
+		defer w.stop() // panic/return safety net (idempotent via once)
 	}
 
 	// Go through the workers, starting with the first one, taking all chunks
 	// from their bucket before moving on to the next. It's possible that a worker
 	// reaches the end of the stream before the following worker does (eof=true),
 	// don't advance to the next worker in that case.
+	var chunkErr error
 	for _, w := range worker {
 		for chunk := range w.results {
 			// Assemble the list of chunks in the index
@@ -138,15 +140,26 @@ func IndexFromFile(ctx context.Context,
 		}
 		// Done reading all chunks from this worker, check for any errors
 		if w.err != nil {
-			return index, stats, w.err
+			chunkErr = w.err
+			break
 		}
 		// Stop if this worker reached the end of the stream (it's not necessarily
 		// the last worker!)
 		if w.eof {
 			break
 		}
 	}
-	return index, stats, nil
+
+	// Stop every worker and wait for all goroutines to fully exit before
+	// copying stats by value into the return. Otherwise the copy races the
+	// workers' atomic stats updates and the count would be incomplete.
+	cancel()
+	for _, w := range worker {
+		w.stop()
+	}
+	wg.Wait()
+
+	return index, stats, chunkErr
 }
 
 // Parallel chunk worker - Splits a stream and stores start, size and ID in

make_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	"github.com/folbricht/tempfile"
+	"github.com/stretchr/testify/require"
 )
 
 func TestParallelChunking(t *testing.T) {
@@ -84,3 +85,51 @@ func TestParallelChunking(t *testing.T) {
 		})
 	}
 }
+
+// TestIndexFromFileStats exercises the parallel chunker's early-EOF/straggler
+// path (null-heavy inputs, multiple workers) and asserts the returned
+// ChunkingStats. It guards the data race where IndexFromFile copied stats by
+// value while worker goroutines were still atomically updating them: with the
+// join in place the counters must be complete and deterministic for every
+// worker count. Run under -race to catch a regression of the join.
+func TestIndexFromFileStats(t *testing.T) {
+	null := make([]byte, 4*ChunkSizeMaxDefault)
+	rnd := make([]byte, 4*ChunkSizeMaxDefault)
+	rand.Read(rnd)
+
+	tests := map[string][][]byte{
+		"trailing null":   {rnd, rnd, null, null, null, null},
+		"middle null":     {rnd, null, null, null, null, rnd},
+		"spread out null": {rnd, null, null, null, rnd, null, null, null, rnd},
+	}
+
+	for name, input := range tests {
+		t.Run(name, func(t *testing.T) {
+			f, err := tempfile.New("", "")
+			require.NoError(t, err)
+			defer os.Remove(f.Name())
+			_, err = f.Write(join(input...))
+			require.NoError(t, err)
+			f.Close()
+
+			for n := 2; n <= 8; n++ {
+				t.Run(fmt.Sprintf("n=%d", n), func(t *testing.T) {
+					index, stats, err := IndexFromFile(
+						context.Background(),
+						f.Name(),
+						n,
+						ChunkSizeMinDefault, ChunkSizeAvgDefault, ChunkSizeMaxDefault,
+						NewProgressBar(""),
+					)
+					require.NoError(t, err)
+					require.Equal(t, uint64(len(index.Chunks)), stats.ChunksAccepted,
+						"ChunksAccepted should equal len(index.Chunks)")
+					require.GreaterOrEqual(t, stats.ChunksProduced, stats.ChunksAccepted,
+						"ChunksProduced should be >= ChunksAccepted")
+					require.NotZero(t, stats.ChunksProduced,
+						"workers should have produced chunks")
+				})
+			}
+		})
+	}
+}