Read memberships from the grants part's state, what the mesh holds from the indexer and placement, and give bundles newest, since and watch on logs, so the tool runner asks the controller for nothing it keeps (issue 1955, ADR 0571 clauses 1, 4, 5, 7)

Pins mesh-sdk bb993dd. The assignments stream stays the membership's
fallback until the lease write; a log another module keeps is read as
<module>.<log>.

node-tools/internal/bus/bus.go

@@ -10,6 +10,7 @@
 package bus
 
 import (
+	"context"
 	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
@@ -17,6 +18,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"log"
 	"regexp"
 	"strings"
@@ -180,9 +182,12 @@ type Conn struct {
 	cred   Credential
 	mu     sync.Mutex
 	issued map[string]*Membership // module → membership; present with nil = followed, none issued
-	onNew  []*func(Membership)
-	subs   []*nats.Subscription
-	Logf   func(format string, args ...any)
+	// fromGrants is every module whose membership the grants part's state memberships has given (novox/hq ADR 0571
+	// clause 7): its assignments stream copy is then a fallback no longer applied.
+	fromGrants map[string]bool
+	onNew      []*func(Membership)
+	subs       []*nats.Subscription
+	Logf       func(format string, args ...any)
 	// answering is every subject this runtime answers on, so a subscription the bus refused can be
 	// asked for again (novox/hq issue 222).
 	answering map[string]*answered
@@ -326,7 +331,7 @@ func Connect(cred Credential) (*Conn, error) {
 		return nil, err
 	}
 	c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred,
-		issued: map[string]*Membership{}, Logf: log.Printf, answering: map[string]*answered{}}
+		issued: map[string]*Membership{}, fromGrants: map[string]bool{}, Logf: log.Printf, answering: map[string]*answered{}}
 	c.Follow(cred.Module)
 	return c, nil
 }
@@ -346,7 +351,10 @@ func MembershipSubject(node, module string) string {
 }
 
 // Follow reads a module's membership on this machine once and follows it live, so its tools are
