10 Commits
Author SHA1 Message Date
mesh-tools 49f9269376 Merge pull request 'node-tools follows memberships from the grants part's state alone (issue 774)' (#54) from feat/l3-removal into main 2026-11-02 00:35:50 +00:00
jschoubben 29fd452628 Pin mesh-sdk at c21d8fd, the removal group's head, so every member builds the same contract (issue 774)
mesh/delivery merged: source.main moved to novox/mesh-tools@49f9269376e3
mesh/delivery-group group feat/l3-removal merged: every member is merged
mesh/check failed: mesh-build-go: the check of novox/mesh-catalog#590: the ask mvyhumdm860c7f's step on shanks did not end within 30m0s of its start |…
Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 00:53:42 +01:00
jschoubben 331fa06b54 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
2026-11-02 00:51:19 +01:00
jschoubben 5b2ff5e321 Test that node-tools' manifest is not refused by the lease write's check (ADR 0571 §13)
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
2026-11-02 00:26:04 +01:00
jschoubben 4fbfdfad13 The tool runner reads its membership from the grants part's state alone: no stream read, no subject heard (ADR 0571 §10)
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
2026-11-02 00:25:19 +01:00
mesh-tools 75943ca7b0 Merge pull request 'The tool runner waits on the self-check's checked, not the controller's (design 47 L7)' (#53) from feat/l5-l7-main into main 2026-11-01 13:54:02 +00:00
jschoubben fb6631aee1 Pin group A's mesh-sdk head on main
mesh/delivery merged: source.main moved to novox/mesh-tools@75943ca7b0c3
mesh/delivery-group group feat/l5-l7-main merged: every member is merged
mesh/check checking fb6631aee1a2 on 75943ca7b0c3 | novox/mesh-tools#53 in group feat/l5-l7-main
Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-01 09:57:49 +01:00
jschoubben d41bee158c node-tools waits on the self-check's checked, not the controller's (ADR 0571 clause 13, issue 1955)
Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-01 09:54:39 +01:00
mesh-tools 110e55cb48 Merge pull request 'The tool runner declares the credential form of its mcp-endpoint (ADR 0579)' (#52) from feat/judge-once into main 2026-11-01 07:55:09 +00:00
jschoubben c494018074 Pin the group's mesh-sdk head after the gofmt fix
mesh/delivery merged: source.main moved to novox/mesh-tools@110e55cb4836
mesh/delivery-group group feat/judge-once merged: every member is merged
mesh/check checking c4940180746a on c2f01bf10b68 | novox/mesh-tools#52 in group feat/judge-once
Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-01 03:27:07 +01:00
9 changed files with 154 additions and 147 deletions
+1 -1
View File
@@ -8,7 +8,7 @@ require (
) )
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101020750-62cddf4efc62 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/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
github.com/google/go-tpm v0.9.8 // indirect github.com/google/go-tpm v0.9.8 // indirect
github.com/klauspost/compress v1.20.0 // indirect github.com/klauspost/compress v1.20.0 // indirect
+4 -6
View File
@@ -1,9 +1,7 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmdPBKqCg5OmXRNb269EkkPe7BsEp+RykEhkgE= 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.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= 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.20261101014310-19c204273bdc h1:kVq7sS4I0EnL8DcAA2a3YstnXS6WmYYzJz7eK7IZ9RY= 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.20261101014310-19c204273bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101235037-c21d8fd9692c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101020750-62cddf4efc62 h1:R5C8ICjiaDITHRFR/uIATwARPPXKxssUC70m0svstb0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101020750-62cddf4efc62/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+63 -73
View File
@@ -37,8 +37,6 @@ import (
// verb still runs, so a caller is never left guessing past this (novox/hq issue 265). // verb still runs, so a caller is never left guessing past this (novox/hq issue 265).
const RequestTimeout = 30 * time.Second const RequestTimeout = 30 * time.Second
const assignmentsStream = "ASSIGNMENTS"
// Credential is a broker credential as the mesh delivers it (novox/hq ADR 0120). // Credential is a broker credential as the mesh delivers it (novox/hq ADR 0120).
type Credential struct { type Credential struct {
URL string `json:"url"` URL string `json:"url"`
@@ -183,12 +181,9 @@ type Conn struct {
cred Credential cred Credential
mu sync.Mutex mu sync.Mutex
issued map[string]*Membership // module → membership; present with nil = followed, none issued 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 onNew []*func(Membership)
// clause 7): its assignments stream copy is then a fallback no longer applied. subs []*nats.Subscription
fromGrants map[string]bool 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 // answering is every subject this runtime answers on, so a subscription the bus refused can be
// asked for again (novox/hq issue 222). // asked for again (novox/hq issue 222).
answering map[string]*answered answering map[string]*answered
@@ -332,7 +327,7 @@ func Connect(cred Credential) (*Conn, error) {
return nil, err return nil, err
} }
c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred, 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) c.Follow(cred.Module)
return c, nil return c, nil
} }
@@ -346,16 +341,10 @@ func (c *Conn) Node() string { return c.node }
// Credential is what this connection was opened with. // Credential is what this connection was opened with.
func (c *Conn) Credential() Credential { return c.cred } 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 // 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 // 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 // `<node>.<module>` (novox/hq ADR 0571 clause 7 and 10, mesh-sdk buscompose.GrantsMembershipsBucket), and nowhere else:
// assignments stream is read beside it as the fallback until the lease write, and applied only while the state has given // the controller's assignments stream is not read and its subject is not heard.
// the module no membership.
func (c *Conn) Follow(module string) { func (c *Conn) Follow(module string) {
c.mu.Lock() c.mu.Lock()
if c.node == "" { if c.node == "" {
@@ -369,53 +358,24 @@ func (c *Conn) Follow(module string) {
c.issued[module] = nil c.issued[module] = nil
c.mu.Unlock() c.mu.Unlock()
c.followGrants(module) 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 { 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) 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. // GrantsPatience bounds the first read of a membership from the grants part's state.
var GrantsPatience = 2 * time.Second 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 // GrantsRetry is the first wait before the memberships state is read again after it could not be; it doubles up to
// passed over once the grants part's state has given the module one. // GrantsRetryMost.
func (c *Conn) issue(module string, m Membership, fromGrants bool) { 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() c.mu.Lock()
if !fromGrants && c.fromGrants[module] {
c.mu.Unlock()
return
}
if fromGrants {
c.fromGrants[module] = true
}
c.issued[module] = &m c.issued[module] = &m
handlers := make([]func(Membership), 0, len(c.onNew)) handlers := make([]func(Membership), 0, len(c.onNew))
for _, h := range c.onNew { for _, h := range c.onNew {
@@ -430,9 +390,38 @@ 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 // 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 // before Follow returns, each change after it live. A state that is not there yet, or not readable (a bus answering 503
// the assignments stream answers alone. // 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) { 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) 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, // 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. // answers at once, where binding the bucket would wait out its bound on a refusal.
@@ -443,27 +432,27 @@ func (c *Conn) followGrants(module string) {
if err != nil { if err != nil {
why = err.Error() why = err.Error()
} }
c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s's membership is read from the "+ logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s has no membership "+
"assignments stream alone", bucket, why, module) "until it can be", bucket, why, module)
return return false
} }
js, err := jetstream.New(c.nc) js, err := jetstream.New(c.nc)
if err != nil { if err != nil {
return return false
} }
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
defer cancel() defer cancel()
kv, err := js.KeyValue(ctx, bucket) kv, err := js.KeyValue(ctx, bucket)
if err != nil { if err != nil {
c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s's membership is read from the "+ logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s has no membership "+
"assignments stream alone", buscompose.GrantsMembershipsBucket, err, module) "until it can be", buscompose.GrantsMembershipsBucket, err, module)
return return false
} }
w, err := kv.Watch(context.Background(), key) w, err := kv.Watch(context.Background(), key)
if err != nil { if err != nil {
c.Logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); read from the "+ logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); %s has no membership "+
"assignments stream alone", module, key, err) "until it can be", module, key, err, module)
return return false
} }
apply := func(e jetstream.KeyValueEntry) { apply := func(e jetstream.KeyValueEntry) {
if e.Operation() != jetstream.KeyValuePut { if e.Operation() != jetstream.KeyValuePut {
@@ -471,7 +460,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 // 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 // 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. // A refused value arriving after a good one keeps the last good membership: nothing here clears it.
var m Membership var m Membership
body, err := membership.BareGrantsMembership(e.Value()) body, err := membership.BareGrantsMembership(e.Value())
@@ -482,7 +471,7 @@ func (c *Conn) followGrants(module string) {
c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err) c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err)
return return
} }
c.issue(module, m, true) c.issue(module, m)
} }
timer := time.NewTimer(StateTimeout) timer := time.NewTimer(StateTimeout)
defer timer.Stop() defer timer.Stop()
@@ -490,7 +479,7 @@ func (c *Conn) followGrants(module string) {
select { select {
case e, ok := <-w.Updates(): case e, ok := <-w.Updates():
if !ok { if !ok {
return return false
} }
if e == nil { if e == nil {
current = false current = false
@@ -499,9 +488,9 @@ func (c *Conn) followGrants(module string) {
apply(e) apply(e)
case <-timer.C: case <-timer.C:
_ = w.Stop() _ = w.Stop()
c.Logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; read from the "+ logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; %s has no membership "+
"assignments stream alone", module, key, StateTimeout) "until it can be", module, key, StateTimeout, module)
return return false
} }
} }
go func() { go func() {
@@ -511,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. // Membership is what the mesh issued a module here, or nil when nothing has been issued.
+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 // **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 // 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 // membership, and served no tool and no state. This is what the live mesh held: read through mesh-sdk's codec, its tools
// held: read through mesh-sdk's codec, its tools are answered where the membership says and its state is reached. // 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) { func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
mesh := mt.New(t) mesh := mt.New(t)
mesh.Bucket(t, "gitea_repos") mesh.Bucket(t, "gitea_repos")
gitea := mt.MembershipOf("gitea", "anchor", false, nil) gitea := mt.MembershipOf("gitea", "anchor", false, nil)
gitea.State = []bus.StateIssued{{Name: "repos", Bucket: "gitea_repos", Writes: true}} gitea.State = []bus.StateIssued{{Name: "repos", Bucket: "gitea_repos", Writes: true}}
mesh.IssueInGrantsWrapped(t, gitea, "published") mesh.IssueInGrantsWrapped(t, gitea, "published")
// The old wrapper of a membership the part held back carries its unpublished composition: refused, so the // The old wrapper of a membership the part held back carries its unpublished composition: refused, never served.
// assignments stream's copy is served, never what the hold kept back.
pages := mt.MembershipOf("pages", "anchor", false, nil)
mesh.Issue(t, pages)
heldBack := mt.MembershipOf("pages", "anchor", true, nil) heldBack := mt.MembershipOf("pages", "anchor", true, nil)
mesh.IssueInGrantsWrapped(t, heldBack, "not published, a shadow difference: publishing would lose the last holder of x") 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"}) conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
if err != nil { if err != nil {
t.Fatal(err) 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) t.Cleanup(conn.Close)
conn.Follow("gitea") conn.Follow("gitea")
mt.Until(t, func() error { mt.Until(t, func() error {
@@ -156,14 +155,9 @@ func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
t.Errorf("the state reads back %+v, %v", e, err) t.Errorf("the state reads back %+v, %v", e, err)
} }
conn.Follow("pages") conn.Follow("pages")
mt.Until(t, func() error { time.Sleep(300 * time.Millisecond)
if m := conn.Membership("pages"); m == nil || len(m.Serves) == 0 { if m := conn.Membership("pages"); m != nil || !strings.Contains(strings.Join(logged, "\n"), "no membership issued for pages") {
return fmt.Errorf("pages' membership is %+v, want the assignments stream's", m) t.Errorf("a held-back membership's wrapper was served or not said: %+v, %q", m, logged)
}
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)
} }
} }
+31 -32
View File
@@ -161,11 +161,11 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
return nil return nil
}) })
// The membership as the grants part keeps it (ADR 0571 clause 7): read first, applied live, and the assignments // The membership as the grants part keeps it (ADR 0571 clause 7, 10): read first and applied live, and nothing
// stream's copy passed over once the state has given one. // published on the controller's subject is heard.
granted := reader granted := reader
granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}} 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"}) again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -183,46 +183,45 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
t.Fatalf("the membership read at the follow names %q, want the grants part's", got) t.Fatalf("the membership read at the follow names %q, want the grants part's", got)
} }
granted.Hears[0].Consumer = "from-grants-2" granted.Hears[0].Consumer = "from-grants-2"
mesh.IssueInGrants(t, granted) mesh.Issue(t, granted)
mt.Until(t, func() error { mt.Until(t, func() error {
if got := consumerOf(); got != "from-grants-2" { if got := consumerOf(); got != "from-grants-2" {
return fmt.Errorf("the membership names %q, want the grants part's change", got) return fmt.Errorf("the membership names %q, want the grants part's change", got)
} }
return nil return nil
}) })
// A later, different copy on the assignments stream changes nothing once the state has given one. // 774/S32/AC1: a copy said on the controller's subject (mesh.assignment) changes nothing; there is no stream to read it from.
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()
streamed := reader streamed := reader
streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-stream"}} streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-subject"}}
mesh.Issue(t, streamed) mesh.Publish(t, "mesh.assignment.anchor.reader", streamed)
select {
case <-arrived:
case <-time.After(5 * time.Second):
t.Fatal("the stream's copy was not published")
}
again.Flush() again.Flush()
time.Sleep(200 * time.Millisecond) time.Sleep(200 * time.Millisecond)
if got := consumerOf(); got != "from-grants-2" { 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)
} }
// 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 // novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once
@@ -26,7 +26,7 @@ func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
want := []string{"mesh-knowledge.reviewed", "mesh-delivery-gate.transition", "gitea.forge.status", want := []string{"mesh-knowledge.reviewed", "mesh-delivery-gate.transition", "gitea.forge.status",
"mesh-controller.checked", "operator-channel.decided.claude-code"} "mesh-self-check.checked", "operator-channel.decided.claude-code"}
if m.Module != "node-tools" || !reflect.DeepEqual(m.Waits, want) { if m.Module != "node-tools" || !reflect.DeepEqual(m.Waits, want) {
t.Errorf("%s waits on %v, want %v", m.Module, m.Waits, want) t.Errorf("%s waits on %v, want %v", m.Module, m.Waits, want)
} }
+28 -18
View File
@@ -86,11 +86,10 @@ func New(t *testing.T) *Mesh {
t.Fatal(err) t.Fatal(err)
} }
js, _ := nc.JetStream() js, _ := nc.JetStream()
for _, s := range []string{"ASSIGNMENTS", "EVENTS"} { _ = js.DeleteStream("EVENTS")
_ = js.DeleteStream(s) // 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.AddStream(&nats.StreamConfig{Name: "ASSIGNMENTS", Subjects: []string{"mesh.assignment.>"}, if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil {
MaxMsgsPerSubject: 1, AllowDirect: true}); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { 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} 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) { 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() t.Helper()
m.putGrants(t, mem, membership.EncodeGrantsMembership(sdkMembership(t, mem))) m.putGrants(t, mem, membership.EncodeGrantsMembership(sdkMembership(t, mem)))
} }
@@ -141,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. // 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 { func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership {
t.Helper() t.Helper()
@@ -319,6 +317,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 // Pending is what a module's consumer still holds: delivered and not acknowledged, and not yet
// delivered. // delivered.
func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelivered uint64) { func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelivered uint64) {
@@ -194,3 +194,19 @@ func TestOnlyTheMachinesRuntimeStartsTheHelper(t *testing.T) {
Start(nil, c.module, c.node, time.Second, said.Logf)() 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)
}
}
+1 -1
View File
@@ -9,7 +9,7 @@
"mesh-knowledge.reviewed", "mesh-knowledge.reviewed",
"mesh-delivery-gate.transition", "mesh-delivery-gate.transition",
"gitea.forge.status", "gitea.forge.status",
"mesh-controller.checked", "mesh-self-check.checked",
"operator-channel.decided.claude-code" "operator-channel.decided.claude-code"
], ],
"provides": [ "provides": [