fix: serialize ingestr installation to prevent file lock conflicts on Windows

When running Bruin with multiple workers on Windows, parallel workers
would simultaneously try to install ingestr via 'uv tool install --force',
causing file lock conflicts on ingestr.exe (OS error 32).

This commit:
- Adds a package-level mutex to serialize ingestr installation calls
- Tracks installed package combinations to avoid redundant reinstalls
- Only uses --force flag when actually needed (first install or local path)
- Adds comprehensive tests for the synchronization mechanism

The fix ensures that ingestr is installed only once per unique package
combination during a Bruin run, preventing the race condition that caused
file locks on Windows.

Fixes: ingestr.exe file lock error when running with multiple workers

Co-authored-by: Arsalan <arsalann@users.noreply.github.com>

Author: cursoragent

pkg/python/uv.go

@@ -25,6 +25,11 @@ import (
 	"github.com/spf13/afero"
 )
 
+var (
+	ingestrInstallMutex     sync.Mutex
+	ingestrInstalledPackages = make(map[string]bool)
+)
+
 var AvailablePythonVersions = map[string]bool{
 	"3.8":  true,
 	"3.9":  true,
@@ -126,12 +131,9 @@ func (u *UvPythonRunner) RunIngestr(ctx context.Context, args, extraPackages []s
 	}
 	u.binaryFullPath = binaryFullPath
 
-	err = u.Cmd.Run(ctx, repo, &CommandInstance{
-		Name: u.binaryFullPath,
-		Args: u.ingestrInstallCmd(ctx, extraPackages),
-	})
+	err = u.ensureIngestrInstalled(ctx, extraPackages, repo)
 	if err != nil {
-		return errors.Wrap(err, "failed to install ingestr")
+		return err
 	}
 
 	flags := []string{"tool", "run", "--no-config", "--prerelease", "allow", "--python", pythonVersionForIngestr, "ingestr"}
@@ -381,12 +383,9 @@ func (u *UvPythonRunner) runWithMaterialization(ctx context.Context, execCtx *ex
 		}
 	}
 
-	err = u.Cmd.Run(ctx, execCtx.repo, &CommandInstance{
-		Name: u.binaryFullPath,
-		Args: u.ingestrInstallCmd(ctx, extraPackages),
-	})
+	err = u.ensureIngestrInstalled(ctx, extraPackages, execCtx.repo)
 	if err != nil {
-		return errors.Wrap(err, "failed to install ingestr")
+		return err
 	}
 
 	runArgs := slices.Concat([]string{"tool", "run", "--no-config", "--prerelease", "allow", "--python", pythonVersionForIngestr, "ingestr"}, cmdArgs)
