diff --git a/go.mod b/go.mod index 4fe78f6..9b04973 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 3df6ce6..2565f13 100644 --- a/go.sum +++ b/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= diff --git a/queue/controls.go b/queue/controls.go index 7461e8e..7345f78 100644 --- a/queue/controls.go +++ b/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 diff --git a/queue/controls_test.go b/queue/controls_test.go index 7ff3f92..ad2fd37 100644 --- a/queue/controls_test.go +++ b/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) + }) +} diff --git a/queue/handlers.go b/queue/handlers.go index 77398be..2a543de 100644 --- a/queue/handlers.go +++ b/queue/handlers.go @@ -57,7 +57,7 @@ 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: @@ -65,15 +65,13 @@ import ( // 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: @@ -81,13 +79,11 @@ func Done() state.NewStep { // 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 - }) - } -} diff --git a/queue/handlers_test.go b/queue/handlers_test.go index d98dcfa..e1a2309 100644 --- a/queue/handlers_test.go +++ b/queue/handlers_test.go @@ -7,6 +7,8 @@ import ( "testing" "time" + "github.com/stretchr/testify/require" + "github.com/authzed/ctxkey" "github.com/authzed/controller-idioms/queue" @@ -15,151 +17,139 @@ import ( ) func TestDone(t *testing.T) { - ctx := t.Context() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - // Create and run pipeline + var afterExecuted bool pipeline := state.Sequence( + state.Do(func(ctx context.Context) context.Context { return ctx }), + queue.Done, state.Do(func(ctx context.Context) context.Context { - // This should execute - return ctx - }), - queue.Done(), - state.Do(func(ctx context.Context) context.Context { - t.Error("This should not execute - pipeline should have terminated") + afterExecuted = true return ctx }), ) state.Run(ctxWithQueue, pipeline) - // Verify Done was called - if fakeQueue.DoneCallCount() != 1 { - t.Errorf("Expected Done to be called once, got %d calls", fakeQueue.DoneCallCount()) + require.Equal(t, 1, fakeQueue.DoneCallCount()) + require.False(t, afterExecuted, "pipeline should terminate after Done") +} + +// Queue operations annotate the outcome observed by outcome-aware +// middleware: Done attests detail "done", the requeue family attests +// "requeued" — with the error, for the error variants. +func TestQueueOperationsAnnotateOutcome(t *testing.T) { + observe := func(step state.NewStep) state.Outcome { + fakeQueue := &fake.FakeInterface{} + ctx := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) + var got state.Outcome + mw := state.AfterOutcome(func(ctx context.Context, o state.Outcome) context.Context { + got = o + return ctx + }) + state.Run(ctx, mw.Wrap(step)) + return got } + + t.Run("Done", func(t *testing.T) { + got := observe(queue.Done) + require.Equal(t, state.OutcomeTerminated, got.Kind) + require.Equal(t, "done", got.Detail) + require.NoError(t, got.Cause) + }) + + t.Run("Requeue", func(t *testing.T) { + got := observe(queue.Requeue) + require.Equal(t, state.OutcomeTerminated, got.Kind) + require.Equal(t, "requeued", got.Detail) + }) + + t.Run("RequeueAfter", func(t *testing.T) { + got := observe(queue.RequeueAfter(time.Minute)) + require.Equal(t, state.OutcomeTerminated, got.Kind) + require.Equal(t, "requeued", got.Detail) + }) + + t.Run("RequeueErr carries the error", func(t *testing.T) { + cause := errors.New("sync failed") + got := observe(queue.RequeueErr(cause)) + require.Equal(t, state.OutcomeTerminated, got.Kind) + require.Equal(t, "requeued", got.Detail) + require.Equal(t, cause, got.Cause) + }) } func TestRequeue(t *testing.T) { - ctx := t.Context() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - executed := false + var afterExecuted bool pipeline := state.Sequence( + state.Do(func(ctx context.Context) context.Context { return ctx }), + queue.Requeue, state.Do(func(ctx context.Context) context.Context { - executed = true - return ctx - }), - queue.Requeue(), - state.Do(func(ctx context.Context) context.Context { - t.Error("This should not execute - pipeline should have terminated") + afterExecuted = true return ctx }), ) state.Run(ctxWithQueue, pipeline) - // Verify execution and requeue - if !executed { - t.Error("Expected first action to execute") - } - if fakeQueue.RequeueCallCount() != 1 { - t.Errorf("Expected Requeue to be called once, got %d calls", fakeQueue.RequeueCallCount()) - } + require.Equal(t, 1, fakeQueue.RequeueCallCount()) + require.False(t, afterExecuted, "pipeline should terminate after Requeue") } func TestRequeueAfter(t *testing.T) { - ctx := t.Context() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) duration := 30 * time.Second - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - pipeline := state.Sequence( - state.Do(func(ctx context.Context) context.Context { - // Setup action - return ctx - }), + state.Do(func(ctx context.Context) context.Context { return ctx }), queue.RequeueAfter(duration), state.Do(func(ctx context.Context) context.Context { - t.Error("This should not execute - pipeline should have terminated") + t.Error("pipeline should terminate after RequeueAfter") return ctx }), ) state.Run(ctxWithQueue, pipeline) - // Verify RequeueAfter was called with correct duration - if fakeQueue.RequeueAfterCallCount() != 1 { - t.Errorf("Expected RequeueAfter to be called once, got %d calls", fakeQueue.RequeueAfterCallCount()) - } - if fakeQueue.RequeueAfterArgsForCall(0) != duration { - t.Errorf("Expected RequeueAfter duration %v, got %v", duration, fakeQueue.RequeueAfterArgsForCall(0)) - } + require.Equal(t, 1, fakeQueue.RequeueAfterCallCount()) + require.Equal(t, duration, fakeQueue.RequeueAfterArgsForCall(0)) } func TestRequeueErr(t *testing.T) { - ctx := t.Context() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) testErr := errors.New("test error") - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - pipeline := state.Sequence( - state.Do(func(ctx context.Context) context.Context { - // Setup action - return ctx - }), + state.Do(func(ctx context.Context) context.Context { return ctx }), queue.RequeueErr(testErr), state.Do(func(ctx context.Context) context.Context { - t.Error("This should not execute - pipeline should have terminated") + t.Error("pipeline should terminate after RequeueErr") return ctx }), ) state.Run(ctxWithQueue, pipeline) - // Verify RequeueErr was called with correct error - if fakeQueue.RequeueErrCallCount() != 1 { - t.Errorf("Expected RequeueErr to be called once, got %d calls", fakeQueue.RequeueErrCallCount()) - } - if !errors.Is(fakeQueue.RequeueErrArgsForCall(0), testErr) { - t.Errorf("Expected RequeueErr error %v, got %v", testErr, fakeQueue.RequeueErrArgsForCall(0)) - } + require.Equal(t, 1, fakeQueue.RequeueErrCallCount()) + require.ErrorIs(t, fakeQueue.RequeueErrArgsForCall(0), testErr) } func TestRequeueErrFromWithinStep(t *testing.T) { - ctx := t.Context() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - var executedBefore bool - var executedAfter bool + var executedBefore, executedAfter bool - // Simulate a step that encounters an error and returns the error handler directly - riskyOperation := state.NewStepFunc(func(ctx context.Context, next state.Step) state.Step { + riskyOperation := state.NewStepFunc(func(ctx context.Context, _ state.Step) state.Step { executedBefore = true - // Simulate some operation that fails err := errors.New("operation failed") - if err != nil { - // Return the error handler directly, short-circuiting the pipeline - return queue.RequeueErr(err).Step().Run(ctx) - } - return state.Continue(ctx, next) + return queue.RequeueErr(err).Step().Run(ctx) }) pipeline := state.Sequence( @@ -172,62 +162,32 @@ func TestRequeueErrFromWithinStep(t *testing.T) { state.Run(ctxWithQueue, pipeline) - // Verify the operation executed but subsequent steps did not - if !executedBefore { - t.Error("Expected risky operation to execute") - } - if executedAfter { - t.Error("Expected pipeline to terminate after error, but subsequent step executed") - } - - // Verify RequeueErr was called - if fakeQueue.RequeueErrCallCount() != 1 { - t.Errorf("Expected RequeueErr to be called once, got %d calls", fakeQueue.RequeueErrCallCount()) - } + require.True(t, executedBefore) + require.False(t, executedAfter, "pipeline should terminate after inline error") + require.Equal(t, 1, fakeQueue.RequeueErrCallCount()) } func TestRequeueAPIErr(t *testing.T) { - ctx := t.Context() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) testErr := errors.New("API error") - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - pipeline := queue.RequeueAPIErr(testErr) - state.Run(ctxWithQueue, pipeline) + state.Run(ctxWithQueue, queue.RequeueAPIErr(testErr)) - // Verify RequeueAPIErr was called with correct error - if fakeQueue.RequeueAPIErrCallCount() != 1 { - t.Errorf("Expected RequeueAPIErr to be called once, got %d calls", fakeQueue.RequeueAPIErrCallCount()) - } - if !errors.Is(fakeQueue.RequeueAPIErrArgsForCall(0), testErr) { - t.Errorf("Expected RequeueAPIErr error %v, got %v", testErr, fakeQueue.RequeueAPIErrArgsForCall(0)) - } + require.Equal(t, 1, fakeQueue.RequeueAPIErrCallCount()) + require.ErrorIs(t, fakeQueue.RequeueAPIErrArgsForCall(0), testErr) } func TestRequeueAPIErrFromWithinStep(t *testing.T) { - ctx := t.Context() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) + var executedBefore, executedAfter bool - var executedBefore bool - var executedAfter bool - - // Simulate a Kubernetes API call that fails - callKubernetesAPI := state.NewStepFunc(func(ctx context.Context, next state.Step) state.Step { + callKubernetesAPI := state.NewStepFunc(func(ctx context.Context, _ state.Step) state.Step { executedBefore = true - // Simulate an API error apiErr := errors.New("deployments.apps \"myapp\" not found") - if apiErr != nil { - // Return the API error handler directly - return queue.RequeueAPIErr(apiErr).Step().Run(ctx) - } - return state.Continue(ctx, next) + return queue.RequeueAPIErr(apiErr).Step().Run(ctx) }) pipeline := state.Sequence( @@ -240,18 +200,9 @@ func TestRequeueAPIErrFromWithinStep(t *testing.T) { state.Run(ctxWithQueue, pipeline) - // Verify the API call executed but subsequent steps did not - if !executedBefore { - t.Error("Expected API call to execute") - } - if executedAfter { - t.Error("Expected pipeline to terminate after API error, but subsequent step executed") - } - - // Verify RequeueAPIErr was called - if fakeQueue.RequeueAPIErrCallCount() != 1 { - t.Errorf("Expected RequeueAPIErr to be called once, got %d calls", fakeQueue.RequeueAPIErrCallCount()) - } + require.True(t, executedBefore) + require.False(t, executedAfter, "pipeline should terminate after inline API error") + require.Equal(t, 1, fakeQueue.RequeueAPIErrCallCount()) } func TestConditionalRequeue(t *testing.T) { @@ -261,34 +212,20 @@ func TestConditionalRequeue(t *testing.T) { expectRequeue bool expectContinuation bool }{ - { - name: "requeue when condition is true", - condition: true, - expectRequeue: true, - expectContinuation: false, - }, - { - name: "continue when condition is false", - condition: false, - expectRequeue: false, - expectContinuation: true, - }, + {"requeue when condition is true", true, true, false}, + {"continue when condition is false", false, false, true}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ctx := context.Background() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - continued := false + var continued bool pipeline := state.Sequence( state.When( func(_ context.Context) bool { return tt.condition }, - queue.Requeue(), + queue.Requeue, ), state.Do(func(ctx context.Context) context.Context { continued = true @@ -298,22 +235,12 @@ func TestConditionalRequeue(t *testing.T) { state.Run(ctxWithQueue, pipeline) - // Verify requeue behavior - requeueCount := fakeQueue.RequeueCallCount() - if tt.expectRequeue && requeueCount != 1 { - t.Errorf("Expected requeue to be called once, got %d calls", requeueCount) - } - if !tt.expectRequeue && requeueCount != 0 { - t.Errorf("Expected requeue not to be called, got %d calls", requeueCount) - } - - // Verify continuation behavior - if tt.expectContinuation && !continued { - t.Error("Expected pipeline to continue") - } - if !tt.expectContinuation && continued { - t.Error("Expected pipeline to terminate") + if tt.expectRequeue { + require.Equal(t, 1, fakeQueue.RequeueCallCount()) + } else { + require.Equal(t, 0, fakeQueue.RequeueCallCount()) } + require.Equal(t, tt.expectContinuation, continued) }) } } @@ -327,30 +254,16 @@ func TestConditionalRequeueAfter(t *testing.T) { expectRequeue bool expectContinuation bool }{ - { - name: "requeue after when condition is true", - condition: true, - expectRequeue: true, - expectContinuation: false, - }, - { - name: "continue when condition is false", - condition: false, - expectRequeue: false, - expectContinuation: true, - }, + {"requeue after when condition is true", true, true, false}, + {"continue when condition is false", false, false, true}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ctx := context.Background() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - continued := false + var continued bool pipeline := state.Sequence( state.When( func(_ context.Context) bool { return tt.condition }, @@ -364,27 +277,13 @@ func TestConditionalRequeueAfter(t *testing.T) { state.Run(ctxWithQueue, pipeline) - // Verify requeue behavior - requeueCount := fakeQueue.RequeueAfterCallCount() - if tt.expectRequeue && requeueCount != 1 { - t.Errorf("Expected requeueAfter to be called once, got %d calls", requeueCount) - } - if !tt.expectRequeue && requeueCount != 0 { - t.Errorf("Expected requeueAfter not to be called, got %d calls", requeueCount) - } - - // Verify duration if requeued - if tt.expectRequeue && fakeQueue.RequeueAfterArgsForCall(0) != duration { - t.Errorf("Expected duration %v, got %v", duration, fakeQueue.RequeueAfterArgsForCall(0)) - } - - // Verify continuation behavior - if tt.expectContinuation && !continued { - t.Error("Expected pipeline to continue") - } - if !tt.expectContinuation && continued { - t.Error("Expected pipeline to terminate") + if tt.expectRequeue { + require.Equal(t, 1, fakeQueue.RequeueAfterCallCount()) + require.Equal(t, duration, fakeQueue.RequeueAfterArgsForCall(0)) + } else { + require.Equal(t, 0, fakeQueue.RequeueAfterCallCount()) } + require.Equal(t, tt.expectContinuation, continued) }) } } @@ -396,34 +295,20 @@ func TestConditionalDone(t *testing.T) { expectDone bool expectContinuation bool }{ - { - name: "done when condition is true", - condition: true, - expectDone: true, - expectContinuation: false, - }, - { - name: "continue when condition is false", - condition: false, - expectDone: false, - expectContinuation: true, - }, + {"done when condition is true", true, true, false}, + {"continue when condition is false", false, false, true}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ctx := context.Background() fakeQueue := &fake.FakeInterface{} + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - continued := false + var continued bool pipeline := state.Sequence( state.When( func(_ context.Context) bool { return tt.condition }, - queue.Done(), + queue.Done, ), state.Do(func(ctx context.Context) context.Context { continued = true @@ -433,99 +318,21 @@ func TestConditionalDone(t *testing.T) { state.Run(ctxWithQueue, pipeline) - // Verify done behavior - doneCount := fakeQueue.DoneCallCount() - if tt.expectDone && doneCount != 1 { - t.Errorf("Expected done to be called once, got %d calls", doneCount) - } - if !tt.expectDone && doneCount != 0 { - t.Errorf("Expected done not to be called, got %d calls", doneCount) - } - - // Verify continuation behavior - if tt.expectContinuation && !continued { - t.Error("Expected pipeline to continue") - } - if !tt.expectContinuation && continued { - t.Error("Expected pipeline to terminate") + if tt.expectDone { + require.Equal(t, 1, fakeQueue.DoneCallCount()) + } else { + require.Equal(t, 0, fakeQueue.DoneCallCount()) } + require.Equal(t, tt.expectContinuation, continued) }) } } -func TestOnError(t *testing.T) { - tests := []struct { - name string - hasError bool - expectDone bool - expectRequeue bool - }{ - { - name: "error handler when error present", - hasError: true, - expectDone: false, - expectRequeue: true, - }, - { - name: "success handler when no error", - hasError: false, - expectDone: true, - expectRequeue: false, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - ctx := context.Background() - fakeQueue := &fake.FakeInterface{} - - // Set up context with fake queue - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - - // Add error to context if needed - if tt.hasError { - cancelCtx, cancel := context.WithCancel(ctxWithQueue) - cancel() // This sets ctx.Err() - ctxWithQueue = cancelCtx - } - - pipeline := queue.OnError( - queue.Requeue(), // Error handler - queue.Done(), // Success handler - ) - - state.Run(ctxWithQueue, pipeline) - - // Verify behavior based on error state - doneCount := fakeQueue.DoneCallCount() - requeueCount := fakeQueue.RequeueCallCount() - - if tt.expectDone && doneCount != 1 { - t.Errorf("Expected done to be called once, got %d calls", doneCount) - } - if !tt.expectDone && doneCount != 0 { - t.Errorf("Expected done not to be called, got %d calls", doneCount) - } - - if tt.expectRequeue && requeueCount != 1 { - t.Errorf("Expected requeue to be called once, got %d calls", requeueCount) - } - if !tt.expectRequeue && requeueCount != 0 { - t.Errorf("Expected requeue not to be called, got %d calls", requeueCount) - } - }) - } -} - -// Test realistic controller scenarios +// TestControllerScenarios tests realistic controller pipelines. func TestControllerScenarios(t *testing.T) { t.Run("successful processing flow", func(t *testing.T) { - ctx := context.Background() fakeQueue := &fake.FakeInterface{} - - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) var executionOrder []string @@ -542,33 +349,18 @@ func TestControllerScenarios(t *testing.T) { executionOrder = append(executionOrder, "finalize") return ctx }), - queue.Done(), + queue.Done, ) state.Run(ctxWithQueue, pipeline) - expectedOrder := []string{"validate", "process", "finalize"} - if len(executionOrder) != len(expectedOrder) { - t.Fatalf("Expected %d executions, got %d: %v", len(expectedOrder), len(executionOrder), executionOrder) - } - - for i, expected := range expectedOrder { - if executionOrder[i] != expected { - t.Errorf("Expected execution[%d] = %q, got %q", i, expected, executionOrder[i]) - } - } - - if fakeQueue.DoneCallCount() != 1 { - t.Errorf("Expected done to be called once, got %d calls", fakeQueue.DoneCallCount()) - } + require.Equal(t, []string{"validate", "process", "finalize"}, executionOrder) + require.Equal(t, 1, fakeQueue.DoneCallCount()) }) t.Run("resource not ready - requeue after delay", func(t *testing.T) { - ctx := context.Background() fakeQueue := &fake.FakeInterface{} - - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) var executionOrder []string @@ -578,46 +370,30 @@ func TestControllerScenarios(t *testing.T) { return ctx }), state.Decision( - func(_ context.Context) bool { - return false // Dependencies not ready - }, - // Dependencies ready + func(_ context.Context) bool { return false }, state.Sequence( state.Do(func(ctx context.Context) context.Context { executionOrder = append(executionOrder, "process") return ctx }), - queue.Done(), + queue.Done, ), - // Dependencies not ready queue.RequeueAfter(5*time.Minute), ), ) state.Run(ctxWithQueue, pipeline) - expectedOrder := []string{"check-dependencies"} - if len(executionOrder) != len(expectedOrder) { - t.Fatalf("Expected %d executions, got %d: %v", len(expectedOrder), len(executionOrder), executionOrder) - } - - if fakeQueue.RequeueAfterCallCount() != 1 { - t.Errorf("Expected requeueAfter to be called once, got %d calls", fakeQueue.RequeueAfterCallCount()) - } - - if fakeQueue.RequeueAfterArgsForCall(0) != 5*time.Minute { - t.Errorf("Expected 5 minute delay, got %v", fakeQueue.RequeueAfterArgsForCall(0)) - } + require.Equal(t, []string{"check-dependencies"}, executionOrder) + require.Equal(t, 1, fakeQueue.RequeueAfterCallCount()) + require.Equal(t, 5*time.Minute, fakeQueue.RequeueAfterArgsForCall(0)) }) t.Run("validation error - requeue with error", func(t *testing.T) { - ctx := context.Background() fakeQueue := &fake.FakeInterface{} - - queueCtx := queue.NewQueueOperationsCtx() - ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - + ctxWithQueue := queue.NewQueueOperationsCtx().WithValue(t.Context(), fakeQueue) testErr := errors.New("validation failed") + var executionOrder []string pipeline := state.Sequence( @@ -626,40 +402,27 @@ func TestControllerScenarios(t *testing.T) { return ctx }), state.Decision( - func(_ context.Context) bool { - return false // Validation failed - }, - // Validation passed + func(_ context.Context) bool { return false }, state.Sequence( state.Do(func(ctx context.Context) context.Context { executionOrder = append(executionOrder, "process") return ctx }), - queue.Done(), + queue.Done, ), - // Validation failed queue.RequeueErr(testErr), ), ) state.Run(ctxWithQueue, pipeline) - expectedOrder := []string{"validate"} - if len(executionOrder) != len(expectedOrder) { - t.Fatalf("Expected %d executions, got %d: %v", len(expectedOrder), len(executionOrder), executionOrder) - } - - if fakeQueue.RequeueErrCallCount() != 1 { - t.Errorf("Expected requeueErr to be called once, got %d calls", fakeQueue.RequeueErrCallCount()) - } - - if !errors.Is(fakeQueue.RequeueErrArgsForCall(0), testErr) { - t.Errorf("Expected error %v, got %v", testErr, fakeQueue.RequeueErrArgsForCall(0)) - } + require.Equal(t, []string{"validate"}, executionOrder) + require.Equal(t, 1, fakeQueue.RequeueErrCallCount()) + require.ErrorIs(t, fakeQueue.RequeueErrArgsForCall(0), testErr) }) } -// Example demonstrating the clean controller pattern this enables +// Example demonstrates the clean controller pattern this enables. func Example() { ctx := context.Background() fakeQueue := &fake.FakeInterface{} @@ -667,43 +430,29 @@ func Example() { queueCtx := queue.NewQueueOperationsCtx() ctxWithQueue := queueCtx.WithValue(ctx, fakeQueue) - // Simulate resource state resourceReadyKey := ctxkey.New[bool]() ctxWithResource := resourceReadyKey.Set(ctxWithQueue, true) - // Clean controller pipeline using queue handlers controllerPipeline := state.Sequence( - // Set finalizer state.Do(func(ctx context.Context) context.Context { fmt.Println("Setting finalizer") return ctx }), - - // Check if resource is ready to process state.When( func(ctx context.Context) bool { return !resourceReadyKey.MustValue(ctx) }, - queue.RequeueAfter(30*time.Second), // Wait 30 seconds if not ready + queue.RequeueAfter(30*time.Second), ), - - // Process the resource state.Do(func(ctx context.Context) context.Context { fmt.Println("Processing resource") return ctx }), - - // Validate processing completed successfully state.When( - func(_ context.Context) bool { - // In real code, this would check if processing completed - return false // Assume success - }, - queue.Requeue(), + func(_ context.Context) bool { return false }, + queue.Requeue, ), - - // Mark as done - queue.Done(), + queue.Done, ) state.Run(ctxWithResource, controllerPipeline) diff --git a/state/FORMAL.md b/state/FORMAL.md deleted file mode 100644 index 7a453a0..0000000 --- a/state/FORMAL.md +++ /dev/null @@ -1,423 +0,0 @@ -# Formal Mathematical Structure of `state` - -The state package provides a mathematically rigorous foundation for compositional computation with context threading. This document formalizes the mathematical structure underlying the system and documents why we might care. - -## Why These Properties Matter - -The mathematical laws guarantee four practical benefits: - -**1. Safe Refactoring** - These transformations are guaranteed identical: - -```go -Sequence(a, b, c, d) ≡ Sequence(a, Sequence(b, c), d) // Extract/inline -Sequence(a, Noop, b) ≡ Sequence(a, b) // Add/remove no-ops -Parallel(a, b, c) ≡ Parallel(c, a, b) // Reorder independent ops -``` - -Without these laws: refactoring could silently break production. - -**2. Compositional Testing** - Test parts, trust composition: - -- 3 steps = 3 unit tests (with laws) vs 6 integration tests (without laws) -- 10 steps = 10 tests vs 3,628,800 tests -- No "works alone, breaks when composed" bugs - -**3. Correct Context Threading** - The monad structure ensures context flows through the pipeline: - -```go -Sequence( - Do(func(ctx) { return context.WithValue(ctx, "user", "alice") }), - Do(func(ctx) { user := ctx.Value("user") /* guaranteed "alice" */ }), -) -``` - -Without laws: context values can vanish, causing hard-to-debug failures. - -**4. Middleware Correctness** - Kleisli category structure tells us middleware must preserve composition: - -- Build middleware from safe combinators (correct by construction) -- Or write property tests verifying endofunctor laws -- No ad-hoc wrappers that break when composed - -## Category Theory Foundation - -### Basic Category - -Our system forms a **category** `𝒞` where: - -- **Objects**: Context states (`context.Context`) -- **Morphisms**: Pure context transformations `ContextFunc = func(Context) Context` -- **Identity**: The identity function `func(ctx Context) Context { return ctx }` -- **Composition**: Function composition of `ContextFunc` - -#### Category Laws - -**Identity Laws**: For any `ContextFunc f`: - -- Left identity: composing identity then `f` equals `f` -- Right identity: composing `f` then identity equals `f` - -**Associativity**: For `ContextFunc` values `f`, `g`, `h`: - -- `(f ∘ g) ∘ h = f ∘ (g ∘ h)` (standard function composition associativity) - -These are the basic laws of function composition in Go. - -## Monadic Structure - -### Step Monad - -A **monad** is a design pattern for chaining operations together in a composable way. It provides two key operations: - -1. **Unit/Return** (our `Do`): wraps a plain value/function into the monadic context -2. **Bind** (our `Sequence`): chains monadic operations together, handling the "plumbing" automatically - -**Why monads matter for us**: They let you write sequential operations (`Sequence(a, b, c)`) where each step can modify context, and the monad handles threading the context through automatically. You don't have to manually pass `ctx` through each step - the monad structure does it for you. - -The Step system forms a **monad** with the following operations: - -#### Unit (η): Do - -```go -// Do is the unit operation - it lifts a pure context transformation into the Step monad -func Do(fn ContextFunc) NewStep -``` - -In our library, `Do` is the unit/return operation that lifts pure context transformations into the monadic context. - -#### Bind (μ): NewStep Composition - -```go -// NewStep is itself the bind operation - composing NewSteps is monadic bind -// Given: f NewStep, g NewStep -// Bind is: Sequence(f, g) -func Sequence(steps ...NewStep) NewStep -``` - -In our library, `NewStep` composition (via `Sequence`) is the bind operation. The `NewStep` type signature `func(next Step) Step` encodes continuation-passing, which is equivalent to monadic bind. - -#### Monad Laws - -**Left Identity**: `Sequence(Do(id), f) = f` where `id(ctx) = ctx` -**Right Identity**: `Sequence(f, Do(id)) = f` where `id(ctx) = ctx` -**Associativity**: `Sequence(Sequence(f, g), h) = Sequence(f, Sequence(g, h))` - -### Kleisli Category - -Now that we've defined the Step monad, we can derive its **Kleisli category**. - -**Definition**: Given a category `C` and a monad `M` on `C`, the **Kleisli category** `𝒦(M)` is derived as follows: - -- **Objects of 𝒦(M)**: Same as objects of `C` -- **Morphisms of 𝒦(M)**: For objects `A` and `B`, a morphism `A → B` in `𝒦(M)` is a morphism `A → M(B)` in `C` -- **Identity in 𝒦(M)**: The monadic unit `η: A → M(A)` -- **Composition in 𝒦(M)**: Given `f: A → M(B)` and `g: B → M(C)`, their Kleisli composition is `g ∘_K f = (μ ∘ M(g) ∘ f): A → M(C)`, where `μ` is the monadic join - -**In simpler terms**: The Kleisli category lets you treat "functions that return wrapped values" as if they were regular functions you can compose. Instead of going `A → B → C`, you have functions `A → M(B)` and `B → M(C)`, and the Kleisli category gives you a way to compose them into `A → M(C)`. - -**In our system**: - -The Step monad gives rise to a **Kleisli category** `𝒦(M)` where: - -- **Base category `C`**: Objects are `Context` states, morphisms are `ContextFunc = func(Context) Context` -- **Monad `M`**: The Step monad (as defined above with `Do` and `Sequence`) -- **Objects of 𝒦(M)**: Context states (same as the base category) -- **Kleisli Arrows (morphisms of 𝒦(M))**: `NewStep = func(next Step) Step` - - In the base category, this would be `Context → M(Context)` where `M` is the Step monad - - In Go, we encode it as `func(next Step) Step` using continuation-passing style -- **Identity in 𝒦(M)**: `Noop` (corresponds to the monadic unit `Do(id)`) -- **Composition in 𝒦(M)**: `Sequence` (the Kleisli composition operator) - -**Why the Kleisli category perspective matters:** - -Every monad automatically generates a Kleisli category - so just having one isn't special. What matters is **what we do with the categorical structure**. - -The Kleisli category perspective becomes valuable when we have **transformations on `NewStep` that preserve composition**: - -**Middleware as Endofunctors**: Functions of type `func(NewStep) NewStep` are endofunctors on the Kleisli category. When middleware preserves the categorical structure (identity and composition), we get: - -```go -// Middleware transforms NewSteps while preserving composition -type Middleware func(NewStep) NewStep - -// Example: Logging middleware -func LoggingMiddleware(name string) Middleware { - return func(step NewStep) NewStep { - return func(next Step) Step { - return StepFunc(func(ctx context.Context) Step { - log.Printf("entering %s", name) - result := step(next).Run(ctx) - log.Printf("exiting %s", name) - return result - }) - } - } -} - -// Middleware composition preserves the categorical laws: -// middleware(Sequence(a, b)) ≈ Sequence(middleware(a), middleware(b)) -``` - -The Kleisli category perspective tells us that **correct middleware must be an endofunctor** - it must preserve identity and composition. While Go's type system doesn't enforce this, the categorical perspective guides us to: - -**1. Build safe middleware combinators** that structurally preserve composition: - -```go -// This combinator guarantees correct behavior by construction -func MakeMiddleware( - before func(context.Context) context.Context, - after func(context.Context) context.Context, -) Middleware { - return func(step NewStep) NewStep { - return Sequence( - Do(before), // ← Built from composition primitives - step, // ← that already satisfy the laws - Do(after), - ) - } -} -``` - -Since `Sequence` and `Do` already satisfy the categorical laws, middleware built from them **inherits** those properties. No manual proof needed - it's correct by construction. - -**2. Test endofunctor properties** for custom middleware: - -```go -// Property-based test that middleware is an endofunctor -func TestMiddlewarePreservesComposition(t *testing.T, mw Middleware) { - // Law 1: mw(Sequence(a, b)) ≈ Sequence(mw(a), mw(b)) - // Law 2: mw(Noop) ≈ Noop -} -``` - -**Without the Kleisli perspective**: We'd write ad-hoc wrappers and hope they work correctly when composed. - -**With the Kleisli perspective**: We know middleware must preserve categorical structure, so we either build it from compositional primitives (correct by construction) or write property tests to verify the endofunctor laws. - -#### Kleisli Laws - -**Identity Laws**: For any NewStep `f`: - -- Left identity: `Sequence(Noop, f) = f` -- Right identity: `Sequence(f, Noop) = f` - -**Associativity**: For NewSteps `f`, `g`, `h`: - -- `Sequence(f, Sequence(g, h)) = Sequence(Sequence(f, g), h)` - -These laws are verified in our test suite (see `formal_test.go`). - -## Algebraic Operations - -These operations give us the algebraic structure to build complex pipelines. They correspond to fundamental categorical constructions. - -### Sequential Composition (Product): Sequence - -```go -// Sequence executes steps sequentially with context threading -func Sequence(steps ...NewStep) NewStep -``` - -**What it does**: Chains steps together - output of one becomes input to the next. - -**Why "product"**: In category theory, the product represents "and then" - you do the first thing AND THEN the second thing. For Kleisli arrows, this is sequential composition. - -**Properties**: - -- **Associative**: `Sequence(a, Sequence(b, c)) = Sequence(Sequence(a, b), c)` - - Grouping doesn't matter: `(a; b); c` is the same as `a; (b; c)` - - This means you can refactor by extracting/inlining subsequences without changing behavior -- **Identity**: `Sequence(Noop, a) = a` and `Sequence(a, Noop) = a` - - Adding a no-op doesn't change behavior - - Safe to add/remove no-ops for debugging - -**Why this matters**: You can compose pipelines like functions, and the same algebraic laws apply. - -### Choice Composition (Coproduct): Decision - -```go -// Decision provides conditional branching (binary choice) -func Decision(predicate func(Context) bool, ifTrue, ifFalse NewStep) NewStep -``` - -**What it does**: Runs one branch or the other based on a condition - exactly one path executes. - -**Why "coproduct"**: In category theory, the coproduct represents "or" - you do the first thing OR the second thing. This is case analysis / conditional branching. - -**Properties**: - -- **Exhaustive**: Exactly one branch executes (no fall-through) -- **Disjoint**: The branches are independent - each sees the same input context -- **Compositional**: Can nest decisions or put them in sequences - -**Why this matters**: Conditional logic composes cleanly with sequential logic: - -```go -Sequence( - validate, - Decision(isValid, processData, handleError), - cleanup, // This runs after whichever branch executes -) -``` - -### Multi-way Choice: Enum and Switch - -```go -// Enum provides multi-way branching on comparable types -func Enum[T comparable]( - selector func(Context) T, - cases map[T]NewStep, - defaultHandler NewStep, -) NewStep - -// Switch is a convenience wrapper for string-based branching -func Switch( - selector func(Context) string, - cases map[string]NewStep, - defaultHandler NewStep, -) NewStep -``` - -**What it does**: Generalizes `Decision` to n-way branching - like a switch statement. - -**Why it exists**: Real-world logic often has more than two cases. Without `Enum`, you'd nest `Decision` calls: - -```go -// Without Enum - messy nested decisions -Decision( - func(ctx) { return getType(ctx) == "A" }, - handleA, - Decision( - func(ctx) { return getType(ctx) == "B" }, - handleB, - Decision(...) // Gets worse with each case - ) -) - -// With Enum - clean and flat -Enum( - getType, - map[string]NewStep{ - "A": handleA, - "B": handleB, - "C": handleC, - }, - handleDefault, -) -``` - -**Why this matters**: Keeps branching logic flat and readable. Each case is at the same level, making the structure clear. - -### Parallel Composition: Parallel - -```go -// Parallel executes stages concurrently while preserving context -func Parallel(steps ...NewStep) NewStep -``` - -**What it does**: Runs multiple steps concurrently, waiting for all to complete before continuing. - -**Why it exists**: Some operations are independent and can run simultaneously: - -```go -Parallel( - validateSchema, // These three operations - checkPermissions, // don't depend on each other - logRequest, // so run them in parallel -) -``` - -**Important limitation**: Each parallel branch receives the same input context. Context modifications within parallel branches are **not** propagated to subsequent steps or to each other (would cause race conditions). - -**Collecting results from parallel branches**: Use `typedctx.Box` to create a shared concurrent-safe space: - -```go -import "github.com/authzed/controller-idioms/typedctx" - -type ValidationResult struct { Valid bool; Errors []error } -type PermissionsResult struct { Allowed bool; Reason string } - -Sequence( - // Set up boxes before parallel execution - Do(func(ctx context.Context) context.Context { - ctx = typedctx.WithBox[ValidationResult](ctx) - ctx = typedctx.WithBox[PermissionsResult](ctx) - return ctx - }), - // Parallel branches write to their boxes - Parallel( - Do(func(ctx context.Context) context.Context { - result := validateSchema(ctx) - typedctx.MustStore(ctx, ValidationResult{Valid: result.Valid, Errors: result.Errors}) - return ctx - }), - Do(func(ctx context.Context) context.Context { - allowed, reason := checkPermissions(ctx) - typedctx.MustStore(ctx, PermissionsResult{Allowed: allowed, Reason: reason}) - return ctx - }), - ), - // After parallel completes, read the results - Do(func(ctx context.Context) context.Context { - validation := typedctx.MustValue[ValidationResult](ctx) - permissions := typedctx.MustValue[PermissionsResult](ctx) - // Both results are now available - if !validation.Valid || !permissions.Allowed { - // handle errors - } - return ctx - }), -) -``` - -The `Box` is thread-safe, so parallel branches can safely write to it without races. - -**Failure Handling in Parallel**: - -- **Panics**: Goroutine panics propagate naturally and crash the program (programming errors should be loud). Wrap a step with `Recover(step)` to opt into panic suppression for that specific step. -- **Context Cancellation**: Checked after all branches complete via `Continue`; cancellation stops the pipeline from advancing to the next step but does not prevent branches from running. -- **Errors**: Should be communicated via typedctx.Box or context values, then handled after parallel completes -- **Assumption**: All branches run to completion. Context cancellation stops the continuation but not the in-flight goroutines. - -**Use patterns**: - -- **Good**: Parallel queries/validations that read context or write to boxes -- **Bad**: Parallel steps that modify shared context values (use Sequence instead) - -**Why this matters**: - -- Performance: Independent operations run concurrently -- Safety: Isolation + typedctx.Box prevents race conditions -- Compositionality: `Parallel` can be nested in `Sequence`, and vice versa - -## Core Abstraction: Do - -The `Do` function is the fundamental lifting operation: - -```go -// Do lifts a pure context transformation into the Step monad -func Do(fn func(Context) Context) NewStep -``` - -**Mathematical Significance**: - -- **Unit Operation**: Lifts pure transformations into the Step monad (the monadic return) -- **Preserves Composition**: Composing functions then lifting is equivalent to lifting then sequencing -- **Context Threading**: Ensures modified contexts flow to subsequent stages in the pipeline - -### Relationship Hierarchy - -```text -Level 1: Pure Functions -func(Context) Context - - ↓ Do (Unit/Return) - -Level 2: Step Monad -NewStep = func(next Step) Step - - ↓ Composition Operators - -Level 3: Complex Workflows -Sequence, Decision, Parallel, etc. -``` diff --git a/state/doc.go b/state/doc.go index be5ff44..aa2bf12 100644 --- a/state/doc.go +++ b/state/doc.go @@ -116,6 +116,12 @@ // }) // Run(ctx, pipeline) // +// A non-nil Step returned by the handler runs with the handler cleared from +// context, so Continue calls inside the recovery path stop quietly instead of +// re-entering it. The context is still cancelled while recovery runs, which +// stops a Continue-linked sequence at its first link: build the recovery path +// from a terminal step or a single step that does its work in the body. +// // # Helper Functions // // The package provides helper functions for common patterns. These are all @@ -135,7 +141,13 @@ // - Decision(predicate, ifTrue, ifFalse) - Binary branching (if/else) // - Enum(selector, cases, default) - Multi-way branching on comparable types // - Switch(selector, cases, default) - Multi-way branching on strings -// - Recover(step) - Wrap a step to suppress panics (opt-in; panics propagate by default) +// +// A panic in a step is a programming error and is left to crash the process +// by default. Parallel branches run on their own goroutines, so their panics +// bypass a recover installed outside the pipeline; a crash handler must run +// inside the branch. Register one as ambient middleware with CrashHandler +// (or Deferred) — Parallel applies the ambient stack inside every branch — +// or apply a raw wrapper with Map. See Parallel. // // # Choosing a Pattern // @@ -253,37 +265,41 @@ // }), // ) // -// To collect results from parallel branches, use typedctx.Box: +// To collect results from parallel branches, allocate a ctxkey box per result +// before the Parallel and let each branch fill its own. The box itself is not +// synchronized: the pattern is race-free because each box has exactly one +// writer, and Parallel's WaitGroup barrier orders the branch writes before the +// post-parallel read. // -// import "github.com/authzed/controller-idioms/typedctx" +// import "github.com/authzed/ctxkey" // // type ValidationResult struct { Valid bool } // type PermissionsResult struct { Allowed bool } // +// var ( +// validationKey = ctxkey.NewBoxedWithDefault(ValidationResult{}) +// permissionsKey = ctxkey.NewBoxedWithDefault(PermissionsResult{}) +// ) +// // pipeline := state.Sequence( -// // Set up boxes before parallel execution +// // Allocate boxes before parallel execution // state.Do(func(ctx context.Context) context.Context { -// ctx = typedctx.WithBox[ValidationResult](ctx) -// ctx = typedctx.WithBox[PermissionsResult](ctx) -// return ctx +// ctx = validationKey.SetBox(ctx) +// return permissionsKey.SetBox(ctx) // }), -// // Parallel branches write to their boxes +// // Each branch fills exactly one box (Set writes through the existing box) // state.Parallel( // state.Do(func(ctx context.Context) context.Context { -// result := validate(ctx) -// typedctx.MustStore(ctx, ValidationResult{Valid: result}) -// return ctx +// return validationKey.Set(ctx, ValidationResult{Valid: validate(ctx)}) // }), // state.Do(func(ctx context.Context) context.Context { -// allowed := checkPermissions(ctx) -// typedctx.MustStore(ctx, PermissionsResult{Allowed: allowed}) -// return ctx +// return permissionsKey.Set(ctx, PermissionsResult{Allowed: checkPermissions(ctx)}) // }), // ), // // After parallel completes, read the results // state.Do(func(ctx context.Context) context.Context { -// validation := typedctx.MustValue[ValidationResult](ctx) -// permissions := typedctx.MustValue[PermissionsResult](ctx) +// validation := validationKey.Value(ctx) +// permissions := permissionsKey.Value(ctx) // fmt.Printf("Valid: %v, Allowed: %v\n", validation.Valid, permissions.Allowed) // return ctx // }), diff --git a/state/formal_test.go b/state/formal_test.go index ffebac6..7162544 100644 --- a/state/formal_test.go +++ b/state/formal_test.go @@ -2,6 +2,7 @@ package state import ( "context" + "errors" "sync" "testing" @@ -482,3 +483,537 @@ func TestCompleteCategoryIntegration(t *testing.T) { // Verify branch-false was not executed require.NotContains(t, trace, "branch-false", "False branch should not execute") } + +// ============================================================================= +// BRACKET MIDDLEWARE LAWS +// +// These tests verify the laws of the sealed Middleware type. A pure Kleisli +// middleware Sequence(Do(before), step, Do(after)) provably cannot run its +// after-hook around a terminal step: terminal steps are left zeros of +// Sequence (Sequence(t, x) = t), so the after-hook is absorbed. The +// interpreter (Wrap, and the dispatch wrappers built on it) therefore +// delimits the step with a sentinel continuation when a stack carries any +// after hook — the package's one control operator. +// +// Because Middleware is sealed (constructible only via the hook generators — +// Before, After, AfterOutcome, Deferred, CrashHandler, the Around pair — and +// Compose, zero value = identity), these laws are universally quantified +// over EVERY value of the type, not just values built through one blessed +// constructor. +// +// The laws (see FORMAL.md "The Bracket Laws"): +// B1 (finalization): every after-form hook closes exactly once per bracket +// entry, on every outcome — continue, terminal, cancel, +// or panic; observation hooks propagate the panic value +// untouched. Befores run in composition order, afters +// close in reverse, so Around's pairing is derived from +// position. +// B2 (nesting): Compose(mwA, mwB) and mwA.Wrap(mwB.Wrap(s)) both +// bracket properly: beforeA, beforeB, s, afterB, afterA +// B3 (transparency): terminality is preserved, and downstream observes +// after(ctx') where ctx' is the context the step threaded +// B4 (identity): the zero Middleware's Wrap returns the step unchanged — +// exact identity, no delimiting at all +// B5 (fidelity): AfterOutcome hooks receive the outcome the pipeline +// acts on, never a fabricated one +// Monoid: Compose is associative with the zero value as unit +// ============================================================================= + +// TestBracketPairingLaw verifies B1: the after-hook fires exactly once +// whenever the before-hook fired, regardless of the wrapped step's outcome. +// +// What this prevents: +// - Spans that never close / timers that never stop on terminal steps, +// which are how every controller pipeline ends (queue.Done, queue.Requeue) +// - Metrics silently dropped on the cancellation (failure) path +func TestBracketPairingLaw(t *testing.T) { + outcomes := map[string]NewStep{ + "continues": Do(func(ctx context.Context) context.Context { return ctx }), + "terminal": Terminal, + "cancels": Do(func(ctx context.Context) context.Context { + c, cancel := context.WithCancel(ctx) + cancel() + return c + }), + } + + for name, step := range outcomes { + t.Run(name, func(t *testing.T) { + var before, after int + mw := Around( + func(ctx context.Context) context.Context { before++; return ctx }, + func(ctx context.Context) context.Context { after++; return ctx }, + ) + Run(t.Context(), mw.Wrap(step)) + require.Equal(t, 1, before, "before-hook should fire exactly once") + require.Equal(t, before, after, "after-hook must pair with before-hook") + }) + } +} + +// TestBracketNestingLaw verifies B2: nested brackets open and close in +// properly nested (LIFO) order, like defer or try/finally — and the two ways +// of nesting (Compose into one stack, or Wrap applied twice) agree. +// +// What this prevents: +// - Interleaved spans (A opens, B opens, A closes, B closes) that break +// tracing tools expecting well-nested extents +func TestBracketNestingLaw(t *testing.T) { + var trace []string + hook := func(s string) ContextFunc { + return func(ctx context.Context) context.Context { + trace = append(trace, s) + return ctx + } + } + mwA := Around(hook("beforeA"), hook("afterA")) + mwB := Around(hook("beforeB"), hook("afterB")) + + step := Do(func(ctx context.Context) context.Context { + trace = append(trace, "step") + return ctx + }) + + want := []string{"beforeA", "beforeB", "step", "afterB", "afterA"} + + Run(t.Context(), Compose(mwA, mwB).Wrap(step)) + require.Equal(t, want, trace, "Compose brackets LIFO") + + trace = nil + Run(t.Context(), mwA.Wrap(mwB.Wrap(step))) + require.Equal(t, want, trace, "nested Wrap agrees with Compose") +} + +// TestMiddlewareMonoidLaws verifies that Middleware forms a monoid under +// Compose: associative, with the zero value as identity. Because the type is +// sealed, these laws hold for every constructible value. +// +// What this prevents: +// - Registration order (one WithAmbientMiddleware call vs several) +// affecting behavior beyond the documented outermost-first ordering +func TestMiddlewareMonoidLaws(t *testing.T) { + var trace []string + hook := func(s string) ContextFunc { + return func(ctx context.Context) context.Context { + trace = append(trace, s) + return ctx + } + } + a := Around(hook("a+"), hook("a-")) + b := Around(hook("b+"), hook("b-")) + c := Around(hook("c+"), hook("c-")) + step := Do(func(ctx context.Context) context.Context { + trace = append(trace, "step") + return ctx + }) + + run := func(m Middleware) []string { + trace = nil + Run(t.Context(), m.Wrap(step)) + return append([]string(nil), trace...) + } + + left := run(Compose(Compose(a, b), c)) + right := run(Compose(a, Compose(b, c))) + require.Equal(t, left, right, "Compose must be associative") + require.Equal(t, []string{"a+", "b+", "c+", "step", "c-", "b-", "a-"}, left) + + require.Equal(t, run(a), run(Compose(Middleware{}, a)), "zero is a left identity") + require.Equal(t, run(a), run(Compose(a, Middleware{})), "zero is a right identity") + require.True(t, Compose().IsZero(), "empty Compose is the zero value") +} + +// TestBracketTransparencyLaw verifies B3: the bracket is transparent to the +// pipeline around it. Terminality is preserved (a bracketed terminal step +// still stops the pipeline), and when the step continues, downstream steps +// observe after(ctx') where ctx' is the context the step threaded forward. +// +// What this prevents: +// - A bracket resurrecting a terminated pipeline (running steps after +// queue.Done) +// - Context values written by the step or the after-hook vanishing before +// downstream steps +func TestBracketTransparencyLaw(t *testing.T) { + t.Run("terminality preserved", func(t *testing.T) { + var downstream bool + mw := Around( + func(ctx context.Context) context.Context { return ctx }, + func(ctx context.Context) context.Context { return ctx }, + ) + Run(t.Context(), Sequence( + mw.Wrap(Terminal), + Do(func(ctx context.Context) context.Context { + downstream = true + return ctx + }), + )) + require.False(t, downstream, "bracketed terminal step must still terminate the pipeline") + }) + + t.Run("context threads through step and after-hook", func(t *testing.T) { + var sawUser string + var sawProcessed bool + mw := After( + func(ctx context.Context) context.Context { + // after-hook sees the context the step threaded forward + return processedCtxKey.Set(ctx, true) + }, + ) + Run(t.Context(), Sequence( + mw.Wrap(Do(func(ctx context.Context) context.Context { + return userCtxKey.Set(ctx, "alice") + })), + Do(func(ctx context.Context) context.Context { + sawUser, _ = userCtxKey.Value(ctx) + sawProcessed, _ = processedCtxKey.Value(ctx) + return ctx + }), + )) + require.Equal(t, "alice", sawUser, "downstream must see context written by the bracketed step") + require.True(t, sawProcessed, "downstream must see context written by the after-hook") + }) +} + +// TestBracketIdentityLaw verifies B4: the zero Middleware is the exact +// identity. Because the type is sealed and Wrap of the zero value returns the +// step unchanged, this law is unqualified — there is no delimiting at all, so +// even a step that does work after invoking its continuation is unaffected. +// (The reordering caveat applies only when a stack carries after hooks, which +// is when delimiting is provably necessary — see the Absorption Theorem.) +func TestBracketIdentityLaw(t *testing.T) { + identity := Around(nil, nil) + require.True(t, identity.IsZero(), "Around(nil, nil) is the zero value") + + var trace []string + step := Do(func(ctx context.Context) context.Context { + trace = append(trace, "step") + return userCtxKey.Set(ctx, "alice") + }) + downstream := Do(func(ctx context.Context) context.Context { + user, _ := userCtxKey.Value(ctx) + trace = append(trace, "downstream:"+user) + return ctx + }) + + Run(t.Context(), Sequence(identity.Wrap(step), downstream)) + require.Equal(t, []string{"step", "downstream:alice"}, trace, + "the zero Middleware must be observationally identity") + + // Terminality is also preserved by the identity. + trace = nil + Run(t.Context(), Sequence(identity.Wrap(Terminal), downstream)) + require.Empty(t, trace, "the zero Middleware must preserve terminality") +} + +// TestBracketReleaseOnPanic verifies the panic side of B1: the bracket closes +// during unwinding whether or not anything above will recover the panic. +// +// What this prevents: +// - A span left open or a timer left running when a step panics, so the +// last thing a crashing controller emits is missing the step that broke +func TestBracketReleaseOnPanic(t *testing.T) { + hook := func(trace *[]string, s string) ContextFunc { + return func(ctx context.Context) context.Context { + *trace = append(*trace, s) + return ctx + } + } + panicStep := NewStepFunc(func(_ context.Context, _ Step) Step { + panic("boom") + }) + + t.Run("panic recovered above", func(t *testing.T) { + var trace []string + mw := Around(hook(&trace, "before"), hook(&trace, "after")) + + recovery := func(step NewStep) NewStep { + return func(next Step) Step { + return StepFunc(func(ctx context.Context) (result Step) { + defer func() { + if r := recover(); r != nil { + trace = append(trace, "recovered") + } + }() + return step(next).Run(ctx) + }) + } + } + + require.NotPanics(t, func() { + Run(t.Context(), recovery(mw.Wrap(panicStep))) + }) + require.Equal(t, []string{"before", "after", "recovered"}, trace, + "bracket must close during unwinding, before recovery observes the panic") + }) + + t.Run("panic not recovered", func(t *testing.T) { + var trace []string + mw := Around(hook(&trace, "before"), hook(&trace, "after")) + + // The bracket does not call recover, so the panic still propagates + // with its original value. + require.PanicsWithValue(t, "boom", func() { + Run(t.Context(), mw.Wrap(panicStep)) + }, "panic must propagate unmodified") + require.Equal(t, []string{"before", "after"}, trace, + "bracket must close during unwinding even when the panic is fatal") + }) +} + +// TestObservationHooksCannotRecover pins the constructor-allocated recover +// capability, denied side. recover is frame-scoped in Go — effective only in +// a function called directly from the panicking defer chain — and +// After interposes an adapter closure as the defer operand, so an +// observation hook calling recover() receives nil and the panic keeps +// propagating. Contrast Deferred, whose hook is the operand +// itself and holds the capability (TestDeferredHooksCanRecover). +func TestObservationHooksCannotRecover(t *testing.T) { + var recovered any = "sentinel" + mw := After(func(ctx context.Context) context.Context { + recovered = recover() + return ctx + }) + panicStep := NewStepFunc(func(_ context.Context, _ Step) Step { + panic("boom") + }) + + require.PanicsWithValue(t, "boom", func() { + Run(t.Context(), mw.Wrap(panicStep)) + }, "panic must propagate despite the hook's recover attempt") + require.Nil(t, recovered, "recover inside an observation hook must return nil") +} + +// TestDeferredHooksCanRecover pins the granted side: a Deferred-constructed +// hook is the defer operand itself, so recover called directly in its body is +// effective. Swallowing converts the panic into a termination outcome; +// re-panicking keeps the crash propagating; and on the normal path the hook +// participates in context threading through the pointer. +func TestDeferredHooksCanRecover(t *testing.T) { + panicStep := NewStepFunc(func(_ context.Context, _ Step) Step { + panic("boom") + }) + + t.Run("swallow converts panic to termination", func(t *testing.T) { + var recovered any + var downstream bool + mw := Deferred(func(_ *context.Context) { + recovered = recover() + }) + require.NotPanics(t, func() { + Run(t.Context(), Sequence( + mw.Wrap(panicStep), + Do(func(ctx context.Context) context.Context { + downstream = true + return ctx + }), + )) + }) + require.Equal(t, "boom", recovered, "the hook must receive the panic value") + require.False(t, downstream, "a swallowed panic terminates the pipeline like a terminal step") + }) + + t.Run("re-panic keeps the crash propagating", func(t *testing.T) { + mw := Deferred(func(_ *context.Context) { + if r := recover(); r != nil { + panic(r) + } + }) + require.PanicsWithValue(t, "boom", func() { + Run(t.Context(), mw.Wrap(panicStep)) + }) + }) + + t.Run("threads context on the normal path", func(t *testing.T) { + var sawUser string + mw := Deferred(func(ctx *context.Context) { + *ctx = processedCtxKey.Set(*ctx, true) + }) + Run(t.Context(), Sequence( + mw.Wrap(Do(func(ctx context.Context) context.Context { + return userCtxKey.Set(ctx, "alice") + })), + Do(func(ctx context.Context) context.Context { + sawUser, _ = userCtxKey.Value(ctx) + processed, _ := processedCtxKey.Value(ctx) + require.True(t, processed, "downstream must see the deferred hook's pointer write") + return ctx + }), + )) + require.Equal(t, "alice", sawUser, "deferred hooks observe the step's threaded context (B3)") + }) +} + +// TestCrashHandlerIsDeferOperand pins that a handler passed whole +// to CrashHandler is itself the defer operand: its own recover is +// effective (the property that a handler wrapped in a closure would lose), +// and it receives the bracket-entry context by value. +func TestCrashHandlerIsDeferOperand(t *testing.T) { + panicStep := NewStepFunc(func(_ context.Context, _ Step) Step { + panic("boom") + }) + + // Same shape as utilruntime.HandleCrashWithContext, without the k8s + // dependency: recovers in its own body, records what it saw. + var got any + var sawUser string + handler := func(ctx context.Context, _ ...func(context.Context, any)) { + sawUser, _ = userCtxKey.Value(ctx) + got = recover() + } + + ctx := userCtxKey.Set(context.Background(), "alice") + require.NotPanics(t, func() { + Run(ctx, CrashHandler(handler).Wrap(panicStep)) + }, "the handler's own recover must intercept the panic") + require.Equal(t, "boom", got) + require.Equal(t, "alice", sawUser, "the handler receives the bracket-entry context") +} + +// TestOutcomeFidelityLaw verifies B5: an outcome-aware after-hook receives +// the outcome the pipeline acts on — never a fabricated one. Each subtest +// drives the wrapped step to one of the four fates and checks that the hook +// observes exactly that fate. +// +// What this prevents: +// - Spans marked Ok around steps that panicked or were cancelled +// - Metrics that count a cancellation as a normal termination +func TestOutcomeFidelityLaw(t *testing.T) { + observe := func() (*Outcome, Middleware) { + var got Outcome + mw := AfterOutcome(func(ctx context.Context, o Outcome) context.Context { + got = o + return ctx + }) + return &got, mw + } + + t.Run("continued", func(t *testing.T) { + got, mw := observe() + Run(t.Context(), mw.Wrap(Do(func(ctx context.Context) context.Context { return ctx }))) + require.Equal(t, OutcomeContinued, got.Kind) + require.NoError(t, got.Cause) + }) + + t.Run("terminated", func(t *testing.T) { + got, mw := observe() + Run(t.Context(), mw.Wrap(Terminal)) + require.Equal(t, OutcomeTerminated, got.Kind) + require.NoError(t, got.Cause) + }) + + t.Run("cancelled with cause", func(t *testing.T) { + cause := errors.New("deadline blown") + got, mw := observe() + Run(t.Context(), mw.Wrap(Do(func(ctx context.Context) context.Context { + c, cancel := context.WithCancelCause(ctx) + cancel(cause) + return c + }))) + require.Equal(t, OutcomeCancelled, got.Kind) + require.Equal(t, cause, got.Cause, "the hook must receive the cancellation cause") + }) + + t.Run("panicked", func(t *testing.T) { + got, mw := observe() + require.PanicsWithValue(t, "boom", func() { + Run(t.Context(), mw.Wrap(NewStepFunc(func(_ context.Context, _ Step) Step { + panic("boom") + }))) + }, "observing the outcome must not stop the panic") + require.Equal(t, OutcomePanicked, got.Kind) + }) + + t.Run("terminated with attested detail and cause", func(t *testing.T) { + cause := errors.New("sync failed") + got, mw := observe() + Run(t.Context(), mw.Wrap(NewTerminalStepFunc(func(ctx context.Context) { + RecordTermination(ctx, "requeued", cause) + }))) + require.Equal(t, OutcomeTerminated, got.Kind) + require.Equal(t, "requeued", got.Detail) + require.Equal(t, cause, got.Cause) + }) + + t.Run("attestation dropped when the step continues", func(t *testing.T) { + // Detail is attested, not derived: a step that records a termination + // but continues anyway reports Continued with the attestation gone. + got, mw := observe() + Run(t.Context(), mw.Wrap(Do(func(ctx context.Context) context.Context { + RecordTermination(ctx, "done", nil) + return ctx + }))) + require.Equal(t, OutcomeContinued, got.Kind) + require.Empty(t, got.Detail) + }) + + t.Run("attested termination outranks recorded cancellation", func(t *testing.T) { + // Queue operations cancel the context as an implementation detail of + // stopping the pipeline; the attested termination is the truth. + got, mw := observe() + Run(t.Context(), mw.Wrap(NewStepFunc(func(ctx context.Context, next Step) Step { + RecordTermination(ctx, "done", nil) + c, cancel := context.WithCancel(ctx) + cancel() + return Continue(c, next) // records cancellation into the same slot + }))) + require.Equal(t, OutcomeTerminated, got.Kind) + require.Equal(t, "done", got.Detail) + require.NoError(t, got.Cause) + }) +} + +// TestOutcomeHooksCannotRecover pins that outcome-aware hooks hold the +// observation capability only: they learn that the step panicked (B5) but, +// like After hooks, run below a library-owned defer operand — so +// recover inside them is a no-op and the panic keeps propagating. +func TestOutcomeHooksCannotRecover(t *testing.T) { + var recovered any = "sentinel" + mw := AfterOutcome(func(ctx context.Context, _ Outcome) context.Context { + recovered = recover() + return ctx + }) + require.PanicsWithValue(t, "boom", func() { + Run(t.Context(), mw.Wrap(NewStepFunc(func(_ context.Context, _ Step) Step { + panic("boom") + }))) + }) + require.Nil(t, recovered, "recover inside an outcome hook must return nil") +} + +// TestParallelCommutativityLaw verifies that branch order in Parallel is +// observationally irrelevant: Parallel(a, b, c) ≡ Parallel(c, a, b). +// +// Why it holds: branches are isolated (each receives the same input context, +// and its context modifications are discarded) and Parallel waits for every +// branch before continuing, so no observable effect depends on the order in +// which branches are listed. +func TestParallelCommutativityLaw(t *testing.T) { + run := func(order ...string) (observed map[string]bool, continued bool) { + var mu sync.Mutex + observed = map[string]bool{} + steps := make([]NewStep, 0, len(order)) + for _, name := range order { + steps = append(steps, Do(func(ctx context.Context) context.Context { + mu.Lock() + defer mu.Unlock() + observed[name] = true + return ctx + })) + } + Run(t.Context(), Sequence( + Parallel(steps...), + Do(func(ctx context.Context) context.Context { + continued = true + return ctx + }), + )) + return observed, continued + } + + abcObserved, abcContinued := run("a", "b", "c") + cabObserved, cabContinued := run("c", "a", "b") + + require.Equal(t, abcObserved, cabObserved, "Parallel commutativity law violated") + require.True(t, abcContinued) + require.True(t, cabContinued) +} diff --git a/state/middleware/doc.go b/state/middleware/doc.go new file mode 100644 index 0000000..d4cb747 --- /dev/null +++ b/state/middleware/doc.go @@ -0,0 +1,32 @@ +// Package middleware provides ready-made state.Middleware implementations +// for common cross-cutting concerns. +// +// # Available Middleware +// +// - Log: logs each step's name, elapsed time, and outcome, escalating the +// level on cancellation and panic +// +// # Usage with AmbientDispatch +// +// The typical pattern is to register middleware via WithAmbientMiddleware and +// run the pipeline with AmbientDispatch: +// +// ctx = state.WithAmbientMiddleware(ctx, middleware.Log(slog.Default())) +// state.Run(ctx, state.AmbientDispatch(step1, step2, step3)) +// +// # Panics +// +// This package deliberately ships no panic-recovery middleware. A panic in a +// step is a programming error, and controller runtimes generally install their +// own crash handler around the worker loop — one that logs the panic with its +// stack and then lets the process die. Recovering inside the pipeline would +// convert that loud, debuggable crash into a silently dropped reconcile. +// +// state.Parallel branches run on their own goroutines, so their panics bypass +// such a handler when it wraps the pipeline from outside — recover cannot +// cross a goroutine boundary. To cover branches, register the crash handler +// as ambient middleware: state.CrashHandler accepts +// utilruntime.HandleCrashWithContext directly, and state.Parallel applies the +// ambient stack inside every branch. Observation hooks like Log's still close +// during the unwind, so the crashing step is logged on the way out. +package middleware diff --git a/state/middleware/middleware.go b/state/middleware/middleware.go new file mode 100644 index 0000000..2de9028 --- /dev/null +++ b/state/middleware/middleware.go @@ -0,0 +1,71 @@ +// Package middleware provides ready-made state.Middleware implementations +// for common cross-cutting concerns. +package middleware + +import ( + "context" + "log/slog" + "time" + + "github.com/authzed/ctxkey" + + "github.com/authzed/controller-idioms/state" +) + +var logStartKey = ctxkey.New[time.Time]() + +// Log returns a Middleware that logs each step's name, elapsed time, and +// outcome after the step executes. It uses state.Named for the step name +// (empty string if the step was not annotated). Elapsed time is logged as +// duration_ms (float64, milliseconds); the outcome is logged as a stable +// label (continued, terminated, cancelled, panicked), plus the terminal +// operation's detail ("done", "requeued") and cause when recorded. The level +// escalates with the outcome: Info normally, Warn on cancellation or an +// error-carrying termination (an error requeue is a failed reconcile +// attempt), Error on panic — so the last line a crashing controller emits +// identifies the step that broke. +// +// Log is built on state.Before and state.AfterOutcome, so it is a bracket +// (see FORMAL.md, "The Bracket Laws"): the log line is emitted on every +// outcome — including while a panic unwinds — and the measured duration +// covers only the wrapped step, not the downstream steps that CPS would +// otherwise run inline within the step's frame. +// +// Typical controller usage: +// +// ctx = state.WithAmbientMiddleware(ctx, middleware.Log(slog.Default())) +func Log(logger *slog.Logger) state.Middleware { + return state.Compose( + state.Before(func(ctx context.Context) context.Context { + ctx = state.WithStepNameCapture(ctx) + return logStartKey.Set(ctx, time.Now()) + }), + state.AfterOutcome(func(ctx context.Context, outcome state.Outcome) context.Context { + start, _ := logStartKey.Value(ctx) + level := slog.LevelInfo + switch { + case outcome.Kind == state.OutcomePanicked: + level = slog.LevelError + case outcome.Kind == state.OutcomeCancelled, + outcome.Kind == state.OutcomeTerminated && outcome.Cause != nil: + level = slog.LevelWarn + } + attrs := []any{ + "step", state.CapturedStepName(ctx), + "duration_ms", float64(time.Since(start).Microseconds()) / 1000.0, + "outcome", outcome.Kind.String(), + } + if outcome.Detail != "" { + attrs = append(attrs, "detail", outcome.Detail) + } + if outcome.Cause != nil { + attrs = append(attrs, "cause", outcome.Cause.Error()) + } + logger.Log(ctx, level, "step executed", attrs...) + return ctx + }), + ) +} + +// Log must be usable wherever a state.Middleware is expected. +var _ state.Middleware = Log(slog.Default()) diff --git a/state/middleware/middleware_test.go b/state/middleware/middleware_test.go new file mode 100644 index 0000000..f0442b5 --- /dev/null +++ b/state/middleware/middleware_test.go @@ -0,0 +1,205 @@ +package middleware_test + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "log/slog" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/authzed/controller-idioms/state" + "github.com/authzed/controller-idioms/state/middleware" +) + +func TestLogRecordsStepNameAndDuration(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + + ctx := state.WithAmbientMiddleware(t.Context(), middleware.Log(logger)) + state.Run(ctx, state.AmbientDispatch( + state.Named("myStep", state.Do(func(ctx context.Context) context.Context { return ctx })), + )) + + var entry map[string]any + require.NoError(t, json.Unmarshal(buf.Bytes(), &entry)) + require.Equal(t, "myStep", entry["step"]) + _, hasDuration := entry["duration_ms"] + require.True(t, hasDuration, "log entry should contain duration_ms") +} + +func TestLogUnnamedStepLogsEmptyName(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + + ctx := state.WithAmbientMiddleware(t.Context(), middleware.Log(logger)) + state.Run(ctx, state.AmbientDispatch( + state.Do(func(ctx context.Context) context.Context { return ctx }), + )) + + var entry map[string]any + require.NoError(t, json.Unmarshal(buf.Bytes(), &entry)) + name, ok := entry["step"] + require.True(t, ok, "log entry should contain step") + require.Empty(t, name) +} + +func TestLogRecordsNameForTerminalStep(t *testing.T) { + // Named terminal steps (those that never call their continuation) must still + // log the step name correctly. + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + + ctx := state.WithAmbientMiddleware(t.Context(), middleware.Log(logger)) + state.Run(ctx, state.AmbientDispatch( + state.Named("cleanup", state.Terminal), + )) + + var entry map[string]any + require.NoError(t, json.Unmarshal(buf.Bytes(), &entry)) + require.Equal(t, "cleanup", entry["step"]) + _, hasDuration := entry["duration_ms"] + require.True(t, hasDuration, "log entry should contain duration_ms") +} + +// decodeEntries reads the JSON log lines emitted by Log. +func decodeEntries(t *testing.T, buf *bytes.Buffer) []map[string]any { + t.Helper() + dec := json.NewDecoder(bytes.NewReader(buf.Bytes())) + var out []map[string]any + for dec.More() { + var e map[string]any + require.NoError(t, dec.Decode(&e)) + out = append(out, e) + } + return out +} + +// Log runs inside each branch goroutine and allocates its own name-capture +// slot per branch, so every branch is reported separately with its own name +// and duration. +func TestLogReportsEachParallelBranch(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + branch := func(name string) state.NewStep { + return state.Named(name, state.Do(func(c context.Context) context.Context { return c })) + } + + ctx := state.WithAmbientMiddleware(t.Context(), middleware.Log(logger)) + state.Run(ctx, state.Parallel(branch("alpha"), branch("beta"), branch("gamma"))) + + entries := decodeEntries(t, &buf) + require.Len(t, entries, 3, "each branch should be logged") + names := make([]string, 0, len(entries)) + for _, e := range entries { + names = append(names, e["step"].(string)) + require.Contains(t, e, "duration_ms") + } + require.ElementsMatch(t, []string{"alpha", "beta", "gamma"}, names) +} + +// Each step is logged individually, and a step's log line is emitted when the +// step completes — before downstream steps run — because Log brackets the step +// rather than timing its whole CPS continuation. +func TestLogAttributesPerStep(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + + ctx := state.WithAmbientMiddleware(t.Context(), middleware.Log(logger)) + state.Run(ctx, state.AmbientDispatch( + state.Named("first", state.Do(func(c context.Context) context.Context { return c })), + state.Named("second", state.Do(func(c context.Context) context.Context { return c })), + )) + + entries := decodeEntries(t, &buf) + require.Len(t, entries, 2) + require.Equal(t, "first", entries[0]["step"], "first step's line is emitted before downstream runs") + require.Equal(t, "second", entries[1]["step"]) +} + +// Log classifies each step's result: the outcome label reflects the fate the +// pipeline acted on, the level escalates on failure, and a cancellation +// carries its cause. +func TestLogRecordsOutcome(t *testing.T) { + newLogger := func() (*bytes.Buffer, *slog.Logger) { + var buf bytes.Buffer + return &buf, slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + } + run := func(logger *slog.Logger, step state.NewStep) { + ctx := state.WithAmbientMiddleware(context.Background(), middleware.Log(logger)) + state.Run(ctx, state.AmbientDispatch(step)) + } + + t.Run("continued at info", func(t *testing.T) { + buf, logger := newLogger() + run(logger, state.Do(func(c context.Context) context.Context { return c })) + entry := decodeEntries(t, buf)[0] + require.Equal(t, "continued", entry["outcome"]) + require.Equal(t, "INFO", entry["level"]) + }) + + t.Run("terminated at info", func(t *testing.T) { + buf, logger := newLogger() + run(logger, state.Terminal) + entry := decodeEntries(t, buf)[0] + require.Equal(t, "terminated", entry["outcome"]) + require.Equal(t, "INFO", entry["level"]) + }) + + t.Run("cancelled at warn with cause", func(t *testing.T) { + buf, logger := newLogger() + run(logger, state.Do(func(c context.Context) context.Context { + cc, cancel := context.WithCancelCause(c) + cancel(errors.New("deadline blown")) + return cc + })) + entry := decodeEntries(t, buf)[0] + require.Equal(t, "cancelled", entry["outcome"]) + require.Equal(t, "WARN", entry["level"]) + require.Equal(t, "deadline blown", entry["cause"]) + }) + + t.Run("terminated with detail and cause at warn", func(t *testing.T) { + buf, logger := newLogger() + run(logger, state.NewTerminalStepFunc(func(ctx context.Context) { + state.RecordTermination(ctx, "requeued", errors.New("sync failed")) + })) + entry := decodeEntries(t, buf)[0] + require.Equal(t, "terminated", entry["outcome"]) + require.Equal(t, "requeued", entry["detail"]) + require.Equal(t, "sync failed", entry["cause"]) + require.Equal(t, "WARN", entry["level"]) + }) + + t.Run("panicked at error", func(t *testing.T) { + buf, logger := newLogger() + require.PanicsWithValue(t, "boom", func() { + run(logger, state.Named("culprit", state.NewStepFunc( + func(_ context.Context, _ state.Step) state.Step { panic("boom") }, + ))) + }) + entry := decodeEntries(t, buf)[0] + require.Equal(t, "panicked", entry["outcome"]) + require.Equal(t, "ERROR", entry["level"]) + require.Equal(t, "culprit", entry["step"], "the last line before the crash names the step") + }) +} + +// Clearing the stack exempts a subtree, so nothing downstream is logged. +func TestLogSilencedByWithoutAmbientMiddleware(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + branch := func(name string) state.NewStep { + return state.Named(name, state.Do(func(c context.Context) context.Context { return c })) + } + + ctx := state.WithAmbientMiddleware(t.Context(), middleware.Log(logger)) + state.Run(ctx, state.Sequence( + state.Do(state.WithoutAmbientMiddleware), + state.Parallel(branch("alpha"), branch("beta")), + )) + + require.Empty(t, decodeEntries(t, &buf), "cleared stack should log nothing") +} diff --git a/state/state.go b/state/state.go index b517024..6631e39 100644 --- a/state/state.go +++ b/state/state.go @@ -37,9 +37,11 @@ package state import ( "context" - "fmt" "slices" "sync" + "sync/atomic" + + "github.com/authzed/ctxkey" ) // Step represents a step in a processing pipeline. @@ -64,6 +66,577 @@ type ContextFunc func(context.Context) context.Context // This is the basic building block for composing step pipelines using continuation-passing style. type NewStep func(next Step) Step +// Middleware is an opaque, immutable stack of paired before/after hooks that +// dispatch applies around each step. The zero value is the identity +// middleware, and Compose makes Middleware a monoid. +// +// Middleware is sealed: the only constructors are the hook generators — +// Before, After, AfterOutcome, Deferred, and CrashHandler, plus the Around +// pair — and Compose. This is what makes the laws in FORMAL.md ("The +// Bracket Laws") theorems about the type rather than conventions for its +// users — a Middleware value is hook data, not code, so it cannot invoke a +// step twice, substitute a continuation, or resurrect a terminated +// pipeline. The one control capability a hook can hold is scoped and +// constructor-chosen: Deferred and CrashHandler hooks run on the unwinding +// defer chain and may recover a panic (converting it into a termination, or +// re-raising it), while Before/After/AfterOutcome hooks are pure +// observations and cannot. Everything else lives in the interpreter (see +// Wrap and AmbientDispatch). +// +// For step transformations that hooks cannot express — retries, panic +// recovery, custom scoping — write a plain func(NewStep) NewStep and apply it +// directly or with Map. Such wrappers are ordinary composition: they make no +// law claims and do not ride the ambient dispatch path. +// +// middleware.Log (in state/middleware) is an example of a Middleware. +type Middleware struct { + // hooks[0] is outermost: its before hook runs first, its after hook last. + hooks []hook +} + +// hook is one paired before/after unit. Any field may be nil. The after +// forms are stored so runBracketed can defer them directly; the constructors +// decide whether the operand is a library adapter (After: observation-only) +// or the user's own function (Deferred and CrashHandler: on the defer +// chain, recover-capable). +type hook struct { + before ContextFunc + after func(*context.Context) + // outcomeAfter is an observation hook that also receives the Outcome of + // the wrapped step (see AfterOutcome). Called via a library + // closure, so it cannot recover. + outcomeAfter func(context.Context, Outcome) context.Context + // crashAfter is deferred with the bracket-entry context as a value. + // Its shape matches crash-handler conventions (see CrashHandler). + crashAfter func(context.Context, ...func(context.Context, any)) +} + +// hasAfter reports whether the hook carries any after form. +func (h hook) hasAfter() bool { + return h.after != nil || h.outcomeAfter != nil || h.crashAfter != nil +} + +// OutcomeKind classifies how a bracketed step's extent ended. +type OutcomeKind int + +const ( + // OutcomeContinued: the step invoked its continuation; the pipeline + // carries on downstream. + OutcomeContinued OutcomeKind = iota + // OutcomeTerminated: the step ended the pipeline deliberately (a + // terminal step such as Terminal, queue.Done, or queue.Requeue). + OutcomeTerminated + // OutcomeCancelled: the step stopped because its context was cancelled. + OutcomeCancelled + // OutcomePanicked: the step panicked; the hook observing this outcome is + // running while the panic unwinds. + OutcomePanicked +) + +// String returns a stable lowercase label suitable for metrics and logs. +func (k OutcomeKind) String() string { + switch k { + case OutcomeContinued: + return "continued" + case OutcomeTerminated: + return "terminated" + case OutcomeCancelled: + return "cancelled" + case OutcomePanicked: + return "panicked" + default: + return "unknown" + } +} + +// Outcome reports how a bracketed step's extent ended, as the pipeline +// itself acted on it (the fidelity law, B5 in FORMAL.md). +type Outcome struct { + Kind OutcomeKind + // Cause is the cancellation cause when Kind is OutcomeCancelled, or the + // error recorded by the terminal operation (e.g. queue.RequeueErr) when + // Kind is OutcomeTerminated with a Detail. The panic value is + // deliberately not exposed here: observation hooks run during the unwind + // without recovering, so the value remains attached to the propagating + // panic. + Cause error + // Detail is a terminal operation's self-reported refinement of a + // Terminated outcome — "done" or "requeued" from the queue package — + // and empty otherwise. See RecordTermination. + Detail string +} + +// outcomeCapture is the slot a bracket plants in its step's context so the +// step's extent can report how it ended: the cancelled context and the +// terminal operation's label are both produced inside the step and would +// otherwise never escape it. The pointers are atomic because Parallel +// branches sharing the step's context may report concurrently; the first +// writer wins. +type outcomeCapture struct { + cancelCause atomic.Pointer[error] + term atomic.Pointer[termination] +} + +// termination is a terminal operation's attested outcome refinement. +type termination struct { + detail string + cause error +} + +var outcomeCaptureKey = ctxkey.New[*outcomeCapture]() + +// recordCancellation fills the nearest outcome-capture slot with the +// cancellation cause, if a bracket planted one. Called by Continue on the +// cancellation path. +func recordCancellation(ctx context.Context, cause error) { + if slot, ok := outcomeCaptureKey.Value(ctx); ok && slot != nil { + if cause == nil { + cause = context.Canceled + } + slot.cancelCause.CompareAndSwap(nil, &cause) + } +} + +// RecordTermination annotates the enclosing bracket's outcome: the pipeline +// is being terminated deliberately, with a short label for observability +// ("done", "requeued") and an optional error. Outcome-aware hooks then +// receive Outcome{Kind: OutcomeTerminated, Detail: detail, Cause: cause} +// instead of a bare termination. +// +// Detail is attested, not derived: the interpreter reports it only when the +// step in fact terminated — a step that records a termination and then +// continues anyway reports OutcomeContinued and the attestation is dropped. +// An attested termination also takes precedence over a recorded +// cancellation, because queue operations cancel the context as an +// implementation detail of stopping the pipeline. +// +// RecordTermination is a no-op when no outcome-aware hook is registered +// around the step. The queue package calls it from its OperationsContext +// methods, so queue.Done, queue.Requeue, and hand-rolled steps that invoke +// queue operations directly are all annotated without further wiring. +func RecordTermination(ctx context.Context, detail string, cause error) { + if slot, ok := outcomeCaptureKey.Value(ctx); ok && slot != nil { + slot.term.CompareAndSwap(nil, &termination{detail: detail, cause: cause}) + } +} + +// Before returns a Middleware with a single before hook: fn runs ahead of +// the wrapped step, and the context it returns threads into the step. A nil +// fn returns the zero (identity) Middleware. +func Before(fn ContextFunc) Middleware { + if fn == nil { + return Middleware{} + } + return Middleware{hooks: []hook{{before: fn}}} +} + +// After returns a Middleware with a single after hook: fn runs once the +// wrapped step completes — including when the step is terminal, cancels the +// context, or panics, none of which invoke the continuation — so finalizers +// (stop a timer, end a span, record a metric) always fire. +// +// The hook is released by a defer, so it also runs while a panic unwinds. It +// is observation-only: a library adapter, not fn itself, is the defer +// operand, so recover inside fn is a no-op and the panic propagates with its +// value and stack untouched (see Deferred for the recover-capable form). As +// with any finalizer, a hook that itself panics masks the original panic. +// +// fn runs for its side effects: the context it returns is threaded to +// subsequent steps only when the wrapped step continued the pipeline. +// +// In a composed stack, before hooks execute in composition order and +// after-form hooks close in reverse composition order (see Compose). A nil +// fn returns the zero (identity) Middleware. +func After(fn ContextFunc) Middleware { + if fn == nil { + return Middleware{} + } + return Middleware{hooks: []hook{{ + after: func(p *context.Context) { *p = fn(*p) }, + }}} +} + +// Around returns the classic paired bracket — Compose(Before(before), +// After(after)): before runs ahead of the step, and after is guaranteed once +// the step completes, on every outcome. Either hook may be nil; if both are +// nil the result is the zero (identity) Middleware. +func Around(before, after ContextFunc) Middleware { + return Compose(Before(before), After(after)) +} + +// AfterOutcome returns a Middleware with a single after hook that also +// receives the Outcome of the wrapped step: continued, terminated (with any +// attested detail), cancelled (with the cancellation cause), or panicked. +// Use it for observability that should classify results — span statuses, +// outcome-labeled metrics, log levels that escalate on failure. +// +// AfterOutcome hooks hold the same capability as After hooks: observation +// only. On the panic path the hook runs while the panic unwinds with Kind +// OutcomePanicked, but the panic value is not exposed and recover inside the +// hook is a no-op — interception requires Deferred or CrashHandler. +// +// As with After, fn runs for its side effects: its returned context is +// threaded onward only when the step continued. A nil fn returns the zero +// (identity) Middleware. +func AfterOutcome(fn func(context.Context, Outcome) context.Context) Middleware { + if fn == nil { + return Middleware{} + } + return Middleware{hooks: []hook{{outcomeAfter: fn}}} +} + +// Deferred returns a Middleware whose single after hook is the defer +// operand itself: it runs as `defer fn(&ctx)` around the wrapped step, so +// the hook sits on the panicking defer chain and a recover() written +// directly in its body is effective. This is a constructor for hand-written +// crash-handler middleware; registered ambient, it covers every Parallel +// branch on the branch's own goroutine. +// +// Semantics beyond After hooks: +// +// - recover() in the hook body intercepts a panicking step. Recovering +// without re-panicking converts the panic into a termination — the +// pipeline stops as if the step were terminal, and responsibility for +// the abandoned work is the hook's. Re-panic (panic(r)) to observe the +// panic and keep the crash loud. +// - The recover must appear literally in the hook body. Go makes recover +// effective only when called directly by the deferred function, so +// helpers that recover internally silently do nothing when called from +// the hook. To use such a helper (utilruntime.HandleCrashWithContext, +// for example) as the hook itself, see CrashHandler, whose hook +// signature matches it directly. +// - On the normal path the hook still runs (recover returns nil when no +// panic is active); read and write the context through the pointer, +// which observes the context the step threaded forward (B3). +// +// A nil fn returns the zero (identity) Middleware. +func Deferred(fn func(*context.Context)) Middleware { + if fn == nil { + return Middleware{} + } + return Middleware{hooks: []hook{{after: fn}}} +} + +// CrashHandler is the deferred-hook variant whose signature matches +// crash-handler conventions: the hook receives the bracket-entry context by +// value and is deferred directly, so a recover() in its body — or in the +// body of a handler passed whole, since the handler itself is the defer +// operand — is effective. +// +// The shape is chosen so that utilruntime.HandleCrashWithContext is +// assignable as-is (the variadic parameter is why a plain func(ctx) variant +// would not accept it): +// +// ctx = state.WithAmbientMiddleware(ctx, state.CrashHandler(utilruntime.HandleCrashWithContext)) +// +// This must be the whole hook, not a call inside one: recover is effective +// only in the deferred function itself, so wrapping the handler in a closure +// silently disables it. +// +// Unlike Deferred's pointer hook, the context is passed by value as observed +// at bracket entry — the hook cannot amend the forwarded context, which +// suits crash handlers: on the panic path there is no threaded context +// anyway, and the entry context carries the loggers and values the handler +// needs. +// +// A nil fn returns the zero (identity) Middleware. +func CrashHandler(fn func(context.Context, ...func(context.Context, any))) Middleware { + if fn == nil { + return Middleware{} + } + return Middleware{hooks: []hook{{crashAfter: fn}}} +} + +// Compose combines middleware into one stack. The first argument is outermost +// — its before hook fires first and its after hook last — matching the +// registration order of WithAmbientMiddleware. In general: before hooks +// execute in composition order, and after-form hooks close in reverse +// composition order, like nested defers. Compose is associative and the +// zero Middleware is its identity, making Middleware a monoid; these laws are +// verified in formal_test.go. +func Compose(mws ...Middleware) Middleware { + var hooks []hook + for _, m := range mws { + hooks = append(hooks, m.hooks...) + } + return Middleware{hooks: hooks} +} + +// IsZero reports whether m is the identity middleware (no hooks). +func (m Middleware) IsZero() bool { return len(m.hooks) == 0 } + +// Wrap applies the middleware to a single step: m's before hooks run +// (outermost first), then the step, then m's after hooks (innermost first), +// with the after hooks guaranteed on every outcome — continue, terminal, +// cancel, or panic unwind. Wrap is the interpreter for Middleware, and it +// uses the least machinery each stack requires: +// +// - zero middleware: returns step unchanged — exact identity. +// - before hooks only: stays in the pure algebra — the result is literally +// Sequence(Do(before)..., step), so it inherits the Kleisli laws. +// - any after hook: delimits the step with the package's one control +// operator (see runBracketed), because a finalizer that survives terminal +// steps is not expressible by Kleisli composition (the Absorption Theorem +// in FORMAL.md). +func (m Middleware) Wrap(step NewStep) NewStep { + if len(m.hooks) == 0 { + return step + } + hasAfter := false + for _, h := range m.hooks { + if h.hasAfter() { + hasAfter = true + break + } + } + if !hasAfter { + steps := make([]NewStep, 0, len(m.hooks)+1) + for _, h := range m.hooks { + if h.before != nil { + steps = append(steps, Do(h.before)) + } + } + return Sequence(append(steps, step)...) + } + hooks := m.hooks + return func(next Step) Step { + return StepFunc(func(ctx context.Context) Step { + continued, fwd := runBracketed(ctx, hooks, step) + if !continued { + return nil + } + return Continue(fwd, next) + }) + } +} + +// runBracketed is the single control operator in the package. It runs step +// against a sentinel continuation so control returns here even when the step +// is terminal or cancels the context, and it defers the after hooks so they +// close in LIFO order and also run while a panic unwinds. The named result +// fwd is threaded through the hooks by pointer: a hook observes the context +// the step threaded forward and may amend it, on the normal path and during +// unwind alike. +// +// The defer operand determines who may recover: Go makes recover effective +// only when called directly by a function on the panicking defer chain, and +// h.after is deferred directly — so the constructors allocate the +// capability. After and AfterOutcome interpose an adapter closure, keeping +// their hooks observation-only (recover inside them is a no-op), while +// Deferred and CrashHandler store the user's function as the operand itself, +// granting it recover deliberately. runBracketed never recovers on its own: +// an unintercepted panic propagates with value and stack untouched, and a +// deferred hook that recovers without re-panicking leaves continued=false, +// which Wrap reports as a termination. +func runBracketed(ctx context.Context, hooks []hook, step NewStep) (continued bool, fwd context.Context) { + fwd = ctx + var completed bool + // Plant an outcome-capture slot only when an outcome-aware hook will + // need to discriminate how the step ended. A fresh slot per bracket + // shadows any slot inherited from an enclosing or preceding bracket, so + // outcomes never bleed across steps or branches. + var slot *outcomeCapture + for _, h := range hooks { + if h.outcomeAfter != nil { + slot = &outcomeCapture{} + fwd = outcomeCaptureKey.Set(fwd, slot) + break + } + } + // outcome is evaluated lazily inside the deferred hooks, when the + // extent's fate is known: a panic leaves completed false; otherwise the + // sentinel and the capture slot discriminate the remaining kinds. An + // attested termination outranks a recorded cancellation, because queue + // operations cancel the context as an implementation detail of stopping + // the pipeline. + outcome := func() Outcome { + switch { + case !completed: + return Outcome{Kind: OutcomePanicked} + case continued: + return Outcome{Kind: OutcomeContinued} + default: + if slot != nil { + if t := slot.term.Load(); t != nil { + return Outcome{Kind: OutcomeTerminated, Detail: t.detail, Cause: t.cause} + } + if cause := slot.cancelCause.Load(); cause != nil { + return Outcome{Kind: OutcomeCancelled, Cause: *cause} + } + } + return Outcome{Kind: OutcomeTerminated} + } + } + for _, h := range hooks { + if h.before != nil { + fwd = h.before(fwd) + } + if h.after != nil { + defer h.after(&fwd) + } + if h.outcomeAfter != nil { + after := h.outcomeAfter + defer func() { fwd = after(fwd, outcome()) }() + } + if h.crashAfter != nil { + // Deferred with the entry context by value: crash handlers + // observe, they do not thread. The handler itself is the operand, + // so its own recover is effective. + defer h.crashAfter(fwd) + } + } + // Delimit: the sentinel records whether the step continued and the + // context it threaded forward, and stops the step's dynamic extent there + // so downstream steps run outside it. + sentinel := StepFunc(func(c context.Context) Step { + continued, fwd = true, c + return nil + }) + step(sentinel).Run(fwd) + completed = true + return continued, fwd +} + +var ambientMiddlewareKey = ctxkey.New[Middleware]() + +// AmbientMiddleware returns the composed middleware from context. +// Returns the zero (identity) Middleware if none has been registered. +func AmbientMiddleware(ctx context.Context) Middleware { + mw, _ := ambientMiddlewareKey.Value(ctx) + return mw +} + +// WithAmbientMiddleware composes mw into the ambient middleware stack and +// returns an updated context. The first-registered middleware is outermost. +// Passing the zero Middleware is a no-op. +func WithAmbientMiddleware(ctx context.Context, mw Middleware) context.Context { + if mw.IsZero() { + return ctx + } + return ambientMiddlewareKey.Set(ctx, Compose(AmbientMiddleware(ctx), mw)) +} + +// WithoutAmbientMiddleware clears the ambient middleware stack, so steps run +// under the returned context — including branches of a Parallel reached from +// it — run with no ambient middleware. +// +// Its signature is a ContextFunc, so it drops straight into a pipeline to +// exempt everything downstream: +// +// Sequence(Do(WithoutAmbientMiddleware), Parallel(a, b, c)) +// +// Use it to carve a subtree out of middleware that is registered globally. +func WithoutAmbientMiddleware(ctx context.Context) context.Context { + return ambientMiddlewareKey.Set(ctx, Middleware{}) +} + +// WithoutAmbientMiddleware must be usable wherever a ContextFunc is expected. +var _ ContextFunc = WithoutAmbientMiddleware + +// AmbientDispatch wraps each step so that ambient middleware registered in +// context fires around every step as the pipeline executes. It is the opt-in +// mechanism for ambient middleware support: +// +// state.Run(ctx, state.AmbientDispatch(a, b, c)) +// +// Each step argument is wrapped individually — middleware fires once per step, +// not once for the whole group. Middleware registered via WithAmbientMiddleware +// before Run applies to all steps. Middleware registered mid-pipeline (inside a +// step) applies to all subsequent steps in the same AmbientDispatch call. +// +// AmbientDispatch works for any NewStep, including raw StepFunc closures and +// struct method steps — no special constructor is required. +// +// Composite steps are opaque to it: a Sequence or Parallel passed to +// AmbientDispatch is one step, so middleware fires once around the whole group +// rather than once per inner step. Parallel additionally dispatches into its own +// branches, so under AmbientDispatch a Parallel fires middleware once for the +// group and once more inside each branch. +func AmbientDispatch(steps ...NewStep) NewStep { + return func(outerNext Step) Step { + // Build the chain right-to-left. Each step's "next" is the already-dispatch- + // wrapped Step for the following step, so middleware fires exactly once per + // step. Wrap of the zero Middleware returns the step unchanged, so dispatch + // is inert when nothing is registered. + current := outerNext + for _, step := range slices.Backward(steps) { + s := step + n := current + current = StepFunc(func(ctx context.Context) Step { + return AmbientMiddleware(ctx).Wrap(s)(n).Run(ctx) + }) + } + return current + } +} + +// Dispatch is the single-step form of AmbientDispatch. It carries the ambient +// middleware stack across a boundary that middleware applied outside cannot +// reach. +// +// Dispatch is a plain step transformer, not a Middleware: it is part of the +// interpreter that applies Middleware, one level up from the sealed type. +// +// Parallel already applies it to its own branches, so most code never needs it +// directly. Reach for it when writing a combinator of your own that crosses a +// goroutine boundary. +// +// Like AmbientDispatch, Dispatch is inert when no middleware is registered. +func Dispatch(step NewStep) NewStep { return AmbientDispatch(step) } + +var ( + stepNameKey = ctxkey.NewWithDefault[string]("") + stepNameCaptureKey = ctxkey.New[*atomic.Pointer[string]]() +) + +// StepName returns the name of the currently-executing step from context. +// Returns the name set by Named if present, otherwise "". +func StepName(ctx context.Context) string { + return stepNameKey.Value(ctx) +} + +// WithStepNameCapture allocates a capture slot in context that the first +// (outermost) Named step fills with its name. This lets observability +// middleware read the step name even for terminal steps that never invoke +// their continuation. The slot is a pointer to an atomic, so concurrent Named +// branches (e.g. under Parallel) fill it without racing and the first writer +// wins. Pair with CapturedStepName to read the result. +func WithStepNameCapture(ctx context.Context) context.Context { + return stepNameCaptureKey.Set(ctx, &atomic.Pointer[string]{}) +} + +// CapturedStepName returns the step name written into the capture slot set up +// by WithStepNameCapture. Returns "" if no slot was allocated or no Named step +// ran. +func CapturedStepName(ctx context.Context) string { + if slot, ok := stepNameCaptureKey.Value(ctx); ok && slot != nil { + if name := slot.Load(); name != nil { + return *name + } + } + return "" +} + +// Named annotates a step with a human-readable name for observability. +// The name is stored in context before the step executes, making it +// available to middleware via StepName(ctx). +// +// Named is entirely optional — pipelines behave identically without it. +func Named(name string, step NewStep) NewStep { + return func(next Step) Step { + return StepFunc(func(ctx context.Context) Step { + ctx = stepNameKey.Set(ctx, name) + // Fill the capture slot (if a middleware allocated one). First + // writer wins, so an outer Named is recorded over any nested ones, + // and concurrent Named branches do not race. + if slot, ok := stepNameCaptureKey.Value(ctx); ok && slot != nil { + slot.CompareAndSwap(nil, &name) + } + return step(next).Run(ctx) + }) + } +} + // Step converts a NewStep to a Step by calling it with nil as the next step. func (ns NewStep) Step() Step { return ns(nil) @@ -78,17 +651,6 @@ func (ns NewStep) Step() Step { // return Continue(ctx, next) // }) // } -// -// Instead of the more verbose: -// -// func MyStep() NewStep { -// return func(next Step) Step { -// return StepFunc(func(ctx context.Context) Step { -// // your logic here -// return Continue(ctx, next) -// }) -// } -// } func NewStepFunc(fn func(ctx context.Context, next Step) Step) NewStep { return func(next Step) Step { return StepFunc(func(ctx context.Context) Step { @@ -122,18 +684,24 @@ func Run(ctx context.Context, newStep NewStep) { } } -type errorHandlerKey struct{} +var errorHandlerKey = ctxkey.New[func(error) Step]() // WithErrorHandler adds an error handler to the context. -// When Continue encounters a cancelled context, it will call this handler -// instead of stopping the pipeline. +// When Continue encounters a cancelled context, it calls this handler with the +// cancellation cause instead of stopping the pipeline. A non-nil Step returned +// by the handler is run as a recovery path; see Continue for its semantics. func WithErrorHandler(ctx context.Context, handler func(error) Step) context.Context { - return context.WithValue(ctx, errorHandlerKey{}, handler) + return errorHandlerKey.Set(ctx, handler) } // Continue runs the next step if it exists and context is not cancelled. // If context is cancelled: -// - If an error handler is set via WithErrorHandler, calls the handler +// - If an error handler is set via WithErrorHandler, calls the handler and, +// when the handler returns a non-nil recovery Step, runs it with the +// handler cleared from context. The context is still cancelled while the +// recovery step runs, so any Continue inside it stops the recovery path +// rather than re-entering the handler; build recovery from a terminal +// step or a single step that does its work in the body. // - Otherwise, stops the pipeline (returns nil) // // This is a helper to avoid the common pattern: @@ -148,8 +716,18 @@ func WithErrorHandler(ctx context.Context, handler func(error) Step) context.Con func Continue(ctx context.Context, next Step) Step { if ctx.Err() != nil { err := context.Cause(ctx) - if handler, ok := ctx.Value(errorHandlerKey{}).(func(error) Step); ok && handler != nil { - return handler(err) + // Report the cancellation to the enclosing bracket's outcome slot + // (if any): the cancelled context never escapes the step, so this is + // how outcome-aware hooks learn the pipeline stopped for + // cancellation rather than termination. + recordCancellation(ctx, err) + if handler, ok := errorHandlerKey.Value(ctx); ok && handler != nil { + if recovery := handler(err); recovery != nil { + // The context is still cancelled: clear the handler so the + // recovery path's own Continue calls stop instead of + // re-entering it. + return recovery.Run(errorHandlerKey.Set(ctx, nil)) + } } return nil } @@ -193,32 +771,55 @@ func Sequence(steps ...NewStep) NewStep { } } -// ParallelWith composes multiple NewStep functions to run in parallel, -// applying wrapper to each branch before execution. This is useful for -// applying a uniform policy to all branches, such as panic recovery: +// Map applies wrapper to each step and returns the resulting slice. +// This is useful for applying a uniform policy to a set of steps before +// passing them to Parallel or other combinators. The wrapper is a plain step +// transformer; pass a Middleware's Wrap method to apply sealed middleware: // -// ParallelWith(Recover, step1, step2, step3) +// Parallel(Map(instrument.Wrap, step1, step2, step3)...) // -// The wrapper is called once per branch at instantiation time (when the -// returned NewStep is called), not at execution time. +// or pass a raw func(NewStep) NewStep for transformations that hooks cannot +// express, such as a deferred crash handler. // -// If using Recover as the wrapper and multiple branches panic concurrently, -// the error handler may be called multiple times — once per panicking branch. -// The context cause will reflect whichever panic cancelled the context first. -func ParallelWith(wrapper func(NewStep) NewStep, steps ...NewStep) NewStep { - wrapped := make([]NewStep, len(steps)) +// The wrapper is called once per step when Map is called. +func Map(wrapper func(NewStep) NewStep, steps ...NewStep) []NewStep { + result := make([]NewStep, len(steps)) for i, s := range steps { - wrapped[i] = wrapper(s) + result[i] = wrapper(s) } - return Parallel(wrapped...) + return result } -// Parallel composes multiple NewStep functions to run in parallel, -// then continues to the next step after all complete. +// Parallel composes multiple NewStep functions to run in parallel, then +// continues to the next step after all complete. Each step runs in a new +// goroutine. +// +// Ambient middleware registered with WithAmbientMiddleware is applied inside +// each branch, on that branch's own goroutine. This is the default because the +// alternative fails silently: middleware whose effect is goroutine-local — a +// deferred crash handler, say — would look installed while never covering a +// branch. It follows that such middleware runs concurrently, once per branch, +// so it must be safe for concurrent use. It is inert when none is registered. +// To run branches without it, clear the stack first: +// +// Sequence(Do(WithoutAmbientMiddleware), Parallel(a, b, c)) +// +// A panicking step crashes the process. Parallel installs no recover, and +// recover cannot cross a goroutine boundary, so a branch panic bypasses any +// recover the caller installed around the pipeline: a crash handler wrapping +// Run never observes it. No information is lost — the process dies with the +// branch's stack. Because each branch is bracketed on its own goroutine, +// ambient after-hooks close during the unwind, and a crash handler +// registered as an ambient CrashHandler observes branch panics from +// inside each branch (see CrashHandler and the state/middleware package +// docs). A raw wrapper applied with Map works too. func Parallel(steps ...NewStep) NewStep { return func(next Step) Step { return &ParallelStep{ - steps: steps, + // Dispatch at composition time, so each branch runs the ambient + // stack on its own goroutine and ParallelStep.Run stays unaware + // that middleware exists. Inert when none is registered. + steps: Map(Dispatch, steps...), next: next, } } @@ -233,6 +834,13 @@ type ParallelStep struct { func (p *ParallelStep) Run(ctx context.Context) Step { var wg sync.WaitGroup + // Branch panics are deliberately not recovered and re-raised here. + // x/sync/errgroup rejects that design for reasons that apply equally to + // this loop: it delays the panic until every sibling finishes, it reduces + // the panic stack to a mere value that crash-monitoring tools cannot see, + // and it risks deadlocking in a way that hides the panic entirely. Callers + // who want structured crash reporting defer a handler inside the branch + // instead — see the Parallel doc comment. for _, step := range p.steps { wg.Go(func() { if s := step.Step(); s != nil { @@ -245,44 +853,6 @@ func (p *ParallelStep) Run(ctx context.Context) Step { return Continue(ctx, p.next) } -// PanicError wraps a recovered panic value as an error. -// It is set as the cause on the context passed to WithErrorHandler when Recover -// catches a panic, so callers can distinguish panics from normal cancellations -// and recover the original panic value via errors.As. -type PanicError struct { - Value any -} - -func (p *PanicError) Error() string { - return fmt.Sprintf("panic: %v", p.Value) -} - -// Recover wraps a step with panic recovery. On panic, it cancels a child -// context with a *PanicError cause and routes through Continue, so any -// WithErrorHandler registered on the context will be called with the -// *PanicError. The pipeline does not continue past the recovered panic. -// In most cases, panics should propagate naturally — only use Recover when -// you explicitly need to handle a panic from a specific step. -// -// Note: Recover cannot catch panics from goroutines spawned by the wrapped -// step (e.g. a Parallel step). Go's runtime does not allow cross-goroutine -// panic recovery; those panics will still crash the program. -func Recover(step NewStep) NewStep { - return func(next Step) Step { - return StepFunc(func(ctx context.Context) (result Step) { - childCtx, cancel := context.WithCancelCause(ctx) - defer cancel(nil) - defer func() { - if r := recover(); r != nil { - cancel(&PanicError{Value: r}) - result = Continue(childCtx, next) - } - }() - return step(next).Run(childCtx) - }) - } -} - // Decision creates a conditional step that chooses between two paths. func Decision(predicate func(context.Context) bool, trueHandler, falseHandler NewStep) NewStep { return func(next Step) Step { diff --git a/state/state_test.go b/state/state_test.go index 5b32e36..957ebc9 100644 --- a/state/state_test.go +++ b/state/state_test.go @@ -990,6 +990,74 @@ func TestEnumWithoutDefault(t *testing.T) { } } +func TestWhenPredicateTrue(t *testing.T) { + ctx := t.Context() + var handlerExecuted, afterExecuted bool + + pipeline := Sequence( + When( + func(_ context.Context) bool { return true }, + Do(func(ctx context.Context) context.Context { + handlerExecuted = true + return ctx + }), + ), + Do(func(ctx context.Context) context.Context { + afterExecuted = true + return ctx + }), + ) + + Run(ctx, pipeline) + + require.True(t, handlerExecuted) + require.True(t, afterExecuted, "pipeline continues after a non-terminal When handler") +} + +func TestWhenPredicateFalse(t *testing.T) { + ctx := t.Context() + var handlerExecuted, afterExecuted bool + + pipeline := Sequence( + When( + func(_ context.Context) bool { return false }, + Do(func(ctx context.Context) context.Context { + handlerExecuted = true + return ctx + }), + ), + Do(func(ctx context.Context) context.Context { + afterExecuted = true + return ctx + }), + ) + + Run(ctx, pipeline) + + require.False(t, handlerExecuted) + require.True(t, afterExecuted, "pipeline continues when predicate is false") +} + +func TestWhenWithTerminalHandlerStopsPipeline(t *testing.T) { + ctx := t.Context() + var afterExecuted bool + + pipeline := Sequence( + When( + func(_ context.Context) bool { return true }, + Terminal, + ), + Do(func(ctx context.Context) context.Context { + afterExecuted = true + return ctx + }), + ) + + Run(ctx, pipeline) + + require.False(t, afterExecuted, "terminal handler prevents continuation to subsequent steps") +} + func TestDecisionRespectsCancellation(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -1070,206 +1138,120 @@ func TestEnumWithoutDefaultRespectsCancellation(t *testing.T) { require.True(t, handlerCalled, "error handler should be called") } -func TestParallelDoesNotRecoverPanics(t *testing.T) { - // Panics in Parallel branches occur in child goroutines and cannot be caught - // by Recover on the parent goroutine. Wrapping Parallel with Recover does NOT - // suppress branch panics — the program will still crash. - // - // This is correct behavior: panics are programming errors and should be loud. - // Recovery at the goroutine boundary is intentionally not supported. - // - // Verified by inspection: ParallelStep.Run spawns goroutines with no recover, - // and Go's runtime does not allow cross-goroutine panic recovery. - t.Log("Parallel panics cannot be caught by Recover (cross-goroutine limitation, verified by inspection)") -} - -func TestParallelRespectsErrorHandler(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - cancel() - - var handlerCalled bool - ctx = WithErrorHandler(ctx, func(_ error) Step { - handlerCalled = true - return nil - }) - - pipeline := Parallel( - Do(func(ctx context.Context) context.Context { return ctx }), - ) - - Run(ctx, pipeline) - - require.True(t, handlerCalled, "Parallel should invoke WithErrorHandler on cancellation, not bypass it") -} - -func TestRecoverStopsPipelineOnPanic(t *testing.T) { - ctx := t.Context() - var afterPanicExecuted bool +// Parallel installs no recover of its own, so a branch panic crashes the +// process (verified by inspection: ParallelStep.Run has no recover, and Go +// does not allow cross-goroutine recovery). What this test pins is the +// escape hatch documented on Parallel: a caller who wants utilruntime's +// HandleCrash semantics can defer a handler inside the branch, and because +// steps run in continuation-passing style that single defer covers the whole +// branch. Modelled here with a plain recover so state's tests stay free of +// Kubernetes dependencies. +func TestParallelBranchPanicIsHandleableInsideTheBranch(t *testing.T) { + var mu sync.Mutex + var observed []any - pipeline := Sequence( - Recover(Do(func(_ context.Context) context.Context { - panic("intentional panic") - })), - Do(func(ctx context.Context) context.Context { - afterPanicExecuted = true - return ctx - }), - ) + // Stand-in for `defer utilruntime.HandleCrashWithContext(ctx)`, minus the + // re-panic that would take the test process down with it. + handleCrash := func(step NewStep) NewStep { + return func(next Step) Step { + return StepFunc(func(ctx context.Context) Step { + defer func() { + if r := recover(); r != nil { + mu.Lock() + observed = append(observed, r) + mu.Unlock() + } + }() + return step(next).Run(ctx) + }) + } + } + var downstreamRan atomic.Bool require.NotPanics(t, func() { - Run(ctx, pipeline) - }, "Recover should suppress the panic") - require.False(t, afterPanicExecuted, "pipeline should stop after recovered panic") -} - -func TestRecoverContinuesNormally(t *testing.T) { - ctx := t.Context() - var executed []string - - pipeline := Sequence( - Recover(Do(func(ctx context.Context) context.Context { - executed = append(executed, "step1") - return ctx - })), - Do(func(ctx context.Context) context.Context { - executed = append(executed, "step2") - return ctx - }), - ) - - Run(ctx, pipeline) - - require.Equal(t, []string{"step1", "step2"}, executed) -} - -func TestRecoverRoutesToErrorHandler(t *testing.T) { - var handlerErr error - ctx := WithErrorHandler(t.Context(), func(err error) Step { - handlerErr = err - return nil + Run(t.Context(), Parallel(Map(handleCrash, + Do(func(c context.Context) context.Context { return c }), + Sequence( + NewStepFunc(func(_ context.Context, _ Step) Step { panic("branch boom") }), + Do(func(c context.Context) context.Context { downstreamRan.Store(true); return c }), + ), + )...)) }) - pipeline := Recover(Do(func(_ context.Context) context.Context { - panic("intentional panic") - })) - - require.NotPanics(t, func() { - Run(ctx, pipeline) - }) - require.Error(t, handlerErr, "error handler should be called after recovered panic") + require.Equal(t, []any{"branch boom"}, observed, + "a handler deferred inside the branch must see the branch panic") + require.False(t, downstreamRan.Load(), + "the rest of the branch must not run after it panics") } -func TestRecoverPanicValueAvailableViaCause(t *testing.T) { - type sentinelType struct{ msg string } - panicValue := sentinelType{"something went wrong"} +// The branch handler must also cover panics raised downstream of the wrapped +// step, since CPS runs the continuation inside the same frame. +func TestParallelBranchHandlerCoversDownstreamOfTheBranch(t *testing.T) { + var caught any + handleCrash := func(step NewStep) NewStep { + return func(next Step) Step { + return StepFunc(func(ctx context.Context) Step { + defer func() { caught = recover() }() + return step(next).Run(ctx) + }) + } + } - var causeErr error - ctx := WithErrorHandler(t.Context(), func(err error) Step { - causeErr = err - return nil + require.NotPanics(t, func() { + Run(t.Context(), Parallel(handleCrash(Sequence( + Do(func(c context.Context) context.Context { return c }), + NewStepFunc(func(_ context.Context, _ Step) Step { panic("late boom") }), + )))) }) - - pipeline := Recover(Do(func(_ context.Context) context.Context { - panic(panicValue) - })) - - Run(ctx, pipeline) - - var p *PanicError - require.ErrorAs(t, causeErr, &p, "error should be a *PanicError") - require.Equal(t, panicValue, p.Value, "PanicError.Value should be the original panic value") -} - -func TestRecoverPanicErrorMessage(t *testing.T) { - p := &PanicError{Value: "boom"} - require.Equal(t, "panic: boom", p.Error()) + require.Equal(t, "late boom", caught) } -func TestRecoverWrappingSequence(t *testing.T) { - var afterPanicExecuted bool - var handlerErr error - - ctx := WithErrorHandler(t.Context(), func(err error) Step { - handlerErr = err - return nil - }) - - pipeline := Sequence( - Recover(Sequence( - Do(func(ctx context.Context) context.Context { return ctx }), - Do(func(_ context.Context) context.Context { panic("mid-sequence panic") }), - Do(func(ctx context.Context) context.Context { - afterPanicExecuted = true - return ctx - }), - )), - Do(func(ctx context.Context) context.Context { - afterPanicExecuted = true - return ctx - }), +// A clean Parallel must not panic. +func TestParallelNoPanicOnCleanBranches(t *testing.T) { + var count atomic.Int32 + pipeline := Parallel( + Do(func(c context.Context) context.Context { count.Add(1); return c }), + Do(func(c context.Context) context.Context { count.Add(1); return c }), ) - require.NotPanics(t, func() { Run(ctx, pipeline) }) - require.False(t, afterPanicExecuted, "steps after panic should not execute") - var p *PanicError - require.ErrorAs(t, handlerErr, &p) - require.Equal(t, "mid-sequence panic", p.Value) + require.NotPanics(t, func() { Run(t.Context(), pipeline) }) + require.Equal(t, int32(2), count.Load()) } -func TestParallelWithRunsAllBranches(t *testing.T) { - ctx := t.Context() - var counter int32 - - pipeline := ParallelWith(Recover, - Do(func(ctx context.Context) context.Context { - atomic.AddInt32(&counter, 1) - return ctx - }), - Do(func(ctx context.Context) context.Context { - atomic.AddInt32(&counter, 1) - return ctx - }), - ) - - Run(ctx, pipeline) - require.Equal(t, int32(2), atomic.LoadInt32(&counter)) -} +func TestParallelRespectsErrorHandler(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() -func TestParallelWithRecoverRoutesPanicToErrorHandler(t *testing.T) { - var handlerErr error - ctx := WithErrorHandler(t.Context(), func(err error) Step { - handlerErr = err + var handlerCalled bool + ctx = WithErrorHandler(ctx, func(_ error) Step { + handlerCalled = true return nil }) - pipeline := ParallelWith(Recover, - Do(func(ctx context.Context) context.Context { return ctx }), - Do(func(_ context.Context) context.Context { panic("branch panic") }), + pipeline := Parallel( Do(func(ctx context.Context) context.Context { return ctx }), ) - require.NotPanics(t, func() { Run(ctx, pipeline) }) + Run(ctx, pipeline) - var p *PanicError - require.ErrorAs(t, handlerErr, &p, "error handler should receive a *PanicError from the panicking branch") - require.Equal(t, "branch panic", p.Value) + require.True(t, handlerCalled, "Parallel should invoke WithErrorHandler on cancellation, not bypass it") } -func TestParallelWithCustomWrapper(t *testing.T) { - // Verify that the wrapper is actually applied to each branch, not just once. +func TestMapAppliesWrapperToEach(t *testing.T) { + // Map applies the wrapper to each step. var wrappedCount int32 countingWrapper := func(step NewStep) NewStep { atomic.AddInt32(&wrappedCount, 1) return step } - ParallelWith(countingWrapper, + Map(countingWrapper, Do(func(ctx context.Context) context.Context { return ctx }), Do(func(ctx context.Context) context.Context { return ctx }), Do(func(ctx context.Context) context.Context { return ctx }), - )(nil) // instantiate to trigger wrapping + ) - require.Equal(t, int32(3), atomic.LoadInt32(&wrappedCount), "wrapper should be applied to each branch") + require.Equal(t, int32(3), atomic.LoadInt32(&wrappedCount), "wrapper should be applied to each step") } func TestParallelCancellation(t *testing.T) { @@ -2135,6 +2117,131 @@ func TestIntegrationContextThreadingShowcase(t *testing.T) { } } +// ============================================================================ +// AMBIENT DISPATCH TESTS +// ============================================================================ + +func TestAmbientDispatchNoMiddleware(t *testing.T) { + // With no middleware registered, AmbientDispatch is transparent + var executed bool + pipeline := AmbientDispatch(Do(func(ctx context.Context) context.Context { + executed = true + return ctx + })) + Run(t.Context(), pipeline) + require.True(t, executed) +} + +func TestAmbientDispatchAppliesMiddleware(t *testing.T) { + var log []string + mw := Around( + func(ctx context.Context) context.Context { log = append(log, "before"); return ctx }, + func(ctx context.Context) context.Context { log = append(log, "after"); return ctx }, + ) + ctx := WithAmbientMiddleware(t.Context(), mw) + + pipeline := AmbientDispatch(Do(func(ctx context.Context) context.Context { + log = append(log, "step") + return ctx + })) + Run(ctx, pipeline) + require.Equal(t, []string{"before", "step", "after"}, log) +} + +func TestAmbientDispatchPropagatesAcrossSteps(t *testing.T) { + // Middleware registered before Run fires for every step in the pipeline + var log []string + mw := Around( + func(ctx context.Context) context.Context { log = append(log, "before"); return ctx }, + func(ctx context.Context) context.Context { log = append(log, "after"); return ctx }, + ) + ctx := WithAmbientMiddleware(t.Context(), mw) + + pipeline := AmbientDispatch( + Do(func(ctx context.Context) context.Context { log = append(log, "step1"); return ctx }), + Do(func(ctx context.Context) context.Context { log = append(log, "step2"); return ctx }), + ) + Run(ctx, pipeline) + require.Equal(t, []string{ + "before", "step1", "after", + "before", "step2", "after", + }, log) +} + +func TestAmbientDispatchMidPipelineInjection(t *testing.T) { + // Middleware registered inside a step applies to subsequent steps only + var log []string + mw := Around( + func(ctx context.Context) context.Context { log = append(log, "mw-before"); return ctx }, + func(ctx context.Context) context.Context { log = append(log, "mw-after"); return ctx }, + ) + + pipeline := AmbientDispatch( + Do(func(ctx context.Context) context.Context { + log = append(log, "step1") + return ctx + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "inject") + return WithAmbientMiddleware(ctx, mw) + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step3") + return ctx + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step4") + return ctx + }), + ) + Run(t.Context(), pipeline) + require.Equal(t, []string{ + "step1", + "inject", + "mw-before", "step3", "mw-after", + "mw-before", "step4", "mw-after", + }, log) +} + +func TestAmbientDispatchWorksWithStructStep(t *testing.T) { + // AmbientDispatch works for steps that are raw StepFuncs (not NewStepFunc), + // simulating the struct method pattern. + var log []string + mw := Around( + func(ctx context.Context) context.Context { log = append(log, "before"); return ctx }, + func(ctx context.Context) context.Context { log = append(log, "after"); return ctx }, + ) + ctx := WithAmbientMiddleware(t.Context(), mw) + + // Simulate a struct-based step: raw NewStep returning a raw StepFunc + structStep := NewStep(func(next Step) Step { + return StepFunc(func(ctx context.Context) Step { + log = append(log, "struct-step") + return Continue(ctx, next) + }) + }) + + pipeline := AmbientDispatch(structStep) + Run(ctx, pipeline) + require.Equal(t, []string{"before", "struct-step", "after"}, log) +} + +func TestAmbientDispatchNamedStepNameVisibleInStepBody(t *testing.T) { + // Named sets the step name in ctx before the inner step executes. + // Middleware wrapping the outer Named step fires before the name is set, + // so middleware "before" hooks see "" for StepName. The name is visible + // within the step body and to any middleware applied to the inner step. + var nameInBody string + pipeline := AmbientDispatch( + Named("myStep", Do(func(ctx context.Context) context.Context { + nameInBody = StepName(ctx) + return ctx + })), + ) + Run(t.Context(), pipeline) + require.Equal(t, "myStep", nameInBody) +} + // ============================================================================ // BENCHMARKS // ============================================================================ @@ -2302,3 +2409,553 @@ func ExampleWithErrorHandler() { // step 1 // handling cancellation } + +// A non-nil Step returned by the error handler is executed as a recovery path. +func TestErrorHandlerRecoveryStepRuns(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + var cleanupRan bool + ctx = WithErrorHandler(ctx, func(_ error) Step { + return StepFunc(func(_ context.Context) Step { + cleanupRan = true + return nil + }) + }) + + Run(ctx, NewStepFunc(Continue)) + require.True(t, cleanupRan, "a non-nil Step returned by the error handler must be executed") +} + +// The recovery step runs with the handler cleared from context: the context is +// still cancelled, so leaving the handler registered would re-enter it on the +// recovery path's first Continue, forever. +func TestErrorHandlerRecoveryDoesNotReenterHandler(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + var handlerCalls int + var recoveryTrace []string + ctx = WithErrorHandler(ctx, func(_ error) Step { + handlerCalls++ + // A recovery path with its own Continue links: the context is still + // cancelled, so it must stop after the first step body instead of + // re-entering the handler. + return Sequence( + Do(func(ctx context.Context) context.Context { + recoveryTrace = append(recoveryTrace, "cleanup") + return ctx + }), + Do(func(ctx context.Context) context.Context { + recoveryTrace = append(recoveryTrace, "unreachable") + return ctx + }), + ).Step() + }) + + Run(ctx, NewStepFunc(Continue)) + require.Equal(t, 1, handlerCalls, "recovery must not re-enter the error handler") + require.Equal(t, []string{"cleanup"}, recoveryTrace, + "recovery runs under the cancelled context, so Continue stops it after the first step body") +} + +// ============================================================================ +// MIDDLEWARE TESTS +// ============================================================================ + +func TestNewStepFuncIsConvenienceWrapper(t *testing.T) { + // NewStepFunc(fn) should be equivalent to: + // func(next Step) Step { return StepFunc(func(ctx context.Context) Step { return fn(ctx, next) }) } + var called bool + step := NewStepFunc(func(ctx context.Context, next Step) Step { + called = true + return Continue(ctx, next) + }) + Run(t.Context(), step) + require.True(t, called) +} + +func TestMiddlewareZeroValueIsIdentity(t *testing.T) { + // Middleware is a sealed type: the zero value is the identity, and Wrap + // of the zero value returns the step unchanged. + var mw Middleware + require.True(t, mw.IsZero()) + + var executed bool + step := Do(func(ctx context.Context) context.Context { executed = true; return ctx }) + Run(t.Context(), mw.Wrap(step)) + require.True(t, executed) +} + +func TestMiddlewareWrapsStep(t *testing.T) { + // A Middleware's hooks fire around the wrapped step. + var called bool + mw := Before(func(ctx context.Context) context.Context { called = true; return ctx }) + + pipeline := mw.Wrap(Do(func(ctx context.Context) context.Context { return ctx })) + Run(t.Context(), pipeline) + require.True(t, called) +} + +func TestMapAcceptsMiddlewareWrap(t *testing.T) { + // Map accepts a Middleware's Wrap method as the wrapper. + var fired int32 + mw := Before(func(ctx context.Context) context.Context { atomic.AddInt32(&fired, 1); return ctx }) + + Run(t.Context(), Sequence(Map(mw.Wrap, + Do(func(ctx context.Context) context.Context { return ctx }), + Do(func(ctx context.Context) context.Context { return ctx }), + )...)) + + require.Equal(t, int32(2), atomic.LoadInt32(&fired), "hooks fire once per wrapped step") +} + +// Parallel applies the ambient stack inside each branch. This is the default +// because the alternative fails silently for goroutine-local middleware. +func TestParallelAppliesAmbientMiddlewarePerBranch(t *testing.T) { + var fired atomic.Int32 + mw := Before(func(ctx context.Context) context.Context { fired.Add(1); return ctx }) + + noop := Do(func(c context.Context) context.Context { return c }) + Run(WithAmbientMiddleware(t.Context(), mw), Parallel(noop, noop, noop)) + + require.Equal(t, int32(3), fired.Load(), "middleware must fire once per branch") +} + +// Under AmbientDispatch, a Parallel fires middleware once for the group and +// once more inside each branch — the granularities compose like nested spans. +func TestParallelUnderAmbientDispatchFiresGroupAndBranches(t *testing.T) { + var fired atomic.Int32 + mw := Before(func(ctx context.Context) context.Context { fired.Add(1); return ctx }) + + noop := Do(func(c context.Context) context.Context { return c }) + Run(WithAmbientMiddleware(t.Context(), mw), AmbientDispatch(Parallel(noop, noop))) + + require.Equal(t, int32(3), fired.Load(), "1 for the group + 1 per branch") +} + +// Dispatching must stay inert when no middleware is registered. +func TestParallelInertWithoutAmbientMiddleware(t *testing.T) { + var ran atomic.Int32 + step := Do(func(c context.Context) context.Context { ran.Add(1); return c }) + + require.NotPanics(t, func() { Run(t.Context(), Parallel(step, step)) }) + require.Equal(t, int32(2), ran.Load(), "branches still run with no middleware registered") +} + +// WithoutAmbientMiddleware is the escape hatch: it clears the stack for +// everything downstream, including Parallel branches reached from it. +func TestWithoutAmbientMiddlewareSeversParallelBranches(t *testing.T) { + var fired, ran atomic.Int32 + mw := Before(func(ctx context.Context) context.Context { fired.Add(1); return ctx }) + step := Do(func(c context.Context) context.Context { ran.Add(1); return c }) + + Run(WithAmbientMiddleware(t.Context(), mw), Sequence( + Do(WithoutAmbientMiddleware), + Parallel(step, step, step), + )) + + require.Equal(t, int32(3), ran.Load(), "branches must still run") + require.Zero(t, fired.Load(), "cleared stack means no middleware anywhere downstream") +} + +func TestWithoutAmbientMiddlewareClearsTheStack(t *testing.T) { + mw := Around(func(ctx context.Context) context.Context { return ctx }, nil) + ctx := WithAmbientMiddleware(t.Context(), mw) + require.False(t, AmbientMiddleware(ctx).IsZero()) + require.True(t, AmbientMiddleware(WithoutAmbientMiddleware(ctx)).IsZero()) +} + +// Clearing is scoped: it must not leak back out to the enclosing context. +func TestWithoutAmbientMiddlewareIsScoped(t *testing.T) { + var fired atomic.Int32 + mw := Before(func(ctx context.Context) context.Context { fired.Add(1); return ctx }) + noop := Do(func(c context.Context) context.Context { return c }) + + ctx := WithAmbientMiddleware(t.Context(), mw) + // A cleared context used for one subtree leaves the original untouched. + _ = WithoutAmbientMiddleware(ctx) + Run(ctx, Parallel(noop, noop)) + + require.Equal(t, int32(2), fired.Load(), "the original context still carries middleware") +} + +func TestAmbientMiddlewareZeroByDefault(t *testing.T) { + require.True(t, AmbientMiddleware(t.Context()).IsZero()) +} + +func TestWithAmbientMiddlewareSingle(t *testing.T) { + var called bool + mw := Before(func(ctx context.Context) context.Context { called = true; return ctx }) + ctx := WithAmbientMiddleware(t.Context(), mw) + got := AmbientMiddleware(ctx) + require.False(t, got.IsZero()) + Run(t.Context(), got.Wrap(Noop)) // trigger it + require.True(t, called) +} + +// hookPair builds an Around middleware whose hooks append to the given log. +func hookPair(log *[]string, name string) Middleware { + return Around( + func(ctx context.Context) context.Context { + *log = append(*log, name+"-before") + return ctx + }, + func(ctx context.Context) context.Context { + *log = append(*log, name+"-after") + return ctx + }, + ) +} + +func TestWithAmbientMiddlewareComposes(t *testing.T) { + // Middleware registered first is outermost at pipeline run time. + // "Outermost" means its before-logic runs first, after-logic runs last. + var log []string + + ctx := WithAmbientMiddleware(t.Context(), hookPair(&log, "mw1")) + ctx = WithAmbientMiddleware(ctx, hookPair(&log, "mw2")) + + // AmbientDispatch is required to apply ambient middleware to each step. + Run(ctx, AmbientDispatch(Do(func(ctx context.Context) context.Context { + log = append(log, "step") + return ctx + }))) + + // mw1 registered first = outermost: Compose(mw1, mw2). + require.Equal(t, []string{"mw1-before", "mw2-before", "step", "mw2-after", "mw1-after"}, log) +} + +func TestWithAmbientMiddlewareComposesDirectly(t *testing.T) { + var log []string + + ctx := WithAmbientMiddleware(t.Context(), hookPair(&log, "mw1")) + ctx = WithAmbientMiddleware(ctx, hookPair(&log, "mw2")) + + // Apply composed middleware directly to a step and run it + composed := AmbientMiddleware(ctx) + require.False(t, composed.IsZero()) + + pipeline := composed.Wrap(Do(func(ctx context.Context) context.Context { + log = append(log, "step") + return ctx + })) + Run(t.Context(), pipeline) + + // mw1 registered first = outermost + require.Equal(t, []string{"mw1-before", "mw2-before", "step", "mw2-after", "mw1-after"}, log) +} + +func TestWithAmbientMiddlewareZeroIsNoOp(t *testing.T) { + // Registering the zero (identity) Middleware should be a no-op. + mw := Around(func(ctx context.Context) context.Context { return ctx }, nil) + ctx := WithAmbientMiddleware(t.Context(), mw) + ctx2 := WithAmbientMiddleware(ctx, Middleware{}) + // zero is no-op — same middleware still present + require.False(t, AmbientMiddleware(ctx2).IsZero()) +} + +func TestAmbientMiddlewareMidPipelineInjection(t *testing.T) { + // Middleware registered inside a step during execution should + // affect all subsequent steps. + var log []string + + mw := hookPair(&log, "mw") + + pipeline := AmbientDispatch( + Do(func(ctx context.Context) context.Context { + log = append(log, "step1") + return ctx + }), + Do(func(ctx context.Context) context.Context { + // Inject mid-pipeline + log = append(log, "inject") + return WithAmbientMiddleware(ctx, mw) + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step3") + return ctx + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step4") + return ctx + }), + ) + + Run(t.Context(), pipeline) + // step1 and inject are before middleware registration — no wrapping. + // step3 and step4 get middleware applied. The interpreter brackets each + // step, so a step's after-hook fires when that step completes, before + // downstream steps run — per-step attribution. + require.Equal(t, []string{ + "step1", + "inject", + "mw-before", "step3", "mw-after", + "mw-before", "step4", "mw-after", + }, log) +} + +func TestAroundFiresBeforeAndAfter(t *testing.T) { + var log []string + + mw := Around( + func(ctx context.Context) context.Context { + log = append(log, "before") + return ctx + }, + func(ctx context.Context) context.Context { + log = append(log, "after") + return ctx + }, + ) + + pipeline := mw.Wrap(Do(func(ctx context.Context) context.Context { + log = append(log, "step") + return ctx + })) + + Run(t.Context(), pipeline) + require.Equal(t, []string{"before", "step", "after"}, log) +} + +func TestAroundNilBeforeOrAfter(t *testing.T) { + // nil before or after should not panic + mw := Around(nil, nil) + pipeline := mw.Wrap(Do(func(ctx context.Context) context.Context { return ctx })) + require.NotPanics(t, func() { Run(t.Context(), pipeline) }) +} + +func TestAroundWrapsSequenceAsUnit(t *testing.T) { + // mw.Wrap(Sequence(a, b)) wraps the whole sequence with one before/after + // pair. This differs from Sequence(mw.Wrap(a), mw.Wrap(b)) which wraps + // each step individually. Wrap treats its step argument as a single unit. + var log []string + mw := Around( + func(ctx context.Context) context.Context { log = append(log, "before"); return ctx }, + func(ctx context.Context) context.Context { log = append(log, "after"); return ctx }, + ) + a := Do(func(ctx context.Context) context.Context { log = append(log, "a"); return ctx }) + b := Do(func(ctx context.Context) context.Context { log = append(log, "b"); return ctx }) + + Run(t.Context(), mw.Wrap(Sequence(a, b))) + require.Equal(t, []string{"before", "a", "b", "after"}, log) +} + +func TestAmbientMiddlewareFiresAroundSubsequentSteps(t *testing.T) { + // Middleware injected at step N fires before/after steps N+1, N+2, etc. + var log []string + + loggingMW := Around( + func(ctx context.Context) context.Context { + log = append(log, "before") + return ctx + }, + func(ctx context.Context) context.Context { + log = append(log, "after") + return ctx + }, + ) + + pipeline := AmbientDispatch( + Do(func(ctx context.Context) context.Context { + return WithAmbientMiddleware(ctx, loggingMW) + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step2") + return ctx + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step3") + return ctx + }), + ) + + Run(t.Context(), pipeline) + require.Equal(t, []string{"before", "step2", "after", "before", "step3", "after"}, log) +} + +func TestAmbientMiddlewareNotAppliedBeforeInjection(t *testing.T) { + // Steps before the injection point are NOT wrapped. + var log []string + + loggingMW := Before(func(ctx context.Context) context.Context { + log = append(log, "mw") + return ctx + }) + + pipeline := AmbientDispatch( + Do(func(ctx context.Context) context.Context { + log = append(log, "step1") + return ctx + }), + Do(func(ctx context.Context) context.Context { + return WithAmbientMiddleware(ctx, loggingMW) + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step3") + return ctx + }), + ) + + Run(t.Context(), pipeline) + // mw should appear only around step3, not step1 + require.Equal(t, []string{"step1", "mw", "step3"}, log) +} + +func TestNamedStepNameAvailableInStepBody(t *testing.T) { + // Named sets the step name in ctx before the inner step body executes. + // The name is visible within the step body via StepName(ctx). + // Middleware wrapping the outer Named step fires before the name is set, + // so outer middleware "before" hooks see "". Use Named to annotate steps + // for structured logging within the step body, not for middleware observability. + var nameInBody string + + pipeline := AmbientDispatch( + Do(func(ctx context.Context) context.Context { + return WithAmbientMiddleware(ctx, Before(func(ctx context.Context) context.Context { return ctx })) + }), + Named("myStep", Do(func(ctx context.Context) context.Context { + nameInBody = StepName(ctx) + return ctx + })), + ) + + Run(t.Context(), pipeline) + require.Equal(t, "myStep", nameInBody) +} + +func TestUnnamedStepHasEmptyName(t *testing.T) { + // Unnamed steps have StepName == "" — no reflection-based fallback. + // Use Named() to provide an explicit name for observability. + var observedName string + + namingMW := Before(func(ctx context.Context) context.Context { + observedName = StepName(ctx) + return ctx + }) + + ctx := WithAmbientMiddleware(t.Context(), namingMW) + Run(ctx, AmbientDispatch(Do(func(ctx context.Context) context.Context { return ctx }))) + require.Empty(t, observedName) +} + +func TestNamedPipelineWorksWithoutMiddleware(t *testing.T) { + // Named should be transparent — pipeline works identically with or without middleware + var executed bool + pipeline := Named("myStep", Do(func(ctx context.Context) context.Context { + executed = true + return ctx + })) + Run(t.Context(), pipeline) + require.True(t, executed) +} + +func TestNamedSetsNameInContext(t *testing.T) { + var observedName string + step := Named("myStep", NewStepFunc(func(ctx context.Context, next Step) Step { + observedName = StepName(ctx) + return Continue(ctx, next) + })) + Run(t.Context(), step) + require.Equal(t, "myStep", observedName) +} + +func TestNamedTransparentWithoutMiddleware(t *testing.T) { + var executed bool + pipeline := Named("myStep", Do(func(ctx context.Context) context.Context { + executed = true + return ctx + })) + Run(t.Context(), pipeline) + require.True(t, executed) +} + +// ============================================================================ +// MIDDLEWARE AMBIENT INTEGRATION TESTS +// ============================================================================ + +func TestMiddlewareAmbient(t *testing.T) { + // Middleware registered as ambient applies to each subsequent step. The + // interpreter brackets each step individually, so afters fire per step — + // not once at the end of the whole remaining pipeline. + var log []string + + mw := Around( + func(ctx context.Context) context.Context { log = append(log, "before"); return ctx }, + func(ctx context.Context) context.Context { log = append(log, "after"); return ctx }, + ) + + pipeline := AmbientDispatch( + Do(func(ctx context.Context) context.Context { + return WithAmbientMiddleware(ctx, mw) + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step2") + return ctx + }), + Do(func(ctx context.Context) context.Context { + log = append(log, "step3") + return ctx + }), + ) + + Run(t.Context(), pipeline) + require.Equal(t, []string{"before", "step2", "after", "before", "step3", "after"}, log) +} + +// ============================================================================ +// STEP NAME CAPTURE TESTS (regression: nested clobber + parallel race) +// ============================================================================ + +// Regression for the capture-clobber bug: when an outer Named step composes an +// inner Named step, observability middleware reading the capture slot must see +// the OUTER (the step it wrapped) name, not the innermost one. +func TestCapturedStepNameUsesOutermostName(t *testing.T) { + ctx := WithStepNameCapture(t.Context()) + pipeline := Named("outer", Sequence( + Named("inner", Do(func(c context.Context) context.Context { return c })), + )) + pipeline.Step().Run(ctx) + require.Equal(t, "outer", CapturedStepName(ctx)) +} + +// Regression for the capture-race bug: parallel Named branches sharing one +// capture slot must not race when writing the name. Run under -race. +func TestCapturedStepNameNoRaceUnderParallel(t *testing.T) { + ctx := WithStepNameCapture(t.Context()) + branches := make([]NewStep, 0, 8) + for i := 0; i < 8; i++ { + branches = append(branches, Named("branch", Do(func(c context.Context) context.Context { return c }))) + } + Parallel(branches...).Step().Run(ctx) + require.Equal(t, "branch", CapturedStepName(ctx)) +} + +// Regression for the after-drop bug: the after hook must fire +// even when the wrapped step is terminal (never invokes its continuation). +func TestAroundAfterFiresOnTerminalStep(t *testing.T) { + var ran []string + mw := Around( + func(ctx context.Context) context.Context { ran = append(ran, "before"); return ctx }, + func(ctx context.Context) context.Context { ran = append(ran, "after"); return ctx }, + ) + Run(t.Context(), mw.Wrap(Terminal)) + require.Equal(t, []string{"before", "after"}, ran) +} + +// Regression: the after hook must fire even when the wrapped step cancels the +// context (the failure path), so metrics/tracing record the step. +func TestAroundAfterFiresOnCancelledStep(t *testing.T) { + var ran []string + mw := Around( + func(ctx context.Context) context.Context { ran = append(ran, "before"); return ctx }, + func(ctx context.Context) context.Context { ran = append(ran, "after"); return ctx }, + ) + canceller := Do(func(ctx context.Context) context.Context { + c, cancel := context.WithCancel(ctx) + cancel() + return c + }) + Run(t.Context(), mw.Wrap(canceller)) + require.Equal(t, []string{"before", "after"}, ran) +}