feat: drain-on-Stop and NewTicker primitive

[klaidliadon]

runnable v0.2 — moves cross-cutting behaviors (drain-on-shutdown, periodic execution, panic recovery, retry) out of the core lifecycle and into the new runnable/adapters subpackage as chi-style middleware. Motivated by the OMSX Frontegg reconciler (and a queue of similar workers across Polygon services) needing shutdown-safe periodic execution. Core Run/Stop semantics for existing callers are unchanged unless they opt in.

Final shape:

r := runnable.New(reconcile, runnable.WithAdapters(
    adapters.Draining(10*time.Second),
    adapters.Recovering(),
    adapters.Retry(3, time.Minute),
    adapters.Ticker(2*time.Second),
))

Each adapter is a config-only constructor returning runnable.Adapter (func(next RunFunc) RunFunc). runnable.WithAdapters wraps runFunc left-to-right (first listed = outermost).

---

What lands

Core (runnable/)

• runnable.RunFunc and runnable.Adapter types — the extension point.
• runnable.WithAdapters(...Adapter) Option — applies a list of adapters, stacking left-to-right.
• Stop's drain logic is gone from core; drain is now an opt-in adapter concern.
• WithDrain, NewTicker, WithRetry, WithRecoverer Options are removed. Replacements live in adapters/.
• Status.Restarts removed (returns in the stacked v0.3 PR via the event Publisher).
• CI workflow: go-version-file: 'go.mod' and bumped actions/checkout@v4 + actions/setup-go@v5 (the previous pin to Go 1.20 broke once context.WithoutCancel was introduced).

Adapters (runnable/adapters/)

