diff --git a/core/git/git.go b/core/git/git.go index 9ff7d528..2b5db659 100644 --- a/core/git/git.go +++ b/core/git/git.go @@ -156,24 +156,29 @@ func (c *impl) ApplyPatch(ctx context.Context, patch []byte) error { // RevParse returns the revision hash of a reference. func (c *impl) RevParse(ctx context.Context, ref string) (string, error) { + ctx, cancel := context.WithTimeout(ctx, _gitTimeout) + defer cancel() args := []string{"rev-parse", ref} out, err := c.runner.output(ctx, c.directory, "git", args...) if err != nil { - return "", err + return "", wrapError(ctx, args, err) } return strings.TrimSpace(string(out)), nil } // IsAncestor reports whether ancestorRef is an ancestor of descendantRef. func (c *impl) IsAncestor(ctx context.Context, ancestorRef, descendantRef string) (bool, error) { - _, err := c.runner.output(ctx, c.directory, "git", "merge-base", "--is-ancestor", ancestorRef, descendantRef) + ctx, cancel := context.WithTimeout(ctx, _gitTimeout) + defer cancel() + args := []string{"merge-base", "--is-ancestor", ancestorRef, descendantRef} + _, err := c.runner.output(ctx, c.directory, "git", args...) if err != nil { if exitErr, ok := err.(*exec.ExitError); ok && exitErr.ExitCode() == 1 { return false, nil } // an exit code other than 1, or a non-ExitError failure (context canceled, // git binary missing, I/O error), indicates the check itself failed. - return false, fmt.Errorf("check if ref %s is ancestor of %s: %w", ancestorRef, descendantRef, err) + return false, wrapError(ctx, args, err) } return true, nil } @@ -226,18 +231,24 @@ func (c *impl) DiffWithStatus(ctx context.Context, baseRef, targetRef string) ([ // GetCommitTimeSecond returns the commit timestamp of the given ref in Unix seconds. func (c *impl) GetCommitTimeSecond(ctx context.Context, ref string) (int64, error) { - out, err := c.runner.output(ctx, c.directory, "git", "log", "-1", "--format=%ct", ref) + ctx, cancel := context.WithTimeout(ctx, _gitTimeout) + defer cancel() + args := []string{"log", "-1", "--format=%ct", ref} + out, err := c.runner.output(ctx, c.directory, "git", args...) if err != nil { - return 0, err + return 0, wrapError(ctx, args, err) } return strconv.ParseInt(strings.TrimSpace(string(out)), 10, 64) } // FileHashes gets a mapping of files to their hashes based on `git ls-tree --full-tree -r `. func (c *impl) FileHashes(ctx context.Context, ref string) (map[string][]byte, error) { - out, err := c.runner.output(ctx, c.directory, "git", "ls-tree", "--full-tree", "-r", "-z", ref) + ctx, cancel := context.WithTimeout(ctx, _gitTimeout) + defer cancel() + args := []string{"ls-tree", "--full-tree", "-r", "-z", ref} + out, err := c.runner.output(ctx, c.directory, "git", args...) if err != nil { - return nil, err + return nil, wrapError(ctx, args, err) } fileHashes := make(map[string][]byte) diff --git a/core/git/git_test.go b/core/git/git_test.go index c2214995..a6ff87f1 100644 --- a/core/git/git_test.go +++ b/core/git/git_test.go @@ -280,10 +280,11 @@ func TestGetCommitTimeSecond_parsesUnixTimestamp(t *testing.T) { } func TestGetCommitTimeSecond_errorPropagates(t *testing.T) { - m := &mockRunner{err: errors.New("git error")} + m := &mockRunner{err: assert.AnError} g := &impl{directory: "/repo", runner: m} _, err := g.GetCommitTimeSecond(context.Background(), "HEAD") require.Error(t, err) + assert.ErrorIs(t, err, assert.AnError) } func TestDefaultGit_FileHashes(t *testing.T) { @@ -314,7 +315,7 @@ func TestDefaultGit_FileHashes(t *testing.T) { }, { name: "git error", - wantError: errors.New(""), + wantError: assert.AnError, }, } @@ -330,7 +331,12 @@ func TestDefaultGit_FileHashes(t *testing.T) { m.out = tt.giveOutput m.err = tt.wantError gotHashes, err := g.FileHashes(ctx, tt.name) - require.Equal(t, tt.wantError, err) + if tt.wantError != nil { + require.Error(t, err) + assert.ErrorIs(t, err, tt.wantError) + } else { + require.NoError(t, err) + } assert.Equal(t, tt.wantHashes, gotHashes) }) } @@ -440,3 +446,55 @@ func runGit(t *testing.T, directory string, args ...string) { output, err := cmd.CombinedOutput() require.NoError(t, err, "git %v: %s", args, output) } + +func TestFatalExitCode_wrapsErrFatal(t *testing.T) { + // A fatal exit (128) from any of the previously-unwrapped methods must + // surface as errors.Is(err, ErrFatal) so the orchestrator's + // classifyGitError can tag it as an infra failure. + fatalErr := exec.Command("sh", "-c", "exit 128").Run() + require.Error(t, fatalErr) + + tests := []struct { + name string + call func(g *impl) error + }{ + { + name: "RevParse", + call: func(g *impl) error { + _, err := g.RevParse(context.Background(), "HEAD") + return err + }, + }, + { + name: "IsAncestor", + call: func(g *impl) error { + _, err := g.IsAncestor(context.Background(), "a", "b") + return err + }, + }, + { + name: "GetCommitTimeSecond", + call: func(g *impl) error { + _, err := g.GetCommitTimeSecond(context.Background(), "HEAD") + return err + }, + }, + { + name: "FileHashes", + call: func(g *impl) error { + _, err := g.FileHashes(context.Background(), "HEAD") + return err + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := &mockRunner{err: fatalErr} + g := &impl{directory: "/repo", runner: m, logger: zap.NewNop().Sugar()} + err := tt.call(g) + require.Error(t, err) + assert.ErrorIs(t, err, ErrFatal, "fatal exit code must wrap ErrFatal") + }) + } +} diff --git a/core/itg/graph/update.go b/core/itg/graph/update.go index b3dc3ede..457f4c15 100644 --- a/core/itg/graph/update.go +++ b/core/itg/graph/update.go @@ -85,7 +85,7 @@ func (g *OptimizedGraph) UpdateGraph( } } // compute hashes for source file, package group, and rule common targets - if err := computeAvailableHashes(sourceHasher, targets); err != nil { + if err := computeAvailableHashes(ctx, sourceHasher, targets); err != nil { return err } @@ -147,6 +147,7 @@ func (g *OptimizedGraph) upsertExternalRuleTarget(target *targethasher.Target, i // computeAvailableHashes computes hashes that are available without dep traversal. func computeAvailableHashes( + ctx context.Context, hasher targethasher.SourceHasher, targets map[string]*targethasher.Target, ) error { @@ -161,7 +162,7 @@ func computeAvailableHashes( h.Write([]byte(name)) hash = h.Sum(nil) case targethasher.SourceFileType: - h, err := hasher.HashSourceFile(target.SourceFile) + h, err := hasher.HashSourceFile(ctx, target.SourceFile) if err != nil { return err } diff --git a/core/itg/graph/update_test.go b/core/itg/graph/update_test.go index 720c9511..6c28bc7c 100644 --- a/core/itg/graph/update_test.go +++ b/core/itg/graph/update_test.go @@ -31,7 +31,7 @@ type fakeSourceHasher struct { err error } -func (f *fakeSourceHasher) HashSourceFile(_ *buildpb.SourceFile) ([]byte, error) { +func (f *fakeSourceHasher) HashSourceFile(_ context.Context, _ *buildpb.SourceFile) ([]byte, error) { return f.result, f.err } @@ -53,7 +53,7 @@ func TestComputeAvailableHashes(t *testing.T) { }, } - require.NoError(t, computeAvailableHashes(hasher, targets)) + require.NoError(t, computeAvailableHashes(context.Background(), hasher, targets)) assert.Equal(t, expected, targets[name].Hash) }) @@ -64,7 +64,7 @@ func TestComputeAvailableHashes(t *testing.T) { name: {Name: name, RuleType: targethasher.PackageGroup}, } - require.NoError(t, computeAvailableHashes(&fakeSourceHasher{}, targets)) + require.NoError(t, computeAvailableHashes(context.Background(), &fakeSourceHasher{}, targets)) h := sha1.New() h.Write([]byte(name)) @@ -78,7 +78,7 @@ func TestComputeAvailableHashes(t *testing.T) { name: {Name: name, RuleType: targethasher.ExternalRuleType}, } - require.NoError(t, computeAvailableHashes(&fakeSourceHasher{}, targets)) + require.NoError(t, computeAvailableHashes(context.Background(), &fakeSourceHasher{}, targets)) assert.Nil(t, targets[name].Hash, "external rule targets should not get a hash here") }) @@ -89,7 +89,7 @@ func TestComputeAvailableHashes(t *testing.T) { name: {Name: name, RuleType: targethasher.GeneratedFileType}, } - require.NoError(t, computeAvailableHashes(&fakeSourceHasher{}, targets)) + require.NoError(t, computeAvailableHashes(context.Background(), &fakeSourceHasher{}, targets)) assert.Nil(t, targets[name].Hash, "generated file hash is resolved later") }) @@ -106,7 +106,7 @@ func TestComputeAvailableHashes(t *testing.T) { }, } - require.NoError(t, computeAvailableHashes(&fakeSourceHasher{}, targets)) + require.NoError(t, computeAvailableHashes(context.Background(), &fakeSourceHasher{}, targets)) assert.NotNil(t, targets[name].HashWithoutDeps, "rule should have HashWithoutDeps after hashing") assert.Nil(t, targets[name].Hash, "full hash is not computed here — deps are needed") }) @@ -118,7 +118,7 @@ func TestComputeAvailableHashes(t *testing.T) { "//pkg:f": {Name: "//pkg:f", RuleType: targethasher.SourceFileType, SourceFile: &buildpb.SourceFile{}}, } - err := computeAvailableHashes(hasher, targets) + err := computeAvailableHashes(context.Background(), hasher, targets) assert.Error(t, err) }) } diff --git a/core/repomanager/repo_manager.go b/core/repomanager/repo_manager.go index 196f265b..2f25fc8d 100644 --- a/core/repomanager/repo_manager.go +++ b/core/repomanager/repo_manager.go @@ -35,7 +35,8 @@ import ( // Sentinel errors for classification by upper layers. var ( // ErrPoolTimeout indicates that all worker slots were leased and the - // caller's context was cancelled while waiting for one to become available. + // caller's context was cancelled or its deadline elapsed while waiting + // for one to become available. ErrPoolTimeout = errors.New("worker pool timeout") ) @@ -174,10 +175,7 @@ func (r *repoManager) Lease(ctx context.Context, desc entity.BuildDescription) ( } recordStep(e, _stepWaitSlot, waitStart, metrics.FastDurationBuckets) if waitErr != nil { - if errors.Is(waitErr, context.DeadlineExceeded) { - return nil, fmt.Errorf("%w: %w", ErrPoolTimeout, waitErr) - } - return nil, fmt.Errorf("pool for repo %s: %w", repo, waitErr) + return nil, fmt.Errorf("%w: %w", ErrPoolTimeout, waitErr) } // Lazily create the worker clone on first use diff --git a/core/repomanager/repo_manager_test.go b/core/repomanager/repo_manager_test.go index c83e8fb1..4ba91d55 100644 --- a/core/repomanager/repo_manager_test.go +++ b/core/repomanager/repo_manager_test.go @@ -213,7 +213,7 @@ func TestLease_CtxCanceled(t *testing.T) { cancel() _, err = rm.Lease(ctx, entity.BuildDescription{Remote: remote}) require.Error(t, err) - assert.False(t, errors.Is(err, ErrPoolTimeout), "cancelled context should not produce ErrPoolTimeout") + assert.True(t, errors.Is(err, ErrPoolTimeout), "cancelled context should produce ErrPoolTimeout") assert.True(t, errors.Is(err, context.Canceled), "expected underlying context.Canceled") require.NoError(t, ws1.Release()) diff --git a/core/targethasher/graph.go b/core/targethasher/graph.go index b49f844e..5dbe4059 100644 --- a/core/targethasher/graph.go +++ b/core/targethasher/graph.go @@ -643,7 +643,7 @@ func HashRecursively(ctx context.Context, p HashParam) ([]byte, error) { switch target.RuleType { case SourceFileType: - h, err := p.Hasher.HashSourceFile(target.SourceFile) + h, err := p.Hasher.HashSourceFile(ctx, target.SourceFile) if err != nil { return nil, err } diff --git a/core/targethasher/graph_test.go b/core/targethasher/graph_test.go index 2cf745a9..034a04f6 100644 --- a/core/targethasher/graph_test.go +++ b/core/targethasher/graph_test.go @@ -300,8 +300,8 @@ func Test_fromProto(t *testing.T) { mockHasher := NewMockSourceHasher(ctrl) mockHasher.EXPECT(). - HashSourceFile(gomock.Any()). - DoAndReturn(func(s *buildpb.SourceFile) ([]byte, error) { + HashSourceFile(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, s *buildpb.SourceFile) ([]byte, error) { h := newHash() io.WriteString(h, s.GetName()) return h.Sum(nil), nil diff --git a/core/targethasher/mock_sourcehasher_test.go b/core/targethasher/mock_sourcehasher_test.go index 6cba64ed..f8f95afc 100755 --- a/core/targethasher/mock_sourcehasher_test.go +++ b/core/targethasher/mock_sourcehasher_test.go @@ -5,6 +5,7 @@ package targethasher import ( + context "context" reflect "reflect" build_proto "github.com/bazelbuild/buildtools/build_proto" @@ -35,16 +36,16 @@ func (m *MockSourceHasher) EXPECT() *MockSourceHasherMockRecorder { } // HashSourceFile mocks base method. -func (m *MockSourceHasher) HashSourceFile(arg0 *build_proto.SourceFile) ([]byte, error) { +func (m *MockSourceHasher) HashSourceFile(arg0 context.Context, arg1 *build_proto.SourceFile) ([]byte, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "HashSourceFile", arg0) + ret := m.ctrl.Call(m, "HashSourceFile", arg0, arg1) ret0, _ := ret[0].([]byte) ret1, _ := ret[1].(error) return ret0, ret1 } // HashSourceFile indicates an expected call of HashSourceFile. -func (mr *MockSourceHasherMockRecorder) HashSourceFile(arg0 interface{}) *gomock.Call { +func (mr *MockSourceHasherMockRecorder) HashSourceFile(arg0, arg1 interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HashSourceFile", reflect.TypeOf((*MockSourceHasher)(nil).HashSourceFile), arg0) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HashSourceFile", reflect.TypeOf((*MockSourceHasher)(nil).HashSourceFile), arg0, arg1) } diff --git a/core/targethasher/sourcehasher.go b/core/targethasher/sourcehasher.go index aef04f9e..c51181db 100644 --- a/core/targethasher/sourcehasher.go +++ b/core/targethasher/sourcehasher.go @@ -15,6 +15,7 @@ package targethasher import ( + "context" "crypto/sha1" "fmt" "hash" @@ -37,11 +38,16 @@ const ( _defaultSourceFileVisibility = "//visibility:private" ) +// cancelCheckInterval is how often the directory walk checks ctx.Err(), +// measured in files hashed. Matches the convention used elsewhere in the +// codebase (controller, mapper, bazel/stream). +const cancelCheckInterval = 1024 + // SourceHasher provides hashes for source nodes in the target graph. These // can be calculated based on disk contents or form other sources such as a // vcs system. type SourceHasher interface { - HashSourceFile(s *buildpb.SourceFile) ([]byte, error) + HashSourceFile(ctx context.Context, s *buildpb.SourceFile) ([]byte, error) } // diskHashHelper is a SourceHasher that provides hashes based on disk @@ -73,11 +79,11 @@ func NewSourceHasher(p Params) SourceHasher { } // HashSourceFile does a no-op hash for the noOpHasher. -func (hh *noOpHasher) HashSourceFile(sourceFile *buildpb.SourceFile) ([]byte, error) { +func (hh *noOpHasher) HashSourceFile(_ context.Context, sourceFile *buildpb.SourceFile) ([]byte, error) { return nil, nil } -func (hh *diskHashHelper) HashSourceFile(sourceFile *buildpb.SourceFile) ([]byte, error) { +func (hh *diskHashHelper) HashSourceFile(ctx context.Context, sourceFile *buildpb.SourceFile) ([]byte, error) { nonDefaultVisibilities := filterVisibilityLabels(sourceFile.GetVisibilityLabel()) // The location may look like /foo/decl.go:1:1 location, _, _ := strings.Cut(sourceFile.GetLocation(), ":") @@ -94,7 +100,7 @@ func (hh *diskHashHelper) HashSourceFile(sourceFile *buildpb.SourceFile) ([]byte var hash hash.Hash if fi.IsDir() { - hash, err = hashDir(location) + hash, err = hashDir(ctx, location) } else { hash, err = hashFile(location) } @@ -131,14 +137,21 @@ func hashFile(path string) (hash.Hash, error) { return hash, nil } -func hashDir(root string) (hash.Hash, error) { +func hashDir(ctx context.Context, root string) (hash.Hash, error) { dirHash := newHash() + var fileCount int walkDirFunc := func(path string, d fs.DirEntry, err error) error { if err != nil { return err } if d.Type().IsRegular() { + fileCount++ + if fileCount%cancelCheckInterval == 0 { + if err := ctx.Err(); err != nil { + return err + } + } fileHash, err := hashFile(path) if err != nil { return err diff --git a/core/targethasher/sourcehasher_test.go b/core/targethasher/sourcehasher_test.go index f6ffba58..d26c6127 100644 --- a/core/targethasher/sourcehasher_test.go +++ b/core/targethasher/sourcehasher_test.go @@ -15,6 +15,8 @@ package targethasher import ( + "context" + "fmt" "os" "path/filepath" "testing" @@ -41,7 +43,7 @@ func TestNewSourceHasher_buildsMaps(t *testing.T) { func TestNoOpHasher_returnsNil(t *testing.T) { h := &noOpHasher{} sf := &buildpb.SourceFile{Name: strPtr("//:dummy")} - got, err := h.HashSourceFile(sf) + got, err := h.HashSourceFile(context.Background(), sf) assert.NoError(t, err, "unexpected error: %v", err) assert.Nil(t, got, "expected nil hash, got %v", got) } @@ -68,7 +70,7 @@ func TestDiskHashHelper_KnownFileHashUsed(t *testing.T) { Location: strPtr(abs + ":1:1"), VisibilityLabel: []string{"//visibility:private"}, } - got, err := h.HashSourceFile(sf) + got, err := h.HashSourceFile(context.Background(), sf) assert.NoError(t, err, "unexpected err: %v", err) assert.Equal(t, string(known), string(got), "expected known hash %q, got %q", known, got) } @@ -94,7 +96,7 @@ func TestDiskHashHelper_NonDefaultVisibilityForcesDiskHash(t *testing.T) { Location: strPtr(abs + ":1:1"), VisibilityLabel: []string{"//visibility:public"}, // non-default } - got, err := h.HashSourceFile(sf) + got, err := h.HashSourceFile(context.Background(), sf) assert.NoError(t, err, "unexpected err: %v", err) assert.NotEqual(t, string(got), "KNOWN", "expected disk hash, but got known hash") assert.NotEqual(t, []byte{}, got, "expected non-empty disk hash") @@ -121,7 +123,7 @@ func TestDiskHashHelper_HashesFileFromDisk(t *testing.T) { Location: strPtr(abs), VisibilityLabel: []string{"//visibility:private"}, } - got, err := h.HashSourceFile(sf) + got, err := h.HashSourceFile(context.Background(), sf) assert.NoError(t, err, "unexpected err: %v", err) assert.NotEmpty(t, got, "expected non-empty hash from disk") } @@ -150,9 +152,28 @@ func TestDiskHashHelper_HashesDirectory(t *testing.T) { Location: strPtr(dirAbs), VisibilityLabel: []string{"//visibility:private"}, } - got, err := h.HashSourceFile(sf) + got, err := h.HashSourceFile(context.Background(), sf) assert.NoError(t, err, "unexpected err: %v", err) assert.NotEmpty(t, got, "expected non-empty hash for directory") } func strPtr(s string) *string { return &s } + +func TestHashDir_RespectsCtxCancellation(t *testing.T) { + // Create a directory with enough files to trigger a cancellation check. + tmp := t.TempDir() + for i := range cancelCheckInterval + 1 { + require.NoError(t, os.WriteFile( + filepath.Join(tmp, fmt.Sprintf("file%d.txt", i)), + []byte("x"), + 0o644, + )) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := hashDir(ctx, tmp) + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled) +} diff --git a/core/workspace/workspace.go b/core/workspace/workspace.go index a084fea1..931b47ca 100644 --- a/core/workspace/workspace.go +++ b/core/workspace/workspace.go @@ -17,12 +17,15 @@ package workspace import ( "context" "fmt" + "sync" "github.com/uber/tango/core/git" "go.uber.org/zap" ) -// Workspace defines interface for workspace +// Workspace defines interface for workspace. +// Release is idempotent: the first call invokes the release callback (if any), +// and subsequent calls are no-ops. type Workspace interface { Path() string Checkout(ctx context.Context, remote string, ref string) error @@ -31,10 +34,11 @@ type Workspace interface { } type workspace struct { - path string - git git.Interface - logger *zap.SugaredLogger - onRelease func() // optional callback invoked on Release + path string + git git.Interface + logger *zap.SugaredLogger + onRelease func() // optional callback invoked on Release + releaseOnce sync.Once } type WorkspaceParams struct { @@ -90,10 +94,13 @@ func (w *workspace) Checkout(ctx context.Context, remote string, ref string) err } // Release invokes the onRelease callback if set (e.g., to return the -// workspace to a pool), otherwise it's a no-op. +// workspace to a pool), otherwise it's a no-op. Release is idempotent: +// only the first call invokes the callback. func (w *workspace) Release() error { - if w.onRelease != nil { - w.onRelease() - } + w.releaseOnce.Do(func() { + if w.onRelease != nil { + w.onRelease() + } + }) return nil } diff --git a/core/workspace/workspace_test.go b/core/workspace/workspace_test.go index f85c051c..3bbff4c8 100644 --- a/core/workspace/workspace_test.go +++ b/core/workspace/workspace_test.go @@ -115,3 +115,23 @@ func TestWorkspace_Release(t *testing.T) { err := w.Release() require.NoError(t, err) } + +func TestWorkspace_Release_Idempotent(t *testing.T) { + // The onRelease callback must be invoked exactly once even if Release + // is called multiple times (e.g. double-release must not push the same + // slot back into a pool twice). + calls := make(chan struct{}, 2) + w := NewWorkspace(WorkspaceParams{ + Path: "/tmp/ws", + Git: gitmock.NewMockInterface(gomock.NewController(t)), + OnRelease: func() { + calls <- struct{}{} + }, + }) + + require.NoError(t, w.Release()) + require.NoError(t, w.Release()) + + // Exactly one callback invocation expected. + assert.Len(t, calls, 1, "onRelease must be called exactly once") +}