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

@@ -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 {

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)
 	}
 }