fix(stages): requeue Stages when Freight is soaking in upstream

Auto-promotion silently stalled whenever a downstream Stage's source
specified requiredSoakTime:

- Regular Stages requeued on a fixed 5 minute cadence, so short soak
  windows (or soak deadlines that did not land on a 5-minute boundary)
  could be missed until a later tick happened to fall past the deadline.
- Control-flow Stages returned ctrl.Result{} with no RequeueAfter at
  all, relying entirely on watch events. No watch fires when 'soak time
  has elapsed', so a control-flow Stage with requiredSoakTime set never
  woke up on its own to re-evaluate the soak and verify its Freight,
  blocking every downstream Stage indefinitely.

Add a calculateNextSoakCheck helper that lists candidate Freight per
upstream source (with the soak filter disabled), computes the remaining
soak per upstream Stage according to the configured
AvailabilityStrategy, and returns the soonest deadline plus a 1 second
buffer.

Wire the helper into both reconcilers:

- The regular reconciler now requeues at min(5m, soakDeadline).
- The control-flow reconciler requeues at the soak deadline when one
  exists, and otherwise keeps the previous watch-driven behavior.

Refs: #4586
Signed-off-by: Justin Chen <mail@justin0u0.com>

Author: justin0u0

pkg/controller/stages/control_flow_stages.go

@@ -259,6 +259,19 @@ func (r *ControlFlowStageReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		}
 		return ctrl.Result{}, fmt.Errorf("failed to update Stage status: %w", err)
 	}
+
+	// Control-flow Stages are normally driven entirely by watches, but when a
+	// source has a RequiredSoakTime, no watch fires once the soak deadline
+	// elapses. In that case we must explicitly schedule a requeue so the soak
+	// can be re-evaluated and the Freight verified for downstream consumers.
+	// See: https://github.com/akuity/kargo/issues/4586
+	if reconcileErr == nil {
+		if soakRequeue, soakErr := calculateNextSoakCheck(ctx, r.client, stage); soakErr != nil {
+			logger.Error(soakErr, "failed to calculate next soak check interval")
+		} else if soakRequeue > 0 {
+			return ctrl.Result{RequeueAfter: soakRequeue}, nil
+		}
+	}
 	return ctrl.Result{}, reconcileErr
 }
 

pkg/controller/stages/regular_stages.go

