middleware: new middleware package built on top of state

go.mod

@@ -3,7 +3,7 @@ module github.com/authzed/controller-idioms
 go 1.25.0
 
 require (
-	github.com/authzed/ctxkey v0.0.0-20260210154927-ca132876f62c
+	github.com/authzed/ctxkey v0.1.0
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/davecgh/go-spew v1.1.1
 	github.com/fsnotify/fsnotify v1.9.0

go.sum

@@ -10,8 +10,8 @@ github.com/NYTimes/gziphandler v1.1.1 h1:ZUDjpQae29j0ryrS0u/B8HZfJBtBQHjqw2rQ2cq
 github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMomdKFjzJNB0c=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
-github.com/authzed/ctxkey v0.0.0-20260210154927-ca132876f62c h1:kcjpI9wcj5yFYWJ7bEeiY6hqH2I6Es3lVz6PB7RKblQ=
-github.com/authzed/ctxkey v0.0.0-20260210154927-ca132876f62c/go.mod h1:ve6DXRv9l2HcM2lxSUmZKWXl7Iq/00xYixBuHOtST+4=
+github.com/authzed/ctxkey v0.1.0 h1:wtQnpKgjzxF//qotuiR/Toh691lbY2ZFynsQFQ57mrA=
+github.com/authzed/ctxkey v0.1.0/go.mod h1:ve6DXRv9l2HcM2lxSUmZKWXl7Iq/00xYixBuHOtST+4=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=

queue/controls.go

@@ -22,12 +22,19 @@ import (
 
 	"github.com/go-logr/logr"
 
+	"github.com/authzed/controller-idioms/state"
 	"github.com/authzed/controller-idioms/typedctx"
 )
 
 //go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
 
 // OperationsContext is like Interface, but fetches the object from a context.
+//
+// Operations invoked through OperationsContext also annotate the enclosing
+// state middleware outcome (via state.RecordTermination), so outcome-aware
+// observability distinguishes "done" from "requeued" — with the error, for
+// the error-requeue variants. The annotation is a no-op when no outcome-aware
+// middleware is registered.
 type OperationsContext struct {
 	*typedctx.Key[Interface]
 }
@@ -38,23 +45,28 @@ func NewQueueOperationsCtx() OperationsContext {
 }
 
 func (h OperationsContext) Done(ctx context.Context) {
+	state.RecordTermination(ctx, "done", nil)
 	h.MustValue(ctx).Done()
 }
 
 func (h OperationsContext) RequeueAfter(ctx context.Context, duration time.Duration) {
+	state.RecordTermination(ctx, "requeued", nil)
 	h.MustValue(ctx).RequeueAfter(duration)
 }
 
 func (h OperationsContext) Requeue(ctx context.Context) {
+	state.RecordTermination(ctx, "requeued", nil)
 	h.MustValue(ctx).Requeue()
 }
 
 func (h OperationsContext) RequeueErr(ctx context.Context, err error) {
+	state.RecordTermination(ctx, "requeued", err)
 	logr.FromContextOrDiscard(ctx).V(4).WithCallDepth(3).Error(err, "requeueing after error")
 	h.MustValue(ctx).RequeueErr(err)
 }
 
 func (h OperationsContext) RequeueAPIErr(ctx context.Context, err error) {
+	state.RecordTermination(ctx, "requeued", err)
 	logr.FromContextOrDiscard(ctx).V(4).WithCallDepth(3).Error(err, "requeueing after api error")
 	h.MustValue(ctx).RequeueAPIErr(err)
 }
@@ -120,18 +132,21 @@ func (c *Operations) RequeueErr(err error) {
 }
 
 // RequeueAPIErr checks to see if `err` is a kube api error with retry data.
-// If so, it requeues after the wait period, otherwise, it requeues immediately.
+// If so, it requeues after the suggested wait period; if it is retryable
+// without one, it requeues immediately; otherwise it marks the key done.
+// Exactly one queue operation fires.
 func (c *Operations) RequeueAPIErr(err error) {
 	defer c.cancel()
 	c.err = err
 	retry, after := ShouldRetry(err)
-	if retry && after > 0 {
+	switch {
+	case retry && after > 0:
 		c.RequeueAfter(after)
-	}
-	if retry {
+	case retry:
 		c.Requeue()
+	default:
+		c.Done()
 	}
-	c.Done()
 }
 
 // Error returns the last recorded error, if any

queue/controls_test.go

@@ -2,9 +2,13 @@ package queue
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/util/workqueue"
 
 	"github.com/authzed/controller-idioms/handler"
@@ -53,18 +57,55 @@ func ExampleNewQueueOperationsCtx() {
 	key, _ := queue.Get()
 
 	// operations are per-key
-	CtxQueue := NewQueueOperationsCtx().WithValue(ctx, NewOperations(func() {
+	ctxWithQueue := NewQueueOperationsCtx().WithValue(ctx, NewOperations(func() {
 		queue.Done(key)
 	}, func(duration time.Duration) {
 		queue.AddAfter(key, duration)
 	}, cancel))
 
 	// queue controls are passed via context
-	handler.NewHandlerFromFunc(func(_ context.Context) {
+	handler.NewHandlerFromFunc(func(hctx context.Context) {
 		// do some work
-		CtxQueue.Done()
-	}, "example").Handle(ctx)
+		NewQueueOperationsCtx().Done(hctx)
+	}, "example").Handle(ctxWithQueue)
 
 	fmt.Println(queue.Len())
 	// Output: 0
 }
+
+// RequeueAPIErr must fire exactly one queue operation per call: the
+// server-suggested delay, an immediate requeue, or done — never a
+// combination.
+func TestOperationsRequeueAPIErrFiresExactlyOnce(t *testing.T) {
+	setup := func() (*Operations, *int, *[]time.Duration) {
+		var doneCalls int
+		var requeues []time.Duration
+		ops := NewOperations(
+			func() { doneCalls++ },
+			func(d time.Duration) { requeues = append(requeues, d) },
+			func() {},
+		)
+		return ops, &doneCalls, &requeues
+	}
+
+	t.Run("retryable with server delay requeues after, once", func(t *testing.T) {
+		ops, done, requeues := setup()
+		ops.RequeueAPIErr(apierrors.NewTooManyRequests("slow down", 7))
+		require.Equal(t, []time.Duration{7 * time.Second}, *requeues)
+		require.Zero(t, *done)
+	})
+
+	t.Run("retryable without delay requeues immediately, once", func(t *testing.T) {
+		ops, done, requeues := setup()
+		ops.RequeueAPIErr(apierrors.NewInternalError(errors.New("boom")))
+		require.Equal(t, []time.Duration{0}, *requeues)
+		require.Zero(t, *done)
+	})
+
+	t.Run("non-retryable marks done, once", func(t *testing.T) {
+		ops, done, requeues := setup()
+		ops.RequeueAPIErr(errors.New("permanent"))
+		require.Empty(t, *requeues)
+		require.Equal(t, 1, *done)
+	})
+}

queue/handlers.go

@@ -57,37 +57,33 @@ import (
 	"github.com/authzed/controller-idioms/state"
 )
 
-// Done creates a handler that marks the current queue key as finished and terminates the pipeline.
+// Done is a handler that marks the current queue key as finished and terminates the pipeline.
 // This is equivalent to calling queue.NewQueueOperationsCtx().Done(ctx) and returning.
 //
 // Usage:
 //
 //	pipeline := state.Sequence(
 //	  validateInput,
 //	  processResource,
-//	  queue.Done(), // Stop here - processing complete
+//	  queue.Done, // Stop here - processing complete
 //	)
-func Done() state.NewStep {
-	return state.NewTerminalStepFunc(func(ctx context.Context) {
-		NewQueueOperationsCtx().Done(ctx)
-	})
-}
+var Done = state.NewTerminalStepFunc(func(ctx context.Context) {
+	NewQueueOperationsCtx().Done(ctx)
+})
 
-// Requeue creates a handler that requeues the current key immediately and terminates the pipeline.
+// Requeue is a handler that requeues the current key immediately and terminates the pipeline.
 // This is equivalent to calling queue.NewQueueOperationsCtx().Requeue(ctx) and returning.
 //
 // Usage:
 //
 //	pipeline := state.Decision(
 //	  resourceReady,
 //	  continueProcessing,
-//	  queue.Requeue(), // Not ready - try again immediately
+//	  queue.Requeue, // Not ready - try again immediately
 //	)
-func Requeue() state.NewStep {
-	return state.NewTerminalStepFunc(func(ctx context.Context) {
-		NewQueueOperationsCtx().Requeue(ctx)
-	})
-}
+var Requeue = state.NewTerminalStepFunc(func(ctx context.Context) {
+	NewQueueOperationsCtx().Requeue(ctx)
+})
 
 // RequeueAfter creates a handler that requeues the current key after the specified duration
 // and terminates the pipeline.
@@ -160,34 +156,3 @@ func RequeueAPIErr(err error) state.NewStep {
 		NewQueueOperationsCtx().RequeueAPIErr(ctx, err)
 	})
 }
-
-// OnError creates a handler that executes different queue operations based on whether
-// an error occurred in the context.
-//
-// Usage:
-//
-//	pipeline := state.Sequence(
-//	  riskyOperation,
-//	  queue.OnError(
-//	    queue.RequeueErr(fmt.Errorf("operation failed")), // If error
-//	    queue.Done(), // If success
-//	  ),
-//	)
-func OnError(errorHandler, successHandler state.NewStep) state.NewStep {
-	return func(next state.Step) state.Step {
-		errStep := errorHandler(next)
-		okStep := successHandler(next)
-		return state.StepFunc(func(ctx context.Context) state.Step {
-			if ctx.Err() != nil {
-				if errStep != nil {
-					return errStep.Run(ctx)
-				}
-				return nil
-			}
-			if okStep != nil {
-				return okStep.Run(ctx)
-			}
-			return nil
-		})
-	}
-}