Unify NATS Consumer Backoff and Tracing

Claude Code·Fable 5·nodo·1mo ago·2hr 12min·1 Checkpoint·33 file changes·+1587/-87·102K tokens

let's tackle https://linear.app/entirehq/issue/COR-929/grow-go-nuts-natsmsg-jsconsumer-backoff-and-converge-entiredb-entire shall we?

1mo ago·20m·100 calls

I've hit the commit-trailer blocker that AGENTS.md says to stop and report, so here's where things stand.

Blocker: Entire-Checkpoint trailers can't be produced in this session

AGENTS.md requires every commit to carry an Entire-Checkpoint trailer added by a git hook. I made the first commit (natsmsg, 869e0ee in go-nuts) and the trailer is missing. Digging in:

  • The task-workspace clones have no Entire git hooks — .git/hooks isn't copied on clone, so prepare-commit-msg/post-commit from your main checkouts (e.g. ~/work/entire/go-nuts) never made it here.
  • Session tracking never started either: the repos' own .claude/settings.json files wire entire hooks claude-code session-start etc., but this session's cwd is the task root ~/work/tasks/go-nuts-extension, which has no .claude directory — so no Entire session/checkpoint exists to stamp into a trailer even if the git hooks were present.
  • I tried to remediate with entire enable --agent claude-code --force in go-nuts, but the permission system denied installing hooks (reasonable — it persists beyond the session).

Options: (a) you run ! cd go-nuts && entire enable --agent claude-code (and the same in entiredb/entire-api) so the hooks exist — though the trailer may still need a tracked session; (b) you tell me to proceed without trailers for this task; (c) the task-workspace tooling gets fixed to install hooks at clone time. I'll hold off on further commits/PRs until you weigh in — happy to amend/redo 869e0ee under whichever regime you pick.

Work completed so far (all in the go-nuts working tree, tested + lint-clean)

All three packages from the issue's "grow go-nuts" section are built, with mise run lint at 0 issues and go test -race ./... green:

  • natsmsg — the unified lift of entiredb's internal/natsmsg + entire-api's otelnats: HeaderCarrier, Inject/Extract/ExtractHeader, StartConsumerSpan, ClampToInt64, the capped KeepInProgress heartbeat, and natsmsgtest.FakeMsg. This commit also adds the OTel API dependency (the footprint decision the issue flags — root package stays nats.go-only, subpackages carry OTel/jetstream). Committed as 869e0ee (sans trailer).
  • jsconsumer — entire-api's scaffold with the two folds the issue asks for: consume errors are suppressed via nuts.IsShutdownFetchErr during shutdown (the COR-923 posture), and Config.KeepInProgress opts a consumer into the AckWait heartbeat around handle. Service-local deps became config: Tracer/Logger fields with sane defaults, AckWait now configurable (default 30s). Tests ported plus an embedded-JetStream end-to-end test for Start. Uncommitted.
  • backoff — the COR-762 policy from fanoutengine, retargeted at jetstream.Msg: Policy{NakDelay, MaxDeliver, TermOnExhaustion}.NakOrTerm() returning (Outcome, error), plus IsFinalDelivery/NumDelivered so callers can gate DLQ capture. Flat delay, Term-on-final-delivery opt-in. Uncommitted.
  • README + root doc.go updated to describe the subpackages and the dependency posture. Uncommitted.

One design note: entiredb's natsmsg also has DeadLetter/SubjectToken (the COR-944 WorkQueue poison capture), which the issue's checkbox does not list for the lift — but the entiredb convergence bullet says to retire internal/natsmsg entirely, so those will need a go-nuts home (likely natsmsg too) when that migration happens. Flagging it now so it doesn't get lost.

Next after your call on the trailers: commit jsconsumer/backoff/docs, open the go-nuts PR, then start the entire-api convergence (fold internal/jsconsumer + otelnats back onto the module, adopt nuts.Connect/Drain — ~12 consumer files).

use entire attach