-// served where the mesh issued them (ADR 0175).
+// served where the mesh issued them (ADR 0175). It is read from the grants part's state memberships, its key
+// `<node>.<module>` (novox/hq ADR 0571 clause 7, mesh-sdk buscompose.GrantsMembershipsBucket); the controller's
+// assignments stream is read beside it as the fallback until the lease write, and applied only while the state has given
+// the module no membership.
 func (c *Conn) Follow(module string) {
 	c.mu.Lock()
 	if c.node == "" {
@@ -359,16 +367,21 @@ func (c *Conn) Follow(module string) {
 	}
 	c.issued[module] = nil
 	c.mu.Unlock()
+	c.followGrants(module)
 	subject := MembershipSubject(c.node, module)
 	// The subject-addressed direct get: the one address the mesh grants this account on the
 	// stream's API.
-	if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil {
-		if got.Header.Get("Status") == "" && len(got.Data) > 0 {
-			var m Membership
-			if json.Unmarshal(got.Data, &m) == nil {
-				c.mu.Lock()
-				c.issued[module] = &m
-				c.mu.Unlock()
+	if c.Membership(module) == nil {
+		if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil {
+			if got.Header.Get("Status") == "" && len(got.Data) > 0 {
+				var m Membership
+				if json.Unmarshal(got.Data, &m) == nil {
+					c.mu.Lock()
+					if !c.fromGrants[module] {
+						c.issued[module] = &m
+					}
+					c.mu.Unlock()
+				}
 			}
 		}
 	}
@@ -381,24 +394,116 @@ func (c *Conn) Follow(module string) {
 			c.Logf("[mesh-tools] a membership arrived that is not one: %v", err)
 			return
 		}
-		c.mu.Lock()
-		c.issued[module] = &m
-		handlers := make([]func(Membership), 0, len(c.onNew))
-		for _, h := range c.onNew {
-			handlers = append(handlers, *h)
-		}
-		c.mu.Unlock()
-		c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node)
-		for _, h := range handlers {
-			h(m)
-		}
-		_ = c.nc.Flush()
+		c.issue(module, m, false)
 	})
 	if err == nil {
 		c.track(sub)
 	}
 }
 
+// GrantsPatience bounds the first read of a membership from the grants part's state.
+var GrantsPatience = 2 * time.Second
+
+// issue applies a membership that arrived for module, live, and re-serves on it. One from the assignments stream is
+// passed over once the grants part's state has given the module one.
+func (c *Conn) issue(module string, m Membership, fromGrants bool) {
+	c.mu.Lock()
+	if !fromGrants && c.fromGrants[module] {
+		c.mu.Unlock()
+		return
+	}
+	if fromGrants {
+		c.fromGrants[module] = true
+	}
+	c.issued[module] = &m
+	handlers := make([]func(Membership), 0, len(c.onNew))
+	for _, h := range c.onNew {
+		handlers = append(handlers, *h)
+	}
+	c.mu.Unlock()
+	c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node)
+	for _, h := range handlers {
+		h(m)
+	}
+	_ = c.nc.Flush()
+}
+
+// followGrants reads module's key of the grants part's state memberships and watches it: the value there now is applied
+// before Follow returns, each change after it live. A state that is not there yet, or not readable, is said once, and
+// the assignments stream answers alone.
+func (c *Conn) followGrants(module string) {
+	key := buscompose.MembershipKey(c.node, module)
+	// First the key's direct get, which the account is granted alone: a bus that refuses it, or a state not there yet,
+	// answers at once, where binding the bucket would wait out its bound on a refusal.
+	bucket := buscompose.GrantsMembershipsBucket
+	probe, err := c.nc.Request("$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+key, nil, GrantsPatience)
+	if err != nil || probe.Header.Get("Status") == "503" {
+		why := "it did not answer"
+		if err != nil {
+			why = err.Error()
+		}
+		c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s's membership is read from the "+
+			"assignments stream alone", bucket, why, module)
+		return
+	}
+	js, err := jetstream.New(c.nc)
+	if err != nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := js.KeyValue(ctx, bucket)
+	if err != nil {
+		c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s's membership is read from the "+
+			"assignments stream alone", buscompose.GrantsMembershipsBucket, err, module)
+		return
+	}
+	w, err := kv.Watch(context.Background(), key)
+	if err != nil {
+		c.Logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); read from the "+
+			"assignments stream alone", module, key, err)
+		return
+	}
+	apply := func(e jetstream.KeyValueEntry) {
+		if e.Operation() != jetstream.KeyValuePut {
+			return // a membership withdrawn is the stream's, or the module's unassignment's, to say
+		}
+		var m Membership
+		if err := json.Unmarshal(e.Value(), &m); err != nil {
+			c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err)
+			return
+		}
+		c.issue(module, m, true)
+	}
+	timer := time.NewTimer(StateTimeout)
+	defer timer.Stop()
+	for current := true; current; {
+		select {
+		case e, ok := <-w.Updates():
+			if !ok {
+				return
+			}
+			if e == nil {
+				current = false
+				continue
+			}
+			apply(e)
+		case <-timer.C:
+			_ = w.Stop()
+			c.Logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; read from the "+
+				"assignments stream alone", module, key, StateTimeout)
+			return
+		}
+	}
+	go func() {
+		for e := range w.Updates() {
+			if e != nil {
+				apply(e)
+			}
+		}
+	}()
+}
+
 // Membership is what the mesh issued a module here, or nil when nothing has been issued.
 func (c *Conn) Membership(module string) *Membership {
 	c.mu.Lock()

node-tools/internal/bus/log.go

@@ -1,6 +1,7 @@
 package bus
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -94,6 +95,10 @@ func logFilter(l LogIssued, pattern string) (string, error) {
 	}
 	parts := strings.Split(pattern, ".")
 	for i, p := range parts {
+		if p == "**" && i == len(parts)-1 {
+			// `**` for the rest, as a state's key pattern and mesh-sdk's log reader say it.
+			parts[i], p = ">", ">"
+		}
 		switch {
 		case p == ">" && i == len(parts)-1, p == "*", logToken.MatchString(p):
 		default:
@@ -104,7 +109,7 @@ func logFilter(l LogIssued, pattern string) (string, error) {
 	if len(pattern) > LogKeyMax {
 		return "", fmt.Errorf("%q is longer than a key can be (%d bytes)", pattern, LogKeyMax)
 	}
-	return l.Subject + "." + pattern, nil
+	return l.Subject + "." + strings.Join(parts, "."), nil
 }
 
 // issuedLog is the log a module may reach by a name, or the reason it may not.
@@ -126,8 +131,9 @@ func (c *Conn) issuedLog(module, name string) (LogIssued, error) {
 	if len(names) > 0 {
 		issued = strings.Join(names, ", ")
 	}
-	return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, "+
-		"and a log is its owner's alone; it was issued: %s (novox/hq ADR 0297)", module, name, issued)
+	return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, and "+
+		"another module's it reads under `reads` as <module>.<log>; it was issued: %s (novox/hq ADR 0297, ADR 0571)",
+		module, name, issued)
 }
 
 // notOnTheBus says a log is issued and its stream is not there yet.
@@ -200,9 +206,10 @@ func (c *Conn) LogAppend(module, name, key string, value json.RawMessage, expect
 	return ack.Seq, nil
 }
 
-// LogRead is the entries from a sequence on, of the keys matching a pattern, at most limit of them and
-// about 4 MiB, by one batch direct get.
-func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) (*LogPage, error) {
+// LogRead is the entries from a sequence on, or from a time on (since, when not zero), of the keys matching a pattern, at
+// most limit of them and about 4 MiB, by one batch direct get. A log another module keeps is read as its own is, by the
+// name its reads declare, `<module>.<log>` (novox/hq ADR 0571).
+func (c *Conn) LogRead(module, name string, from uint64, since time.Time, keys string, limit int) (*LogPage, error) {
 	l, err := c.issuedLog(module, name)
 	if err != nil {
 		return nil, err
@@ -218,52 +225,24 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int)
 	if err != nil {
 		return nil, err
 	}
+	if from != 0 && !since.IsZero() {
+		return nil, fmt.Errorf("a read of %s starts from a sequence or from a time, not both", name)
+	}
 	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)
+	req := map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes}
+	if !since.IsZero() {
+		delete(req, "seq")
+		req["start_time"] = since.UTC()
+	}
+	entries, more, err := c.logBatch(l, req)
 	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))
