fix: hydration errors not set on applications (#24755)

Signed-off-by: Alexandre Gaudreault <alexandre_gaudreault@intuit.com>
Co-authored-by: Michael Crenshaw <350466+crenshaw-dev@users.noreply.github.com>

controller/appcontroller.go

@@ -1878,7 +1878,7 @@ func (ctrl *ApplicationController) processAppHydrateQueueItem() (processNext boo
 		return processNext
 	}
 
-	ctrl.hydrator.ProcessAppHydrateQueueItem(origApp)
+	ctrl.hydrator.ProcessAppHydrateQueueItem(origApp.DeepCopy())
 
 	log.WithFields(applog.GetAppLogFields(origApp)).Debug("Successfully processed app hydrate queue item")
 	return processNext

controller/hydrator/hydrator.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"maps"
 	"path/filepath"
+	"slices"
 	"sync"
 	"time"
 
@@ -99,47 +101,41 @@ func NewHydrator(dependencies Dependencies, statusRefreshTimeout time.Duration,
 // It's likely that multiple applications will trigger hydration at the same time. The hydration queue key is meant to
 // dedupe these requests.
 func (h *Hydrator) ProcessAppHydrateQueueItem(origApp *appv1.Application) {
-	origApp = origApp.DeepCopy()
 	app := origApp.DeepCopy()
-
 	if app.Spec.SourceHydrator == nil {
 		return
 	}
 
 	logCtx := log.WithFields(applog.GetAppLogFields(app))
-
 	logCtx.Debug("Processing app hydrate queue item")
 
-	// TODO: don't reuse statusRefreshTimeout. Create a new timeout for hydration.
-	needsHydration, reason := appNeedsHydration(origApp, h.statusRefreshTimeout)
-	if !needsHydration {
-		return
+	needsHydration, reason := appNeedsHydration(app)
+	if needsHydration {
+		app.Status.SourceHydrator.CurrentOperation = &appv1.HydrateOperation{
+			StartedAt:      metav1.Now(),
+			FinishedAt:     nil,
+			Phase:          appv1.HydrateOperationPhaseHydrating,
+			SourceHydrator: *app.Spec.SourceHydrator,
+		}
+		h.dependencies.PersistAppHydratorStatus(origApp, &app.Status.SourceHydrator)
 	}
 
-	logCtx.WithField("reason", reason).Info("Hydrating app")
-
-	app.Status.SourceHydrator.CurrentOperation = &appv1.HydrateOperation{
-		StartedAt:      metav1.Now(),
-		FinishedAt:     nil,
-		Phase:          appv1.HydrateOperationPhaseHydrating,
-		SourceHydrator: *app.Spec.SourceHydrator,
+	needsRefresh := app.Status.SourceHydrator.CurrentOperation.Phase == appv1.HydrateOperationPhaseHydrating && metav1.Now().Sub(app.Status.SourceHydrator.CurrentOperation.StartedAt.Time) > h.statusRefreshTimeout
+	if needsHydration || needsRefresh {
+		logCtx.WithField("reason", reason).Info("Hydrating app")
+		h.dependencies.AddHydrationQueueItem(getHydrationQueueKey(app))
+	} else {
+		logCtx.WithField("reason", reason).Debug("Skipping hydration")
 	}
-	h.dependencies.PersistAppHydratorStatus(origApp, &app.Status.SourceHydrator)
-	origApp.Status.SourceHydrator = app.Status.SourceHydrator
-	h.dependencies.AddHydrationQueueItem(getHydrationQueueKey(app))
 
 	logCtx.Debug("Successfully processed app hydrate queue item")
 }
 
 func getHydrationQueueKey(app *appv1.Application) types.HydrationQueueKey {
-	destinationBranch := app.Spec.SourceHydrator.SyncSource.TargetBranch
-	if app.Spec.SourceHydrator.HydrateTo != nil {
-		destinationBranch = app.Spec.SourceHydrator.HydrateTo.TargetBranch
-	}
 	key := types.HydrationQueueKey{
 		SourceRepoURL:        git.NormalizeGitURLAllowInvalid(app.Spec.SourceHydrator.DrySource.RepoURL),
 		SourceTargetRevision: app.Spec.SourceHydrator.DrySource.TargetRevision,
-		DestinationBranch:    destinationBranch,
+		DestinationBranch:    app.Spec.GetHydrateToSource().TargetRevision,
 	}
 	return key
 }
@@ -148,43 +144,92 @@ func getHydrationQueueKey(app *appv1.Application) types.HydrationQueueKey {
 // hydration key, hydrates their latest commit, and updates their status accordingly. If the hydration fails, it marks
 // the operation as failed and logs the error. If successful, it updates the operation to indicate that hydration was
 // successful and requests a refresh of the applications to pick up the new hydrated commit.
-func (h *Hydrator) ProcessHydrationQueueItem(hydrationKey types.HydrationQueueKey) (processNext bool) {
+func (h *Hydrator) ProcessHydrationQueueItem(hydrationKey types.HydrationQueueKey) {
 	logCtx := log.WithFields(log.Fields{
 		"sourceRepoURL":        hydrationKey.SourceRepoURL,
 		"sourceTargetRevision": hydrationKey.SourceTargetRevision,
 		"destinationBranch":    hydrationKey.DestinationBranch,
 	})
 
-	relevantApps, drySHA, hydratedSHA, err := h.hydrateAppsLatestCommit(logCtx, hydrationKey)
-	if len(relevantApps) == 0 {
-		// return early if there are no relevant apps found to hydrate
-		// otherwise you'll be stuck in hydrating
-		logCtx.Info("Skipping hydration since there are no relevant apps found to hydrate")
-		return processNext
+	// Get all applications sharing the same hydration key
+	apps, err := h.getAppsForHydrationKey(hydrationKey)
+	if err != nil {
+		// If we get an error here, we cannot proceed with hydration and we do not know
+		// which apps to update with the failure. The best we can do is log an error in
+		// the controller and wait for statusRefreshTimeout to retry
+		logCtx.WithError(err).Error("failed to get apps for hydration")
+		return
+	}
+	logCtx.WithField("appCount", len(apps))
+
+	// FIXME: we might end up in a race condition here where an HydrationQueueItem is processed
+	// before all applications had their CurrentOperation set by ProcessAppHydrateQueueItem.
+	// This would cause this method to update "old" CurrentOperation.
+	// It should only start hydration if all apps are in the HydrateOperationPhaseHydrating phase.
+	raceDetected := false
+	for _, app := range apps {
+		if app.Status.SourceHydrator.CurrentOperation == nil || app.Status.SourceHydrator.CurrentOperation.Phase != appv1.HydrateOperationPhaseHydrating {
+			raceDetected = true
+			break
+		}
+	}
+	if raceDetected {
+		logCtx.Warn("race condition detected: not all apps are in HydrateOperationPhaseHydrating phase")
+	}
+
+	// validate all the applications to make sure they are all correctly configured.
+	// All applications sharing the same hydration key must succeed for the hydration to be processed.
+	projects, validationErrors := h.validateApplications(apps)
+	if len(validationErrors) > 0 {
+		// For the applications that have an error, set the specific error in their status.
+		// Applications without error will still fail with a generic error since the hydration cannot be partial
+		genericError := genericHydrationError(validationErrors)
+		for _, app := range apps {
+			if err, ok := validationErrors[app.QualifiedName()]; ok {
+				logCtx = logCtx.WithFields(applog.GetAppLogFields(app))
+				logCtx.Errorf("failed to validate hydration app: %v", err)
+				h.setAppHydratorError(app, err)
+			} else {
+				h.setAppHydratorError(app, genericError)
+			}
+		}
+		return
+	}
+
+	// Hydrate all the apps
+	drySHA, hydratedSHA, appErrors, err := h.hydrate(logCtx, apps, projects)
+	if err != nil {
+		// If there is a single error, it affects each applications
+		for i := range apps {
+			appErrors[apps[i].QualifiedName()] = err
+		}
 	}
 	if drySHA != "" {
 		logCtx = logCtx.WithField("drySHA", drySHA)
 	}
-	if err != nil {
-		logCtx.WithField("appCount", len(relevantApps)).WithError(err).Error("Failed to hydrate apps")
-		for _, app := range relevantApps {
-			origApp := app.DeepCopy()
-			app.Status.SourceHydrator.CurrentOperation.Phase = appv1.HydrateOperationPhaseFailed
-			failedAt := metav1.Now()
-			app.Status.SourceHydrator.CurrentOperation.FinishedAt = &failedAt
-			app.Status.SourceHydrator.CurrentOperation.Message = fmt.Sprintf("Failed to hydrate revision %q: %v", drySHA, err.Error())
-			// We may or may not have gotten far enough in the hydration process to get a non-empty SHA, but set it just
-			// in case we did.
-			app.Status.SourceHydrator.CurrentOperation.DrySHA = drySHA
-			h.dependencies.PersistAppHydratorStatus(origApp, &app.Status.SourceHydrator)
-			logCtx = logCtx.WithFields(applog.GetAppLogFields(app))
-			logCtx.Errorf("Failed to hydrate app: %v", err)
+	if len(appErrors) > 0 {
+		// For the applications that have an error, set the specific error in their status.
+		// Applications without error will still fail with a generic error since the hydration cannot be partial
+		genericError := genericHydrationError(appErrors)
+		for _, app := range apps {
+			if drySHA != "" {
+				// If we have a drySHA, we can set it on the app status
+				app.Status.SourceHydrator.CurrentOperation.DrySHA = drySHA
+			}
+			if err, ok := appErrors[app.QualifiedName()]; ok {
+				logCtx = logCtx.WithFields(applog.GetAppLogFields(app))
+				logCtx.Errorf("failed to hydrate app: %v", err)
+				h.setAppHydratorError(app, err)
+			} else {
+				h.setAppHydratorError(app, genericError)
+			}
 		}
-		return processNext
+		return
 	}
-	logCtx.WithField("appCount", len(relevantApps)).Debug("Successfully hydrated apps")
+
+	logCtx.Debug("Successfully hydrated apps")
 	finishedAt := metav1.Now()
-	for _, app := range relevantApps {
+	for _, app := range apps {
 		origApp := app.DeepCopy()
 		operation := &appv1.HydrateOperation{
 			StartedAt:      app.Status.SourceHydrator.CurrentOperation.StartedAt,
@@ -202,118 +247,123 @@ func (h *Hydrator) ProcessHydrationQueueItem(hydrationKey types.HydrationQueueKe
 			SourceHydrator: app.Status.SourceHydrator.CurrentOperation.SourceHydrator,
 		}
 		h.dependencies.PersistAppHydratorStatus(origApp, &app.Status.SourceHydrator)
+
 		// Request a refresh since we pushed a new commit.
 		err := h.dependencies.RequestAppRefresh(app.Name, app.Namespace)
 		if err != nil {
-			logCtx.WithField("app", app.QualifiedName()).WithError(err).Error("Failed to request app refresh after hydration")
+			logCtx.WithFields(applog.GetAppLogFields(app)).WithError(err).Error("Failed to request app refresh after hydration")
 		}
 	}
-	return processNext
 }
 
-func (h *Hydrator) hydrateAppsLatestCommit(logCtx *log.Entry, hydrationKey types.HydrationQueueKey) ([]*appv1.Application, string, string, error) {
-	relevantApps, projects, err := h.getRelevantAppsAndProjectsForHydration(logCtx, hydrationKey)
-	if err != nil {
-		return nil, "", "", fmt.Errorf("failed to get relevant apps for hydration: %w", err)
+// setAppHydratorError updates the CurrentOperation with the error information.
+func (h *Hydrator) setAppHydratorError(app *appv1.Application, err error) {
+	// if the operation is not in progress, we do not update the status
+	if app.Status.SourceHydrator.CurrentOperation.Phase != appv1.HydrateOperationPhaseHydrating {
+		return
 	}
 
-	dryRevision, hydratedRevision, err := h.hydrate(logCtx, relevantApps, projects)
-	if err != nil {
-		return relevantApps, dryRevision, "", fmt.Errorf("failed to hydrate apps: %w", err)
-	}
-
-	return relevantApps, dryRevision, hydratedRevision, nil
+	origApp := app.DeepCopy()
+	app.Status.SourceHydrator.CurrentOperation.Phase = appv1.HydrateOperationPhaseFailed
+	failedAt := metav1.Now()
+	app.Status.SourceHydrator.CurrentOperation.FinishedAt = &failedAt
+	app.Status.SourceHydrator.CurrentOperation.Message = fmt.Sprintf("Failed to hydrate: %v", err.Error())
+	h.dependencies.PersistAppHydratorStatus(origApp, &app.Status.SourceHydrator)
 }
 
-func (h *Hydrator) getRelevantAppsAndProjectsForHydration(logCtx *log.Entry, hydrationKey types.HydrationQueueKey) ([]*appv1.Application, map[string]*appv1.AppProject, error) {
+// getAppsForHydrationKey returns the applications matching the hydration key.
+func (h *Hydrator) getAppsForHydrationKey(hydrationKey types.HydrationQueueKey) ([]*appv1.Application, error) {
 	// Get all apps
 	apps, err := h.dependencies.GetProcessableApps()
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to list apps: %w", err)
+		return nil, fmt.Errorf("failed to list apps: %w", err)
 	}
 
 	var relevantApps []*appv1.Application
-	projects := make(map[string]*appv1.AppProject)
-	uniquePaths := make(map[string]bool, len(apps.Items))
 	for _, app := range apps.Items {
 		if app.Spec.SourceHydrator == nil {
 			continue
 		}
-
-		if !git.SameURL(app.Spec.SourceHydrator.DrySource.RepoURL, hydrationKey.SourceRepoURL) ||
-			app.Spec.SourceHydrator.DrySource.TargetRevision != hydrationKey.SourceTargetRevision {
-			continue
-		}
-		destinationBranch := app.Spec.SourceHydrator.SyncSource.TargetBranch
-		if app.Spec.SourceHydrator.HydrateTo != nil {
-			destinationBranch = app.Spec.SourceHydrator.HydrateTo.TargetBranch
-		}
-		if destinationBranch != hydrationKey.DestinationBranch {
+		appKey := getHydrationQueueKey(&app)
+		if appKey != hydrationKey {
 			continue
 		}
+		relevantApps = append(relevantApps, &app)
+	}
+	return relevantApps, nil
+}
 
-		path := app.Spec.SourceHydrator.SyncSource.Path
-		// ensure that the path is always set to a path that doesn't resolve to the root of the repo
-		if IsRootPath(path) {
-			return nil, nil, fmt.Errorf("app %q has path %q which resolves to repository root", app.QualifiedName(), path)
-		}
+// validateApplications checks that all applications are valid for hydration.
+func (h *Hydrator) validateApplications(apps []*appv1.Application) (map[string]*appv1.AppProject, map[string]error) {
+	projects := make(map[string]*appv1.AppProject)
+	errors := make(map[string]error)
+	uniquePaths := make(map[string]string, len(apps))
 
-		var proj *appv1.AppProject
+	for _, app := range apps {
+		// Get the project for the app and validate if the app is allowed to use the source.
 		// We can't short-circuit this even if we have seen this project before, because we need to verify that this
-		// particular app is allowed to use this project. That logic is in GetProcessableAppProj.
-		proj, err = h.dependencies.GetProcessableAppProj(&app)
+		// particular app is allowed to use this project.
+		proj, err := h.dependencies.GetProcessableAppProj(app)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to get project %q for app %q: %w", app.Spec.Project, app.QualifiedName(), err)
+			errors[app.QualifiedName()] = fmt.Errorf("failed to get project %q: %w", app.Spec.Project, err)
+			continue
 		}
 		permitted := proj.IsSourcePermitted(app.Spec.GetSource())
 		if !permitted {
-			// Log and skip. We don't want to fail the entire operation because of one app.
-			logCtx.Warnf("App %q is not permitted to use source %q", app.QualifiedName(), app.Spec.Source.String())
+			errors[app.QualifiedName()] = fmt.Errorf("application repo %s is not permitted in project '%s'", app.Spec.GetSource().RepoURL, proj.Name)
 			continue
 		}
 		projects[app.Spec.Project] = proj
 
+		// Disallow hydrating to the repository root.
+		// Hydrating to root would overwrite or delete files at the top level of the repo,
+		// which can break other applications or shared configuration.
+		// Every hydrated app must write into a subdirectory instead.
+		destPath := app.Spec.SourceHydrator.SyncSource.Path
+		if IsRootPath(destPath) {
+			errors[app.QualifiedName()] = fmt.Errorf("app is configured to hydrate to the repository root (branch %q, path %q) which is not allowed", app.Spec.GetHydrateToSource().TargetRevision, destPath)
+			continue
+		}
+
 		// TODO: test the dupe detection
 		// TODO: normalize the path to avoid "path/.." from being treated as different from "."
-		if _, ok := uniquePaths[path]; ok {
-			return nil, nil, fmt.Errorf("multiple app hydrators use the same destination: %v", app.Spec.SourceHydrator.SyncSource.Path)
+		if appName, ok := uniquePaths[destPath]; ok {
+			errors[app.QualifiedName()] = fmt.Errorf("app %s hydrator use the same destination: %v", appName, app.Spec.SourceHydrator.SyncSource.Path)
+			errors[appName] = fmt.Errorf("app %s hydrator use the same destination: %v", app.QualifiedName(), app.Spec.SourceHydrator.SyncSource.Path)
+			continue
 		}
-		uniquePaths[path] = true
-
-		relevantApps = append(relevantApps, &app)
+		uniquePaths[destPath] = app.QualifiedName()
 	}
-	return relevantApps, projects, nil
+
+	// If there are any errors, return nil for projects to avoid possible partial processing.
+	if len(errors) > 0 {
+		projects = nil
+	}
+
+	return projects, errors
 }
 
-func (h *Hydrator) hydrate(logCtx *log.Entry, apps []*appv1.Application, projects map[string]*appv1.AppProject) (string, string, error) {
+func (h *Hydrator) hydrate(logCtx *log.Entry, apps []*appv1.Application, projects map[string]*appv1.AppProject) (string, string, map[string]error, error) {
+	errors := make(map[string]error)
 	if len(apps) == 0 {
-		return "", "", nil
+		return "", "", nil, nil
 	}
 
 	// These values are the same for all apps being hydrated together, so just get them from the first app.
-	repoURL := apps[0].Spec.SourceHydrator.DrySource.RepoURL
-	syncBranch := apps[0].Spec.SourceHydrator.SyncSource.TargetBranch
+	repoURL := apps[0].Spec.GetHydrateToSource().RepoURL
 	targetBranch := apps[0].Spec.GetHydrateToSource().TargetRevision
-
-	// Disallow hydrating to the repository root.
-	// Hydrating to root would overwrite or delete files at the top level of the repo,
-	// which can break other applications or shared configuration.
-	// Every hydrated app must write into a subdirectory instead.
-
-	for _, app := range apps {
-		destPath := app.Spec.SourceHydrator.SyncSource.Path
-		if IsRootPath(destPath) {
-			return "", "", fmt.Errorf(
-				"app %q is configured to hydrate to the repository root (branch %q, path %q) which is not allowed",
-				app.QualifiedName(), targetBranch, destPath,
-			)
-		}
-	}
+	// FIXME: As a convenience, the commit server will create the syncBranch if it does not exist. If the
+	// targetBranch does not exist, it will create it based on the syncBranch. On the next line, we take
+	// the `syncBranch` from the first app and assume that they're all configured the same. Instead, if any
+	// app has a different syncBranch, we should send the commit server an empty string and allow it to
+	// create the targetBranch as an orphan since we can't reliable determine a reasonable base.
+	syncBranch := apps[0].Spec.SourceHydrator.SyncSource.TargetBranch
 
 	// Get a static SHA revision from the first app so that all apps are hydrated from the same revision.
 	targetRevision, pathDetails, err := h.getManifests(context.Background(), apps[0], "", projects[apps[0].Spec.Project])
 	if err != nil {
-		return "", "", fmt.Errorf("failed to get manifests for app %q: %w", apps[0].QualifiedName(), err)
+		errors[apps[0].QualifiedName()] = fmt.Errorf("failed to get manifests: %w", err)
+		return "", "", errors, nil
 	}
 	paths := []*commitclient.PathDetails{pathDetails}
 
@@ -324,18 +374,18 @@ func (h *Hydrator) hydrate(logCtx *log.Entry, apps []*appv1.Application, project
 		app := app
 		eg.Go(func() error {
 			_, pathDetails, err = h.getManifests(ctx, app, targetRevision, projects[app.Spec.Project])
-			if err != nil {
-				return fmt.Errorf("failed to get manifests for app %q: %w", app.QualifiedName(), err)
-			}
 			mu.Lock()
+			defer mu.Unlock()
+			if err != nil {
+				errors[app.QualifiedName()] = fmt.Errorf("failed to get manifests: %w", err)
+				return errors[app.QualifiedName()]
+			}
 			paths = append(paths, pathDetails)
-			mu.Unlock()
 			return nil
 		})
 	}
-	err = eg.Wait()
-	if err != nil {
-		return "", "", fmt.Errorf("failed to get manifests for apps: %w", err)
+	if err := eg.Wait(); err != nil {
+		return targetRevision, "", errors, nil
 	}
 
 	// If all the apps are under the same project, use that project. Otherwise, use an empty string to indicate that we
@@ -344,18 +394,19 @@ func (h *Hydrator) hydrate(logCtx *log.Entry, apps []*appv1.Application, project
 	if len(projects) == 1 {
 		for p := range projects {
 			project = p
+			break
 		}
 	}
 
 	// Get the commit metadata for the target revision.
 	revisionMetadata, err := h.getRevisionMetadata(context.Background(), repoURL, project, targetRevision)
 	if err != nil {
-		return "", "", fmt.Errorf("failed to get revision metadata for %q: %w", targetRevision, err)
+		return targetRevision, "", errors, fmt.Errorf("failed to get revision metadata for %q: %w", targetRevision, err)
 	}
 
 	repo, err := h.dependencies.GetWriteCredentials(context.Background(), repoURL, project)
 	if err != nil {
-		return "", "", fmt.Errorf("failed to get hydrator credentials: %w", err)
+		return targetRevision, "", errors, fmt.Errorf("failed to get hydrator credentials: %w", err)
 	}
 	if repo == nil {
 		// Try without credentials.
@@ -367,11 +418,11 @@ func (h *Hydrator) hydrate(logCtx *log.Entry, apps []*appv1.Application, project
 	// get the commit message template
 	commitMessageTemplate, err := h.dependencies.GetHydratorCommitMessageTemplate()
 	if err != nil {
-		return "", "", fmt.Errorf("failed to get hydrated commit message template: %w", err)
+		return targetRevision, "", errors, fmt.Errorf("failed to get hydrated commit message template: %w", err)
 	}
 	commitMessage, errMsg := getTemplatedCommitMessage(repoURL, targetRevision, commitMessageTemplate, revisionMetadata)
 	if errMsg != nil {
-		return "", "", fmt.Errorf("failed to get hydrator commit templated message: %w", errMsg)
+		return targetRevision, "", errors, fmt.Errorf("failed to get hydrator commit templated message: %w", errMsg)
 	}
 
 	manifestsRequest := commitclient.CommitHydratedManifestsRequest{
@@ -386,14 +437,14 @@ func (h *Hydrator) hydrate(logCtx *log.Entry, apps []*appv1.Application, project
 
 	closer, commitService, err := h.commitClientset.NewCommitServerClient()
 	if err != nil {
-		return targetRevision, "", fmt.Errorf("failed to create commit service: %w", err)
+		return targetRevision, "", errors, fmt.Errorf("failed to create commit service: %w", err)
 	}
 	defer utilio.Close(closer)
 	resp, err := commitService.CommitHydratedManifests(context.Background(), &manifestsRequest)
 	if err != nil {
-		return targetRevision, "", fmt.Errorf("failed to commit hydrated manifests: %w", err)
+		return targetRevision, "", errors, fmt.Errorf("failed to commit hydrated manifests: %w", err)
 	}
-	return targetRevision, resp.HydratedSha, nil
+	return targetRevision, resp.HydratedSha, errors, nil
 }
 
 // getManifests gets the manifests for the given application and target revision. It returns the resolved revision
@@ -456,34 +507,27 @@ func (h *Hydrator) getRevisionMetadata(ctx context.Context, repoURL, project, re
 }
 
 // appNeedsHydration answers if application needs manifests hydrated.
-func appNeedsHydration(app *appv1.Application, statusHydrateTimeout time.Duration) (needsHydration bool, reason string) {
-	if app.Spec.SourceHydrator == nil {
-		return false, "source hydrator not configured"
-	}
-
-	var hydratedAt *metav1.Time
-	if app.Status.SourceHydrator.CurrentOperation != nil {
-		hydratedAt = &app.Status.SourceHydrator.CurrentOperation.StartedAt
-	}
-
+func appNeedsHydration(app *appv1.Application) (needsHydration bool, reason string) {
 	switch {
-	case app.IsHydrateRequested():
-		return true, "hydrate requested"
+	case app.Spec.SourceHydrator == nil:
+		return false, "source hydrator not configured"
 	case app.Status.SourceHydrator.CurrentOperation == nil:
 		return true, "no previous hydrate operation"
+	case app.Status.SourceHydrator.CurrentOperation.Phase == appv1.HydrateOperationPhaseHydrating:
+		return false, "hydration operation already in progress"
+	case app.IsHydrateRequested():
+		return true, "hydrate requested"
 	case !app.Spec.SourceHydrator.DeepEquals(app.Status.SourceHydrator.CurrentOperation.SourceHydrator):
 		return true, "spec.sourceHydrator differs"
 	case app.Status.SourceHydrator.CurrentOperation.Phase == appv1.HydrateOperationPhaseFailed && metav1.Now().Sub(app.Status.SourceHydrator.CurrentOperation.FinishedAt.Time) > 2*time.Minute:
 		return true, "previous hydrate operation failed more than 2 minutes ago"
-	case hydratedAt == nil || hydratedAt.Add(statusHydrateTimeout).Before(time.Now().UTC()):
-		return true, "hydration expired"
 	}
 
-	return false, ""
+	return false, "hydration not needed"
 }
 
-// Gets the multi-line commit message based on the template defined in the configmap. It is a two step process:
-// 1.  Get the metadata template engine would use to render the template
+// getTemplatedCommitMessage gets the multi-line commit message based on the template defined in the configmap. It is a two step process:
+// 1. Get the metadata template engine would use to render the template
 // 2. Pass the output of Step 1 and Step 2 to template Render
 func getTemplatedCommitMessage(repoURL, revision, commitMessageTemplate string, dryCommitMetadata *appv1.RevisionMetadata) (string, error) {
 	hydratorCommitMetadata, err := hydrator.GetCommitMetadata(repoURL, revision, dryCommitMetadata)
@@ -497,6 +541,20 @@ func getTemplatedCommitMessage(repoURL, revision, commitMessageTemplate string,
 	return templatedCommitMsg, nil
 }
 
+// genericHydrationError returns an error that summarizes the hydration errors for all applications.
+func genericHydrationError(validationErrors map[string]error) error {
+	if len(validationErrors) == 0 {
+		return nil
+	}
+
+	keys := slices.Sorted(maps.Keys(validationErrors))
+	remainder := "has an error"
+	if len(keys) > 1 {
+		remainder = fmt.Sprintf("and %d more have errors", len(keys)-1)
+	}
+	return fmt.Errorf("cannot hydrate because application %s %s", keys[0], remainder)
+}
+
 // IsRootPath returns whether the path references a root path
 func IsRootPath(path string) bool {
 	clean := filepath.Clean(path)

controller/hydrator/mocks/RepoGetter.go

@@ -0,0 +1,113 @@
+// Code generated by mockery; DO NOT EDIT.
+// github.com/vektra/mockery
+// template: testify
+
+package mocks
+
+import (
+	"context"
+
+	"github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
+	mock "github.com/stretchr/testify/mock"
+)
+
+// NewRepoGetter creates a new instance of RepoGetter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewRepoGetter(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *RepoGetter {
+	mock := &RepoGetter{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
+
+// RepoGetter is an autogenerated mock type for the RepoGetter type
+type RepoGetter struct {
+	mock.Mock
+}
+
+type RepoGetter_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *RepoGetter) EXPECT() *RepoGetter_Expecter {
+	return &RepoGetter_Expecter{mock: &_m.Mock}
+}
+
+// GetRepository provides a mock function for the type RepoGetter
+func (_mock *RepoGetter) GetRepository(ctx context.Context, repoURL string, project string) (*v1alpha1.Repository, error) {
+	ret := _mock.Called(ctx, repoURL, project)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetRepository")
+	}
+
+	var r0 *v1alpha1.Repository
+	var r1 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) (*v1alpha1.Repository, error)); ok {
+		return returnFunc(ctx, repoURL, project)
+	}
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) *v1alpha1.Repository); ok {
+		r0 = returnFunc(ctx, repoURL, project)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*v1alpha1.Repository)
+		}
+	}
+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string) error); ok {
+		r1 = returnFunc(ctx, repoURL, project)
+	} else {
+		r1 = ret.Error(1)
+	}
+	return r0, r1
+}
+
+// RepoGetter_GetRepository_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetRepository'
+type RepoGetter_GetRepository_Call struct {
+	*mock.Call
+}
+
+// GetRepository is a helper method to define mock.On call
+//   - ctx context.Context
+//   - repoURL string
+//   - project string
+func (_e *RepoGetter_Expecter) GetRepository(ctx interface{}, repoURL interface{}, project interface{}) *RepoGetter_GetRepository_Call {
+	return &RepoGetter_GetRepository_Call{Call: _e.mock.On("GetRepository", ctx, repoURL, project)}
+}
+
+func (_c *RepoGetter_GetRepository_Call) Run(run func(ctx context.Context, repoURL string, project string)) *RepoGetter_GetRepository_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 context.Context
+		if args[0] != nil {
+			arg0 = args[0].(context.Context)
+		}
+		var arg1 string
+		if args[1] != nil {
+			arg1 = args[1].(string)
+		}
+		var arg2 string
+		if args[2] != nil {
+			arg2 = args[2].(string)
+		}
+		run(
+			arg0,
+			arg1,
+			arg2,
+		)
+	})
+	return _c
+}
+
+func (_c *RepoGetter_GetRepository_Call) Return(repository *v1alpha1.Repository, err error) *RepoGetter_GetRepository_Call {
+	_c.Call.Return(repository, err)
+	return _c
+}
+
+func (_c *RepoGetter_GetRepository_Call) RunAndReturn(run func(ctx context.Context, repoURL string, project string) (*v1alpha1.Repository, error)) *RepoGetter_GetRepository_Call {
+	_c.Call.Return(run)
+	return _c
+}