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