1 files changed, 497 insertions, 0 deletions

diff --git a/cli/internal/taskhash/taskhash.go b/cli/internal/taskhash/taskhash.go
new file mode 100644
index 0000000..a912ad9
--- /dev/null
+++ b/cli/internal/taskhash/taskhash.go
@@ -0,0 +1,497 @@
+// Package taskhash handles calculating dependency hashes for nodes in the task execution graph.
+package taskhash
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/hashicorp/go-hclog"
+	"github.com/pyr-sh/dag"
+	gitignore "github.com/sabhiram/go-gitignore"
+	"github.com/vercel/turbo/cli/internal/doublestar"
+	"github.com/vercel/turbo/cli/internal/env"
+	"github.com/vercel/turbo/cli/internal/fs"
+	"github.com/vercel/turbo/cli/internal/hashing"
+	"github.com/vercel/turbo/cli/internal/inference"
+	"github.com/vercel/turbo/cli/internal/nodes"
+	"github.com/vercel/turbo/cli/internal/runsummary"
+	"github.com/vercel/turbo/cli/internal/turbopath"
+	"github.com/vercel/turbo/cli/internal/util"
+	"github.com/vercel/turbo/cli/internal/workspace"
+	"golang.org/x/sync/errgroup"
+)
+
+// Tracker caches package-inputs hashes, as well as package-task hashes.
+// package-inputs hashes must be calculated before package-task hashes,
+// and package-task hashes must be calculated in topographical order.
+// package-task hashing is threadsafe, provided topographical order is
+// respected.
+type Tracker struct {
+	rootNode   string
+	globalHash string
+	pipeline   fs.Pipeline
+
+	packageInputsHashes packageFileHashes
+
+	// packageInputsExpandedHashes is a map of a hashkey to a list of files that are inputs to the task.
+	// Writes to this map happen during CalculateFileHash(). Since this happens synchronously
+	// before walking the task graph, it does not need to be protected by a mutex.
+	packageInputsExpandedHashes map[packageFileHashKey]map[turbopath.AnchoredUnixPath]string
+
+	// mu is a mutex that we can lock/unlock to read/write from maps
+	// the fields below should be protected by the mutex.
+	mu                     sync.RWMutex
+	packageTaskEnvVars     map[string]env.DetailedMap // taskId -> envvar pairs that affect the hash.
+	packageTaskHashes      map[string]string          // taskID -> hash
+	packageTaskFramework   map[string]string          // taskID -> inferred framework for package
+	packageTaskOutputs     map[string][]turbopath.AnchoredSystemPath
+	packageTaskCacheStatus map[string]runsummary.TaskCacheSummary
+}
+
+// NewTracker creates a tracker for package-inputs combinations and package-task combinations.
+func NewTracker(rootNode string, globalHash string, pipeline fs.Pipeline) *Tracker {
+	return &Tracker{
+		rootNode:               rootNode,
+		globalHash:             globalHash,
+		pipeline:               pipeline,
+		packageTaskHashes:      make(map[string]string),
+		packageTaskFramework:   make(map[string]string),
+		packageTaskEnvVars:     make(map[string]env.DetailedMap),
+		packageTaskOutputs:     make(map[string][]turbopath.AnchoredSystemPath),
+		packageTaskCacheStatus: make(map[string]runsummary.TaskCacheSummary),
+	}
+}
+
+// packageFileSpec defines a combination of a package and optional set of input globs
+type packageFileSpec struct {
+	pkg    string
+	inputs []string
+}
+
+func specFromPackageTask(packageTask *nodes.PackageTask) packageFileSpec {
+	return packageFileSpec{
+		pkg:    packageTask.PackageName,
+		inputs: packageTask.TaskDefinition.Inputs,
+	}
+}
+
+// packageFileHashKey is a hashable representation of a packageFileSpec.
+type packageFileHashKey string
+
+// hashes the inputs for a packageTask
+func (pfs packageFileSpec) ToKey() packageFileHashKey {
+	sort.Strings(pfs.inputs)
+	return packageFileHashKey(fmt.Sprintf("%v#%v", pfs.pkg, strings.Join(pfs.inputs, "!")))
+}
+
+func safeCompileIgnoreFile(filepath string) (*gitignore.GitIgnore, error) {
+	if fs.FileExists(filepath) {
+		return gitignore.CompileIgnoreFile(filepath)
+	}
+	// no op
+	return gitignore.CompileIgnoreLines([]string{}...), nil
+}
+
+func (pfs *packageFileSpec) getHashObject(pkg *fs.PackageJSON, repoRoot turbopath.AbsoluteSystemPath) map[turbopath.AnchoredUnixPath]string {
+	hashObject, pkgDepsErr := hashing.GetPackageDeps(repoRoot, &hashing.PackageDepsOptions{
+		PackagePath:   pkg.Dir,
+		InputPatterns: pfs.inputs,
+	})
+	if pkgDepsErr != nil {
+		manualHashObject, err := manuallyHashPackage(pkg, pfs.inputs, repoRoot)
+		if err != nil {
+			return make(map[turbopath.AnchoredUnixPath]string)
+		}
+		hashObject = manualHashObject
+	}
+
+	return hashObject
+}
+
+func (pfs *packageFileSpec) hash(hashObject map[turbopath.AnchoredUnixPath]string) (string, error) {
+	hashOfFiles, otherErr := fs.HashObject(hashObject)
+	if otherErr != nil {
+		return "", otherErr
+	}
+	return hashOfFiles, nil
+}
+
+func manuallyHashPackage(pkg *fs.PackageJSON, inputs []string, rootPath turbopath.AbsoluteSystemPath) (map[turbopath.AnchoredUnixPath]string, error) {
+	hashObject := make(map[turbopath.AnchoredUnixPath]string)
+	// Instead of implementing all gitignore properly, we hack it. We only respect .gitignore in the root and in
+	// the directory of a package.
+	ignore, err := safeCompileIgnoreFile(rootPath.UntypedJoin(".gitignore").ToString())
+	if err != nil {
+		return nil, err
+	}
+
+	ignorePkg, err := safeCompileIgnoreFile(rootPath.UntypedJoin(pkg.Dir.ToStringDuringMigration(), ".gitignore").ToString())
+	if err != nil {
+		return nil, err
+	}
+
+	pathPrefix := rootPath.UntypedJoin(pkg.Dir.ToStringDuringMigration())
+	includePattern := ""
+	excludePattern := ""
+	if len(inputs) > 0 {
+		var includePatterns []string
+		var excludePatterns []string
+		for _, pattern := range inputs {
+			if len(pattern) > 0 && pattern[0] == '!' {
+				excludePatterns = append(excludePatterns, pathPrefix.UntypedJoin(pattern[1:]).ToString())
+			} else {
+				includePatterns = append(includePatterns, pathPrefix.UntypedJoin(pattern).ToString())
+			}
+		}
+		if len(includePatterns) > 0 {
+			includePattern = "{" + strings.Join(includePatterns, ",") + "}"
+		}
+		if len(excludePatterns) > 0 {
+			excludePattern = "{" + strings.Join(excludePatterns, ",") + "}"
+		}
+	}
+
+	err = fs.Walk(pathPrefix.ToStringDuringMigration(), func(name string, isDir bool) error {
+		convertedName := turbopath.AbsoluteSystemPathFromUpstream(name)
+		rootMatch := ignore.MatchesPath(convertedName.ToString())
+		otherMatch := ignorePkg.MatchesPath(convertedName.ToString())
+		if !rootMatch && !otherMatch {
+			if !isDir {
+				if includePattern != "" {
+					val, err := doublestar.PathMatch(includePattern, convertedName.ToString())
+					if err != nil {
+						return err
+					}
+					if !val {
+						return nil
+					}
+				}
+				if excludePattern != "" {
+					val, err := doublestar.PathMatch(excludePattern, convertedName.ToString())
+					if err != nil {
+						return err
+					}
+					if val {
+						return nil
+					}
+				}
+				hash, err := fs.GitLikeHashFile(convertedName.ToString())
+				if err != nil {
+					return fmt.Errorf("could not hash file %v. \n%w", convertedName.ToString(), err)
+				}
+
+				relativePath, err := convertedName.RelativeTo(pathPrefix)
+				if err != nil {
+					return fmt.Errorf("File path cannot be made relative: %w", err)
+				}
+				hashObject[relativePath.ToUnixPath()] = hash
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return hashObject, nil
+}
+
+// packageFileHashes is a map from a package and optional input globs to the hash of
+// the matched files in the package.
+type packageFileHashes map[packageFileHashKey]string
+
+// CalculateFileHashes hashes each unique package-inputs combination that is present
+// in the task graph. Must be called before calculating task hashes.
+func (th *Tracker) CalculateFileHashes(
+	allTasks []dag.Vertex,
+	workerCount int,
+	workspaceInfos workspace.Catalog,
+	taskDefinitions map[string]*fs.TaskDefinition,
+	repoRoot turbopath.AbsoluteSystemPath,
+) error {
+	hashTasks := make(util.Set)
+
+	for _, v := range allTasks {
+		taskID, ok := v.(string)
+		if !ok {
+			return fmt.Errorf("unknown task %v", taskID)
+		}
+		if taskID == th.rootNode {
+			continue
+		}
+		pkgName, _ := util.GetPackageTaskFromId(taskID)
+		if pkgName == th.rootNode {
+			continue
+		}
+
+		taskDefinition, ok := taskDefinitions[taskID]
+		if !ok {
+			return fmt.Errorf("missing pipeline entry %v", taskID)
+		}
+
+		pfs := &packageFileSpec{
+			pkg:    pkgName,
+			inputs: taskDefinition.Inputs,
+		}
+
+		hashTasks.Add(pfs)
+	}
+
+	hashes := make(map[packageFileHashKey]string, len(hashTasks))
+	hashObjects := make(map[packageFileHashKey]map[turbopath.AnchoredUnixPath]string, len(hashTasks))
+	hashQueue := make(chan *packageFileSpec, workerCount)
+	hashErrs := &errgroup.Group{}
+
+	for i := 0; i < workerCount; i++ {
+		hashErrs.Go(func() error {
+			for packageFileSpec := range hashQueue {
+				pkg, ok := workspaceInfos.PackageJSONs[packageFileSpec.pkg]
+				if !ok {
+					return fmt.Errorf("cannot find package %v", packageFileSpec.pkg)
+				}
+				hashObject := packageFileSpec.getHashObject(pkg, repoRoot)
+				hash, err := packageFileSpec.hash(hashObject)
+				if err != nil {
+					return err
+				}
+				th.mu.Lock()
+				pfsKey := packageFileSpec.ToKey()
+				hashes[pfsKey] = hash
+				hashObjects[pfsKey] = hashObject
+				th.mu.Unlock()
+			}
+			return nil
+		})
+	}
+	for ht := range hashTasks {
+		hashQueue <- ht.(*packageFileSpec)
+	}
+	close(hashQueue)
+	err := hashErrs.Wait()
+	if err != nil {
+		return err
+	}
+	th.packageInputsHashes = hashes
+	th.packageInputsExpandedHashes = hashObjects
+	return nil
+}
+
+type taskHashable struct {
+	packageDir           turbopath.AnchoredUnixPath
+	hashOfFiles          string
+	externalDepsHash     string
+	task                 string
+	outputs              fs.TaskOutputs
+	passThruArgs         []string
+	envMode              util.EnvMode
+	passthroughEnv       []string
+	hashableEnvPairs     []string
+	globalHash           string
+	taskDependencyHashes []string
+}
+
+type oldTaskHashable struct {
+	packageDir           turbopath.AnchoredUnixPath
+	hashOfFiles          string
+	externalDepsHash     string
+	task                 string
+	outputs              fs.TaskOutputs
+	passThruArgs         []string
+	hashableEnvPairs     []string
+	globalHash           string
+	taskDependencyHashes []string
+}
+
+// calculateTaskHashFromHashable returns a hash string from the taskHashable
+func calculateTaskHashFromHashable(full *taskHashable, useOldTaskHashable bool) (string, error) {
+	// The user is not using the strict environment variables feature.
+	if useOldTaskHashable {
+		return fs.HashObject(&oldTaskHashable{
+			packageDir:           full.packageDir,
+			hashOfFiles:          full.hashOfFiles,
+			externalDepsHash:     full.externalDepsHash,
+			task:                 full.task,
+			outputs:              full.outputs,
+			passThruArgs:         full.passThruArgs,
+			hashableEnvPairs:     full.hashableEnvPairs,
+			globalHash:           full.globalHash,
+			taskDependencyHashes: full.taskDependencyHashes,
+		})
+	}
+
+	switch full.envMode {
+	case util.Loose:
+		// Remove the passthroughs from hash consideration if we're explicitly loose.
+		full.passthroughEnv = nil
+		return fs.HashObject(full)
+	case util.Strict:
+		// Collapse `nil` and `[]` in strict mode.
+		if full.passthroughEnv == nil {
+			full.passthroughEnv = make([]string, 0)
+		}
+		return fs.HashObject(full)
+	case util.Infer:
+		panic("task inferred status should have already been resolved")
+	default:
+		panic("unimplemented environment mode")
+	}
+}
+
+func (th *Tracker) calculateDependencyHashes(dependencySet dag.Set) ([]string, error) {
+	dependencyHashSet := make(util.Set)
+
+	rootPrefix := th.rootNode + util.TaskDelimiter
+	th.mu.RLock()
+	defer th.mu.RUnlock()
+	for _, dependency := range dependencySet {
+		if dependency == th.rootNode {
+			continue
+		}
+		dependencyTask, ok := dependency.(string)
+		if !ok {
+			return nil, fmt.Errorf("unknown task: %v", dependency)
+		}
+		if strings.HasPrefix(dependencyTask, rootPrefix) {
+			continue
+		}
+		dependencyHash, ok := th.packageTaskHashes[dependencyTask]
+		if !ok {
+			return nil, fmt.Errorf("missing hash for dependent task: %v", dependencyTask)
+		}
+		dependencyHashSet.Add(dependencyHash)
+	}
+	dependenciesHashList := dependencyHashSet.UnsafeListOfStrings()
+	sort.Strings(dependenciesHashList)
+	return dependenciesHashList, nil
+}
+
+// CalculateTaskHash calculates the hash for package-task combination. It is threadsafe, provided
+// that it has previously been called on its task-graph dependencies. File hashes must be calculated
+// first.
+func (th *Tracker) CalculateTaskHash(packageTask *nodes.PackageTask, dependencySet dag.Set, logger hclog.Logger, args []string, useOldTaskHashable bool) (string, error) {
+	pfs := specFromPackageTask(packageTask)
+	pkgFileHashKey := pfs.ToKey()
+
+	hashOfFiles, ok := th.packageInputsHashes[pkgFileHashKey]
+	if !ok {
+		return "", fmt.Errorf("cannot find package-file hash for %v", pkgFileHashKey)
+	}
+
+	var keyMatchers []string
+	framework := inference.InferFramework(packageTask.Pkg)
+	if framework != nil && framework.EnvMatcher != "" {
+		// log auto detected framework and env prefix
+		logger.Debug(fmt.Sprintf("auto detected framework for %s", packageTask.PackageName), "framework", framework.Slug, "env_prefix", framework.EnvMatcher)
+		keyMatchers = append(keyMatchers, framework.EnvMatcher)
+	}
+
+	envVars, err := env.GetHashableEnvVars(
+		packageTask.TaskDefinition.EnvVarDependencies,
+		keyMatchers,
+		"TURBO_CI_VENDOR_ENV_KEY",
+	)
+	if err != nil {
+		return "", err
+	}
+	hashableEnvPairs := envVars.All.ToHashable()
+	outputs := packageTask.HashableOutputs()
+	taskDependencyHashes, err := th.calculateDependencyHashes(dependencySet)
+	if err != nil {
+		return "", err
+	}
+	// log any auto detected env vars
+	logger.Debug(fmt.Sprintf("task hash env vars for %s:%s", packageTask.PackageName, packageTask.Task), "vars", hashableEnvPairs)
+
+	hash, err := calculateTaskHashFromHashable(&taskHashable{
+		packageDir:           packageTask.Pkg.Dir.ToUnixPath(),
+		hashOfFiles:          hashOfFiles,
+		externalDepsHash:     packageTask.Pkg.ExternalDepsHash,
+		task:                 packageTask.Task,
+		outputs:              outputs.Sort(),
+		passThruArgs:         args,
+		envMode:              packageTask.EnvMode,
+		passthroughEnv:       packageTask.TaskDefinition.PassthroughEnv,
+		hashableEnvPairs:     hashableEnvPairs,
+		globalHash:           th.globalHash,
+		taskDependencyHashes: taskDependencyHashes,
+	}, useOldTaskHashable)
+	if err != nil {
+		return "", fmt.Errorf("failed to hash task %v: %v", packageTask.TaskID, hash)
+	}
+	th.mu.Lock()
+	th.packageTaskEnvVars[packageTask.TaskID] = envVars
+	th.packageTaskHashes[packageTask.TaskID] = hash
+	if framework != nil {
+		th.packageTaskFramework[packageTask.TaskID] = framework.Slug
+	}
+	th.mu.Unlock()
+	return hash, nil
+}
+
+// GetExpandedInputs gets the expanded set of inputs for a given PackageTask
+func (th *Tracker) GetExpandedInputs(packageTask *nodes.PackageTask) map[turbopath.AnchoredUnixPath]string {
+	pfs := specFromPackageTask(packageTask)
+	expandedInputs := th.packageInputsExpandedHashes[pfs.ToKey()]
+	inputsCopy := make(map[turbopath.AnchoredUnixPath]string, len(expandedInputs))
+
+	for path, hash := range expandedInputs {
+		inputsCopy[path] = hash
+	}
+
+	return inputsCopy
+}
+
+// GetEnvVars returns the hashed env vars for a given taskID
+func (th *Tracker) GetEnvVars(taskID string) env.DetailedMap {
+	th.mu.RLock()
+	defer th.mu.RUnlock()
+	return th.packageTaskEnvVars[taskID]
+}
+
+// GetFramework returns the inferred framework for a given taskID
+func (th *Tracker) GetFramework(taskID string) string {
+	th.mu.RLock()
+	defer th.mu.RUnlock()
+	return th.packageTaskFramework[taskID]
+}
+
+// GetExpandedOutputs returns a list of outputs for a given taskID
+func (th *Tracker) GetExpandedOutputs(taskID string) []turbopath.AnchoredSystemPath {
+	th.mu.RLock()
+	defer th.mu.RUnlock()
+	outputs, ok := th.packageTaskOutputs[taskID]
+
+	if !ok {
+		return []turbopath.AnchoredSystemPath{}
+	}
+
+	return outputs
+}
+
+// SetExpandedOutputs a list of outputs for a given taskID so it can be read later
+func (th *Tracker) SetExpandedOutputs(taskID string, outputs []turbopath.AnchoredSystemPath) {
+	th.mu.Lock()
+	defer th.mu.Unlock()
+	th.packageTaskOutputs[taskID] = outputs
+}
+
+// SetCacheStatus records the task status for the given taskID
+func (th *Tracker) SetCacheStatus(taskID string, cacheSummary runsummary.TaskCacheSummary) {
+	th.mu.Lock()
+	defer th.mu.Unlock()
+	th.packageTaskCacheStatus[taskID] = cacheSummary
+}
+
+// GetCacheStatus records the task status for the given taskID
+func (th *Tracker) GetCacheStatus(taskID string) runsummary.TaskCacheSummary {
+	th.mu.Lock()
+	defer th.mu.Unlock()
+
+	if status, ok := th.packageTaskCacheStatus[taskID]; ok {
+		return status
+	}
+
+	// Return an empty one, all the fields will be false and 0
+	return runsummary.TaskCacheSummary{}
+}