fix: pass through correctly requeue errors via qtransform

The controller should unwrap the error to allow `Modify` call to
succeed, but it should re-wrap it back into RequeueError when returning
to qruntime.

Signed-off-by: Andrey Smirnov <[email protected]>
Signed-off-by: Utku Ozdemir <[email protected]>

.codecov.yml

@@ -1,6 +1,6 @@
 # THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
 #
-# Generated on 2021-05-24T20:18:47Z by kres d88b53b-dirty.
+# Generated on 2024-01-30T09:06:08Z by kres latest.
 
 codecov:
   require_ci_to_pass: false
@@ -9,7 +9,7 @@ coverage:
   status:
     project:
       default:
-        target: 50%
+        target: 45%
         threshold: 0.5%
         base: auto
         if_ci_failed: success

.kres.yaml

@@ -29,3 +29,7 @@ spec:
 kind: auto.CI
 spec:
   provider: drone
+---
+kind: service.CodeCov
+spec:
+  targetThreshold: 45

pkg/controller/generic/qtransform/qtransform.go

@@ -7,6 +7,7 @@ package qtransform
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/siderolabs/gen/optional"
@@ -225,8 +226,17 @@ func (ctrl *QController[Input, Output]) reconcileRunning(ctx context.Context, lo
 		}
 	}
 
+	var requeueError *controller.RequeueError
+
 	if err := safe.WriterModify(ctx, r, mappedOut, func(out Output) error {
-		return ctrl.transformFunc(ctx, r, logger, in, out)
+		transformError := ctrl.transformFunc(ctx, r, logger, in, out)
+
+		// unwrap requeue error, so that we don't fail the modify if requeue was done without an explicit error
+		if errors.As(transformError, &requeueError) {
+			return requeueError.Err()
+		}
+
+		return transformError
 	}); err != nil {
 		if state.IsConflictError(err) {
 			// conflict due to wrong phase, skip it
@@ -237,9 +247,19 @@ func (ctrl *QController[Input, Output]) reconcileRunning(ctx context.Context, lo
 			return nil
 		}
 
+		if requeueError != nil && requeueError.Err() == err { //nolint:errorlint
+			// if requeueError was specified, and Modify returned it unmodified, return it
+			// otherwise Modify failed for its own reasons, and use that error
+			return requeueError
+		}
+
 		return err
 	}
 
+	if requeueError != nil {
+		return requeueError
+	}
+
 	return nil
 }
 

pkg/controller/generic/qtransform/qtransform_test.go

