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

@@ -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 <url> --credential <file>`; the archive is also what the nats CLI

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

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

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

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"`

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",