fix: data races in tests and NFS cache scanner (#2256)

* fix: data races in tests and NFS cache scanner

- Serialize goose migrations with mutex (race on package globals)
- Use RunAndReturn in envd conversion test (race on connect.Response headers)
- Replace *os.File fd with path string in NFS cache scanner (race between Close/Fd)
- Capture loop vars in NBD path_direct goroutine closure
- Use sync.Map and mutexes in multipart upload tests
- Use atomic.Bool and separate context vars in errorcollector test

These races were exposed by ARM64 testing but are architecture-independent bugs.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: include full path in statInDir error message

Use filepath.Join(dirPath, filename) in the error instead of bare
filename, matching the sibling stat() function's error format.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: remove unnecessary loop var capture (Go 1.22+)

Go 1.22+ changed loop variable semantics — each iteration gets its
own copy. No need to capture deviceIndex/i before the goroutine.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: restore deviceIndex sentinel value (math.MaxUint32)

The MaxUint32 initialization was intentional as a sentinel value.
Reverting the unnecessary change per review feedback.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: e2b <e2b@Onsites-MacBook-Pro.local>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

packages/db/pkg/testutils/db.go

@@ -5,6 +5,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -106,6 +107,12 @@ func SetupDatabase(t *testing.T) *Database {
 	}
 }
 
