From bff850f7983865b21ed708c40ef7008971209895 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 10:45:00 +0100 Subject: [PATCH] nats: nats_consumer_remove reads a push consumer too (issue 1126) js.Consumer refuses a push consumer ("consumer is not a pull consumer"), so the controller's leftovers on EVENTS and CONTROL could not be read and were not removed. The tool now reads a consumer of either kind (CONSUMER.INFO only, which is all its grant gives), before the delete and for the confirmation after it. Quick fix: this change, in the tool, one function, about an hour with its test rows (a pull and a push consumer). Thorough fix: the same tool also needs its grant to cover the KV_* streams' consumers (mesh-controller-take-in, mesh-controller-config-follow), and that grant is composed by mesh-sdk's BusRemovalGrants from a closed set (ControllerLeftoverConsumersOn), not by the manifest, so it is not in this commit: it needs a change in mesh-sdk and a group delivery. Chosen here: the quick fix for the read, since it is one place and blocks the operator's removal now; the grant is reported to the lead. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- modules/nats/cmd/nats-tools/removal.go | 24 +++++++++++++++++--- modules/nats/cmd/nats-tools/removal_test.go | 25 ++++++++++++++++++--- 2 files changed, 43 insertions(+), 6 deletions(-) diff --git a/modules/nats/cmd/nats-tools/removal.go b/modules/nats/cmd/nats-tools/removal.go index fd64382c..21888b0f 100644 --- a/modules/nats/cmd/nats-tools/removal.go +++ b/modules/nats/cmd/nats-tools/removal.go @@ -72,6 +72,25 @@ func step(ctx context.Context) (context.Context, context.CancelFunc) { return context.WithTimeout(ctx, answerWithin) } +// readConsumer reads one consumer of any kind. **A push consumer is read as well as a pull consumer**: js.Consumer +// refuses one ("consumer is not a pull consumer", the controller's leftovers on EVENTS and CONTROL were push consumers), +// so the push reading is tried when the pull reading refuses. Both ask only CONSUMER.INFO of that consumer, which is +// all the grant gives (never STREAM.INFO of a stream the tool may not read). +func readConsumer(ctx context.Context, js jetstream.JetStream, stream, consumer string) (*jetstream.ConsumerInfo, error) { + c, err := js.Consumer(ctx, stream, consumer) + if errors.Is(err, jetstream.ErrNotPullConsumer) { + p, perr := js.PushConsumer(ctx, stream, consumer) + if perr != nil { + return nil, perr + } + return p.CachedInfo(), nil + } + if err != nil { + return nil, err + } + return c.CachedInfo(), nil +} + // ConsumerRemoved is what nats_consumer_remove answers and says. type ConsumerRemoved struct { Stream string `json:"stream"` @@ -98,7 +117,7 @@ func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) ( } defer closeBus() sctx, cancel := step(ctx) - c, err := js.Consumer(sctx, stream, consumer) + info, err := readConsumer(sctx, js, stream, consumer) cancel() if err != nil { if errors.Is(err, jetstream.ErrConsumerNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) { @@ -106,7 +125,6 @@ func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) ( } return nil, unanswered(err, fmt.Sprintf("the consumer %s of %s could not be read, so it was not removed", consumer, stream)) } - info := c.CachedInfo() was := map[string]any{ "durable": info.Config.Durable, "filter_subject": info.Config.FilterSubject, "filter_subjects": info.Config.FilterSubjects, "deliver_subject": info.Config.DeliverSubject, "pending": info.NumPending, "ack_pending": info.NumAckPending, @@ -119,7 +137,7 @@ func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) ( return nil, unanswered(err, fmt.Sprintf("consumer %s of %s was not removed", consumer, stream)) } sctx, cancel = step(ctx) - _, err = js.Consumer(sctx, stream, consumer) + _, err = readConsumer(sctx, js, stream, consumer) cancel() // Only the bus saying it is not there is a removal; a read that failed otherwise confirms nothing. if err == nil { diff --git a/modules/nats/cmd/nats-tools/removal_test.go b/modules/nats/cmd/nats-tools/removal_test.go index 3e211b27..b11f0929 100644 --- a/modules/nats/cmd/nats-tools/removal_test.go +++ b/modules/nats/cmd/nats-tools/removal_test.go @@ -189,8 +189,27 @@ accounts { MESH { jetstream: enabled, users: [ if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "EVENTS", "consumer": "mesh-grants", "why": "x"}); err == nil { t.Fatal("a consumer of EVENTS outside the grant was removed") } - if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "EVENTS", "consumer": "controller-merges", "why": "the controller ended"}); err != nil { - t.Fatalf("the controller's consumer of EVENTS: %v", err) + // Issue 1126: the controller's leftovers were push consumers too, which js.Consumer refuses ("not a pull consumer"): + // a pull and a push consumer are read, removed and confirmed alike. + for _, row := range []struct{ name, kind, deliver string }{ + {"controller-merges", "pull", ""}, + {"controller", "push", "deliver.controller"}, + } { + if row.deliver != "" { + if _, err := ajs.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: row.name, DeliverSubject: row.deliver, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + } + got, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "EVENTS", "consumer": row.name, "why": "the controller ended"}) + if err != nil { + t.Fatalf("the controller's %s consumer %s of EVENTS: %v", row.kind, row.name, err) + } + if r := got.(ConsumerRemoved); !r.Removed || r.Was["deliver_subject"] != row.deliver { + t.Fatalf("the %s consumer %s: %+v", row.kind, row.name, r) + } + if _, err := ajs.Consumer(ctx, "EVENTS", row.name); err == nil { + t.Fatalf("the %s consumer %s is still there", row.kind, row.name) + } } if _, err := ajs.Consumer(ctx, "EVENTS", "mesh-grants"); err != nil { t.Fatalf("the other consumer of EVENTS went: %v", err) @@ -241,7 +260,7 @@ accounts { MESH { jetstream: enabled, users: [ if k := keys(other); k != "g14.1" { t.Fatalf("another bucket changed: %s", k) } - if strings.Join(said, ",") != "consumer-removed,stream-removed,consumer-removed,stream-removed,bucket-keys-removed" { + if strings.Join(said, ",") != "consumer-removed,stream-removed,consumer-removed,consumer-removed,stream-removed,bucket-keys-removed" { t.Fatalf("the audit events said: %v", said) } }