fix(retry): implement two-stage select to prioritize context cancellation

- Add non-blocking select stage to check context/closer before timer
- Remove skipUnderDuress logic from tests as no longer needed
- Revert timing tolerance changes since root cause is fixed
- Eliminates race condition in retry.Next() select statement

Implements solution suggested by @kev-cao in #154764

Author: sanki92

pkg/util/retry/retry.go

@@ -192,6 +192,15 @@ func (r *Retry) Next() bool {
 		r.backingOffHook(actualWait)
 	}
 
+	// Check for cancellation first to prioritize over timer.
+	select {
+	case <-r.opts.Closer:
+		return false
+	case <-r.ctx.Done():
+		return false
+	default:
+	}
+
 	select {
 	case <-r.opts.Closer:
 		return false

pkg/util/retry/retry_test.go

@@ -406,7 +406,6 @@ func TestRetryWithMaxDuration(t *testing.T) {
 		// an upper bound instead (e.g. for context or closer).
 		maxExpectedTimeSpent time.Duration
 		expectedErr          bool
-		skipUnderDuress      bool
 	}
 
 	testCases := []testcase{
@@ -474,11 +473,8 @@ func TestRetryWithMaxDuration(t *testing.T) {
 			preRetryFunc: func() {
 				cancelCtxFunc()
 			},
-			maxExpectedTimeSpent: time.Millisecond * 30,
+			maxExpectedTimeSpent: time.Millisecond * 20,
 			expectedErr:          true,
-			// Under duress, closing a context will not necessarily stop the retry
-			// loop immediately, so we skip this test under duress.
-			skipUnderDuress: true,
 		},
 		{
 			name: "errors with opt.Closer that is closed",
@@ -494,20 +490,13 @@ func TestRetryWithMaxDuration(t *testing.T) {
 			preRetryFunc: func() {
 				close(closeCh)
 			},
-			maxExpectedTimeSpent: time.Millisecond * 30,
+			maxExpectedTimeSpent: time.Millisecond * 20,
 			expectedErr:          true,
-			// Under duress, closing a channel will not necessarily stop the retry
-			// loop immediately, so we skip this test under duress.
-			skipUnderDuress: true,
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			if tc.skipUnderDuress && skip.Duress() {
-				skip.UnderDuress(t, "skipping test under duress: %s", tc.name)
-			}
-
 			timeSource := timeutil.NewManualTime(time.Now())
 			tc.opts.Clock = timeSource
 			// Disable randomization for deterministic tests.