nats_stream_messages shows a body only on an event stream (issue 1966, review 1)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-09 21:20:19 +01:00
parent 3d70bd3660
commit 4fbe6e347f
2 changed files with 45 additions and 2 deletions
+16 -2
View File
@@ -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_<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
}},
}
}
@@ -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")
}