novox/mesh-tools
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

The tool runner reads its membership from the grants part's state alone: no stream read, no subject heard (ADR 0571 §10)

Browse Source 
Follow no longer asks the ASSIGNMENTS stream for a copy nor subscribes mesh.assignment; a module with no membership key is
said so and served in its derived shape. The test mesh issues memberships where the grants part keeps them.
Tests: none added; the stream-fallback assertions of two existing tests are replaced by their state-only counterparts, and
the test mesh no longer makes the stream. The two integration tests of the bus were run locally against the embedded bus.

(issue 774, 774/S32)
Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
jschoubben
2026-11-02 00:25:19 +01:00
parent 75943ca7b0
commit 4fbfdfad13
4 changed files with 56 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node-tools/internal/bus/bus.go
+19 -66
View File
@@ -37,8 +37,6 @@ import (
// verb still runs, so a caller is never left guessing past this (novox/hq issue 265).
const RequestTimeout = 30 * time.Second
const assignmentsStream = "ASSIGNMENTS"
// Credential is a broker credential as the mesh delivers it (novox/hq ADR 0120).
type Credential struct {
URL string `json:"url"`
@@ -183,12 +181,9 @@ type Conn struct {
cred Credential
mu sync.Mutex
issued map[string]*Membership // module → membership; present with nil = followed, none issued
// 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)
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
@@ -332,7 +327,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{}, fromGrants: map[string]bool{}, Logf: log.Printf, answering: map[string]*answered{}}
issued: map[string]*Membership{}, Logf: log.Printf, answering: map[string]*answered{}}
c.Follow(cred.Module)
return c, nil
}
@@ -346,16 +341,10 @@ func (c *Conn) Node() string { return c.node }
// Credential is what this connection was opened with.
func (c *Conn) Credential() Credential { return c.cred }
// MembershipSubject is the one address a runtime derives for an assignment (ADR 0160).
func MembershipSubject(node, module string) string {
return "mesh.assignment." + node + "." + module
}
// 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). 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.
// `<node>.<module>` (novox/hq ADR 0571 clause 7 and 10, mesh-sdk buscompose.GrantsMembershipsBucket), and nowhere else:
// the controller's assignments stream is not read and its subject is not heard.
func (c *Conn) Follow(module string) {
c.mu.Lock()
if c.node == "" {
@@ -369,53 +358,17 @@ 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 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()
}
}
}
}
if c.Membership(module) == nil {
c.Logf("[mesh-tools] no membership issued for %s on %s yet; serving the derived shape until one arrives", module, c.node)
}
sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
var m Membership
if err := json.Unmarshal(msg.Data, &m); err != nil {
c.Logf("[mesh-tools] a membership arrived that is not one: %v", err)
return
}
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) {
// issue applies a membership that arrived for module, live, and re-serves on it.
func (c *Conn) issue(module string, m Membership) {
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 {
@@ -431,7 +384,7 @@ func (c *Conn) issue(module string, m Membership, fromGrants bool) {
// 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.
// the module has no membership until it can be read.
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,
@@ -443,8 +396,8 @@ func (c *Conn) followGrants(module string) {
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)
c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s has no membership "+
"until it can be", bucket, why, module)
return
}
js, err := jetstream.New(c.nc)
@@ -455,14 +408,14 @@ func (c *Conn) followGrants(module string) {
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)
c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s has no membership "+
"until it can be", 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)
c.Logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); %s has no membership "+
"until it can be", module, key, err, module)
return
}
apply := func(e jetstream.KeyValueEntry) {
@@ -471,7 +424,7 @@ func (c *Conn) followGrants(module string) {
}
// Read through mesh-sdk's one codec of this state, the grants part's writer's (issue 774): it takes the bare
// membership and a wrapper the part wrote before of one published or unchanged, and refuses a held-back one's
// wrapper or a value naming no node or module, which then leaves the assignments stream's copy in force instead.
// wrapper or a value naming no node or module, which then leaves the module with no membership.
// A refused value arriving after a good one keeps the last good membership: nothing here clears it.
var m Membership
body, err := membership.BareGrantsMembership(e.Value())
@@ -482,7 +435,7 @@ func (c *Conn) followGrants(module string) {
c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err)
return
}
c.issue(module, m, true)
c.issue(module, m)
}
timer := time.NewTimer(StateTimeout)
defer timer.Stop()
@@ -499,8 +452,8 @@ func (c *Conn) followGrants(module string) {
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)
c.Logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; %s has no membership "+
"until it can be", module, key, StateTimeout, module)
return
}
}
node-tools/internal/bus/state_watch_test.go
+9 -15
View File
@@ -119,25 +119,24 @@ func TestAWatchIsRefusedWhenNoCurrentValueIsTakenWithinTheBound(t *testing.T) {
// **A membership the grants part wrote in its old wrapper is served whole** (issue 774). Until 2026-10-30 the part wrote
// its pass record, the membership inside, into its state memberships; the tool runner read the value as a bare
// membership, took it over the assignments stream's copy, and served no tool and no state. This is what the live mesh
// held: read through mesh-sdk's codec, its tools are answered where the membership says and its state is reached.
// membership, and served no tool and no state. This is what the live mesh held: read through mesh-sdk's codec, its tools
// are answered where the membership says and its state is reached. A held-back one's wrapper is refused, and the module has
// no membership then (774/S32/AC2: no stream is read in its place).
func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
mesh := mt.New(t)
mesh.Bucket(t, "gitea_repos")
gitea := mt.MembershipOf("gitea", "anchor", false, nil)
gitea.State = []bus.StateIssued{{Name: "repos", Bucket: "gitea_repos", Writes: true}}
mesh.IssueInGrantsWrapped(t, gitea, "published")
// The old wrapper of a membership the part held back carries its unpublished composition: refused, so the
// assignments stream's copy is served, never what the hold kept back.
pages := mt.MembershipOf("pages", "anchor", false, nil)
mesh.Issue(t, pages)
// The old wrapper of a membership the part held back carries its unpublished composition: refused, never served.
heldBack := mt.MembershipOf("pages", "anchor", true, nil)
mesh.IssueInGrantsWrapped(t, heldBack, "not published, a shadow difference: publishing would lose the last holder of x")
conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
if err != nil {
t.Fatal(err)
}
conn.Logf = func(string, ...any) {}
var logged []string
conn.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
t.Cleanup(conn.Close)
conn.Follow("gitea")
mt.Until(t, func() error {
@@ -156,14 +155,9 @@ func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
t.Errorf("the state reads back %+v, %v", e, err)
}
conn.Follow("pages")
mt.Until(t, func() error {
if m := conn.Membership("pages"); m == nil || len(m.Serves) == 0 {
return fmt.Errorf("pages' membership is %+v, want the assignments stream's", m)
}
return nil
})
if got, want := fmt.Sprint(conn.ServedOn("pages", "pages_list")), "[{mesh.mod.pages.tool.pages_list.anchor }]"; got != want {
t.Errorf("a held-back membership's wrapper is served: pages_list on %s, want the stream's %s", got, want)
time.Sleep(300 * time.Millisecond)
if m := conn.Membership("pages"); m != nil || !strings.Contains(strings.Join(logged, "\n"), "no membership issued for pages") {
t.Errorf("a held-back membership's wrapper was served or not said: %+v, %q", m, logged)
}
}
node-tools/internal/bus/streams_test.go
+8 -32
View File
@@ -161,11 +161,11 @@ 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.
// The membership as the grants part keeps it (ADR 0571 clause 7, 10): read first and applied live, and nothing
// published on the controller's subject is heard.
granted := reader
granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}}
mesh.IssueInGrants(t, granted)
mesh.Issue(t, granted)
again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
if err != nil {
t.Fatal(err)
@@ -183,45 +183,21 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
t.Fatalf("the membership read at the follow names %q, want the grants part's", got)
}
granted.Hears[0].Consumer = "from-grants-2"
mesh.IssueInGrants(t, granted)
mesh.Issue(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
})
// A later, different copy on the assignments stream changes nothing once the state has given one.
arrived := make(chan struct{}, 1)
probe, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "probe", Node: "elsewhere"})
if err != nil {
t.Fatal(err)
}
probe.Logf = func(string, ...any) {}
t.Cleanup(probe.Close)
stopProbe, err := probe.Raw(bus.MembershipSubject("anchor", "reader"), func(string, []byte) []byte {
select {
case arrived <- struct{}{}:
default:
}
return nil
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(stopProbe)
probe.Flush()
// 774/S32/AC1: a copy said on the controller's subject (mesh.assignment) changes nothing; there is no stream to read it from.
streamed := reader
streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-stream"}}
mesh.Issue(t, streamed)
select {
case <-arrived:
case <-time.After(5 * time.Second):
t.Fatal("the stream's copy was not published")
}
streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-subject"}}
mesh.Publish(t, "mesh.assignment.anchor.reader", streamed)
again.Flush()
time.Sleep(200 * time.Millisecond)
if got := consumerOf(); got != "from-grants-2" {
t.Fatalf("the stream's copy was applied over the grants part's: the membership names %q", got)
t.Fatalf("a membership said on the controller's subject was applied: the membership names %q", got)
}
}
node-tools/internal/meshtest/meshtest.go
+20 -18
View File
@@ -86,11 +86,10 @@ func New(t *testing.T) *Mesh {
t.Fatal(err)
}
js, _ := nc.JetStream()
for _, s := range []string{"ASSIGNMENTS", "EVENTS"} {
_ = js.DeleteStream(s)
}
if _, err := js.AddStream(&nats.StreamConfig{Name: "ASSIGNMENTS", Subjects: []string{"mesh.assignment.>"},
MaxMsgsPerSubject: 1, AllowDirect: true}); err != nil {
_ = js.DeleteStream("EVENTS")
// The grants part's state memberships, where a runtime reads what is issued it (ADR 0571 clause 10).
_ = js.DeleteKeyValue(buscompose.GrantsMembershipsBucket)
if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil {
t.Fatal(err)
}
if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
@@ -100,20 +99,11 @@ func New(t *testing.T) *Mesh {
return &Mesh{nc: nc, js: js}
}
// Issue publishes a membership.
// Issue writes a membership where the grants part keeps it, its state memberships, key `<node>.<module>` (novox/hq ADR
// 0571 clause 7 and 10; mesh-sdk buscompose.GrantsMembershipsBucket), the one place a runtime reads it. It writes with
// mesh-sdk's encoder, the grants part's own, never json of this package's type: a value built here by hand agreed with
// the reader's mistake and hid that the part wrote another shape (issue 774).
func (m *Mesh) Issue(t *testing.T, mem bus.Membership) {
t.Helper()
body, _ := json.Marshal(mem)
if _, err := m.js.Publish(bus.MembershipSubject(mem.Node, mem.Module), body); err != nil {
t.Fatal(err)
}
}
// IssueInGrants writes a membership where the grants part keeps it, its state memberships, key `<node>.<module>`
// (novox/hq ADR 0571 clause 7; mesh-sdk buscompose.GrantsMembershipsBucket), making the bucket when it is not there.
// It writes with mesh-sdk's encoder, the grants part's own, never json of this package's type: a value built here by
// hand agreed with the reader's mistake and hid that the part wrote another shape (issue 774).
func (m *Mesh) IssueInGrants(t *testing.T, mem bus.Membership) {
t.Helper()
m.putGrants(t, mem, membership.EncodeGrantsMembership(sdkMembership(t, mem)))
}
@@ -319,6 +309,18 @@ func (m *Mesh) Emit(t *testing.T, subject string, body any) {
}
}
// Publish says body on a plain subject, as a core publish with no stream behind it.
func (m *Mesh) Publish(t *testing.T, subject string, body any) {
t.Helper()
data, _ := json.Marshal(body)
if err := m.nc.Publish(subject, data); err != nil {
t.Fatal(err)
}
if err := m.nc.Flush(); err != nil {
t.Fatal(err)
}
}
// Pending is what a module's consumer still holds: delivered and not acknowledged, and not yet
// delivered.
func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelivered uint64) {