Repository: akuity/kargo
Pull request #3744 (2025-03-28 16:24:39)

chore(backport release-1.3): fix: always honor availability strategy (#3740)
Manual backport of #3740

netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *docs-kargo-io* processing.


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | ca2c976f28a3997df76e971f4712a629a24b5efc |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/docs-kargo-io/deploys/67e6cd4b292ef600081bdc45 |
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/akuity/kargo/pull/3744?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 52.75%. Comparing base [(`05da370`)](https://app.codecov.io/gh/akuity/kargo/commit/05da37021e615359f287bbd551bf6bdbcc3d84cc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity) to head [(`ca2c976`)](https://app.codecov.io/gh/akuity/kargo/commit/ca2c976f28a3997df76e971f4712a629a24b5efc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity).

<details><summary>Additional details and impacted files</summary>


```diff
@@               Coverage Diff               @@
##           release-1.3    #3744      +/-   ##
===============================================
+ Coverage        52.74%   52.75%   +0.01%     
===============================================
  Files              296      296              
  Lines            26856    26870      +14     
===============================================
+ Hits             14165    14176      +11     
- Misses           11918    11920       +2     
- Partials           773      774       +1     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/akuity/kargo/pull/3744?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=akuity).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>

Patch:
diff --git a/api/v1alpha1/freight_helpers.go b/api/v1alpha1/freight_helpers.go
index 6d8de7f5..bb4b34ec 100644
--- a/api/v1alpha1/freight_helpers.go
+++ b/api/v1alpha1/freight_helpers.go
@@ -158,6 +158,19 @@ func (f *Freight) GetLongestSoak(stage string) time.Duration {
 	return time.Duration(max(longestCompleted.Nanoseconds(), current.Nanoseconds()))
 }
 
+// HasSoakedIn returns whether the Freight has soaked in the specified Stage for
+// at least the specified duration. If the specified duration is nil, this
+// method will return true.
+func (f *Freight) HasSoakedIn(stage string, dur *metav1.Duration) bool {
+	if f == nil {
+		return false
+	}
+	if dur == nil {
+		return true
+	}
+	return f.GetLongestSoak(stage) >= dur.Duration
+}
+
 // AddCurrentStage updates the Freight status to reflect that the Freight is
 // currently in the specified Stage.
 func (f *FreightStatus) AddCurrentStage(stage string, since time.Time) {
diff --git a/api/v1alpha1/stage_helpers.go b/api/v1alpha1/stage_helpers.go
index 19dbc53f..38d3fb77 100644
--- a/api/v1alpha1/stage_helpers.go
+++ b/api/v1alpha1/stage_helpers.go
@@ -139,9 +139,7 @@ func (s *Stage) ListAvailableFreight(
 				ApprovedFor:          s.Name,
 				VerifiedIn:           req.Sources.Stages,
 				AvailabilityStrategy: req.Sources.AvailabilityStrategy,
-			}
-			if requiredSoak := req.Sources.RequiredSoakTime; requiredSoak != nil {
-				listOpts.RequiredSoakTime = &requiredSoak.Duration
+				RequiredSoakTime:     req.Sources.RequiredSoakTime,
 			}
 		}
 		freightFromWarehouse, err := warehouse.ListFreight(ctx, c, listOpts)
@@ -172,16 +170,26 @@ func (s *Stage) IsFreightAvailable(freight *Freight) bool {
 		return true
 	}
 	for _, req := range s.Spec.RequestedFreight {
-		if freight.Origin.Equals(&req.Origin) {
-			if req.Sources.Direct {
-				return true
-			}
-			for _, source := range req.Sources.Stages {
-				if freight.IsVerifiedIn(source) {
-					return req.Sources.RequiredSoakTime == nil ||
-						freight.GetLongestSoak(source) >= req.Sources.RequiredSoakTime.Duration
+		if !freight.Origin.Equals(&req.Origin) {
+			continue
+		}
+		if req.Sources.Direct {
+			return true
+		}
+		if req.Sources.AvailabilityStrategy == FreightAvailabilityStrategyAll {
+			// Make sure Freight is verified and soaked in all upstream Stages
+			for _, upstream := range req.Sources.Stages {
+				if !freight.IsVerifiedIn(upstream) || !freight.HasSoakedIn(upstream, req.Sources.RequiredSoakTime) {
+					return false
 				}
 			}
+			return true
+		}
+		// Make sure Freight is verified and soaked in at least one upstream Stage
+		for _, upstream := range req.Sources.Stages {
+			if freight.IsVerifiedIn(upstream) && freight.HasSoakedIn(upstream, req.Sources.RequiredSoakTime) {
+				return true
+			}
 		}
 	}
 	return false
diff --git a/api/v1alpha1/warehouse_helpers.go b/api/v1alpha1/warehouse_helpers.go
index 185c3467..2b2ea3ca 100644
--- a/api/v1alpha1/warehouse_helpers.go
+++ b/api/v1alpha1/warehouse_helpers.go
@@ -85,7 +85,7 @@ type ListWarehouseFreightOptions struct {
 	// RequiredSoakTime optionally specifies a minimum duration that a piece of
 	// Freight must have continuously remained in a Stage at any time after being
 	// verified.
-	RequiredSoakTime *time.Duration
+	RequiredSoakTime *metav1.Duration
 	// AvailabilityStrategy specifies the semantics for how Freight is determined
 	// to be available. If not set, the default is to consider Freight available
 	// if it has been verified in any of the provided VerifiedIn stages.
@@ -211,7 +211,7 @@ func (w *Warehouse) ListFreight(
 		// for the Freight.
 		verifiedStages := sets.New[string]()
 		for stage := range f.Status.VerifiedIn {
-			if f.GetLongestSoak(stage) >= *opts.RequiredSoakTime {
+			if f.HasSoakedIn(stage, opts.RequiredSoakTime) {
 				verifiedStages.Insert(stage)
 			}
 		}


Test patch:
diff --git a/api/v1alpha1/stage_helpers_test.go b/api/v1alpha1/stage_helpers_test.go
index 74eacedd..e1b15c26 100644
--- a/api/v1alpha1/stage_helpers_test.go
+++ b/api/v1alpha1/stage_helpers_test.go
@@ -571,6 +571,65 @@ func TestStage_IsFreightAvailable(t *testing.T) {
 			},
 			expected: true,
 		},
+		{
+			name: "availability strategy all + freight is verified and soaked in one upstream stage, but not all",
+			stage: &Stage{
+				ObjectMeta: testStageMeta,
+				Spec: StageSpec{
+					RequestedFreight: []FreightRequest{{
+						Origin: testOrigin,
+						Sources: FreightSources{
+							AvailabilityStrategy: FreightAvailabilityStrategyAll,
+							Stages:               []string{"upstream-stage-1", "upstream-stage-2"},
+							RequiredSoakTime:     &metav1.Duration{Duration: time.Hour},
+						},
+					}},
+				},
+			},
+			freight: &Freight{
+				ObjectMeta: testFreightMeta,
+				Origin:     testOrigin,
+				Status: FreightStatus{
+					VerifiedIn: map[string]VerifiedStage{
+						"upstream-stage-1": {
+							LongestCompletedSoak: &metav1.Duration{Duration: 2 * time.Hour},
+						},
+					},
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "availability strategy all + freight is verified and soaked in all upstream stages",
+			stage: &Stage{
+				ObjectMeta: testStageMeta,
+				Spec: StageSpec{
+					RequestedFreight: []FreightRequest{{
+						Origin: testOrigin,
+						Sources: FreightSources{
+							AvailabilityStrategy: FreightAvailabilityStrategyAll,
+							Stages:               []string{"upstream-stage-1", "upstream-stage-2"},
+							RequiredSoakTime:     &metav1.Duration{Duration: time.Hour},
+						},
+					}},
+				},
+			},
+			freight: &Freight{
+				ObjectMeta: testFreightMeta,
+				Origin:     testOrigin,
+				Status: FreightStatus{
+					VerifiedIn: map[string]VerifiedStage{
+						"upstream-stage-1": {
+							LongestCompletedSoak: &metav1.Duration{Duration: 2 * time.Hour},
+						},
+						"upstream-stage-2": {
+							LongestCompletedSoak: &metav1.Duration{Duration: 2 * time.Hour},
+						},
+					},
+				},
+			},
+			expected: true,
+		},
 		{
 			name: "freight from origin not requested",
 			stage: &Stage{
diff --git a/api/v1alpha1/warehouse_helpers_test.go b/api/v1alpha1/warehouse_helpers_test.go
index 86dce29d..42c7a751 100644
--- a/api/v1alpha1/warehouse_helpers_test.go
+++ b/api/v1alpha1/warehouse_helpers_test.go
@@ -10,7 +10,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
@@ -176,7 +175,8 @@ func TestWarehouse_ListFreight(t *testing.T) {
 			opts: &ListWarehouseFreightOptions{
 				ApprovedFor:      testStage,
 				VerifiedIn:       []string{testUpstreamStage},
-				RequiredSoakTime: ptr.To(time.Hour)},
+				RequiredSoakTime: &metav1.Duration{Duration: time.Hour},
+			},
 			objects: []client.Object{
 				&Freight{ // This should not be returned
 					ObjectMeta: metav1.ObjectMeta{
@@ -277,7 +277,7 @@ func TestWarehouse_ListFreight(t *testing.T) {
 				AvailabilityStrategy: FreightAvailabilityStrategyAll,
 				ApprovedFor:          testStage,
 				VerifiedIn:           []string{testUpstreamStage, testUpstreamStage2},
-				RequiredSoakTime:     ptr.To(time.Hour),
+				RequiredSoakTime:     &metav1.Duration{Duration: time.Hour},
 			},
 			objects: []client.Object{
 				&Freight{ // This should be returned as its approved for the Stage
@@ -376,7 +376,7 @@ func TestWarehouse_ListFreight(t *testing.T) {
 				AvailabilityStrategy: "Invalid",
 				ApprovedFor:          testStage,
 				VerifiedIn:           []string{testUpstreamStage, testUpstreamStage2},
-				RequiredSoakTime:     ptr.To(time.Hour),
+				RequiredSoakTime:     &metav1.Duration{Duration: time.Hour},
 			},
 			assertions: func(t *testing.T, _ []Freight, err error) {
 				require.ErrorContains(t, err, "unsupported AvailabilityStrategy")