middleware: new middleware package built on top of state

[ecordell]

The state package composes controller reconcile loops out of small steps. This PR adds a way to wrap those steps with cross-cutting behavior: time a step, open a tracing span around it, log what it did and how it ended, catch a crash. It also adds a state/middleware package with a logging middleware ready to drop in.

The hard part of wrapping a controller step is the "after" half. A step often ends by stopping the pipeline (queue.Done, a requeue) or by its context being cancelled, and none of those run a normal next step. So an after-hook written the obvious way, as just another step in the sequence, silently never fires on exactly the outcomes you most want to observe. Middleware here is built so the after-hook fires on every outcome the program survives: a normal continue, a terminal step, a cancellation, or a panic on its way up the stack. You register it once and it applies to whichever steps you opt in.

Using it

Register middleware on the context, then wrap the pipeline in AmbientDispatch to opt in. Middleware fires once per step, and pipelines that don't use AmbientDispatch are completely unaffected.

// Register once, at controller setup. First registered is outermost.
ctx = state.WithAmbientMiddleware(ctx, middleware.Log(slog.Default()))

// Around builds a paired before/after bracket, for example a tracing span:
ctx = state.WithAmbientMiddleware(ctx, state.Around(
    func(ctx context.Context) context.Context {
        return startSpan(ctx) // before each step
    },
    func(ctx context.Context) context.Context {
        endSpan(ctx) // after each step, guaranteed on every outcome
        return ctx
    },
))

// The middleware is "ambient" because it rides in the ctx and applies to any
// machine run with that context. You still opt in by wrapping the pipeline:
state.Run(ctx, state.AmbientDispatch(
    state.Named("validate", c.validateSecret),
    state.Named("ensureDeployment", c.ensureDeployment),
    state.Named("updateStatus", c.updateStatus),
))

Named labels a step so middleware can report it. With middleware.Log, each step now produces a line like:

level=INFO msg="step executed" step=ensureDeployment duration_ms=12.4 outcome=continued

Middleware can also be registered mid-pipeline, where it applies to every subsequent step in the same AmbientDispatch call. This is handy for attaching a logger scoped to the resource being reconciled:

state.Run(ctx, state.AmbientDispatch(
    state.Do(func(ctx context.Context) context.Context {
        return state.WithAmbientMiddleware(ctx, middleware.Log(loggerFor(ctx)))
    }),
    c.stepOne, // logged
    c.stepTwo, // logged
))

Reacting to how a step ended

Timers and spans want to fire regardless of outcome, but some middleware wants to know the outcome: set a span's status, label a metric, escalate a log level when a reconcile fails. An AfterOutcome hook receives that: whether the step continued, terminated, was cancelled (with the cause), or panicked.

Two of those are genuinely hard to see from outside a step, so the pipeline reports them rather than leaving you to guess. A cancellation is noticed inside the step and never escapes, so the bracket captures the cause on the way out. And "done" and "requeued" look identical from the outside (both just stop the pipeline), so the terminal operation attests which one it was: the queue operations call RecordTermination, carrying the error for the error-requeue variants. middleware.Log uses this to log at Info normally, Warn on cancellation or an error requeue, and Error on panic, so the last line a crashing controller emits names the step that broke.

Crashes

A panicking step is a programming error and stays loud by default: nothing recovers it, and it crashes the process with its stack intact. Observation hooks (Before, After, AfterOutcome) still run as the panic unwinds, so spans and timers close on the way out, but they cannot swallow it.

When you do want to handle a crash, Deferred and CrashHandler are the recover-capable forms: the hook is the defer operand itself, so a recover() in its body is effective. CrashHandler's signature accepts utilruntime.HandleCrashWithContext as-is. This matters most under Parallel, whose branches run on their own goroutines: a recover wrapped around Run cannot reach them, but a crash handler registered as ambient middleware covers every branch from inside.

Why a sealed type

Middleware could have been a plain func(NewStep) NewStep, and for control-shaped transformations (retries, custom scoping) that is still exactly what you write, applied directly or with Map. But a raw function can do anything: run a step twice, swap out its continuation, resurrect a pipeline a terminal step deliberately ended. There is no way to promise the "fires on every outcome" guarantee about a type that open.

So Middleware is instead a sealed type: an opaque, immutable stack of hooks, built only through the constructors (Before, After, Around, AfterOutcome, Deferred, CrashHandler) and combined with Compose (which makes it a monoid, first argument outermost). A value of this type is data, not code. It cannot redirect the pipeline, so the guarantee holds for every middleware you can build, not just the ones built through a blessed helper. The one scoped exception is the deliberate one above: Deferred and CrashHandler hooks may recover a panic, and nothing else can.

API changes

• ParallelWith(wrapper, steps...) → Parallel(Map(wrapper, steps...)...).
• queue.Done/queue.Requeue are now package vars (use queue.Done, not queue.Done()); queue.OnError removed, since it makes more sense to branch on context directly.
• ctxkey bumped to tagged v0.1.0.

The bracket guarantees (fires once, nests correctly, changes nothing downstream, tells the truth about the outcome) are verified in state/formal_test.go. A prose writeup of the laws, the reason a finalizer cannot live in the plain step algebra, and the correspondence with hierarchical state machines is in progress and will land separately.

[tstirrat15]

See comments. I'm generally psyched on where this is headed.

[tstirrat15]

Do we expect controller panics more often than we expect non-controller panics? I get that this can be a nice escape hatch, but my sense is that panic recovery is not something that you typically want to do in golang programs if you can avoid it.

[ecordell]

I removed this after some research, thanks for poking at it.

Originally I was trying to match / make it possible to fully remove the need for utilruntime.HandleCrash and friends from apimachinery.

Then I went looking at errgroup and found this nice comment:

		// It is tempting to propagate panics from f()
		// up to the goroutine that calls Wait, but
		// it creates more problems than it solves:
		// - it delays panics arbitrarily,
		//   making bugs harder to detect;
		// - it turns f's panic stack into a mere value,
		//   hiding it from crash-monitoring tools;
		// - it risks deadlocks that hide the panic entirely,
		//   if f's panic leaves the program in a state
		//   that prevents the Wait call from being reached.

So basically: the pattern that kube uses is called out as a not great pattern by the go team.

Since I want state to be fully de-kubed, I removed these helpers and instead added an Example test that demonstrates how, if you want it, you an directly re-use utilruntime.HandleCrash with Parallel + Map.

[ecordell]

check out this really long standing bug