@@ -14,6 +14,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/siderolabs/gen/channel"
 	"github.com/siderolabs/gen/optional"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -35,7 +36,7 @@ import (
 
 type ABController = qtransform.QController[*A, *B]
 
-func NewABController(reconcileTeardownCh <-chan string, opts ...qtransform.ControllerOption) *ABController {
+func NewABController(reconcileTeardownCh <-chan string, requeueErrorCh <-chan error, opts ...qtransform.ControllerOption) *ABController {
 	var allowedFinalizerRemovals sync.Map
 
 	return qtransform.NewQController(
@@ -59,6 +60,15 @@ func NewABController(reconcileTeardownCh <-chan string, opts ...qtransform.Contr
 				out.TypedSpec().Out = fmt.Sprintf("%q-%d", in.TypedSpec().Str, in.TypedSpec().Int)
 				out.TypedSpec().TransformCount++
 
+				if requeueErrorCh != nil {
+					select {
+					case err := <-requeueErrorCh:
+						return err
+					case <-ctx.Done():
+						return ctx.Err()
+					}
+				}
+
 				return nil
 			},
 			FinalizerRemovalFunc: func(ctx context.Context, r controller.Reader, l *zap.Logger, in *A) error {
@@ -401,7 +411,7 @@ func TestDestroyInputFinalizers(t *testing.T) {
 	setup(t, func(ctx context.Context, st state.State, runtime *runtime.Runtime) {
 		teardownCh := make(chan string)
 
-		require.NoError(t, runtime.RegisterQController(NewABController(teardownCh)))
+		require.NoError(t, runtime.RegisterQController(NewABController(teardownCh, nil)))
 
 		for _, a := range []*A{
 			NewA("1", ASpec{Str: "reconcile-teardown"}),
@@ -466,7 +476,7 @@ func TestDestroyReconcileTeardown(t *testing.T) {
 	setup(t, func(ctx context.Context, st state.State, runtime *runtime.Runtime) {
 		teardownCh := make(chan string)
 
-		require.NoError(t, runtime.RegisterQController(NewABController(teardownCh)))
+		require.NoError(t, runtime.RegisterQController(NewABController(teardownCh, nil)))
 
 		for _, a := range []*A{
 			NewA("1", ASpec{Str: "reconcile-teardown"}),
@@ -514,7 +524,7 @@ func TestDestroyReconcileTeardown(t *testing.T) {
 
 func TestOutputShared(t *testing.T) {
 	setup(t, func(ctx context.Context, st state.State, runtime *runtime.Runtime) {
-		require.NoError(t, runtime.RegisterQController(NewABController(nil, qtransform.WithOutputKind(controller.OutputShared))))
+		require.NoError(t, runtime.RegisterQController(NewABController(nil, nil, qtransform.WithOutputKind(controller.OutputShared))))
 		require.NoError(t, runtime.RegisterQController(NewABNoFinalizerRemovalController(qtransform.WithOutputKind(controller.OutputShared))))
 	})
 }
@@ -559,6 +569,48 @@ func TestRemappedInput(t *testing.T) {
 	})
 }
 
+func TestRequeueErrorBackoff(t *testing.T) {
+	setup(t, func(ctx context.Context, st state.State, runtime *runtime.Runtime) {
+		errorCh := make(chan error, 1)
+
+		require.NoError(t, runtime.RegisterQController(NewABController(nil, errorCh)))
+
+		// send a RequeueError containing an actual error - the output resource must not be created
+
+		if !channel.SendWithContext[error](ctx, errorCh, controller.NewRequeueError(fmt.Errorf("first error"), 100*time.Millisecond)) {
+			t.FailNow()
+		}
+
+		require.NoError(t, st.Create(ctx, NewA("1", ASpec{Str: "foo", Int: 1})))
+
+		sleep(ctx, 250*time.Millisecond)
+
+		rtestutils.AssertNoResource[*B](ctx, t, st, "transformed-1")
+
+		// send a RequeueError without an inner error (simple requeue request) - the output must be created
+
+		if !channel.SendWithContext[error](ctx, errorCh, controller.NewRequeueInterval(100*time.Millisecond)) {
+			t.FailNow()
+		}
+
+		rtestutils.AssertResource(ctx, t, st, "transformed-1", func(r *B, assert *assert.Assertions) {
+			assert.Equal(`"foo"-1`, r.TypedSpec().Out)
+			assert.Equal(1, r.TypedSpec().TransformCount)
+		})
+
+		// send a nil error - because a requeue was requested above, transform is called again, waiting to receive on the errorCh
+
+		if !channel.SendWithContext[error](ctx, errorCh, nil) {
+			t.FailNow()
+		}
+
+		rtestutils.AssertResource(ctx, t, st, "transformed-1", func(r *B, assert *assert.Assertions) {
+			assert.Equal(`"foo"-1`, r.TypedSpec().Out)
+			assert.Equal(2, r.TypedSpec().TransformCount)
+		})
+	})
+}
+
 func setup(t *testing.T, f func(ctx context.Context, st state.State, rt *runtime.Runtime)) {
 	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
@@ -587,3 +639,13 @@ func setup(t *testing.T, f func(ctx context.Context, st state.State, rt *runtime
 
 	f(ctx, st, rt)
 }
+
+func sleep(ctx context.Context, d time.Duration) {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+
+	select {
+	case <-ctx.Done():
+	case <-timer.C:
+	}
+}