From 4fbe6e347f11884f0d573abf4f7c11fa213cf49b Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:20:19 +0100 Subject: [PATCH] nats_stream_messages shows a body only on an event stream (issue 1966, review 1) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/nats/cmd/nats-tools/consumer.go | 18 ++++++++++-- modules/nats/cmd/nats-tools/consumer_test.go | 29 ++++++++++++++++++++ 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/modules/nats/cmd/nats-tools/consumer.go b/modules/nats/cmd/nats-tools/consumer.go index c0335ca0f..93cb8e2ab 100644 --- a/modules/nats/cmd/nats-tools/consumer.go +++ b/modules/nats/cmd/nats-tools/consumer.go @@ -11,6 +11,7 @@ import ( "encoding/json" "errors" "fmt" + "regexp" "strings" "time" @@ -109,6 +110,11 @@ const ( maxAsks = 3000 ) +// eventStream is a stream that keeps events: the shared EVENTS and a seat's kept events SEAT__EVENTS. Only these show +// a body; every other stream holds calls and replies, asks, mail, dead letters, records or logs, whose bodies carry what +// is not an event's to show. It fails closed: a stream not named here shows sequence, subject, time and size only. +var eventStream = regexp.MustCompile(`^(EVENTS|SEAT_[A-Z0-9]+(_[A-Z0-9]+)*_EVENTS)$`) + // MessageRow is one message of the answer. type MessageRow struct { Seq uint64 `json:"seq"` @@ -266,7 +272,7 @@ func readTools(jsz func() ([]byte, error), open MsgOpen) []stdio.Tool { return Consumer(doc, arg(a, "stream"), arg(a, "consumer")) }}, {Name: "nats_stream_messages", - Description: "The messages a stream keeps for one subject (a filter with * or > is allowed), newest first: sequence, subject, time and size of each, and the body only when asked for and no larger than 4 KiB. Bounded by count (default 20, at most 200) and bytes (default 1 MiB); says when it stopped before the oldest the stream keeps. Read on the bus module's own credential. Refuses the streams of a module's kept state (KV_*, OBJ_*): their values are read through their owners' tools; this tool is for lost events.", + Description: "The messages a stream keeps for one subject (a filter with * or > is allowed), newest first: sequence, subject, time and size of each, and the body only when asked for, no larger than 4 KiB, and only on an event stream (EVENTS or a seat's kept events); on every other stream (asks, calls and replies, mail, dead letters, records, logs) only sequence, subject, time and size. Bounded by count (default 20, at most 200) and bytes (default 1 MiB); says when it stopped before the oldest the stream keeps. Read on the bus module's own credential. Refuses the streams of a module's kept state (KV_*, OBJ_*): their values are read through their owners' tools; this tool is for lost events.", Input: map[string]any{"stream": str("the stream, e.g. EVENTS"), "subject": str("a subject or filter, e.g. mesh.mod.mesh-delivery-indexer.event.view-changed"), "count": num("at most this many (default 20, at most 200)"), "bytes": num("at most this many bytes of messages (default 1048576)"), "body": map[string]any{"type": "boolean", "description": "carry the body of each message of at most 4 KiB"}}, @@ -289,6 +295,10 @@ func readTools(jsz func() ([]byte, error), open MsgOpen) []stdio.Tool { count, _ := a["count"].(float64) bytes, _ := a["bytes"].(float64) body, _ := a["body"].(bool) + bodyHidden := "" + if body && !eventStream.MatchString(stream) { + body, bodyHidden = false, stream+" is not an event stream (EVENTS or a seat's kept events): its bodies are not shown, only sequence, subject, time and size" + } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() get, closeBus, err := open(ctx) @@ -296,7 +306,11 @@ func readTools(jsz func() ([]byte, error), open MsgOpen) []stdio.Tool { return nil, err } defer closeBus() - return NewestMessages(ctx, get, stream, subject, first, last, int(count), int(bytes), body) + out, err := NewestMessages(ctx, get, stream, subject, first, last, int(count), int(bytes), body) + if err == nil && bodyHidden != "" { + out["body_not_shown"] = bodyHidden + } + return out, err }}, } } diff --git a/modules/nats/cmd/nats-tools/consumer_test.go b/modules/nats/cmd/nats-tools/consumer_test.go index c74d14df8..e17017bb0 100644 --- a/modules/nats/cmd/nats-tools/consumer_test.go +++ b/modules/nats/cmd/nats-tools/consumer_test.go @@ -128,6 +128,35 @@ func TestTheReadToolsAreServedAndRefuseBadArguments(t *testing.T) { t.Fatalf("%s was read: %v", st, err) } } + jszOf := strings.Replace(consumerJsz, `"name":"EVENTS"`, `"name":"ASKS"`, 1) + ask := readTools(func() ([]byte, error) { return []byte(jszOf), nil }, + func(context.Context) (MsgGet, func(), error) { + return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil + }) + for _, tl := range ask { + if tl.Name != "nats_stream_messages" { + continue + } + o, err := tl.Run(map[string]any{"stream": "ASKS", "subject": "a.b", "body": true}) + m := o.(map[string]any) + if err != nil || m["body_not_shown"] == nil || m["messages"].([]MessageRow)[0].Body != "" { + t.Fatalf("an ask stream showed a body: %v %v", o, err) + } + } + ev, _ := by["nats_stream_messages"](map[string]any{"stream": "EVENTS", "subject": "a.b", "body": true}) + if rows := ev.(map[string]any)["messages"].([]MessageRow); len(rows) != 1 || rows[0].Body != "body-50" { + t.Fatalf("an event stream showed no body: %v", ev) + } + for _, s := range []string{"EVENTS", "SEAT_NODE_BUILD_GO_EVENTS"} { + if !eventStream.MatchString(s) { + t.Fatalf("%s is an event stream", s) + } + } + for _, s := range []string{"SEAT_CHANNEL", "ASKS", "MAIL", "DEAD_LETTERS", "LOG_X", "SEAT_X_worker"} { + if eventStream.MatchString(s) { + t.Fatalf("%s is not an event stream", s) + } + } if _, err := by["nats_stream_messages"](map[string]any{"stream": "EVENTS"}); err == nil { t.Fatal("no subject answered") }