fix(repomanager): sanitize reused workers for BUG-003

core/git/git.go

@@ -89,6 +89,40 @@ func New(directory string, logger *zap.Logger) Interface {
 	}
 }
 
+// RestoreWorktree discards tracked, untracked, and ignored changes from a Git
+// worktree and restores all initialized submodules to their recorded commits.
+func RestoreWorktree(ctx context.Context, directory string) error {
+	return restoreWorktree(ctx, directory, &osExecRunner{})
+}
+
+func restoreWorktree(ctx context.Context, directory string, runner commandRunner) error {
+	ctx, cancel := context.WithTimeout(ctx, _gitTimeout)
+	defer cancel()
+
+	statusArgs := []string{"status", "--porcelain", "--untracked-files=all", "--ignored", "--ignore-submodules=none"}
+	status, err := runner.output(ctx, directory, "git", statusArgs...)
+	if err != nil {
+		return wrapError(ctx, statusArgs, err)
+	}
+	if len(bytes.TrimSpace(status)) == 0 {
+		return nil
+	}
+
+	commands := [][]string{
+		{"reset", "--hard", "HEAD"},
+		{"clean", "-ffdx"},
+		{"submodule", "foreach", "--recursive", "git", "reset", "--hard", "HEAD"},
+		{"submodule", "foreach", "--recursive", "git", "clean", "-ffdx"},
+		{"submodule", "update", "--init", "--recursive", "--force"},
+	}
+	for _, args := range commands {
+		if err := runner.run(ctx, directory, "git", args...); err != nil {
+			return wrapError(ctx, args, err)
+		}
+	}
+	return nil
+}
+
 // wrapError wraps a non-nil err with the failing git command's arguments,
 // additionally wrapping ErrFatal if the command exited with a fatal (128) or
 // usage (129) exit code, as opposed to a non-fatal, conditional exit code

core/git/git_test.go

@@ -220,6 +220,67 @@ func TestSubmoduleUpdate_usesRunnerWithDirAndArgs(t *testing.T) {
 	assert.EqualValues(t, []string{"submodule", "update", "--init", "--recursive"}, c.args)
 }
 
