Keep the builder's history as its own log that readers declare and follow, the memberships as the grants part's state, and refuse a module's own stream, so every module's data is in the bus's predefined forms and readable by its tools (issue 1955, ADR 0571 clauses 1, 4, 5, 7)

2026-10-29 23:28:21 +01:00
parent 196c5b6af2
commit bb993dd606
16 changed files with 695 additions and 51 deletions
@@ -43,6 +43,14 @@ var sendGrants = []string{
 	"mesh.seat.operator-channel.call.tell.novox~mesh-mail~send",
 	"mesh.seat.operator-channel.tool.tell",
 	"mesh.seat.operator-channel.tool.tell.*",
+	// And the builder's log, as buscompose's readLogGrants grants a declared read of it (novox/hq ADR 0571 clause 1).
+	"$JS.API.STREAM.INFO.LOG_mesh-delivery-builder_build-log",
+	"$JS.API.DIRECT.GET.LOG_mesh-delivery-builder_build-log",
+	"$JS.API.DIRECT.GET.LOG_mesh-delivery-builder_build-log.>",
+	"$JS.API.CONSUMER.CREATE.LOG_mesh-delivery-builder_build-log.>",
+	"$JS.API.CONSUMER.MSG.NEXT.LOG_mesh-delivery-builder_build-log.*",
+	"$JS.API.CONSUMER.INFO.LOG_mesh-delivery-builder_build-log.*",
+	"$JS.API.CONSUMER.DELETE.LOG_mesh-delivery-builder_build-log.*",
 }

 // grantedSendProcess is a test bus with JetStream, with two users: novox.mesh-mail.send, permissioned to
@@ -0,0 +1,152 @@
+package process
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// **Another module's log, read on this process's own credential** (novox/hq ADR 0571 clause 1; the lead, 2026-10-29): a
+// log the module's manifest names in `reads`, `<module>.<log>`, read through an ordered consumer this connection creates
+// and lets go of, never a durable one. Newest is the newest entry of each key, Since every entry from a time, and Watch
+// those or every entry, then each new one as it is appended: event-driven, nothing reads the log again on a timer. The
+// grants are buscompose's readLogGrants, given for a declared read of a log alone.
+
+// LogReader is one module's log this connection reads.
+type LogReader struct {
+	Module, Name string
+	js           func() (jetstream.JetStream, error)
+	closed       func() bool
+}
+
+// ReadLog is module's log name, read on this connection.
+func (c *Conn) ReadLog(module, name string) LogReader {
+	return LogReader{Module: module, Name: name, closed: func() bool { return c.nc.IsClosed() },
+		js: func() (jetstream.JetStream, error) { return jetstream.New(c.nc) }}
+}
+
+func (l LogReader) stream() string  { return "LOG_" + l.Module + "_" + l.Name }
+func (l LogReader) subject() string { return "mesh.log." + l.Module + "." + l.Name }
+
+// filter is keys as a subject filter: a key, `*` for one name, `**` for the rest; empty is every key.
+func (l LogReader) filter(keys string) string {
+	if keys == "" {
+		return l.subject() + ".>"
+	}
+	return l.subject() + "." + watchPattern(keys)
+}
+
+func (l LogReader) config(keys string, from stdio.LogFrom) (jetstream.OrderedConsumerConfig, error) {
+	cfg := jetstream.OrderedConsumerConfig{FilterSubjects: []string{l.filter(keys)}}
+	switch from.Kind {
+	case stdio.LogFromNewestKind:
+		cfg.DeliverPolicy = jetstream.DeliverLastPerSubjectPolicy
+	case stdio.LogFromAllKind:
+		cfg.DeliverPolicy = jetstream.DeliverAllPolicy
+	case stdio.LogFromSinceKind:
+		at := from.Since
+		cfg.DeliverPolicy, cfg.OptStartTime = jetstream.DeliverByStartTimePolicy, &at
+	case stdio.LogFromNowKind:
+		cfg.DeliverPolicy = jetstream.DeliverNewPolicy
+	default:
+		return cfg, fmt.Errorf("%s's log %s: a read starts from newest, all, since or now, not %q", l.Module, l.Name, from.Kind)
+	}
+	return cfg, nil
+}
+
+func (l LogReader) entry(m jetstream.Msg) (stdio.LogEntry, error) {
+	meta, err := m.Metadata()
+	if err != nil {
+		return stdio.LogEntry{}, err
+	}
+	return stdio.LogEntry{Sequence: meta.Sequence.Stream, Key: strings.TrimPrefix(m.Subject(), l.subject()+"."),
+		Value: append([]byte(nil), m.Data()...), At: meta.Timestamp}, nil
+}
+
+// Newest is the newest entry of each key keys names, in the log's order.
+func (l LogReader) Newest(keys string) ([]stdio.LogEntry, error) {
+	return l.collect(keys, stdio.LogFromNewest)
+}
+
+// Since is every entry of keys from at on, in the log's order.
+func (l LogReader) Since(at time.Time, keys string) ([]stdio.LogEntry, error) {
+	return l.collect(keys, stdio.LogFromSince(at))
+}
+
+func (l LogReader) collect(keys string, from stdio.LogFrom) ([]stdio.LogEntry, error) {
+	var out []stdio.LogEntry
+	stop, err := l.watch(keys, from, func(e stdio.LogEntry) error { out = append(out, e); return nil }, false)
+	if stop != nil {
+		stop()
+	}
+	return out, err
+}
+
+// Watch hands handle each entry from where from says and then every new one, in the log's order, and returns once the
+// entries there at its start have been handled. The watch lasts as long as the connection; an error from handle ends
+// it, as the connection closing does. Before the entries at its start are handled, either is returned as the error.
+func (l LogReader) Watch(keys string, from stdio.LogFrom, handle func(stdio.LogEntry) error) error {
+	_, err := l.watch(keys, from, handle, true)
+	return err
+}
+
+func (l LogReader) watch(keys string, from stdio.LogFrom, handle func(stdio.LogEntry) error, follow bool) (func(), error) {
+	cfg, err := l.config(keys, from)
+	if err != nil {
+		return nil, err
+	}
+	js, err := l.js()
+	if err != nil {
+		return nil, err
+	}
+	bind, stopBind := context.WithTimeout(context.Background(), stateTimeout)
+	defer stopBind()
+	cons, err := js.OrderedConsumer(bind, l.stream(), cfg)
+	if err != nil {
+		return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err)
+	}
+	info, err := cons.Info(bind)
+	if err != nil {
+		return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err)
+	}
+	pending := info.NumPending
+	it, err := cons.Messages()
+	if err != nil {
+		return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err)
+	}
+	for ; pending > 0; pending-- {
+		m, err := it.Next()
+		if err == nil {
+			var e stdio.LogEntry
+			if e, err = l.entry(m); err == nil {
+				err = handle(e)
+			}
+		}
+		if err != nil {
+			it.Stop()
+			return nil, fmt.Errorf("%s's log %s: %w", l.Module, l.Name, err)
+		}
+	}
+	if !follow {
+		return it.Stop, nil
+	}
+	go func() {
+		defer it.Stop()
+		for {
+			m, err := it.Next()
+			if err != nil || (l.closed != nil && l.closed()) {
+				return
+			}
+			e, err := l.entry(m)
+			if err != nil || handle(e) != nil {
+				return
+			}
+		}
+	}()
+	return it.Stop, nil
+}
@@ -274,6 +274,44 @@ func TestStateWatchOnAGrantedBucket(t *testing.T) {
 		t.Fatalf("narrowed watch got %+v for a key it did not name", c)
 	case <-time.After(500 * time.Millisecond):
 	}
