fix(core): close contract edges in git/repomanager/workspace/targethasher (audit #14)

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 <ref>`.
 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)

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")
+		})
+	}
+}

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
 			}

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)
 	})
 }

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

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())

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
 		}

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

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