+func TestRestoreWorktree_usesDestructiveCleanupSequence(t *testing.T) {
+	m := &mockRunner{out: []byte(" M tracked.txt\n")}
+	require.NoError(t, restoreWorktree(context.Background(), "/repo", m))
+
+	require.Len(t, m.calls, 6)
+	assert.Equal(t, "output", m.calls[0].kind)
+	assert.EqualValues(t, []string{"status", "--porcelain", "--untracked-files=all", "--ignored", "--ignore-submodules=none"}, m.calls[0].args)
+	assert.EqualValues(t, []string{"reset", "--hard", "HEAD"}, m.calls[1].args)
+	assert.EqualValues(t, []string{"clean", "-ffdx"}, m.calls[2].args)
+	assert.EqualValues(t, []string{"submodule", "foreach", "--recursive", "git", "reset", "--hard", "HEAD"}, m.calls[3].args)
+	assert.EqualValues(t, []string{"submodule", "foreach", "--recursive", "git", "clean", "-ffdx"}, m.calls[4].args)
+	assert.EqualValues(t, []string{"submodule", "update", "--init", "--recursive", "--force"}, m.calls[5].args)
+	for _, call := range m.calls[1:] {
+		assert.Equal(t, "run", call.kind)
+		assert.Equal(t, "/repo", call.dir)
+		assert.Equal(t, "git", call.name)
+	}
+}
+
+func TestRestoreWorktree_skipsCleanupWhenAlreadyClean(t *testing.T) {
+	m := &mockRunner{}
+	require.NoError(t, restoreWorktree(context.Background(), "/repo", m))
+
+	require.Len(t, m.calls, 1)
+	assert.Equal(t, "output", m.calls[0].kind)
+	assert.Equal(t, "/repo", m.calls[0].dir)
+	assert.Equal(t, "git", m.calls[0].name)
+	assert.EqualValues(t, []string{"status", "--porcelain", "--untracked-files=all", "--ignored", "--ignore-submodules=none"}, m.calls[0].args)
+}
+
+func TestRestoreWorktree_removesTrackedAndUntrackedChanges(t *testing.T) {
+	repoDir := t.TempDir()
+	runGit(t, repoDir, "init")
+	runGit(t, repoDir, "config", "user.email", "test@example.com")
+	runGit(t, repoDir, "config", "user.name", "Test User")
+	runGit(t, repoDir, "config", "commit.gpgsign", "false")
+	require.NoError(t, os.WriteFile(filepath.Join(repoDir, ".gitignore"), []byte("ignored.txt\n"), 0o644))
+	require.NoError(t, os.WriteFile(filepath.Join(repoDir, "tracked.txt"), []byte("original\n"), 0o644))
+	runGit(t, repoDir, "add", ".gitignore", "tracked.txt")
+	runGit(t, repoDir, "commit", "-m", "initial")
+
+	require.NoError(t, os.WriteFile(filepath.Join(repoDir, "tracked.txt"), []byte("dirty\n"), 0o644))
+	require.NoError(t, os.WriteFile(filepath.Join(repoDir, "untracked.txt"), []byte("untracked\n"), 0o644))
+	require.NoError(t, os.WriteFile(filepath.Join(repoDir, "ignored.txt"), []byte("ignored\n"), 0o644))
+	runGit(t, repoDir, "add", "tracked.txt")
+
+	require.NoError(t, RestoreWorktree(context.Background(), repoDir))
+
+	content, err := os.ReadFile(filepath.Join(repoDir, "tracked.txt"))
+	require.NoError(t, err)
+	assert.Equal(t, "original\n", string(content))
+	assert.NoFileExists(t, filepath.Join(repoDir, "untracked.txt"))
+	assert.NoFileExists(t, filepath.Join(repoDir, "ignored.txt"))
+
+	cmd := exec.Command("git", "status", "--porcelain", "--untracked-files=all")
+	cmd.Dir = repoDir
+	output, err := cmd.Output()
+	require.NoError(t, err)
+	assert.Empty(t, output)
+}
+
 func TestDiffWithStatus_parsesNameStatusOutput(t *testing.T) {
 	tests := []struct {
 		name    string

core/repomanager/BUILD.bazel

@@ -24,7 +24,9 @@ go_test(
     srcs = ["repo_manager_test.go"],
     embed = [":repomanager"],
     deps = [
+        "//core/git",
         "//core/git/gitmock",
+        "//core/workspace",
         "//entity",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",

core/repomanager/metrics.go

@@ -25,9 +25,10 @@ import (
 const (
 	_opLease = "lease"
 
-	_stepEnsureOrigin = "ensure_origin_duration"
-	_stepWaitSlot     = "wait_slot_duration"
-	_stepCreateWorker = "create_worker_duration"
+	_stepEnsureOrigin  = "ensure_origin_duration"
+	_stepWaitSlot      = "wait_slot_duration"
+	_stepCreateWorker  = "create_worker_duration"
+	_stepRestoreWorker = "restore_worker_duration"
 )
 
 func recordStep(e *metrics.Emitter, name string, start time.Time, buckets tally.DurationBuckets) {

core/repomanager/repo_manager.go

@@ -50,6 +50,7 @@ type repoManager struct {
 	logger               *zap.Logger
 	emitter              *metrics.Emitter
 	poolSize             int
+	restoreWorker        func(context.Context, string) error
 
 	mu    sync.Mutex
 	pools map[string]*workerPool
@@ -109,6 +110,7 @@ func NewRepoManager(appCtx context.Context, p Params) (RepoManager, error) {
 		logger:               p.Logger,
 		emitter:              metrics.New(p.Scope).SubScope("repo_manager"),
 		poolSize:             p.PoolSize,
+		restoreWorker:        git.RestoreWorktree,
 		pools:                make(map[string]*workerPool),
 		appCtx:               appCtx,
 	}, nil
@@ -177,13 +179,31 @@ func (r *repoManager) Lease(ctx context.Context, desc entity.BuildDescription) (
 		return nil, fmt.Errorf("pool for repo %s: %w", repo, waitErr)
 	}
 
-	// Lazily create the worker clone on first use
+	// Restore reused workers before handing them to another request. If the
+	// worktree cannot be verified clean, quarantine it by recreating the clone.
+	var restoreErr error
+	if slot.created {
+		restoreStart := time.Now()
+		restoreErr = r.restoreWorker(ctx, slot.dir)
+		recordStep(e, _stepRestoreWorker, restoreStart, metrics.FastDurationBuckets)
+		if restoreErr != nil {
+			r.logger.Warn("failed to restore worker; recreating it",
+				zap.String("directory", slot.dir),
+				zap.Error(restoreErr))
+			slot.created = false
+		}
+	}
+
+	// Lazily create new workers and synchronously recreate quarantined workers.
 	if !slot.created {
 		createStart := time.Now()
 		err := r.createWorker(ctx, pool.originDir, slot.dir)
 		recordStep(e, _stepCreateWorker, createStart, metrics.FastDurationBuckets)
 		if err != nil {
 			pool.avail <- slot // return slot so others can retry
+			if restoreErr != nil {
+				return nil, fmt.Errorf("restore worker failed (%v); recreate worker: %w", restoreErr, err)
+			}
 			return nil, fmt.Errorf("create worker: %w", err)
 		}
 		slot.created = true
@@ -227,7 +247,9 @@ func (p *workerPool) ensureOrigin(ctx context.Context, g git.Interface, remote s
 // createWorker creates a worker by cloning the origin with --local
 // (fast and space-efficient).
 func (r *repoManager) createWorker(ctx context.Context, originDir, workerDir string) error {
-	os.RemoveAll(workerDir) // clean up any partial/corrupted previous state
+	if err := os.RemoveAll(workerDir); err != nil {
+		return fmt.Errorf("remove previous worker: %w", err)
+	}
 	if err := os.MkdirAll(filepath.Dir(workerDir), 0o755); err != nil {
 		return err
 	}

core/repomanager/repo_manager_test.go

@@ -19,25 +19,43 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	tangogit "github.com/uber/tango/core/git"
 	gitmock "github.com/uber/tango/core/git/gitmock"
+	"github.com/uber/tango/core/workspace"
 	"github.com/uber/tango/entity"
 	"go.uber.org/mock/gomock"
 	"go.uber.org/zap"
 )
 
+type requestFunc func(context.Context) error
+
+func (f requestFunc) Apply(ctx context.Context) error {
+	return f(ctx)
+}
+
 func newTestRepoManager(t *testing.T, appCtx context.Context, p Params) RepoManager {
 	t.Helper()
 	rm, err := NewRepoManager(appCtx, p)
 	require.NoError(t, err)
+	rm.(*repoManager).restoreWorker = func(context.Context, string) error { return nil }
 	return rm
 }
 
+func runRepoGit(t *testing.T, directory string, args ...string) {
+	t.Helper()
+	cmd := exec.CommandContext(t.Context(), "git", args...)
+	cmd.Dir = directory
+	output, err := cmd.CombinedOutput()
+	require.NoError(t, err, "git %v: %s", args, output)
+}
+
 func TestNewRepoManager_InvalidPoolSize(t *testing.T) {
 	t.Parallel()
 	_, err := NewRepoManager(context.Background(), Params{
@@ -117,6 +135,91 @@ func TestLease_ReusesWorker_AfterRelease(t *testing.T) {
 	require.NoError(t, ws2.Release())
 }
 
+func TestLease_RestoresWorkerAfterFailedMaterialization(t *testing.T) {
+	sourceDir := filepath.Join(t.TempDir(), "source")
+	require.NoError(t, os.MkdirAll(sourceDir, 0o755))
+	runRepoGit(t, sourceDir, "init")
+	runRepoGit(t, sourceDir, "config", "user.email", "test@example.com")
+	runRepoGit(t, sourceDir, "config", "user.name", "Test User")
+	runRepoGit(t, sourceDir, "config", "commit.gpgsign", "false")
+	require.NoError(t, os.WriteFile(filepath.Join(sourceDir, ".gitignore"), []byte("generated.txt\n"), 0o644))
+	require.NoError(t, os.WriteFile(filepath.Join(sourceDir, "tracked.txt"), []byte("original\n"), 0o644))
+	runRepoGit(t, sourceDir, "add", ".gitignore", "tracked.txt")
+	runRepoGit(t, sourceDir, "commit", "-m", "initial")
+
+	root := t.TempDir()
+	rm, err := NewRepoManager(context.Background(), Params{
+		Git:                  tangogit.New(t.TempDir(), zap.NewNop()),
+		Logger:               zap.NewNop(),
+		RepoManagerClonePath: root,
+		PoolSize:             1,
+	})
+	require.NoError(t, err)
+
+	ws1, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: sourceDir})
+	require.NoError(t, err)
+	workerDir := ws1.Path()
+	reuseMarker := filepath.Join(workerDir, ".git", "reuse-marker")
+	require.NoError(t, os.WriteFile(reuseMarker, []byte("keep"), 0o644))
+
+	materializeErr := ws1.ApplyRequests(context.Background(), []workspace.Request{
+		requestFunc(func(context.Context) error {
+			require.NoError(t, os.WriteFile(filepath.Join(workerDir, "tracked.txt"), []byte("dirty\n"), 0o644))
+			require.NoError(t, os.WriteFile(filepath.Join(workerDir, "untracked.txt"), []byte("untracked\n"), 0o644))
+			require.NoError(t, os.WriteFile(filepath.Join(workerDir, "generated.txt"), []byte("generated\n"), 0o644))
+			runRepoGit(t, workerDir, "add", "tracked.txt")
+			return assert.AnError
+		}),
+	})
+	require.ErrorIs(t, materializeErr, assert.AnError)
+	require.NoError(t, ws1.Release())
+
+	ws2, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: sourceDir})
+	require.NoError(t, err)
+	assert.Equal(t, workerDir, ws2.Path())
+	assert.FileExists(t, reuseMarker, "a successfully restored worker should be reused rather than recloned")
+
+	content, err := os.ReadFile(filepath.Join(workerDir, "tracked.txt"))
+	require.NoError(t, err)
+	assert.Equal(t, "original\n", string(content))
+	assert.NoFileExists(t, filepath.Join(workerDir, "untracked.txt"))
+	assert.NoFileExists(t, filepath.Join(workerDir, "generated.txt"))
+
+	cmd := exec.Command("git", "status", "--porcelain", "--untracked-files=all")
+	cmd.Dir = workerDir
+	output, err := cmd.Output()
+	require.NoError(t, err)
+	assert.Empty(t, output)
+	require.NoError(t, ws2.Release())
+}
+
+func TestLease_RecreatesWorker_WhenRestoreFails(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	g := gitmock.NewMockInterface(ctrl)
+
+	root := t.TempDir()
+	remote := "git@github.com:org/repo"
+	originDir := filepath.Join(root, "org/repo")
+	workerDir := filepath.Join(root, ".workers", "org/repo", "worker-1")
+
+	g.EXPECT().Clone(gomock.Any(), remote, originDir, "-c", "gc.auto=0").Return(nil)
+	g.EXPECT().Clone(gomock.Any(), originDir, workerDir, "--local", "-c", "gc.auto=0").Return(nil).Times(2)
+
+	rm := newTestRepoManager(t, context.Background(), Params{Git: g, Logger: zap.NewNop(), RepoManagerClonePath: root, PoolSize: 1})
+	manager := rm.(*repoManager)
+	manager.restoreWorker = func(context.Context, string) error { return assert.AnError }
+
+	ws1, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: remote})
+	require.NoError(t, err)
+	require.NoError(t, ws1.Release())
+
+	ws2, err := rm.Lease(context.Background(), entity.BuildDescription{Remote: remote})
+	require.NoError(t, err)
+	assert.Equal(t, workerDir, ws2.Path())
+	require.NoError(t, ws2.Release())
+}
+
 func TestLease_CreatesMultipleWorkers(t *testing.T) {
 	t.Parallel()
 	ctrl := gomock.NewController(t)