• Draining(timeout) — graceful shutdown. Outer ctx cancellation triggers drain: work has timeout to return via adapters.Stopping(ctx) before its ctx is force-cancelled and ErrDrainTimedOut is returned. Internally recovers panics on its spawned goroutine as a safety net.
• Ticker(interval) — periodic execution. Composes with Draining: an in-flight tick is allowed to finish before the loop exits.
• Recovering() — turns panics in the wrapped work into errors. Pairs with runnable.WithPublisher if you want a handler (lands in stacked feat: Publisher interface and adapter event types #7).
• Retry(maxRetries, resetAfter) — re-invokes on non-context errors; budget resets after a quiet period.
• adapters.Stopping(ctx) <-chan struct{} — drain-start signal for long-running work to select on.
• adapters.ErrDrainTimedOut — sentinel returned when the drain window expires.

Migration

The migration table and Migrating from v0.1 to v0.2 section live in the README. v0.2 is not yet released, so the prior intermediate API in earlier commits of this branch never escaped.

Test plan

• go test -race -count=1 ./... — clean (runnable ~6s, runnable/adapters ~3s).
• golangci-lint run ./... — clean.
• examples/ticker-with-drain/main.go builds end-to-end (full SIGTERM-safe shape).
• Hardening through ~7 rounds of Codex review covered: concurrent Stop double-close, drain timer inheriting caller ctx, runCancel field race vs subsequent Run, WithRetry re-entry after drain (now moot — retry is adapter-side), panics in Draining's goroutine.
• Vacuous tests pruned (one for retry-budget-per-cycle; one for late runCancel).

Notes for review

• Adapter type lives in core (runnable.Adapter) so runnable.WithAdapters doesn't have to import the adapters subpackage. Dependency direction is adapters → runnable, never the reverse.
• NewGroup interaction — drain-enabled children of NewGroup now drain correctly when the group is stopped. In v0.1 this was silently broken (child observed groupCtx.Done() and exited before seeing its drain signal); the adapter design fixes it by construction. Load-bearing test at group_drain_test.go.
• Behavioral change vs v0.1: Stop(ctx)'s ctx no longer shortens the drain window. In v0.1 a shorter caller ctx would force-cancel mid-drain; in v0.2, Stop's ctx governs only how long the caller waits to hear back — the drain timer is fixed by Draining's argument.
• Earlier commits on this branch show the v0.1 Option-based shape (WithDrain, NewTicker) being introduced and then removed. They're retained for the audit trail; only the final state lands externally.

Stacked on top

#7 adds runnable.Publisher and typed events (RetryEvent, DrainStartedEvent, PanicRecoveredEvent, etc.). Restores Status.Restarts via subscription rather than the old WithRetry onStart side-channel.

Closes #6 (chi-style middleware reshape, now folded into this PR).

[marino39]

Alternative design: drain as a runFunc adapter

TL;DR

Same user-visible drain semantics — Stopping, ErrDrainTimedOut, configurable timeout, in-flight tick allowed to complete — built as a runFunc adapter in a runfunc/ subpackage instead of a core Option. Net effect:

• runnable.go keeps its pre-branch Stop (single-purpose, ~30 lines).
• with_retry.go loses its Stopping coupling.
• Two of the recent concurrency fixes become unnecessary; one stays as defense-in-depth.
• NewGroup's drain gap closes by construction (currently silently broken — see below).
• The "Stopping foot-gun" warning in the README and the "pristine ctx for Run" workaround in examples both go away.

What the PR gets right

• The drain concept itself: signal → grace window → force-cancel is the right shape.
• Stopping(ctx) exposed as a <-chan struct{} ctx-value, returning nil when not configured, is exactly the right helper signature.
• ErrDrainTimedOut distinct from context.Canceled is useful for callers.
• The recent concurrency fixes in Stop are technically correct.
• NewTicker's in-flight-tick semantics are right.

The pushback below is about where drain logic lives, not whether it's needed.

Concerns with the current design

1. Drain logic is interleaved into the core lifecycle

runnable.go:135-205 now branches on drainEnabled, with three execution paths in Stop (primary-drain, secondary-drain, non-drain) and three new fields on runnable (drainEnabled, drainTimeout, stoppingChan). The lifecycle of r.stoppingChan is now tied to Run-cycle index — set in Run, snapshotted-then-nil'd in Stop — which any future reader has to re-derive.

2. Two recent fixes exist only because of the coupling

• 4dbbc85 (preserve drain under concurrent Stop): only needed because Stop has the primary/secondary split. With drain external, Stop is single-purpose and idempotent like before — concurrent Stops just both cancel ctx and wait on runStop.
• 00b32bc (WithRetry observes Stopping): only needed because runFunc's ctx is kept alive during drain, so WithRetry's ctx-based exit doesn't fire. With drain external, WithRetry exits on ctx.Done() like always — drain is invisible to it.

fe7b1fd (runCancel snapshot) is real defense-in-depth and worth keeping regardless. 0758ed9 (lastTime per Run cycle) is independent of drain and should also be kept.

3. WithRetry shouldn't need to know about drain

with_retry.go:55-59 selecting on Stopping(ctx) is a coupling between two concepts that were previously orthogonal. Tomorrow's WithJitter or WithBackoff will need the same hook; the day after that, every adapter knows about drain.

4. NewGroup + WithDrain is silently broken

A drain-enabled child of NewGroup has its own r.stoppingChan = make(...) (runnable.go:102), which shadows whatever the group set. Calling Stop on the group does not propagate to drain-enabled children — they hang on the group's ctx until their own drain timeout. The README doesn't mention this, and there is no test for it. With drain as a runFunc adapter, children participate in the parent's drain via ctx cancellation by default; no special case needed.

5. SIGTERM-via-ctx bypasses drain

Documented in commit 6db7242 ("use pristine ctx for Run so SIGTERM doesn't bypass drain") and the README's "Stopping foot-gun" warning. Both exist because in the current design, drain only fires when Stop is called explicitly — outer-ctx cancellation skips it. With the adapter, drain is triggered by outer-ctx cancellation directly, so SIGTERM-via-ctx initiates drain naturally and the foot-gun disappears.

6. Stop(ctx)'s ctx-as-budget semantics are subtle

The PR's comment at runnable.go:160 calls out "shortest deadline among concurrent callers wins." That's a real design choice, but it emerges from the implementation rather than being obvious from the API. Worth being intentional about (see Open Questions).

Proposed alternative

runnable/
  runnable.go         (lifecycle: Run, Stop, isRunning, runStop, runCancel)
  runnable_group.go
  with_retry.go       (Option)
  with_recoverer.go   (Option)
  with_status.go      (Option)
  runfunc/
    doc.go
    draining.go       (Draining, Stopping, ErrDrainTimedOut)
    ticker.go         (Ticker)

runfunc/draining.go

package runfunc

var ErrDrainTimedOut = errors.New("runfunc: drain timed out")

type stoppingKey struct{}

func Stopping(ctx context.Context) <-chan struct{} {
    ch, _ := ctx.Value(stoppingKey{}).(<-chan struct{})
    return ch
}

func Draining(timeout time.Duration, work func(context.Context) error) func(context.Context) error {
    return func(outerCtx context.Context) error {
        innerCtx, cancel := context.WithCancel(context.WithoutCancel(outerCtx))
        defer cancel()

        stopping := make(chan struct{})
        innerCtx = context.WithValue(innerCtx, stoppingKey{}, (<-chan struct{})(stopping))

        done := make(chan error, 1)
        go func() { done <- work(innerCtx) }()

        select {
        case err := <-done:
            return err
        case <-outerCtx.Done():
            close(stopping)
        }

        timer := time.NewTimer(timeout)
        defer timer.Stop()
        select {
        case err := <-done:
            return err
        case <-timer.C:
            cancel()
            <-done
            return ErrDrainTimedOut
        }
    }
}

runfunc/ticker.go

package runfunc

func Ticker(interval time.Duration, tick func(context.Context) error) func(context.Context) error {
    return func(ctx context.Context) error {
        t := time.NewTicker(interval)
        defer t.Stop()
        stopping := Stopping(ctx)

        for {
            select {
            case <-ctx.Done():
                return ctx.Err()
            case <-stopping:
                return nil
            case <-t.C:
                select {
                case <-ctx.Done():
                    return ctx.Err()
                case <-stopping:
                    return nil
                default:
                }
                if err := tick(ctx); err != nil {
                    return err
                }
            }
        }
    }
}

Call site

// plain
r := runnable.New(runfunc.Ticker(time.Second, doWork))

// with drain
r := runnable.New(runfunc.Draining(30*time.Second,
    runfunc.Ticker(time.Second, doWork),
))

// inside doWork
select {
case <-runfunc.Stopping(ctx):
    // drain has begun; wrap up
default:
}

What this lets the PR drop

• with_drain.go, with_drain_test.go
• All drainEnabled / drainTimeout / stoppingChan fields on runnable
• The primary/secondary Stop split (runnable.go:157-191); Stop returns to its pre-branch single-purpose form
• The Stopping(ctx) observation in with_retry.go:55-59 (commit 00b32bc)
• The README "Stopping foot-gun" warning
• The "pristine ctx for Run" workaround in examples/main.go (commit 6db7242)

What the PR keeps

• lastTime per-cycle scoping in with_retry.go (commit 0758ed9) — independent bug fix.
• runCancel snapshot in Stop (commit fe7b1fd) — real defense-in-depth, valid even without drain.
• NewTicker's tick semantics — unchanged, just lives at runfunc.Ticker.
• Public API names: Stopping, ErrDrainTimedOut, Ticker (under runfunc).

Honest cons of this proposal

1. One extra goroutine per draining runnable. Draining spawns a goroutine for work. Cheap, but real.
2. Stop(ctx)'s ctx can no longer hurry drain. Today, if the caller's ctx expires during drain, Stop force-cancels runFunc's ctx to shorten shutdown. With the adapter, drain runs on its own fixed timer; the caller's ctx only governs how long they wait for Stop to return. Drain continues in the background until its timer fires. (Note: in both designs, Stop returning ctx.Err() does not guarantee runFunc has exited — so the practical difference is "how aggressively does Stop push runFunc to exit." The PR is more aggressive.)
3. Verbose at call sites. runnable.New(runfunc.Draining(30*time.Second, runfunc.Ticker(...))) reads heavier than runnable.NewTicker(..., WithDrain(...)). The composability buys generality (same Draining works for HTTP server loops, queue consumers, etc.) — judgment call whether that's worth the verbosity.
4. Two imports. Trivial but worth noting.

Open questions

• Is anyone relying on Stop(ctx)'s ctx-as-drain-budget today? If yes, the alternative changes that semantic. If no, the alternative is strictly simpler.
• Should WithRetry and WithRecoverer eventually move to runfunc? They're conceptually runFunc adapters dressed as Options. Out of scope for this PR but the boundary would be cleaner.
• Tick vs Ticker naming. Ticker is a noun and conflicts mentally with time.Ticker (a struct value); runfunc.Tick(interval, fn) reads more honestly as "make a runFunc that ticks." Bikeshed.