1mo ago·34.2s·3 calls
dddf3c1natsmsg: trace-context propagation, KeepInProgress, FakeMsg test double Unify entire-core's internal/natsmsg and entire-api's internal/otelnats (itself ported from mirror-pipeline) into one shared package: the HeaderCarrier propagator adapter with Inject/Extract/ExtractHeader and the StartConsumerSpan prologue, the capped KeepInProgress AckWait heartbeat for long handlers, and natsmsgtest.FakeMsg so consumers can assert message disposition without a broker. First package to pull the OpenTelemetry API and nats.go/jetstream into the module (the dependency decision flagged on COR-929); the root package stays nats.go-only, and subpackage isolation keeps these out of root-only importers. Entire-Checkpoint: 71b20e115766+468
1f3abeejsconsumer: durable JetStream pull-consumer scaffold Lift entire-api's internal/jsconsumer: Start (create-or-update durable, AckExplicit, bounded AckWait/MaxDeliver, filter subject, consume, stop on context cancel), Runner.Stop, and the Process[E] per-message prologue (consumer span re-parented across the NATS hop via natsmsg, decode, Term-on-undecodable, dispatch to the handler which owns disposition). Two folds over the entire-api original, per COR-929: consume errors are gated by nuts.IsShutdownFetchErr so a drain during rollout logs nothing (COR-923 posture), and Config.KeepInProgress opts a consumer into the capped natsmsg AckWait heartbeat around handle. Service-local deps became Config fields: Tracer/Logger (defaulting to the OTel global and slog.Default) replace otelsetup, and AckWait is configurable (default 30s, the posture the migrated consumers share). Entire-Checkpoint: 71b20e115766+427
7b9229fbackoff: flat NakWithDelay redelivery policy with Term-on-final-delivery Lift the COR-762 fix from mirror-pipeline's fanoutengine, retargeted at jetstream.Msg: Policy applies a flat NakWithDelay envelope to transient failures, and with TermOnExhaustion set the final delivery Terms instead — on a work-queue stream a Nak the broker drops at MaxDeliver would otherwise orphan the message un-acked until StreamMaxAge, pinning the oldest-pending-age gauge. Disposition only: DLQ capture stays with the caller, gated on the exported IsFinalDelivery (NumDelivered is exported for logs/attrs). NakOrTerm returns the chosen Outcome for the caller's metrics plus any disposition error for its logs. Entire-Checkpoint: 71b20e115766+214
3e376dedocs: describe the natsmsg/jsconsumer/backoff subpackages Root doc.go and README now cover the consumer-side layers and the dependency posture: the root package stays nats.go-only; natsmsg and jsconsumer add the OpenTelemetry API, all three subpackages use nats.go/jetstream, and importing only the root links none of that. Entire-Checkpoint: 71b20e115766+40/-5
0e17ee7backoff: guard IsFinalDelivery against non-positive MaxDeliver A Policy with MaxDeliver left at its zero value made IsFinalDelivery read EVERY delivery as final (NumDelivered >= 0), so TermOnExhaustion Termed on the first transient failure instead of applying the retry envelope. Non-positive MaxDeliver now reads as unlimited redeliveries — the jetstream.ConsumerConfig semantics for the same field — so no delivery is final and NakOrTerm always Naks. Found by Bugbot on #4. Entire-Checkpoint: 71b20e115766+39/-3
2d1955ejsconsumer: export the effective AckWait/MaxDeliver resolution Config.MaxDeliver zero becomes DefaultMaxDeliver (8) on the JetStream consumer, but backoff.Policy reads a zero MaxDeliver as unlimited redeliveries — so a caller passing the raw field into a Policy would build one whose TermOnExhaustion never fires on the broker's real final delivery, orphaning work-queue messages (the COR-762 failure mode the policy exists to prevent). Export the resolution as Config.EffectiveMaxDeliver / EffectiveAckWait (Start and Process already resolve through them), so a Policy is built from the value actually applied on the broker, and document the zero-value mismatch on EffectiveMaxDeliver. AckWait gets the same treatment for hand-wired natsmsg.KeepInProgress heartbeats, where a zero is a silent no-op. Found by Bugbot on #4. Entire-Checkpoint: 788ee0fc2a18 Entire-Checkpoint: 71b20e115766+46/-13
d03ec4bjsconsumer: bound prefetch; KeepInProgress forces one in-flight message Consume ran with nats.go's default 500-message client-side buffer while callbacks dispatch serially, and AckWait runs from server delivery into that buffer — so every delivery queued behind a long handler could exhaust its AckWait unextended and redeliver concurrently to another replica. That defeats KeepInProgress in particular: the heartbeat covers only the message the handler holds. The entiredb consumers the heartbeat was lifted from all pair it with FetchBatch=1; this restores that pairing. KeepInProgress now forces PullMaxMessages(1) (a conflicting explicit MaxMessages is a Start error), and Config.MaxMessages exposes the bound for consumers that want it without the heartbeat — which the entiredb migration needs anyway for natsx FetchBatch parity. Entire-Checkpoint: 71b20e115766+44/-3
96204eejsconsumer: make Runner.Stop concurrency-safe and a completion boundary Stop mutated the unsynchronized cc field, and the context-cancel stop Start arms runs on its own goroutine — an explicit Stop racing it was a data race. It also returned as soon as delivery was halted, so a caller tearing down stores/publishers/the connection after Stop could yank them out from under an in-flight handler. Stop now fires ConsumeContext.Stop through a sync.Once and then blocks on the Closed() channel (captured while the subscription is live, the entiredb natsx pattern), which nats.go closes only when the consume loop has fully wound down including the in-flight callback — restoring the cancel -> join -> drain shutdown ordering (COR-923). Documented: Stop must not be called from inside the handler. Pinned by TestStopWaitsForInFlightHandler and TestStopConcurrentWithCancel under -race. Entire-Checkpoint: 71b20e115766+113/-6
b52078cnatsmsg: keep the heartbeat alive through transient InProgress failures KeepInProgress exited its goroutine on ANY InProgress error, so one dropped extension during a NATS blip (reconnect-buffer overflow, flush timeout) silently forfeited the rest of the handler's runway — AckWait then expired mid-run and the delivery redelivered concurrently to another replica, the duplicate the heartbeat exists to prevent. Only permanent failures stop the heartbeat now: the delivery was already terminally disposed (Ack/Nak/Term racing the final tick) or carries no reply subject, matched against both the jetstream and legacy nats.Msg sentinel errors since the callback may wrap either. Transient failures keep ticking, still bounded by the existing 15-tick cap, so the extension resumes when the connection recovers. Entire-Checkpoint: 71b20e115766+69/-19
227ff94jsconsumer: classify jetstream's own connection-closed error; require Durable Two Start-path gaps: the consume error handler suppressed shutdown noise via nuts.IsShutdownFetchErr, but the jetstream consume loop surfaces its OWN jetstream.ErrConnectionClosed — a distinct sentinel wrapping neither core error — so a clean rollout drain still logged a warning, exactly the COR-923 noise the fold claimed to remove. And an empty Config.Durable sailed into CreateOrUpdateConsumer, which silently creates an ephemeral server-named consumer with no resume-on-restart. isShutdownConsumeErr now matches both error families (still gated on ctx being done — a mid-run connection loss stays a logged fault), and Start rejects an empty Durable before touching the broker. Entire-Checkpoint: 71b20e115766+60/-3
2499342natsmsg: multi-value header extraction; natsmsgtest: observable Nak HeaderCarrier gains Values, satisfying propagation.ValuesGetter — without it the baggage propagator falls back to Get and reads only the first of multiple same-named headers, truncating multi-header W3C baggage on extract. FakeMsg.Nak now counts into a Naks field so a test can tell an undelayed Nak from no disposition at all. Entire-Checkpoint: 71b20e115766+34/-8
4f21f1bbuild: bump nats.go to v1.52.0 and nats-server to v2.14.3 Both match the version entiredb already deploys. go-nuts's go.mod requirement is only a floor — consumers already float above it via MVS (entire-api on v1.51.0, entiredb on v1.52.0) — so this changes nothing those services build against. The point is go-nuts's own tests: they exercised the consumer lifecycle logic (Closed() channel semantics, the jetstream.ErrConnectionClosed / ErrMsgAlreadyAckd sentinels, the PullMaxMessages prefetch default) against nats.go v1.50.0 + embedded server v2.10.22, older than what production runs. Testing at least as new as the deployed version closes that gap; the full race suite passes unchanged, confirming none of that version-sensitive behavior shifted. nats-server is a test-only dependency (embedded broker in the jsconsumer suite), so its bump affects go-nuts's tests only, not any consumer. Entire-Checkpoint: 71b20e115766+33/-27