+
+	// ADR 0571 clause 1: another module's log, read on the process's own credential with exactly the read grants: the
+	// newest per key, every entry from a time, and a watch from the newest that then hears an appended entry.
+	if _, err := adminJetStream.CreateStream(ctx, jetstream.StreamConfig{Name: "LOG_mesh-delivery-builder_build-log",
+		Subjects: []string{"mesh.log.mesh-delivery-builder.build-log.>"}, AllowDirect: true}); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{"letta", "gitea", "letta"} {
+		if _, err := adminJetStream.Publish(ctx, "mesh.log.mesh-delivery-builder.build-log."+k, []byte(`{"module":"`+k+`"}`)); err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	builds := conn.ReadLog("mesh-delivery-builder", "build-log")
+	newest, err := builds.Newest("")
+	if err != nil || len(newest) != 2 || newest[0].Key != "gitea" || newest[1].Sequence != 3 {
+		t.Fatalf("the newest per key are %+v: %v", newest, err)
+	}
+	since, err := builds.Since(newest[0].At, "letta")
+	if err != nil || len(since) != 1 || since[0].Sequence != 3 {
+		t.Fatalf("letta since gitea's build is %+v: %v", since, err)
+	}
+	heard := make(chan stdio.LogEntry, 4)
+	if err := builds.Watch("letta", stdio.LogFromNewest, func(e stdio.LogEntry) error { heard <- e; return nil }); err != nil || len(heard) != 1 {
+		t.Fatalf("a watch from the newest handed %d at its start: %v", len(heard), err)
+	}
+	<-heard
+	if _, err := adminJetStream.Publish(ctx, "mesh.log.mesh-delivery-builder.build-log.letta", []byte(`{}`)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case e := <-heard:
+		if e.Sequence != 4 || e.Key != "letta" {
+			t.Fatalf("the watch heard %+v", e)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("the watch did not hear the appended entry")
+	}
 }

 // 1840/AC8: given an account not granted a bucket, when it watches it, then the refusal names the bucket.