-		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 := &LogPage{Entries: entries}
+	if more && len(entries) > 0 {
+		page.Next = entries[len(entries)-1].Sequence + 1
 	}
 	info, err := c.logInfo(l)
 	if err != nil {
@@ -273,6 +252,149 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int)
 	return page, nil
 }
 
+// LogNewest is the newest entry of each key matching a pattern, in the log's order, by one batch direct get (mesh-sdk
+// process.LogReader.Newest).
+func (c *Conn) LogNewest(module, name, keys string) ([]LogEntry, error) {
+	l, err := c.issuedLog(module, name)
+	if err != nil {
+		return nil, err
+	}
+	filter, err := logFilter(l, keys)
+	if err != nil {
+		return nil, err
+	}
+	entries, more, err := c.logBatch(l, map[string]any{"multi_last": []string{filter}, "batch": LogReadMax})
+	if err != nil {
+		return nil, err
+	}
+	if more {
+		return nil, fmt.Errorf("%s holds more than %d keys matching %q: name fewer", name, LogReadMax, keys)
+	}
+	return entries, nil
+}
+
+// logBatch is one batch direct get of a log: its entries, and whether more are pending after them.
+func (c *Conn) logBatch(l LogIssued, request map[string]any) ([]LogEntry, bool, error) {
+	req, _ := json.Marshal(request)
+	inbox := c.nc.NewInbox()
+	sub, err := c.nc.SubscribeSync(inbox)
+	if err != nil {
+		return nil, false, err
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil {
+		return nil, false, err
+	}
+	entries := []LogEntry{}
+	deadline := time.Now().Add(StateTimeout)
+	for {
+		msg, err := sub.NextMsg(time.Until(deadline))
+		if err != nil {
+			if errors.Is(err, nats.ErrNoResponders) {
+				return nil, false, notOnTheBus(l)
+			}
+			if errors.Is(err, nats.ErrTimeout) {
+				return nil, false, fmt.Errorf("the read of %s did not end in %s", l.Name, StateTimeout)
+			}
+			return nil, false, err
+		}
+		switch status := msg.Header.Get("Status"); status {
+		case "":
+			e, err := logEntryOf(l, msg)
+			if err != nil {
+				return nil, false, err
+			}
+			entries = append(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)
+			return entries, pending > 0, nil
+		case "404": // nothing from there on
+			return entries, false, nil
+		case "503":
+			return nil, false, notOnTheBus(l)
+		default:
+			return nil, false, fmt.Errorf("the read of %s was refused by the bus: %s %s", l.Name, status, msg.Header.Get("Description"))
+		}
+	}
+}
+
+// LogWatch hands deliver every entry of the keys matching a pattern from a sequence on (from), from a time on (since),
+// or, with neither, each new one, in the log's order, one at a time, through an ordered consumer it lets go of when
+// stopped (mesh-sdk process.LogReader.Watch). It returns once the entries there at its start are handed over; an error
+// from deliver is said, and the watch goes on with the next entry.
+func (c *Conn) LogWatch(module, name, keys string, from uint64, since time.Time, deliver func(LogEntry) error) (func(), error) {
+	l, err := c.issuedLog(module, name)
+	if err != nil {
+		return nil, err
+	}
+	filter, err := logFilter(l, keys)
+	if err != nil {
+		return nil, err
+	}
+	cfg := jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverNewPolicy}
+	switch {
+	case from != 0 && !since.IsZero():
+		return nil, fmt.Errorf("a watch of %s starts from a sequence or from a time, not both", name)
+	case from != 0:
+		cfg.DeliverPolicy, cfg.OptStartSeq = jetstream.DeliverByStartSequencePolicy, from
+	case !since.IsZero():
+		at := since.UTC()
+		cfg.DeliverPolicy, cfg.OptStartTime = jetstream.DeliverByStartTimePolicy, &at
+	}
+	js, err := jetstream.New(c.nc)
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	cons, err := js.OrderedConsumer(ctx, l.Stream, cfg)
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		return nil, notOnTheBus(l)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
+	}
+	info, err := cons.Info(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
+	}
+	it, err := cons.Messages()
+	if err != nil {
+		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
+	}
+	hand := func(m jetstream.Msg) error {
+		meta, err := m.Metadata()
+		if err != nil {
+			return err
+		}
+		e := LogEntry{Sequence: meta.Sequence.Stream, Key: strings.TrimPrefix(m.Subject(), l.Subject+"."),
+			Value: valueOf(m.Data()), At: meta.Timestamp.UTC()}
+		return deliver(e)
+	}
+	for pending := info.NumPending; pending > 0; pending-- {
+		m, err := it.Next()
+		if err == nil {
+			err = hand(m)
+		}
+		if err != nil {
+			it.Stop()
+			return nil, fmt.Errorf("the watch of %s: %w", name, err)
+		}
+	}
+	go func() {
+		for {
+			m, err := it.Next()
+			if err != nil {
+				return // stopped, or the connection closed
+			}
+			if err := hand(m); err != nil {
+				c.Logf("[mesh-tools] %s did not take entry of %s: %v; it is in the log, a read finds it", module, name, err)
+			}
+		}
+	}()
+	return it.Stop, 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)

node-tools/internal/bus/streams_test.go

@@ -160,6 +160,37 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
 		}
 		return nil
 	})
+
+	// The membership as the grants part keeps it (ADR 0571 clause 7): read first, applied live, and the assignments
+	// stream's copy passed over once the state has given one.
+	granted := reader
+	granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}}
+	mesh.IssueInGrants(t, granted)
+	again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	again.Logf = func(string, ...any) {}
+	t.Cleanup(again.Close)
+	again.Follow("reader")
+	consumerOf := func() string {
+		if m := again.Membership("reader"); m != nil && len(m.Hears) == 1 {
+			return m.Hears[0].Consumer
+		}
+		return ""
+	}
+	if got := consumerOf(); got != "from-grants" {
+		t.Fatalf("the membership read at the follow names %q, want the grants part's", got)
+	}
+	mesh.Issue(t, reader)
+	granted.Hears[0].Consumer = "from-grants-2"
+	mesh.IssueInGrants(t, granted)
+	mt.Until(t, func() error {
+		if got := consumerOf(); got != "from-grants-2" {
+			return fmt.Errorf("the membership names %q, want the grants part's change", got)
+		}
+		return nil
+	})
 }
 
 // novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once