From 13dea6b1aa13098f4ec8a95dda21b3d48c74b194 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:13:02 +0100 Subject: [PATCH 1/5] The bus module reads one consumer in full and a stream's messages by subject (issue 1966) A lost event between a shared stream and a module's consumer could not be located: no tool read a consumer's delivered and acknowledged sequence or a stream's messages for one subject. nats_consumer reads from the monitoring document; nats_stream_messages asks STREAM.MSG.GET on the module's own credential, newest first and bounded. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/nats/README.md | 2 +- modules/nats/cmd/nats-tools/consumer.go | 299 +++++++++++++++++++ modules/nats/cmd/nats-tools/consumer_test.go | 127 ++++++++ modules/nats/cmd/nats-tools/main.go | 1 + modules/nats/cmd/nats-tools/tools.go | 14 + modules/nats/module.json | 2 + 6 files changed, 444 insertions(+), 1 deletion(-) create mode 100644 modules/nats/cmd/nats-tools/consumer.go create mode 100644 modules/nats/cmd/nats-tools/consumer_test.go diff --git a/modules/nats/README.md b/modules/nats/README.md index aeab2e77b..104e79e20 100644 --- a/modules/nats/README.md +++ b/modules/nats/README.md @@ -78,7 +78,7 @@ live one**, which a person swaps in. docker start mesh-broker-nats The store kept aside is removed by a person once the bus is seen whole (`nats_streams`, - `nats_backlog`). + `nats_backlog`; one consumer in full: `nats_consumer`; the messages of a stream for a subject, newest first: `nats_stream_messages`). A single stream can be restored into any server that does not hold it with `mesh-nats-snapshot restore --server --credential `; the archive is also what the nats CLI diff --git a/modules/nats/cmd/nats-tools/consumer.go b/modules/nats/cmd/nats-tools/consumer.go new file mode 100644 index 000000000..5bb9f8ea0 --- /dev/null +++ b/modules/nats/cmd/nats-tools/consumer.go @@ -0,0 +1,299 @@ +package main + +// Where a lost event is located (issue 1966, tools-are-how-a-module-is-spoken-to): one consumer of a stream as the server +// holds it, and the messages a stream keeps for one subject. Both read; neither changes the bus. The consumer is read from +// the server's monitoring document, as nats_streams and nats_backlog are, so it needs no grant; the messages are asked of +// the bus on the module's own credential (STREAM.MSG.GET), as the removal tools are, so what the credential is not +// granted shows as an unanswered request and is said so (unanswered). + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Consumer is one consumer of one stream: how it is set and where it stands. +func Consumer(jsz []byte, stream, durable string) (map[string]any, error) { + var doc jszDoc + if err := json.Unmarshal(jsz, &doc); err != nil { + return nil, err + } + streamFound := false + for _, a := range doc.Accounts { + for _, s := range a.Streams { + if s.Name != stream { + continue + } + streamFound = true + for _, c := range s.Consumers { + if c.Name != durable { + continue + } + filters := append([]string(nil), c.Config.FilterSubjects...) + if c.Config.FilterSubject != "" { + filters = append(filters, c.Config.FilterSubject) + } + kind := "pull" + if c.Config.DeliverSubject != "" { + kind = "push" + } + return map[string]any{ + "stream": stream, "consumer": c.Name, "kind": kind, "created": c.Created, + "config": map[string]any{ + "filter_subjects": filters, "deliver_policy": c.Config.DeliverPolicy, "ack_policy": c.Config.AckPolicy, + "ack_wait_seconds": c.Config.AckWait / int64(time.Second), "max_ack_pending": c.Config.MaxAckPending, + "max_deliver": c.Config.MaxDeliver, "replay_policy": c.Config.ReplayPolicy, + "deliver_subject": c.Config.DeliverSubject, "deliver_group": c.Config.DeliverGroup, + }, + "state": map[string]any{ + "delivered_stream_seq": c.Delivered.Stream, "delivered_consumer_seq": c.Delivered.Consumer, + "ack_floor_stream_seq": c.AckFloor.Stream, "ack_floor_consumer_seq": c.AckFloor.Consumer, + "num_pending": c.NumPending, "num_ack_pending": c.NumAckPending, "num_redelivered": c.NumRedelivered, + "num_waiting": c.NumWaiting, "last_delivered": c.Delivered.LastActive, "last_acked": c.AckFloor.LastActive, + }, + "stream_state": map[string]any{"first_seq": s.State.FirstSeq, "last_seq": s.State.LastSeq, "messages": s.State.Messages}, + }, nil + } + } + } + if !streamFound { + return nil, fmt.Errorf("no stream called %s; nats_streams lists them", stream) + } + return nil, fmt.Errorf("stream %s has no consumer called %s; nats_streams with the stream lists its consumers", stream, durable) +} + +// StreamRange is a stream's first and last sequence, from the monitoring document. +func StreamRange(jsz []byte, stream string) (first, last uint64, err error) { + var doc jszDoc + if err := json.Unmarshal(jsz, &doc); err != nil { + return 0, 0, err + } + for _, a := range doc.Accounts { + for _, s := range a.Streams { + if s.Name == stream { + return uint64(s.State.FirstSeq), uint64(s.State.LastSeq), nil + } + } + } + return 0, 0, fmt.Errorf("no stream called %s; nats_streams lists them", stream) +} + +// Stored is one message a stream keeps. +type Stored struct { + Seq uint64 + Subject string + Time time.Time + Data []byte +} + +// errNoMessage is what a MsgGet answers when no message at or after the sequence matches the subject. +var errNoMessage = errors.New("no message found") + +// MsgGet asks the stream for the first message at or after seq whose subject matches (any when subject is empty). +type MsgGet func(ctx context.Context, stream string, seq uint64, subject string) (*Stored, error) + +const ( + defaultMessages = 20 + maxMessages = 200 + defaultBytes = 1 << 20 + // bodyLimit is the largest body an answer carries, and only when asked. + bodyLimit = 4096 + // block is how many sequences are scanned at once going back from the newest, and maxAsks the most requests one + // answer makes: a subject rarely kept stops the answer, said as truncated, never a read of the whole stream. + block = 4096 + maxAsks = 3000 +) + +// MessageRow is one message of the answer. +type MessageRow struct { + Seq uint64 `json:"seq"` + Subject string `json:"subject"` + Time time.Time `json:"time"` + Size int `json:"size"` + Body string `json:"body,omitempty"` + BodyOmitted string `json:"body_omitted,omitempty"` +} + +// NewestMessages is the messages of the stream on the subject (a filter may hold wildcards), newest first, at most count +// and at most maxBytes of their sizes; first and last are the stream's own range. Said: whether it stopped before the +// oldest the stream keeps. +func NewestMessages(ctx context.Context, get MsgGet, stream, subject string, first, last uint64, count, maxBytes int, body bool) (map[string]any, error) { + if count <= 0 { + count = defaultMessages + } + if count > maxMessages { + count = maxMessages + } + if maxBytes <= 0 { + maxBytes = defaultBytes + } + var found []*Stored + asks, bytes, stopped := 0, 0, "" + end := last + 1 // exclusive +scan: + for end > first && len(found) < count { + start := first + if end-first > block { + start = end - block + } + var chunk []*Stored + for seq := start; seq < end; { + if asks++; asks > maxAsks { + stopped = fmt.Sprintf("stopped after %d requests", maxAsks) + break scan + } + m, err := get(ctx, stream, seq, subject) + if errors.Is(err, errNoMessage) { + break + } + if err != nil { + return nil, unanswered(err, fmt.Sprintf("the messages of %s could not be read", stream)) + } + if m.Seq >= end { + break + } + chunk = append(chunk, m) + seq = m.Seq + 1 + } + for i := len(chunk) - 1; i >= 0; i-- { + if len(found) >= count { + stopped = fmt.Sprintf("stopped at the %d messages asked for", count) + break scan + } + if bytes+len(chunk[i].Data) > maxBytes && len(found) > 0 { + stopped = fmt.Sprintf("stopped at %d bytes", maxBytes) + break scan + } + bytes += len(chunk[i].Data) + found = append(found, chunk[i]) + } + end = start + } + rows := make([]MessageRow, 0, len(found)) + for _, m := range found { + r := MessageRow{Seq: m.Seq, Subject: m.Subject, Time: m.Time, Size: len(m.Data)} + switch { + case !body: + case len(m.Data) > bodyLimit: + r.BodyOmitted = fmt.Sprintf("larger than %d bytes", bodyLimit) + default: + r.Body = string(m.Data) + } + rows = append(rows, r) + } + out := map[string]any{"stream": stream, "subject": subject, "first_seq": first, "last_seq": last, "returned": len(rows), "messages": rows} + if stopped != "" { + out["stopped"] = stopped + } + return out, nil +} + +// MsgOpen opens the bus for one answer and gives the reader of messages on it; close ends the connection. +type MsgOpen func(ctx context.Context) (get MsgGet, close func(), err error) + +// LiveMsgGet reads messages by STREAM.MSG.GET on the module's own credential, one connection for the whole answer. +func LiveMsgGet(bus Bus) MsgOpen { + return func(ctx context.Context) (MsgGet, func(), error) { + js, closeBus, err := bus(ctx) + if err != nil { + return nil, nil, err + } + return func(ctx context.Context, stream string, seq uint64, subject string) (*Stored, error) { + sctx, cancel := step(ctx) + defer cancel() + req := map[string]any{"seq": seq} + if subject != "" { + req["next_by_subj"] = subject + } + raw, _ := json.Marshal(req) + rep, err := js.Conn().RequestWithContext(sctx, "$JS.API.STREAM.MSG.GET."+stream, raw) + if err != nil { + return nil, err + } + return decodeStored(rep.Data) + }, closeBus, nil + } +} + +func decodeStored(b []byte) (*Stored, error) { + var doc struct { + Error *struct { + Code int + ErrCode int `json:"err_code"` + Description string `json:"description"` + } `json:"error"` + Message *struct { + Seq uint64 `json:"seq"` + Subject string `json:"subject"` + Data []byte `json:"data"` + Time time.Time `json:"time"` + } `json:"message"` + } + if err := json.Unmarshal(b, &doc); err != nil { + return nil, err + } + if doc.Error != nil { + if doc.Error.Code == 404 { + return nil, errNoMessage + } + return nil, fmt.Errorf("the bus refused: %s (%d)", doc.Error.Description, doc.Error.ErrCode) + } + if doc.Message == nil { + return nil, errNoMessage + } + return &Stored{Seq: doc.Message.Seq, Subject: doc.Message.Subject, Time: doc.Message.Time, Data: doc.Message.Data}, nil +} + +// readTools are the two reads that locate a lost event. +func readTools(jsz func() ([]byte, error), open MsgOpen) []stdio.Tool { + return []stdio.Tool{ + {Name: "nats_consumer", + Description: "One consumer of one stream, in full: how it is set (filters, deliver policy, ack policy, ack wait, max ack pending, max deliver) and where it stands (delivered and acknowledged-up-to sequence of the stream and of the consumer, pending, unacknowledged, redelivered, waiting pulls, last delivery), with the stream's own first and last sequence beside it — the question behind an event a module did not hear.", + Input: map[string]any{"stream": str("the stream, e.g. EVENTS"), "consumer": str("the consumer's durable name, e.g. novox_mesh-module-assigner")}, + Run: func(a map[string]any) (any, error) { + if !name.MatchString(arg(a, "stream")) || !name.MatchString(arg(a, "consumer")) { + return nil, errors.New("stream and consumer are required, each one name (letters, digits, - and _)") + } + doc, err := jsz() + if err != nil { + return nil, err + } + 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.", + 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"}}, + Run: func(a map[string]any) (any, error) { + stream, subject := arg(a, "stream"), strings.TrimSpace(arg(a, "subject")) + if !name.MatchString(stream) || subject == "" || strings.ContainsAny(subject, " \t") { + return nil, errors.New("stream (one name) and subject (a subject or filter, no space) are required") + } + doc, err := jsz() + if err != nil { + return nil, err + } + first, last, err := StreamRange(doc, stream) + if err != nil { + return nil, err + } + count, _ := a["count"].(float64) + bytes, _ := a["bytes"].(float64) + body, _ := a["body"].(bool) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + get, closeBus, err := open(ctx) + if err != nil { + return nil, err + } + defer closeBus() + return NewestMessages(ctx, get, stream, subject, first, last, int(count), int(bytes), body) + }}, + } +} diff --git a/modules/nats/cmd/nats-tools/consumer_test.go b/modules/nats/cmd/nats-tools/consumer_test.go new file mode 100644 index 000000000..0b6b796e4 --- /dev/null +++ b/modules/nats/cmd/nats-tools/consumer_test.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" +) + +const consumerJsz = `{"account_details":[{"stream_detail":[{"name":"EVENTS","state":{"messages":9,"first_seq":1,"last_seq":50}, + "consumer_detail":[{"name":"ace_assigner","created":"2026-11-09T10:00:00Z", + "config":{"filter_subjects":["mesh.mod.a.event.b"],"deliver_policy":"new","ack_policy":"explicit","ack_wait":30000000000,"max_ack_pending":1000,"max_deliver":5}, + "delivered":{"consumer_seq":40,"stream_seq":48,"last_active":"2026-11-09T11:00:00Z"},"ack_floor":{"consumer_seq":39,"stream_seq":47}, + "num_pending":2,"num_ack_pending":1,"num_redelivered":3,"num_waiting":0}]}]}]}` + +// fake is a stream of subjects by sequence 1..n. +func fake(subjects map[uint64]string, fail error) MsgGet { + return func(_ context.Context, _ string, seq uint64, subject string) (*Stored, error) { + if fail != nil { + return nil, fail + } + for s := seq; s <= 50; s++ { + if sub, ok := subjects[s]; ok && (subject == "" || sub == subject) { + return &Stored{Seq: s, Subject: sub, Time: time.Unix(int64(s), 0).UTC(), Data: []byte(fmt.Sprintf("body-%d", s))}, nil + } + } + return nil, errNoMessage + } +} + +func TestAConsumerIsReadWithItsConfigAndItsState(t *testing.T) { + got, err := Consumer([]byte(consumerJsz), "EVENTS", "ace_assigner") + if err != nil { + t.Fatal(err) + } + cfg, st := got["config"].(map[string]any), got["state"].(map[string]any) + if cfg["ack_policy"] != "explicit" || cfg["deliver_policy"] != "new" || cfg["max_ack_pending"].(int64) != 1000 || cfg["ack_wait_seconds"].(int64) != 30 { + t.Fatalf("%v", cfg) + } + if st["delivered_stream_seq"].(int64) != 48 || st["ack_floor_stream_seq"].(int64) != 47 || st["num_pending"].(int64) != 2 || st["num_redelivered"].(int64) != 3 { + t.Fatalf("%v", st) + } + for _, c := range [][2]string{{"NOPE", "x"}, {"EVENTS", "nope"}} { + if _, err := Consumer([]byte(consumerJsz), c[0], c[1]); err == nil { + t.Fatalf("%v answered", c) + } + } +} + +func TestMessagesComeNewestFirstBoundedAndBodiesOnlyWhenAsked(t *testing.T) { + ctx := context.Background() + subjects := map[uint64]string{} + for s := uint64(1); s <= 50; s++ { + subjects[s] = "a.b" + if s%10 == 0 { + subjects[s] = "a.c" + } + } + got, _ := NewestMessages(ctx, fake(subjects, nil), "EVENTS", "a.c", 1, 50, 3, 0, false) + rows := got["messages"].([]MessageRow) + if len(rows) != 3 || rows[0].Seq != 50 || rows[1].Seq != 40 || rows[2].Seq != 30 || rows[0].Body != "" || rows[0].Size != 7 || got["stopped"] == nil { + t.Fatalf("%+v", got) + } + got, _ = NewestMessages(ctx, fake(subjects, nil), "EVENTS", "a.c", 1, 50, 20, 0, true) + rows = got["messages"].([]MessageRow) + if len(rows) != 5 || rows[4].Seq != 10 || rows[0].Body != "body-50" || got["stopped"] != nil { + t.Fatalf("%+v", got) + } + got, _ = NewestMessages(ctx, fake(subjects, nil), "EVENTS", "a.c", 1, 50, 20, 16, false) + if n := len(got["messages"].([]MessageRow)); n != 2 || !strings.Contains(got["stopped"].(string), "bytes") { + t.Fatalf("%+v", got) + } + big := func(_ context.Context, _ string, seq uint64, _ string) (*Stored, error) { + if seq > 5 { + return nil, errNoMessage + } + return &Stored{Seq: 5, Subject: "x", Data: []byte(strings.Repeat("z", bodyLimit+1))}, nil + } + got, _ = NewestMessages(ctx, big, "EVENTS", "x", 1, 5, 5, 0, true) + if r := got["messages"].([]MessageRow)[0]; r.Body != "" || r.BodyOmitted == "" { + t.Fatalf("%+v", r) + } +} + +func TestMessagesScanBackAcrossBlocksAndSayAnUnansweredRead(t *testing.T) { + ctx := context.Background() + got, _ := NewestMessages(ctx, func(_ context.Context, _ string, seq uint64, _ string) (*Stored, error) { + if seq <= 3 { + return &Stored{Seq: 3, Subject: "old"}, nil + } + return nil, errNoMessage + }, "EVENTS", "old", 1, 3*block, 5, 0, false) + if rows := got["messages"].([]MessageRow); len(rows) != 1 || rows[0].Seq != 3 { + t.Fatalf("a message three blocks back was not found: %+v", got) + } + _, err := NewestMessages(ctx, fake(nil, context.DeadlineExceeded), "EVENTS", "a", 1, 50, 5, 0, false) + if err == nil || !strings.Contains(err.Error(), "grant") { + t.Fatalf("%v", err) + } + if _, err := NewestMessages(ctx, fake(nil, errors.New("boom")), "EVENTS", "a", 1, 50, 5, 0, false); err == nil { + t.Fatal("a failed read answered") + } +} + +func TestTheReadToolsAreServedAndRefuseBadArguments(t *testing.T) { + ts := readTools(func() ([]byte, error) { return []byte(consumerJsz), nil }, + func(context.Context) (MsgGet, func(), error) { return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil }) + by := map[string]func(map[string]any) (any, error){} + for _, tl := range ts { + by[tl.Name] = tl.Run + } + if _, err := by["nats_consumer"](map[string]any{"stream": "EVENTS", "consumer": "ace_assigner"}); err != nil { + t.Fatal(err) + } + if _, err := by["nats_consumer"](map[string]any{"stream": "EVENTS.x"}); err == nil { + t.Fatal("bad names answered") + } + out, err := by["nats_stream_messages"](map[string]any{"stream": "EVENTS", "subject": "a.>"}) + if err != nil || out.(map[string]any)["returned"].(int) != 0 && out.(map[string]any)["returned"].(int) != 1 { + t.Fatalf("%v %v", out, err) + } + if _, err := by["nats_stream_messages"](map[string]any{"stream": "EVENTS"}); err == nil { + t.Fatal("no subject answered") + } +} diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index f16222c37..034c5be3b 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -189,5 +189,6 @@ func allTools() []stdio.Tool { out = append(out, expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...) out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz)) + out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...) return append(out, requestTools(RequestsInState())...) } diff --git a/modules/nats/cmd/nats-tools/tools.go b/modules/nats/cmd/nats-tools/tools.go index 88d5b8b97..8235f97de 100644 --- a/modules/nats/cmd/nats-tools/tools.go +++ b/modules/nats/cmd/nats-tools/tools.go @@ -317,12 +317,26 @@ type jszDoc struct { FilterSubjects []string `json:"filter_subjects"` DeliverSubject string `json:"deliver_subject"` MaxDeliver int64 `json:"max_deliver"` + AckPolicy string `json:"ack_policy"` + AckWait int64 `json:"ack_wait"` + DeliverPolicy string `json:"deliver_policy"` + MaxAckPending int64 `json:"max_ack_pending"` + ReplayPolicy string `json:"replay_policy"` + DeliverGroup string `json:"deliver_group"` } `json:"config"` + Created string `json:"created"` + AckFloor struct { + Consumer int64 `json:"consumer_seq"` + Stream int64 `json:"stream_seq"` + LastActive string `json:"last_active"` + } `json:"ack_floor"` NumPending int64 `json:"num_pending"` NumAckPending int64 `json:"num_ack_pending"` NumRedelivered int64 `json:"num_redelivered"` NumWaiting int64 `json:"num_waiting"` Delivered struct { + Consumer int64 `json:"consumer_seq"` + Stream int64 `json:"stream_seq"` LastActive string `json:"last_active"` } `json:"delivered"` } `json:"consumer_detail"` diff --git a/modules/nats/module.json b/modules/nats/module.json index aa0824633..ad9d08019 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -116,6 +116,8 @@ "nats_subscriptions", "nats_streams", "nats_backlog", + "nats_consumer", + "nats_stream_messages", "nats_buckets", "nats_users", "nats_user_can", -- 2.54.0 From a4e5a8ffef64e1d32f1659d686ae4a3b5e6d4ddb Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:14:00 +0100 Subject: [PATCH 2/5] Format the bus tools' new test (issue 1966) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/nats/cmd/nats-tools/consumer_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/modules/nats/cmd/nats-tools/consumer_test.go b/modules/nats/cmd/nats-tools/consumer_test.go index 0b6b796e4..858efd80c 100644 --- a/modules/nats/cmd/nats-tools/consumer_test.go +++ b/modules/nats/cmd/nats-tools/consumer_test.go @@ -106,7 +106,9 @@ func TestMessagesScanBackAcrossBlocksAndSayAnUnansweredRead(t *testing.T) { func TestTheReadToolsAreServedAndRefuseBadArguments(t *testing.T) { ts := readTools(func() ([]byte, error) { return []byte(consumerJsz), nil }, - func(context.Context) (MsgGet, func(), error) { return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil }) + func(context.Context) (MsgGet, func(), error) { + return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil + }) by := map[string]func(map[string]any) (any, error){} for _, tl := range ts { by[tl.Name] = tl.Run -- 2.54.0 From 3d70bd36601424c8d2505d79e14e9999a0ec2805 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:18:43 +0100 Subject: [PATCH 3/5] nats_stream_messages refuses a module's kept state (KV_*, OBJ_*), which can hold secrets (issue 1966, review 1) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/nats/cmd/nats-tools/consumer.go | 5 ++++- modules/nats/cmd/nats-tools/consumer_test.go | 5 +++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/modules/nats/cmd/nats-tools/consumer.go b/modules/nats/cmd/nats-tools/consumer.go index 5bb9f8ea0..c0335ca0f 100644 --- a/modules/nats/cmd/nats-tools/consumer.go +++ b/modules/nats/cmd/nats-tools/consumer.go @@ -266,7 +266,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.", + 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.", 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"}}, @@ -275,6 +275,9 @@ func readTools(jsz func() ([]byte, error), open MsgOpen) []stdio.Tool { if !name.MatchString(stream) || subject == "" || strings.ContainsAny(subject, " \t") { return nil, errors.New("stream (one name) and subject (a subject or filter, no space) are required") } + if strings.HasPrefix(stream, "KV_") || strings.HasPrefix(stream, "OBJ_") { + return nil, fmt.Errorf("%s is a module's kept state (a bucket or an object store), which can hold secrets: it is read through its owner's tools, never here; this tool reads the events of a stream", stream) + } doc, err := jsz() if err != nil { return nil, err diff --git a/modules/nats/cmd/nats-tools/consumer_test.go b/modules/nats/cmd/nats-tools/consumer_test.go index 858efd80c..c74d14df8 100644 --- a/modules/nats/cmd/nats-tools/consumer_test.go +++ b/modules/nats/cmd/nats-tools/consumer_test.go @@ -123,6 +123,11 @@ func TestTheReadToolsAreServedAndRefuseBadArguments(t *testing.T) { if err != nil || out.(map[string]any)["returned"].(int) != 0 && out.(map[string]any)["returned"].(int) != 1 { t.Fatalf("%v %v", out, err) } + for _, st := range []string{"KV_mesh-vault_signers", "OBJ_mesh-knowledge_sections"} { + if _, err := by["nats_stream_messages"](map[string]any{"stream": st, "subject": ">", "body": true}); err == nil || !strings.Contains(err.Error(), "kept state") { + t.Fatalf("%s was read: %v", st, err) + } + } if _, err := by["nats_stream_messages"](map[string]any{"stream": "EVENTS"}); err == nil { t.Fatal("no subject answered") } -- 2.54.0 From 4fbe6e347f11884f0d573abf4f7c11fa213cf49b Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:20:19 +0100 Subject: [PATCH 4/5] 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") } -- 2.54.0 From a3d141e79351ea9da28d6a71d9b11a89f68d2065 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:21:03 +0100 Subject: [PATCH 5/5] nats_stream_messages shows a body only for the stream EVENTS (issue 1966, review 1) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/nats/cmd/nats-tools/consumer.go | 15 +++++++-------- modules/nats/cmd/nats-tools/consumer_test.go | 17 ++++++++++------- 2 files changed, 17 insertions(+), 15 deletions(-) diff --git a/modules/nats/cmd/nats-tools/consumer.go b/modules/nats/cmd/nats-tools/consumer.go index 93cb8e2ab..ccaa14555 100644 --- a/modules/nats/cmd/nats-tools/consumer.go +++ b/modules/nats/cmd/nats-tools/consumer.go @@ -11,7 +11,6 @@ import ( "encoding/json" "errors" "fmt" - "regexp" "strings" "time" @@ -110,10 +109,10 @@ 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)$`) +// bodyStream is the one stream that shows a body: the shared EVENTS, which holds every event, the seats' kept event streams +// (SEAT__) being sourced from it. Every other stream holds calls and replies, asks, mail, dead letters, records +// or logs, whose bodies are not an event's to show: it answers sequence, subject, time and size only. +const bodyStream = "EVENTS" // MessageRow is one message of the answer. type MessageRow struct { @@ -272,7 +271,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, 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.", + 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 the stream EVENTS (every kept event is there); on every other stream (kept events, 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"}}, @@ -296,8 +295,8 @@ func readTools(jsz func() ([]byte, error), open MsgOpen) []stdio.Tool { 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" + if body && stream != bodyStream { + body, bodyHidden = false, stream+" is not the stream EVENTS: its bodies are not shown, only sequence, subject, time and size" } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() diff --git a/modules/nats/cmd/nats-tools/consumer_test.go b/modules/nats/cmd/nats-tools/consumer_test.go index e17017bb0..b8adf93d5 100644 --- a/modules/nats/cmd/nats-tools/consumer_test.go +++ b/modules/nats/cmd/nats-tools/consumer_test.go @@ -147,14 +147,17 @@ func TestTheReadToolsAreServedAndRefuseBadArguments(t *testing.T) { 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) + seat := strings.Replace(consumerJsz, `"name":"EVENTS"`, `"name":"SEAT_node-engine_running"`, 1) + for _, tl := range readTools(func() ([]byte, error) { return []byte(seat), nil }, + func(context.Context) (MsgGet, func(), error) { + return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil + }) { + if tl.Name != "nats_stream_messages" { + continue } - } - 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) + o, err := tl.Run(map[string]any{"stream": "SEAT_node-engine_running", "subject": "a.b", "body": true}) + if err != nil || o.(map[string]any)["messages"].([]MessageRow)[0].Body != "" { + t.Fatalf("a kept event stream showed a body: %v %v", o, err) } } if _, err := by["nats_stream_messages"](map[string]any{"stream": "EVENTS"}); err == nil { -- 2.54.0