tasks/patches/TASK_CR_013_source.patch

diff --git a/cache/cache.go b/cache/cache.go
index 0948b6de..a5a403bb 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -24,7 +24,7 @@ func (nopAdapter) GetGoCloudURL(ctx context.Context, upload bool) (GoCloudURL, e
 
 var createAdapter = getCreateAdapter
 
-func GetAdapter(config *cacheconfig.Config, timeout time.Duration, shortToken, projectId, key string) Adapter {
+func GetAdapter(config *cacheconfig.Config, timeout time.Duration, shortToken, projectId, key string, sharded bool) Adapter {
 	if config == nil {
 		return nopAdapter{}
 	}
@@ -42,7 +42,22 @@ func GetAdapter(config *cacheconfig.Config, timeout time.Duration, shortToken, p
 		namespace = path.Join("runner", shortToken)
 	}
 	basePath := path.Join(config.GetPath(), namespace, "project", projectId)
-	fullPath := path.Join(basePath, key)
+
+	// When sharded (i.e. FF_HASH_CACHE_KEYS is enabled), insert the first two
+	// hex characters of the key as an intermediate path component. This
+	// distributes objects across 256 distinct S3 prefixes per project, avoiding
+	// 503 Slow Down responses caused by all cache objects sharing the same
+	// prefix and landing on the same partition.
+	var fullPath string
+	if sharded {
+		if len(key) < 2 {
+			logrus.WithError(fmt.Errorf("cache key too short to shard (length %d)", len(key))).Error("Error while generating cache bucket.")
+			return nopAdapter{}
+		}
+		fullPath = path.Join(basePath, key[:2], key)
+	} else {
+		fullPath = path.Join(basePath, key)
+	}
 
 	// The typical concerns regarding the use of strings.HasPrefix to detect
 	// path traversal do not apply here. The detection here is made easier
diff --git a/common/build_step_dispatch.go b/common/build_step_dispatch.go
index 7120df28..898c2414 100644
--- a/common/build_step_dispatch.go
+++ b/common/build_step_dispatch.go
@@ -166,7 +166,7 @@ func stagesToConcreteStep(ctx context.Context, executor Executor) ([]schema.Step
 	if build.Runner.Cache != nil {
 		opts = append(opts, builder.WithCacheMaxArchiveSize(build.Runner.Cache.MaxUploadedArchiveSize),
 			builder.WithCacheDownloadDescriptor(func(cacheKey string) (cacheprovider.Descriptor, error) {
-				adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey)
+				adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey, build.IsFeatureFlagOn(featureflags.HashCacheKeys))
 
 				goCloudURL, err := adapter.GetGoCloudURL(ctx, false)
 				if goCloudURL.URL != nil {
@@ -187,7 +187,7 @@ func stagesToConcreteStep(ctx context.Context, executor Executor) ([]schema.Step
 				return cacheprovider.Descriptor{}, nil
 			}),
 			builder.WithCacheUploadDescriptor(func(cacheKey string) (cacheprovider.Descriptor, error) {
-				adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey)
+				adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey, build.IsFeatureFlagOn(featureflags.HashCacheKeys))
 
 				goCloudURL, err := adapter.GetGoCloudURL(ctx, true)
 				if err != nil {
diff --git a/shells/abstract.go b/shells/abstract.go
index a8b242ad..70bae0e8 100644
--- a/shells/abstract.go
+++ b/shells/abstract.go
@@ -379,7 +379,7 @@ func (b *AbstractShell) addExtractCacheCommand(
 // getCacheDownloadURLAndEnv will first try to generate the GoCloud URL if it's
 // available then fallback to a pre-signed URL.
 func getCacheDownloadURLAndEnv(ctx context.Context, build *common.Build, cacheKey string) ([]string, map[string]string, error) {
-	adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey)
+	adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey, build.IsFeatureFlagOn(featureflags.HashCacheKeys))
 
 	// Prefer Go Cloud URL if supported
 	goCloudURL, err := adapter.GetGoCloudURL(ctx, false)
@@ -1458,7 +1458,7 @@ func (b *AbstractShell) addCacheUploadCommand(
 // getCacheUploadURLAndEnv will first try to generate the GoCloud URL if it's
 // available then fallback to a pre-signed URL.
 func getCacheUploadURLAndEnv(ctx context.Context, build *common.Build, cacheKey string, metadata map[string]string) ([]string, map[string]string, error) {
-	adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey)
+	adapter := cache.GetAdapter(build.Runner.Cache, build.GetBuildTimeout(), build.Runner.ShortDescription(), fmt.Sprintf("%d", build.JobInfo.ProjectID), cacheKey, build.IsFeatureFlagOn(featureflags.HashCacheKeys))
 
 	// Prefer Go Cloud URL if supported
 	goCloudURL, err := adapter.GetGoCloudURL(ctx, true)