@@ -394,7 +394,17 @@ func (r *RegularStageReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 	// Otherwise, requeue after a delay.
 	// TODO: Make the requeue delay configurable.
-	return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil
+	requeueAfter := 5 * time.Minute
+
+	// If any candidate Freight is still soaking in an upstream Stage, requeue
+	// sooner so auto-promotion can fire as soon as the soak time elapses
+	// rather than waiting for the next default-interval tick.
+	if soakRequeue, soakErr := calculateNextSoakCheck(ctx, r.client, stage); soakErr != nil {
+		logger.Error(soakErr, "failed to calculate next soak check interval")
+	} else if soakRequeue > 0 && soakRequeue < requeueAfter {
+		requeueAfter = soakRequeue
+	}
+	return ctrl.Result{RequeueAfter: requeueAfter}, nil
 }
 
 func (r *RegularStageReconciler) reconcile(

pkg/controller/stages/soak.go

@@ -0,0 +1,143 @@
+package stages
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kargoapi "github.com/akuity/kargo/api/v1alpha1"
+	"github.com/akuity/kargo/pkg/api"
+)
+
+// soakRequeueBuffer adds a small delay to avoid requeuing immediately after
+// the soak time elapses to avoid getting no available Freight.
+const soakRequeueBuffer = time.Second
+
+// calculateNextSoakCheck computes how long the reconciler should wait before
+// re-evaluating the given Stage for auto-promotion when one or more pieces of
+// candidate Freight are still soaking in upstream Stages.
+//
+// It returns 0 when no soak deadline is pending.
+func calculateNextSoakCheck(
+	ctx context.Context,
+	c client.Client,
+	stage *kargoapi.Stage,
+) (time.Duration, error) {
+	var soonest time.Duration
+
+	for _, req := range stage.Spec.RequestedFreight {
+		if req.Sources.RequiredSoakTime == nil ||
+			req.Sources.RequiredSoakTime.Duration <= 0 ||
+			len(req.Sources.Stages) == 0 {
+			continue
+		}
+
+		warehouse, err := api.GetWarehouse(ctx, c, client.ObjectKey{
+			Namespace: stage.Namespace,
+			Name:      req.Origin.Name,
+		})
+		if err != nil {
+			return 0, fmt.Errorf("get Warehouse %q: %w", req.Origin.Name, err)
+		}
+		if warehouse == nil {
+			continue
+		}
+
+		// List candidates with the soak filter disabled so we can see Freight
+		// that is verified in upstream Stages but still soaking.
+		candidates, err := api.ListFreightFromWarehouse(
+			ctx,
+			c,
+			warehouse,
+			&api.ListWarehouseFreightOptions{
+				ApprovedFor:          stage.Name,
+				VerifiedIn:           req.Sources.Stages,
+				AvailabilityStrategy: req.Sources.AvailabilityStrategy,
+			},
+		)
+		if err != nil {
+			return 0, fmt.Errorf("list freight from Warehouse %q: %w", warehouse.Name, err)
+		}
+
+		for i := range candidates {
+			remaining := remainingSoakForFreight(
+				&candidates[i],
+				req.Sources.Stages,
+				req.Sources.RequiredSoakTime.Duration,
+				req.Sources.AvailabilityStrategy,
+			)
+			if remaining <= 0 {
+				continue
+			}
+			if soonest == 0 || remaining < soonest {
+				soonest = remaining
+			}
+		}
+	}
+
+	if soonest == 0 {
+		return 0, nil
+	}
+	return soonest + soakRequeueBuffer, nil
+}
+
+// remainingSoakForFreight returns how much longer the given Freight must soak
+// before it satisfies the requirement of being soaked in the configured
+// upstream Stages according to the given AvailabilityStrategy. A return value
+// of 0 means the Freight either already satisfies the requirement or is not
+// eligible to satisfy it (e.g. with the All strategy when the Freight has not
+// been verified in every upstream Stage).
+func remainingSoakForFreight(
+	freight *kargoapi.Freight,
+	upstreamStages []string,
+	required time.Duration,
+	strategy kargoapi.FreightAvailabilityStrategy,
+) time.Duration {
+	switch strategy {
+	case kargoapi.FreightAvailabilityStrategyAll:
+		// Freight must satisfy the soak in EVERY upstream Stage. The remaining
+		// wait is the longest remaining wait across those Stages. If any
+		// upstream Stage has not verified the Freight, the Freight will never
+		// satisfy this requirement on its own; return 0 to indicate "no
+		// deadline".
+		var longest time.Duration
+		for _, stage := range upstreamStages {
+			if !freight.IsVerifiedIn(stage) {
+				return 0
+			}
+			elapsed := freight.GetLongestSoak(stage)
+			if elapsed >= required {
+				continue
+			}
+			wait := required - elapsed
+			if wait > longest {
+				longest = wait
+			}
+		}
+		return longest
+
+	default:
+		// OneOf (and empty/unset, which is treated as OneOf): the Freight
+		// becomes eligible as soon as ANY upstream Stage's soak completes.
+		// Return the smallest positive remaining wait across the upstream
+		// Stages where the Freight is verified.
+		var shortest time.Duration
+		for _, stage := range upstreamStages {
+			if !freight.IsVerifiedIn(stage) {
+				continue
+			}
+			elapsed := freight.GetLongestSoak(stage)
+			if elapsed >= required {
+				// Already soaked somewhere — no deadline needed.
+				return 0
+			}
+			wait := required - elapsed
+			if shortest == 0 || wait < shortest {
+				shortest = wait
+			}
+		}
+		return shortest
+	}
+}