+// gooseMu serializes goose operations across parallel tests.
+// goose.OpenDBWithDriver calls goose.SetDialect which writes to package-level
+// globals (dialect, store) without synchronization. Concurrent test goroutines
+// race on these globals, triggering the race detector on ARM64.
+var gooseMu sync.Mutex
+
 // runDatabaseMigrations executes all required database migrations
 func runDatabaseMigrations(t *testing.T, connStr string) {
 	t.Helper()
@@ -115,6 +122,9 @@ func runDatabaseMigrations(t *testing.T, connStr string) {
 	require.NoError(t, err, "Failed to find git root")
 	repoRoot := strings.TrimSpace(string(output))
 
+	gooseMu.Lock()
+	defer gooseMu.Unlock()
+
 	db, err := goose.OpenDBWithDriver("pgx", connStr)
 	require.NoError(t, err)
 	t.Cleanup(func() {

packages/envd/internal/services/legacy/conversion_test.go

@@ -2,6 +2,7 @@ package legacy
 
 import (
 	"bytes"
+	"context"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -22,12 +23,19 @@ import (
 func TestFilesystemClient_FieldFormatter(t *testing.T) {
 	t.Parallel()
 	fsh := filesystemconnectmocks.NewMockFilesystemHandler(t)
-	fsh.EXPECT().Move(mock.Anything, mock.Anything).Return(connect.NewResponse(&filesystem.MoveResponse{
-		Entry: &filesystem.EntryInfo{
-			Name:  "test-name",
-			Owner: "new-extra-field",
+	// Use RunAndReturn to create a fresh response per call. Using Return()
+	// shares one Response across parallel subtests, causing a data race on
+	// the lazily-initialized header/trailer maps inside connect.Response.
+	fsh.EXPECT().Move(mock.Anything, mock.Anything).RunAndReturn(
+		func(_ context.Context, _ *connect.Request[filesystem.MoveRequest]) (*connect.Response[filesystem.MoveResponse], error) {
+			return connect.NewResponse(&filesystem.MoveResponse{
+				Entry: &filesystem.EntryInfo{
+					Name:  "test-name",
+					Owner: "new-extra-field",
+				},
+			}), nil
 		},
-	}), nil)
+	)
 
 	_, handler := filesystemconnect.NewFilesystemHandler(fsh,
 		connect.WithInterceptors(

packages/orchestrator/cmd/clean-nfs-cache/cleaner/clean.go

@@ -99,7 +99,7 @@ type Candidate struct {
 }
 
 type statReq struct {
-	df       *os.File
+	dirPath  string
 	name     string
 	response chan *statReq
 	f        *File

packages/orchestrator/cmd/clean-nfs-cache/cleaner/scan.go

@@ -60,7 +60,7 @@ func (c *Cleaner) Statter(ctx context.Context, done *sync.WaitGroup) {
 		case <-ctx.Done():
 			return
 		case req := <-c.statRequestCh:
-			f, err := c.statInDir(req.df, req.name)
+			f, err := c.statInDir(req.dirPath, req.name)
 			req.f = f
 			req.err = err
 			req.response <- req
@@ -201,13 +201,16 @@ func (c *Cleaner) scanDir(ctx context.Context, path []*Dir) (out *Dir, err error
 		}
 	}
 
-	// submit all stat requests
+	// Submit stat requests using the directory path (not the *os.File).
+	// The file descriptor df is closed when scanDir returns (defer above),
+	// but Statter goroutines may still be processing requests concurrently.
+	// Passing the path avoids a race between df.Close() and df.Fd().
 	responseCh := make(chan *statReq, len(filenames))
 	for _, name := range filenames {
 		select {
 		case <-ctx.Done():
 			return nil, ctx.Err()
-		case c.statRequestCh <- &statReq{df: df, name: name, response: responseCh}:
+		case c.statRequestCh <- &statReq{dirPath: absPath, name: name, response: responseCh}:
 			// submitted
 		}
 	}

packages/orchestrator/cmd/clean-nfs-cache/cleaner/stat_linux.go

@@ -4,7 +4,7 @@ package cleaner
 
 import (
 	"fmt"
-	"os"
+	"path/filepath"
 
 	"golang.org/x/sys/unix"
 )
@@ -30,17 +30,18 @@ func (c *Cleaner) stat(fullPath string) (*Candidate, error) {
 	}, nil
 }
 
-func (c *Cleaner) statInDir(df *os.File, filename string) (*File, error) {
+func (c *Cleaner) statInDir(dirPath string, filename string) (*File, error) {
 	c.StatxC.Add(1)
 	c.StatxInDirC.Add(1)
 	var statx unix.Statx_t
-	err := unix.Statx(int(df.Fd()), filename,
+	fullPath := filepath.Join(dirPath, filename)
+	err := unix.Statx(unix.AT_FDCWD, fullPath,
 		unix.AT_STATX_DONT_SYNC|unix.AT_SYMLINK_NOFOLLOW|unix.AT_NO_AUTOMOUNT,
 		unix.STATX_ATIME|unix.STATX_SIZE,
 		&statx,
 	)
 	if err != nil {
-		return nil, fmt.Errorf("failed to statx %q: %w", filename, err)
+		return nil, fmt.Errorf("failed to statx %q: %w", fullPath, err)
 	}
 
 	return &File{

packages/orchestrator/cmd/clean-nfs-cache/cleaner/stat_osx.go

@@ -31,10 +31,10 @@ func (c *Cleaner) stat(path string) (*Candidate, error) {
 	}, nil
 }
 
-func (c *Cleaner) statInDir(df *os.File, filename string) (*File, error) {
+func (c *Cleaner) statInDir(dirPath string, filename string) (*File, error) {
 	c.StatxInDirC.Add(1)
-	// performance on OS X doeas not matter, so just use the full stat
-	cand, err := c.stat(filepath.Join(df.Name(), filename))
+	// performance on OS X does not matter, so just use the full stat
+	cand, err := c.stat(filepath.Join(dirPath, filename))
 	if err != nil {
 		return nil, err
 	}

packages/shared/pkg/storage/gcp_multipart_test.go

@@ -172,7 +172,7 @@ func TestMultipartUploader_UploadFileInParallel_Success(t *testing.T) {
 
 	var uploadID string
 	var initiateCount, uploadPartCount, completeCount int32
-	receivedParts := make(map[int]string)
+	receivedParts := sync.Map{}
 
 	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch {
@@ -194,7 +194,7 @@ func TestMultipartUploader_UploadFileInParallel_Success(t *testing.T) {
 			// Upload part
 			partNum := atomic.AddInt32(&uploadPartCount, 1)
 			body, _ := io.ReadAll(r.Body)
-			receivedParts[int(partNum)] = string(body)
+			receivedParts.Store(int(partNum), string(body))
 
 			w.Header().Set("ETag", fmt.Sprintf(`"etag%d"`, partNum))
 			w.WriteHeader(http.StatusOK)
@@ -217,7 +217,9 @@ func TestMultipartUploader_UploadFileInParallel_Success(t *testing.T) {
 	// Verify all parts were uploaded and content matches
 	var reconstructed strings.Builder
 	for i := 1; i <= int(atomic.LoadInt32(&uploadPartCount)); i++ {
-		reconstructed.WriteString(receivedParts[i])
+		if part, ok := receivedParts.Load(i); ok {
+			reconstructed.WriteString(part.(string))
+		}
 	}
 	require.Equal(t, testContent, reconstructed.String())
 }
@@ -655,6 +657,7 @@ func TestMultipartUploader_BoundaryConditions_ExactChunkSize(t *testing.T) {
 	require.NoError(t, err)
 
 	var partSizes []int
+	var partSizesMu sync.Mutex
 
 	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch {
@@ -670,7 +673,9 @@ func TestMultipartUploader_BoundaryConditions_ExactChunkSize(t *testing.T) {
 
 		case strings.Contains(r.URL.RawQuery, "partNumber"):
 			body, _ := io.ReadAll(r.Body)
+			partSizesMu.Lock()
 			partSizes = append(partSizes, len(body))
+			partSizesMu.Unlock()
 
 			partNum := strings.Split(strings.Split(r.URL.RawQuery, "partNumber=")[1], "&")[0]
 			w.Header().Set("ETag", fmt.Sprintf(`"boundary-etag-%s"`, partNum))
@@ -904,10 +909,13 @@ func TestRetryableClient_ActualRetryBehavior(t *testing.T) {
 	var requestCount int32
 	var retryDelays []time.Duration
 	var retryTimes []time.Time
+	var retryMu sync.Mutex
 
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		count := atomic.AddInt32(&requestCount, 1)
+		retryMu.Lock()
 		retryTimes = append(retryTimes, time.Now())
+		retryMu.Unlock()
 
 		if count < 3 {
 			w.WriteHeader(http.StatusInternalServerError)

packages/shared/pkg/utils/errorcollector_test.go

@@ -3,6 +3,7 @@ package utils
 import (
 	"context"
 	"errors"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -49,23 +50,31 @@ func TestErrorCollector(t *testing.T) {
 
 		ec := NewErrorCollector(1)
 
-		// Block the collector's only slot
+		// Block the collector's only slot.
+		// ctx1 and ctx2 must be distinct variables: the closure passed to ec.Go
+		// captures the context variable by reference. If we reused a single "ctx"
+		// variable, the first closure's <-ctx.Done() would race with the main
+		// goroutine's reassignment of ctx on the second WithCancel call.
 		started := make(chan struct{})
-		ctx, cancel1 := context.WithCancel(t.Context())
-		ec.Go(ctx, func() error {
+		ctx1, cancel1 := context.WithCancel(t.Context())
+		ec.Go(ctx1, func() error {
 			close(started)
-			<-ctx.Done()
+			<-ctx1.Done()
 
 			return nil
 		})
 
 		<-started
 
-		// This Go call should block on the semaphore
-		var wasCalled bool
-		ctx, cancel2 := context.WithCancel(t.Context())
-		ec.Go(ctx, func() error {
-			wasCalled = true
+		// This Go call should block on the semaphore.
+		// wasCalled must be atomic: the goroutine spawned by ec.Go may write it
+		// concurrently with the main goroutine's read in assert.False below.
+		// A plain bool causes a data race that the -race detector catches on ARM64
+		// (weaker memory model) even though it appears safe on x86.
+		var wasCalled atomic.Bool
+		ctx2, cancel2 := context.WithCancel(t.Context())
+		ec.Go(ctx2, func() error {
+			wasCalled.Store(true)
 
 			return nil
 		})
@@ -78,6 +87,6 @@ func TestErrorCollector(t *testing.T) {
 
 		err := ec.Wait()
 		require.ErrorIs(t, err, context.Canceled)
-		assert.False(t, wasCalled)
+		assert.False(t, wasCalled.Load())
 	})
 }