diff --git a/node-tools/internal/bus/log.go b/node-tools/internal/bus/log.go index 6ac1e0e..6f680b6 100644 --- a/node-tools/internal/bus/log.go +++ b/node-tools/internal/bus/log.go @@ -221,49 +221,17 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) if from == 0 { from = 1 } - req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes}) - inbox := c.nc.NewInbox() - sub, err := c.nc.SubscribeSync(inbox) + msgs, next, err := c.directBatch(l.Stream, filter, from, limit, name, func() error { return notOnTheBus(l) }) if err != nil { return nil, err } - defer func() { _ = sub.Unsubscribe() }() - if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil { - return nil, err - } - page := &LogPage{Entries: []LogEntry{}} - deadline := time.Now().Add(StateTimeout) - for done := false; !done; { - msg, err := sub.NextMsg(time.Until(deadline)) + page := &LogPage{Entries: []LogEntry{}, Next: next} + for _, msg := range msgs { + e, err := logEntryOf(l, msg) if err != nil { - if errors.Is(err, nats.ErrNoResponders) { - return nil, notOnTheBus(l) - } - if errors.Is(err, nats.ErrTimeout) { - return nil, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout) - } return nil, err } - switch status := msg.Header.Get("Status"); status { - case "": - e, err := logEntryOf(l, msg) - if err != nil { - return nil, err - } - page.Entries = append(page.Entries, *e) - case "204": // the end of the batch: what is still pending after the last entry sent - pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64) - if pending > 0 && len(page.Entries) > 0 { - page.Next = page.Entries[len(page.Entries)-1].Sequence + 1 - } - done = true - case "404": // nothing from that sequence on - done = true - case "503": - return nil, notOnTheBus(l) - default: - return nil, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description")) - } + page.Entries = append(page.Entries, *e) } info, err := c.logInfo(l) if err != nil { @@ -273,6 +241,58 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) return page, nil } +// directBatch is the messages of a stream from a sequence on, on the subjects matching filter, at most limit and about +// LogPageBytes, by one batch direct get (NATS 2.11), and the sequence to read on from, 0 when nothing follows. name is +// what is read, for its errors; missing is the error of a stream that is not on the bus. +func (c *Conn) directBatch(stream, filter string, from uint64, limit int, name string, missing func() error) ([]*nats.Msg, uint64, error) { + if from == 0 { + from = 1 + } + req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes}) + inbox := c.nc.NewInbox() + sub, err := c.nc.SubscribeSync(inbox) + if err != nil { + return nil, 0, err + } + defer func() { _ = sub.Unsubscribe() }() + if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+stream, inbox, req); err != nil { + return nil, 0, err + } + var msgs []*nats.Msg + var next uint64 + deadline := time.Now().Add(StateTimeout) + for done := false; !done; { + msg, err := sub.NextMsg(time.Until(deadline)) + if err != nil { + if errors.Is(err, nats.ErrNoResponders) { + return nil, 0, missing() + } + if errors.Is(err, nats.ErrTimeout) { + return nil, 0, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout) + } + return nil, 0, err + } + switch status := msg.Header.Get("Status"); status { + case "": + msgs = append(msgs, msg) + case "204": // the end of the batch: what is still pending after the last message sent + pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64) + if pending > 0 && len(msgs) > 0 { + last, _ := strconv.ParseUint(msgs[len(msgs)-1].Header.Get("Nats-Sequence"), 10, 64) + next = last + 1 + } + done = true + case "404": // nothing from that sequence on + done = true + case "503": + return nil, 0, missing() + default: + return nil, 0, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description")) + } + } + return msgs, next, nil +} + // LogLast is a key's last entry, or nil when it has none. func (c *Conn) LogLast(module, name, key string) (*LogEntry, error) { l, err := c.issuedLog(module, name) diff --git a/node-tools/internal/bus/streamread.go b/node-tools/internal/bus/streamread.go new file mode 100644 index 0000000..720d2d6 --- /dev/null +++ b/node-tools/internal/bus/streamread.go @@ -0,0 +1,126 @@ +package bus + +import ( + "encoding/json" + "fmt" + "sort" + "strconv" + "strings" + "time" + + "github.com/nats-io/nats.go" +) + +// A module reading its own stream back (novox/hq ADR 0363; the operator, 2026-10-30: a build step's whole failure output +// is said on its build module's stream, and the module's tools read it there). As a log's read: one batch direct get, +// no consumer, and only a stream the module's own membership names (`streams`); the bus grants its owner that stream's +// info and direct get and no other's (mesh-sdk buscompose ownStreamGrants). + +// StreamEntry is one event of a module's own stream: its sequence, its event (the subject after the stream's +// `…event.`), its body and when the bus took it. +type StreamEntry struct { + Sequence uint64 `json:"sequence"` + Event string `json:"event"` + Body json.RawMessage `json:"body"` + At time.Time `json:"at"` +} + +// StreamPage is one page of a stream's read: Next is the sequence to continue from, 0 when nothing follows; Last the +// stream's last sequence. +type StreamPage struct { + Entries []StreamEntry `json:"entries"` + Next uint64 `json:"next"` + Last uint64 `json:"last"` +} + +// streamBusName is a stream's local name as the bus holds it (mesh-sdk catalogue.StreamBusName): capitals, a hyphen an +// underscore; a bus name is itself. +func streamBusName(name string) string { return strings.ToUpper(strings.ReplaceAll(name, "-", "_")) } + +// issuedStream is the stream of its own a module may read by a name, local or the bus's, or the reason it may not. +func (c *Conn) issuedStream(module, name string) (StreamIssued, error) { + m := c.Membership(module) + if m == nil { + return StreamIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no stream of it is readable "+ + "until the mesh issues one (novox/hq ADR 0363)", module, c.node) + } + var names []string + for _, s := range m.Streams { + if s.Stream == streamBusName(name) && strings.Contains(s.Subject, "{event}") { + return s, nil + } + names = append(names, s.Stream) + } + sort.Strings(names) + issued := "none" + if len(names) > 0 { + issued = strings.Join(names, ", ") + } + return StreamIssued{}, fmt.Errorf("%s keeps no stream called %q: a module reads the streams it declares under `streams`, "+ + "and no other's; it was issued: %s (novox/hq ADR 0363)", module, name, issued) +} + +// StreamRead is a module's own stream's events from a sequence on, whose event names match a pattern (`*` one name, `>` +// the rest, last; empty is every event), at most limit (LogReadDefault when 0, at most LogReadMax) and about LogPageBytes. +func (c *Conn) StreamRead(module, name, events string, from uint64, limit int) (*StreamPage, error) { + s, err := c.issuedStream(module, name) + if err != nil { + return nil, err + } + switch { + case limit == 0: + limit = LogReadDefault + case limit < 0 || limit > LogReadMax: + return nil, fmt.Errorf("a read's limit is 1 to %d events, %d when none is given, not %d", LogReadMax, LogReadDefault, limit) + } + if events == "" { + events = ">" + } + parts := strings.Split(events, ".") + for i, p := range parts { + if !(p == ">" && i == len(parts)-1) && p != "*" && !logToken.MatchString(p) { + return nil, fmt.Errorf("%q is not an event pattern: names, `*` for one and `>` for the rest, last", events) + } + } + prefix, _, _ := strings.Cut(s.Subject, "{event}") + missing := func() error { + return fmt.Errorf("the stream %s is issued and not on the bus yet: the controller creates it from the catalogue "+ + "on its next raise, and reads it directly once it allows that (novox/hq ADR 0363)", s.Stream) + } + msgs, next, err := c.directBatch(s.Stream, prefix+events, from, limit, s.Stream, missing) + if err != nil { + return nil, err + } + page := &StreamPage{Entries: []StreamEntry{}, Next: next} + for _, msg := range msgs { + e, err := streamEntryOf(s.Stream, prefix, msg) + if err != nil { + return nil, err + } + page.Entries = append(page.Entries, *e) + } + info, err := c.logInfo(LogIssued{Name: s.Stream, Stream: s.Stream}) + if err != nil { + return nil, err + } + page.Last = info.Last + return page, nil +} + +// streamEntryOf is an event a direct get answered: its event is its subject after prefix. +func streamEntryOf(stream, prefix string, msg *nats.Msg) (*StreamEntry, error) { + subject := msg.Header.Get("Nats-Subject") + event, ok := strings.CutPrefix(subject, prefix) + if !ok { + return nil, fmt.Errorf("the bus answered an event of %s on %q, which is not under %s", stream, subject, prefix) + } + seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64) + if err != nil { + return nil, fmt.Errorf("the bus answered an event of %s with no sequence", stream) + } + at, err := time.Parse(time.RFC3339Nano, msg.Header.Get("Nats-Time-Stamp")) + if err != nil { + return nil, fmt.Errorf("the bus answered event %d of %s with no time: %v", seq, stream, err) + } + return &StreamEntry{Sequence: seq, Event: event, Body: valueOf(msg.Data), At: at.UTC()}, nil +} diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index 05d8594..1184b99 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -38,6 +38,7 @@ func (b *subscribing) Publish(json.RawMessage) error { func (b *subscribing) Ask(json.RawMessage) (json.RawMessage, error) { return nil, nil } func (b *subscribing) State(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } func (b *subscribing) Log(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } +func (b *subscribing) StreamRead(json.RawMessage) (json.RawMessage, error) { return nil, nil } func (b *subscribing) Watch(json.RawMessage, func(json.RawMessage) error) (func(), error) { return func() {}, nil } diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index fecb1c2..a944e6e 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -80,6 +80,9 @@ type Bus interface { Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) // Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297). Log(verb string, params json.RawMessage) (json.RawMessage, error) + // StreamRead answers a bundle's `read` of one of its module's own streams (novox/hq ADR 0363): {stream, subject, + // from, limit}, as a log's read. + StreamRead(params json.RawMessage) (json.RawMessage, error) // Seat answers a bundle's seat traffic (novox/hq ADR 0259 §3): `publish`, `prove` and `record` at once; // `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child // of the module asked last — a restarted child asks again as its code runs again. @@ -333,6 +336,11 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil { result = answered } + case "mesh/stream.read": + var answered json.RawMessage + if answered, err = mesh.StreamRead(m.Params); err == nil { + result = answered + } case "mesh/provision.get": var answered json.RawMessage if answered, err = mesh.Provision("get", m.Params); err == nil { diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 9edec5e..a99b81e 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -278,7 +278,7 @@ func (m *Mesh) ModuleStream(t *testing.T, module, name string) { t.Helper() _ = m.js.DeleteStream(name) if _, err := m.js.AddStream(&nats.StreamConfig{Name: name, Subjects: []string{"mesh.stream." + module + ".event.>"}, - Retention: nats.LimitsPolicy, Discard: nats.DiscardOld}); err != nil { + Retention: nats.LimitsPolicy, Discard: nats.DiscardOld, AllowDirect: true}); err != nil { t.Fatal(err) } } diff --git a/node-tools/internal/runtime/log_test.go b/node-tools/internal/runtime/log_test.go index f664783..3495c65 100644 --- a/node-tools/internal/runtime/log_test.go +++ b/node-tools/internal/runtime/log_test.go @@ -21,6 +21,8 @@ func logBuses(t *testing.T, maxBytes int64) (issues, other launch.Bus, mesh *mt. mesh = mt.New(t) m := mt.MembershipOf("issues", "anchor", false, nil) m.Logs = []bus.LogIssued{mesh.Log(t, "issues", "changes", maxBytes)} + mesh.ModuleStream(t, "issues", "ISSUE_MOVES") + m.Streams = []bus.StreamIssued{{Stream: "ISSUE_MOVES", Events: []string{"moved.>"}, Subject: "mesh.stream.issues.event.{event}"}} mesh.Issue(t, m) mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil)) conn := connect(t, "node-tools", "anchor") @@ -111,7 +113,7 @@ func infoOf(t *testing.T, b launch.Bus) map[string]uint64 { // RFC 3339; last is a key's last entry or null; info says entries, bytes, cap, first and last — and no // consumer is made for any of it. func TestALogAnswersEachVerbOnTheBus(t *testing.T) { - b, _, mesh := logBuses(t, 1<<20) + b, other, mesh := logBuses(t, 1<<20) before := time.Now().Add(-time.Second) var seqs []uint64 @@ -189,6 +191,45 @@ func TestALogAnswersEachVerbOnTheBus(t *testing.T) { if n := mesh.Consumers(t, "LOG_issues_changes"); n != 0 { t.Fatalf("the runtime made %d consumers of the log; it uses none", n) } + + // A module reads its own stream back as it reads its log (mesh/stream.read): by its local name, an event pattern, + // in pages; another module's stream is refused with the reason. + for i, ev := range []string{"moved.a.1", "moved.b.1", "moved.a.1"} { + if err := b.Publish(json.RawMessage(fmt.Sprintf(`{"key":%q,"body":{"n":%d}}`, ev, i))); err != nil { + t.Fatal(err) + } + } + streamOf := func(b launch.Bus, params string) (wireStream, error) { + var p wireStream + raw, err := b.StreamRead(json.RawMessage(params)) + if err == nil { + err = json.Unmarshal(raw, &p) + } + return p, err + } + page, err := streamOf(b, `{"stream":"issue-moves","subject":"moved.a.>","limit":1}`) + if err != nil || len(page.Entries) != 1 || page.Entries[0].Event != "moved.a.1" || string(page.Entries[0].Body) != `{"n":0}` || page.Next == 0 || page.Last != 3 { + t.Fatalf("the first page of moved.a: %+v, %v", page, err) + } + page, err = streamOf(b, fmt.Sprintf(`{"stream":"ISSUE_MOVES","subject":"moved.a.>","from":%d}`, page.Next)) + if err != nil || len(page.Entries) != 1 || string(page.Entries[0].Body) != `{"n":2}` || page.Next != 0 { + t.Fatalf("the second page of moved.a: %+v, %v", page, err) + } + if _, err := streamOf(other, `{"stream":"issue-moves"}`); err == nil || !strings.Contains(err.Error(), "keeps no stream") { + t.Fatalf("another module's stream: %v", err) + } +} + +// wireStream is a stream read's answer with exactly the contract's field names (mesh-sdk stdio.StreamPage). +type wireStream struct { + Entries []struct { + Sequence uint64 `json:"sequence"` + Event string `json:"event"` + Body json.RawMessage `json:"body"` + At string `json:"at"` + } `json:"entries"` + Next uint64 `json:"next"` + Last uint64 `json:"last"` } // A page holds about 4 MiB at most, whatever its limit, and says where to continue. diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 5cd47f9..4a1fdcf 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -988,6 +988,26 @@ func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, e return json.Marshal(answer) } +// StreamRead answers a bundle's read of one of its module's own streams (ADR 0363): {stream, subject, from, limit}, the +// stream by its local name, the subject an event pattern. The runtime refuses, with the reason, a stream the module's +// membership does not name. +func (b *moduleBus) StreamRead(params json.RawMessage) (json.RawMessage, error) { + var asked struct { + Stream string `json:"stream"` + Subject string `json:"subject"` + From uint64 `json:"from"` + Limit int `json:"limit"` + } + if err := json.Unmarshal(params, &asked); err != nil || asked.Stream == "" { + return nil, fmt.Errorf("mesh/stream.read names no stream: {stream, subject, from, limit}") + } + page, err := b.all.conn.StreamRead(b.module, asked.Stream, asked.Subject, asked.From, asked.Limit) + if err != nil { + return nil, err + } + return json.Marshal(page) +} + // Watch hands a bundle its module's state as it is and as it changes (ADR 0201): `{state, key}`, the // key a pattern with `*` and `**`, empty for every key. func (b *moduleBus) Watch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {