You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
mq: propagate terminal JetStream consumer errors to the ingest worker #587
Area: mq / ingest — follow-up from CodeRabbit on #586
jsConsumer.Consume (internal/mq/embedded.go) does not pass jetstream.ConsumeErrHandler. In nats.go v1.53.1 the pull consumer reports terminal errors such as ErrConsumerDeleted through that handler and then stops the subscription. The ingest worker's dispatchLoop (internal/ingest/worker.go) only sees messages, so after the client-side buffer drains it sits on msgChan forever without knowing that delivery has stopped. Ingest would silently halt with no log line and no restart.
This predates #586: the worker never passed the handler when it called nats.go directly either. It is deferred out of that PR because it is a design change to the mq.Consumer contract, not a boundary move.
Exposure today
Narrow. The broker is in-process with DontListen, so there is no connection to drop and no external client to delete the buffer-consumer durable. It becomes real once the boundary is backed by a remote broker or once anything in-process can delete or recreate consumers (multi-tenant per-tenant streams, #583 story 5).
Proposed shape
mq.Consumer gains an asynchronous error path (an error channel or an OnError callback alongside Consume), fed from jetstream.ConsumeErrHandler. Non-terminal errors (ErrNoHeartbeat, leadership changes) are logged; terminal ones (ErrConsumerDeleted, ErrBadRequest, ErrConnectionClosed) are forwarded.
dispatchLoop selects on that path next to ctx.Done() and msgChan. On a terminal error it logs at ERROR, drains what it has, and either re-creates the consumer and resumes or returns and lets the process fail loud. Which of the two is the decision to make here; fail-loud is consistent with how boot config is treated.
testutil mock and internal/mq/embedded_test.go coverage for the forwarded error, plus a worker test that a terminal error ends or restarts the loop rather than blocking.
Related: #583 (story 4 sealed the boundary; story 5 changes subjects and streams), #586.
Activity