@@ -425,19 +424,21 @@ func (u *UvPythonRunner) ingestrPackage(ctx context.Context) (string, bool) {
 
 // ingestrInstallCmd returns the uv tool commandline
 // args necessary for installing ingestr.
-func (u *UvPythonRunner) ingestrInstallCmd(ctx context.Context, pkgs []string) []string {
+func (u *UvPythonRunner) ingestrInstallCmd(ctx context.Context, pkgs []string, forceReinstall bool) []string {
 	ingestrPackageName, isLocal := u.ingestrPackage(ctx)
 	cmdline := []string{
 		"tool",
 		"install",
 		"--no-config",
-		"--force",
 		"--quiet",
 		"--prerelease",
 		"allow",
 		"--python",
 		pythonVersionForIngestr,
 	}
+	if forceReinstall {
+		cmdline = append(cmdline, "--force")
+	}
 	for _, pkg := range pkgs {
 		cmdline = append(cmdline, "--with", pkg)
 	}
@@ -448,6 +449,52 @@ func (u *UvPythonRunner) ingestrInstallCmd(ctx context.Context, pkgs []string) [
 	return cmdline
 }
 
+// buildIngestrPackageKey creates a unique key for the combination of ingestr version and extra packages.
+func (u *UvPythonRunner) buildIngestrPackageKey(ctx context.Context, extraPackages []string) string {
+	ingestrPackageName, _ := u.ingestrPackage(ctx)
+	sortedPkgs := make([]string, len(extraPackages))
+	copy(sortedPkgs, extraPackages)
+	sort.Strings(sortedPkgs)
+	return ingestrPackageName + ":" + strings.Join(sortedPkgs, ",")
+}
+
+// ensureIngestrInstalled installs ingestr with the specified extra packages in a thread-safe manner.
+// It prevents parallel workers from installing ingestr simultaneously, which causes file lock
+// conflicts on Windows (OS error 32).
+func (u *UvPythonRunner) ensureIngestrInstalled(ctx context.Context, extraPackages []string, repo *git.Repo) error {
+	packageKey := u.buildIngestrPackageKey(ctx, extraPackages)
+
+	ingestrInstallMutex.Lock()
+	defer ingestrInstallMutex.Unlock()
+
+	_, isLocal := u.ingestrPackage(ctx)
+	if !isLocal {
+		if ingestrInstalledPackages[packageKey] {
+			return nil
+		}
+	}
+
+	forceReinstall := isLocal || !ingestrInstalledPackages[packageKey]
+	err := u.Cmd.Run(ctx, repo, &CommandInstance{
+		Name: u.binaryFullPath,
+		Args: u.ingestrInstallCmd(ctx, extraPackages, forceReinstall),
+	})
+	if err != nil {
+		return errors.Wrap(err, "failed to install ingestr")
+	}
+
+	ingestrInstalledPackages[packageKey] = true
+	return nil
+}
+
+// ResetIngestrInstallCache resets the ingestr installation cache.
+// This is primarily useful for testing.
+func ResetIngestrInstallCache() {
+	ingestrInstallMutex.Lock()
+	defer ingestrInstallMutex.Unlock()
+	ingestrInstalledPackages = make(map[string]bool)
+}
+
 const PythonArrowTemplate = `
 # /// script
 # dependencies = [

pkg/python/uv_test.go

@@ -2,6 +2,8 @@ package python
 
 import (
 	"context"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/bruin-data/bruin/pkg/git"
@@ -131,3 +133,162 @@ func Test_uvPythonRunner_Run(t *testing.T) {
 		})
 	}
 }
+
+func Test_buildIngestrPackageKey(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name          string
+		extraPackages []string
+		expected      string
+	}{
+		{
+			name:          "no extra packages",
+			extraPackages: []string{},
+			expected:      "ingestr@" + ingestrVersion + ":",
+		},
+		{
+			name:          "single package",
+			extraPackages: []string{"pyodbc"},
+			expected:      "ingestr@" + ingestrVersion + ":pyodbc",
+		},
+		{
+			name:          "multiple packages sorted",
+			extraPackages: []string{"pyodbc", "duckdb"},
+			expected:      "ingestr@" + ingestrVersion + ":duckdb,pyodbc",
+		},
+		{
+			name:          "packages in different order produce same key",
+			extraPackages: []string{"duckdb", "pyodbc"},
+			expected:      "ingestr@" + ingestrVersion + ":duckdb,pyodbc",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			u := &UvPythonRunner{}
+			key := u.buildIngestrPackageKey(context.Background(), tt.extraPackages)
+			assert.Equal(t, tt.expected, key)
+		})
+	}
+}
+
+func Test_ensureIngestrInstalled_OnlyInstallsOnce(t *testing.T) {
+	ResetIngestrInstallCache()
+	defer ResetIngestrInstallCache()
+
+	repo := &git.Repo{}
+	var installCount atomic.Int32
+
+	cmd := new(mockCmd)
+	cmd.On("Run", mock.Anything, repo, mock.MatchedBy(func(c *CommandInstance) bool {
+		return len(c.Args) > 0 && c.Args[0] == "tool" && c.Args[1] == "install"
+	})).Run(func(args mock.Arguments) {
+		installCount.Add(1)
+	}).Return(nil)
+
+	inst := new(mockUvInstaller)
+	inst.On("EnsureUvInstalled", mock.Anything).Return("~/.bruin/uv", nil)
+
+	u := &UvPythonRunner{
+		Cmd:            cmd,
+		UvInstaller:    inst,
+		binaryFullPath: "~/.bruin/uv",
+	}
+
+	ctx := context.Background()
+	extraPackages := []string{"pyodbc"}
+
+	err := u.ensureIngestrInstalled(ctx, extraPackages, repo)
+	assert.NoError(t, err)
+
+	err = u.ensureIngestrInstalled(ctx, extraPackages, repo)
+	assert.NoError(t, err)
+
+	err = u.ensureIngestrInstalled(ctx, extraPackages, repo)
+	assert.NoError(t, err)
+
+	assert.Equal(t, int32(1), installCount.Load(), "ingestr should only be installed once for the same package combination")
+}
+
+func Test_ensureIngestrInstalled_InstallsForDifferentPackages(t *testing.T) {
+	ResetIngestrInstallCache()
+	defer ResetIngestrInstallCache()
+
+	repo := &git.Repo{}
+	var installCount atomic.Int32
+
+	cmd := new(mockCmd)
+	cmd.On("Run", mock.Anything, repo, mock.MatchedBy(func(c *CommandInstance) bool {
+		return len(c.Args) > 0 && c.Args[0] == "tool" && c.Args[1] == "install"
+	})).Run(func(args mock.Arguments) {
+		installCount.Add(1)
+	}).Return(nil)
+
+	inst := new(mockUvInstaller)
+	inst.On("EnsureUvInstalled", mock.Anything).Return("~/.bruin/uv", nil)
+
+	u := &UvPythonRunner{
+		Cmd:            cmd,
+		UvInstaller:    inst,
+		binaryFullPath: "~/.bruin/uv",
+	}
+
+	ctx := context.Background()
+
+	err := u.ensureIngestrInstalled(ctx, []string{"pyodbc"}, repo)
+	assert.NoError(t, err)
+
+	err = u.ensureIngestrInstalled(ctx, []string{"duckdb"}, repo)
+	assert.NoError(t, err)
+
+	err = u.ensureIngestrInstalled(ctx, []string{"pyodbc", "duckdb"}, repo)
+	assert.NoError(t, err)
+
+	assert.Equal(t, int32(3), installCount.Load(), "ingestr should be installed once for each unique package combination")
+}
+
+func Test_ensureIngestrInstalled_ConcurrentCalls(t *testing.T) {
+	ResetIngestrInstallCache()
+	defer ResetIngestrInstallCache()
+
+	repo := &git.Repo{}
+	var installCount atomic.Int32
+
+	cmd := new(mockCmd)
+	cmd.On("Run", mock.Anything, repo, mock.MatchedBy(func(c *CommandInstance) bool {
+		return len(c.Args) > 0 && c.Args[0] == "tool" && c.Args[1] == "install"
+	})).Run(func(args mock.Arguments) {
+		installCount.Add(1)
+	}).Return(nil)
+
+	inst := new(mockUvInstaller)
+	inst.On("EnsureUvInstalled", mock.Anything).Return("~/.bruin/uv", nil)
+
+	u := &UvPythonRunner{
+		Cmd:            cmd,
+		UvInstaller:    inst,
+		binaryFullPath: "~/.bruin/uv",
+	}
+
+	ctx := context.Background()
+	extraPackages := []string{"pyodbc"}
+
+	var wg sync.WaitGroup
+	numGoroutines := 10
+
+	for range numGoroutines {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			err := u.ensureIngestrInstalled(ctx, extraPackages, repo)
+			assert.NoError(t, err)
+		}()
+	}
+
+	wg.Wait()
+
+	assert.Equal(t, int32(1), installCount.Load(), "ingestr should only be installed once even with concurrent calls")
+}