From 4fbfdfad136a8f3190450e53c3b314748d1e5392 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 00:25:19 +0100 Subject: [PATCH 1/4] =?UTF-8?q?The=20tool=20runner=20reads=20its=20members?= =?UTF-8?q?hip=20from=20the=20grants=20part's=20state=20alone:=20no=20stre?= =?UTF-8?q?am=20read,=20no=20subject=20heard=20(ADR=200571=20=C2=A710)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- node-tools/internal/bus/bus.go | 85 +++++---------------- node-tools/internal/bus/state_watch_test.go | 24 +++--- node-tools/internal/bus/streams_test.go | 40 ++-------- node-tools/internal/meshtest/meshtest.go | 38 ++++----- 4 files changed, 56 insertions(+), 131 deletions(-) diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ee01017..ad508f0 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -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 -// `.` (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. +// `.` (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 } } diff --git a/node-tools/internal/bus/state_watch_test.go b/node-tools/internal/bus/state_watch_test.go index 81dbc43..3743d0e 100644 --- a/node-tools/internal/bus/state_watch_test.go +++ b/node-tools/internal/bus/state_watch_test.go @@ -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) } } diff --git a/node-tools/internal/bus/streams_test.go b/node-tools/internal/bus/streams_test.go index b56b568..87deff4 100644 --- a/node-tools/internal/bus/streams_test.go +++ b/node-tools/internal/bus/streams_test.go @@ -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) } } diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index dfbecd4..a058c77 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -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 `.` (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 `.` -// (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) { From 5b2ff5e321bef48a597a737edba45eda2a34fc1f Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 00:26:04 +0100 Subject: [PATCH 2/4] =?UTF-8?q?Test=20that=20node-tools'=20manifest=20is?= =?UTF-8?q?=20not=20refused=20by=20the=20lease=20write's=20check=20(ADR=20?= =?UTF-8?q?0571=20=C2=A713)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tests: one added, the first of its kind in this repository; the repository holds 128 unit tests, within its budget of 500. Built against the group's mesh-sdk through a go.work outside the tree; go.mod is not re-pinned here. (issue 774, 774/S33) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- node-tools/internal/provide/provide_test.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/node-tools/internal/provide/provide_test.go b/node-tools/internal/provide/provide_test.go index 89fd480..216e1ef 100644 --- a/node-tools/internal/provide/provide_test.go +++ b/node-tools/internal/provide/provide_test.go @@ -194,3 +194,19 @@ func TestOnlyTheMachinesRuntimeStartsTheHelper(t *testing.T) { Start(nil, c.module, c.node, time.Second, said.Logf)() } } + +// 774/S33 (ADR 0571 clause 13): this manifest, read as the lease write's check reads it, is not refused at the write: it +// reads and invokes nothing of the controller and consumes nothing of it. +func TestTheManifestIsNotRefusedAtTheLeaseWrite(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + m, err := catalogue.Decode(raw) + if err != nil { + t.Fatal(err) + } + if refused := catalogue.ManifestRefusals(map[string]catalogue.Manifest{Module: m}, catalogue.PhaseWrite); len(refused) != 0 { + t.Errorf("the lease write's check refuses node-tools' manifest: %v", refused) + } +} From 331fa06b54c709151d3433f7260ceb0eaa068328 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 00:51:19 +0100 Subject: [PATCH 3/4] The tool runner reads the memberships state again until it can, instead of giving a module up at its one follow A JetStream 503 or a timeout at the follow (a bus right after its boot) left the module with no membership for good, since Follow runs once per module and the stream fallback is gone. The read is now tried again with a growing wait (1 s doubling to 30 s) until it watches the key or the connection closes; the failure is said once, and its reading once it succeeds. Tests: none added; a step added to an existing unit test (TestAModuleHearsEveryStreamItsMembershipNames), which fails without the retry. (issue 774, 774/S32, review g1 round 1) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- node-tools/internal/bus/bus.go | 61 +++++++++++++++++++----- node-tools/internal/bus/streams_test.go | 23 +++++++++ node-tools/internal/meshtest/meshtest.go | 8 ++++ 3 files changed, 80 insertions(+), 12 deletions(-) diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ad508f0..55d73ce 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -366,6 +366,13 @@ func (c *Conn) Follow(module string) { // GrantsPatience bounds the first read of a membership from the grants part's state. var GrantsPatience = 2 * time.Second +// GrantsRetry is the first wait before the memberships state is read again after it could not be; it doubles up to +// GrantsRetryMost. +var ( + GrantsRetry = time.Second + GrantsRetryMost = 30 * time.Second +) + // 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() @@ -383,9 +390,38 @@ func (c *Conn) issue(module string, m Membership) { } // 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 module has no membership until it can be read. +// before Follow returns, each change after it live. A state that is not there yet, or not readable (a bus answering 503 +// right after its boot, a timeout), is said once and read again, with a growing wait, until it can be or the +// connection closes: Follow runs once per module, so nothing else would ever read it again. func (c *Conn) followGrants(module string) { + if c.watchGrants(module, true) { + return + } + go func() { + wait := GrantsRetry + for !c.nc.IsClosed() { + time.Sleep(wait) + if c.nc.IsClosed() { + return + } + if c.watchGrants(module, false) { + c.Logf("[mesh-tools] the grants part's memberships are read again; %s on %s follows its key", module, c.node) + return + } + if wait *= 2; wait > GrantsRetryMost { + wait = GrantsRetryMost + } + } + }() +} + +// watchGrants is one attempt of followGrants: true once the key is watched. A failure is said when say is true. +func (c *Conn) watchGrants(module string, say bool) bool { + logf := func(format string, a ...any) { + if say { + c.Logf(format, a...) + } + } 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. @@ -396,27 +432,27 @@ 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 has no membership "+ + 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 + return false } js, err := jetstream.New(c.nc) if err != nil { - return + return false } 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 has no membership "+ + 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 + return false } 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); %s has no membership "+ + 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 + return false } apply := func(e jetstream.KeyValueEntry) { if e.Operation() != jetstream.KeyValuePut { @@ -443,7 +479,7 @@ func (c *Conn) followGrants(module string) { select { case e, ok := <-w.Updates(): if !ok { - return + return false } if e == nil { current = false @@ -452,9 +488,9 @@ 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; %s has no membership "+ + 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 + return false } } go func() { @@ -464,6 +500,7 @@ func (c *Conn) followGrants(module string) { } } }() + return true } // Membership is what the mesh issued a module here, or nil when nothing has been issued. diff --git a/node-tools/internal/bus/streams_test.go b/node-tools/internal/bus/streams_test.go index 87deff4..94a6d7c 100644 --- a/node-tools/internal/bus/streams_test.go +++ b/node-tools/internal/bus/streams_test.go @@ -199,6 +199,29 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) { if got := consumerOf(); got != "from-grants-2" { t.Fatalf("a membership said on the controller's subject was applied: the membership names %q", got) } + // Review g1 round 1: memberships that cannot be read at the follow (a bus not ready after its boot) are read again until + // they can be; the follow, which runs once per module, does not give the module up for good. + was := bus.GrantsRetry + bus.GrantsRetry = 50 * time.Millisecond + t.Cleanup(func() { bus.GrantsRetry = was }) + mesh.DropGrants(t) + late, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + late.Logf = func(string, ...any) {} + t.Cleanup(late.Close) + late.Follow("reader") + if m := late.Membership("reader"); m != nil { + t.Fatalf("a membership was served from a state not there: %+v", m) + } + mesh.Issue(t, granted) + mt.Until(t, func() error { + if m := late.Membership("reader"); m == nil || len(m.Hears) != 1 || m.Hears[0].Consumer != "from-grants-2" { + return fmt.Errorf("the membership is %+v, want the one written once the state is there", m) + } + return nil + }) } // novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index a058c77..448e27a 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -131,6 +131,14 @@ func (m *Mesh) putGrants(t *testing.T, mem bus.Membership, body []byte) { } } +// DropGrants deletes the grants part's state memberships, as a bus that has not made it yet holds none; Issue makes it again. +func (m *Mesh) DropGrants(t *testing.T) { + t.Helper() + if err := m.js.DeleteKeyValue(buscompose.GrantsMembershipsBucket); err != nil { + t.Fatal(err) + } +} + // sdkMembership is a membership of this package's type as mesh-sdk's, which the grants part composes and writes. func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership { t.Helper() From 29fd452628ec9ccf527e534b1b2706cf843973ee Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 00:53:42 +0100 Subject: [PATCH 4/4] Pin mesh-sdk at c21d8fd, the removal group's head, so every member builds the same contract (issue 774) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- node-tools/go.mod | 2 +- node-tools/go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/node-tools/go.mod b/node-tools/go.mod index 326f9ed..d0e4637 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -8,7 +8,7 @@ require ( ) require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101085317-790d1c82d027 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect diff --git a/node-tools/go.sum b/node-tools/go.sum index 37e06a8..60e9ab1 100644 --- a/node-tools/go.sum +++ b/node-tools/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101085317-790d1c82d027 h1:a33dsavoaqllh40eVICbhxi6Kero1EGSjt1V+2fa2Hk= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101085317-790d1c82d027/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c h1:/YCxh+NYfqry3PxZYY1nQXaccupobxpsFt8c9BYuGBg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=