feat(jsconsumer): report consume-loop state and retryable start errors

Commit

nodo3w ago

A supervised consumer is currently unobservable from outside. Run retries every failure it can and returns only for a configuration the broker will never accept, so an adopter cannot answer either question its operators ask: am I consuming right now, and how long have I been failing.

Both facts are already in the runner. This exposes them.

  • Config.OnConsuming(ctx, bool) — true once the durable is bound and delivery has begun, false when that loop closes (consumer deleted on the server, subscription invalidated, Runner.Stop). Assign an atomic; read it from the health handler.
  • Config.OnStartError(ctx, err) — each start attempt Run will RETRY, with the error. Permanent failures are returned instead, so they never arrive here: an adopter's failure counter and its fatal path stay describing different events.

Also: IsTransientSubscribeErr now knows the modern client's error values. The jetstream package does not reuse the nats.* ones — its ErrStreamNotFound is a distinct type carrying an APIError — so errors.Is finds neither across the two, and a consumer migrated to the modern client had every rollout-ordering failure classified as a FAULT. That is the alert erosion this predicate exists to prevent, arriving through the door it holds shut. A missing CONSUMER stays absent in both spellings, for the same reason nats.ErrConsumerDeleted is: recreating the durable replays the stream's retained backlog, and the log is the only evidence.

WHY NOW

mirror-pipeline's first migrated consumer (COR-1254 loop 1) reconstructed both facts from outside: a CONSUMER.INFO probe on a ticker, a two-phase cadence, a first-bind flag, a shutdown guard, and a local copy of the error classification — ~130 lines of production code and five of its ten broker tests. Every one of that trail's five review rounds found a defect in that scaffolding; none found one in the migration it was wrapped around. Ten more consumers were queued behind the same copy.

With these two callbacks that scaffolding is deleted, not extracted.

NOTES

OnConsuming is reported from Start (not Run) so a one-shot caller gets it too, and the true is emitted before the ctx watcher is armed — a caller cannot observe the false that follows a close without first having seen the true. Both callbacks run on the runner's own goroutines: documented as must-not-block, and calling Runner.Stop from inside OnConsuming deadlocks.

Deliberately NOT a heartbeat. consuming=true says the loop is live, never that a message arrived; a consumer on a quiet stream is healthy.

Runner.Drain — the other gap COR-1254 recorded, needed for a worker-pool adopter to stop discarding its prefetch buffer on shutdown — is not here. It changes consume semantics rather than exposing existing state, and belongs with the consumer that needs it (loop 9).

TESTS

Three, against the embedded broker, each verified to fail for its own reason: dropping the false transition, dropping the start-error report, and dropping the modern classification each turn exactly one of them red. The retryable fixture withholds the STREAM, which is retryable by construction, and also asserts the error it hands back is classifiable — an error IsTransientSubscribeErr cannot recognise would be logged as a fault by every adopter.

Gate: build, vet, gofmt -s, go test ./..., go test -race ./..., golangci-lint (0 issues) — all clean.

Co-Authored-By: Claude Opus 5 (1M context) noreply@anthropic.com Claude-Session: https://claude.ai/code/session_019zK85NVAs1Enxh2eFidFpo Entire-Checkpoint: 58d6df711574

Checkpoints

Migrate Mirror-Pipeline Consumers to go-nuts

Claude CodeOpus 5.[1m]
View session
Checkpoint 1

Add Consumer Loop State Reporting

Claude CodeOpus 5.[1m]
View session
Checkpoint 1