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= diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ee01017..55d73ce 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,24 @@ 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) { +// 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() - 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 { @@ -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 -// 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. +// 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. @@ -443,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's membership is read from the "+ - "assignments stream alone", bucket, why, module) - return + 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 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's membership is read from the "+ - "assignments stream alone", buscompose.GrantsMembershipsBucket, err, module) - return + 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 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); read from the "+ - "assignments stream alone", module, key, err) - return + 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 false } apply := func(e jetstream.KeyValueEntry) { 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 // 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 +471,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() @@ -490,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 @@ -499,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; read from the "+ - "assignments stream alone", module, key, StateTimeout) - return + 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 false } } 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. 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..94a6d7c 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,46 +183,45 @@ 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) } + // 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 dfbecd4..448e27a 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))) } @@ -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. func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership { 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 // delivered. func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelivered uint64) { 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) + } +}