From 1ff5049d03e1ce469a72d9e49a8faa5094d7e8ae Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 22:37:06 +0100 Subject: [PATCH] mesh-grants publishes memberships on change; the user list, assertions and grants stay in shadow (issue 1945) Since release 6 nobody publishes memberships, so a module placed since cannot serve its seat or reach a new state (the operator, 2026-10-25 22:31). The part publishes each membership on the assignments stream only when it differs from the one last published, composes from the manifest the wanted build names, and says a refused publish and tries again at the next event. Pins mesh-sdk 881da7f, whose composer grants mesh-grants that publish. --- modules/mesh-grants/cmd/mesh-grants/main.go | 37 ++++-- .../mesh-grants/cmd/mesh-grants/publish.go | 55 +++++++++ modules/mesh-grants/cmd/mesh-grants/tools.go | 16 +-- modules/mesh-grants/go.mod | 6 +- modules/mesh-grants/go.sum | 4 +- modules/mesh-grants/grants_test.go | 113 ++++++++++++++++-- modules/mesh-grants/internal/grants/follow.go | 23 +++- modules/mesh-grants/internal/grants/inputs.go | 50 +++++++- .../internal/grants/memberships.go | 88 ++++++++++++-- modules/mesh-grants/internal/grants/part.go | 23 +++- .../mesh-grants/internal/grants/records.go | 11 +- modules/mesh-grants/internal/grants/tools.go | 39 +++++- modules/mesh-grants/module.json | 22 +++- 13 files changed, 425 insertions(+), 62 deletions(-) create mode 100644 modules/mesh-grants/cmd/mesh-grants/publish.go diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 25c7402..740e700 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -1,14 +1,21 @@ -// Command mesh-grants is the grants part (novox/hq ADR 0547 clauses 5 to 8, issue 1945), in shadow for the first -// milestone: a process on its own bus credential that watches placement, the indexer's manifests, mesh-config's -// resolved values and the node agents' haves, and on every change composes the bus's user list, each assignment's -// membership and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk -// buscompose), compares them with what the bus loaded and holds (through the bus module's own read tools), and writes -// only its own state. It publishes no membership, asserts nothing, writes no grant and no user list the bus node reads. -// It holds no timer, and never exits for a refused read or write: a refusal is said and tried again at the next event. +// Command mesh-grants is the grants part (novox/hq ADR 0547 clauses 5 to 8, issue 1945): a process on its own bus +// credential that watches placement, the assigner's wants, the indexer's manifests, the builder's builds, mesh-config's +// resolved values and the node agents' haves, and on every change composes each assignment's membership, the bus's user +// list and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk buscompose). +// +// **Memberships are published for real** (the operator, 2026-10-25 22:31): since release 6 nobody publishes them, so a +// module placed since cannot serve its seat or reach a new state. Each is published on the assignments stream only when +// it differs from the one last published there. **The user list, the assertions and the grants stay in shadow** until +// the operator's take-over, which this milestone always refuses: they are composed, compared with what the bus loaded +// and holds (through the bus module's own read tools) and kept in the part's own state, and take no effect. +// +// It holds no timer, and never exits for a refused read, write or publish: a refusal is said and tried again at the next +// event. // // **No seat in this milestone.** ADR 0547 clause 5 gives the part the seat `mesh-grants`; a `mesh-*` seat must be in // mesh-sdk's own seat list, which the running controller's composer does not know, so the part serves module tools -// (grants_*) and claims no seat until take-over. +// (grants_*) and claims no seat until take-over. The composer grants the membership publish by the module's name, +// mesh-grants. // // Started by the tool runner, with no credential of its own, it serves those tools (a process never serves, ADR 0436). package main @@ -27,6 +34,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/process" @@ -103,9 +111,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } bus := newBusReader(ctx, js) - f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn}} + f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn}, + Publish: memberships{ctx: ctx, nc: nc, js: js}} f.Watch = bus - logger.Printf("started on the account %s, in shadow: composing and comparing, writing only its own state", account.User) + f.Asked = func(wake func()) error { + return conn.State(grants.StateUserList).Watch(grants.KeyAsked, func(c stdio.StateChange) error { + if !c.Current { + wake() + } + return nil + }) + } + logger.Printf("started on the account %s: publishing memberships; the user list, assertions and grants in shadow", account.User) f.Run(ctx) return nil } diff --git a/modules/mesh-grants/cmd/mesh-grants/publish.go b/modules/mesh-grants/cmd/mesh-grants/publish.go new file mode 100644 index 0000000..340a80a --- /dev/null +++ b/modules/mesh-grants/cmd/mesh-grants/publish.go @@ -0,0 +1,55 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strconv" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/membership" +) + +// memberships publishes each membership on the assignments stream, last per subject, over the one last read there +// (ExpectLastSequencePerSubject, so a membership another writer published meanwhile is not overwritten, issues 930 and +// 934), and reads the last one by a direct get of its subject. The one place the part publishes on the bus. +type memberships struct { + ctx context.Context + nc *nats.Conn + js jetstream.JetStream +} + +func (m memberships) LastMembership(node, module string) ([]byte, uint64, bool, error) { + ctx, cancel := context.WithTimeout(m.ctx, requestWithin) + defer cancel() + msg, err := m.nc.RequestWithContext(ctx, membership.DirectGetSubject(node, module), nil) + if err != nil { + if errors.Is(err, context.DeadlineExceeded) { + err = fmt.Errorf("no answer in %s: a read the bus refuses is answered by none (%w)", requestWithin, err) + } + return nil, 0, false, err + } + switch status := msg.Header.Get("Status"); status { + case "": + case "404": + return nil, 0, false, nil + default: + return nil, 0, false, fmt.Errorf("the assignments stream answered %s %s", status, msg.Header.Get("Description")) + } + seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64) + if err != nil { + return nil, 0, false, fmt.Errorf("the membership of %s on %s carries no sequence", module, node) + } + return msg.Data, seq, true, nil +} + +func (m memberships) PublishMembership(node, module string, body []byte, after uint64) error { + ctx, cancel := context.WithTimeout(m.ctx, requestWithin) + defer cancel() + if _, err := m.js.Publish(ctx, membership.Subject(node, module), body, jetstream.WithExpectLastSequencePerSubject(after)); err != nil { + return fmt.Errorf("the membership of %s on %s after sequence %d: %w", module, node, after, err) + } + return nil +} diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index decd379..0ab9305 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -66,8 +66,8 @@ func toolsOf(p *grants.Part) []stdio.Tool { Input: schema(map[string]string{"module": "one module"}), Run: func(a map[string]any) (any, error) { return p.Grants(grants.Answer(a, "module")) }}, {Name: grants.ToolMembership, Description: "One module's membership on one node (ADR 0547 clause 6): as composed now from the " + - "bus's records with the bus composer, as last kept in the part's state, and as published (not readable from here: the bus " + - "grants each module its own alone), with what the composition cannot read. Read-only.", + "bus's records with the bus composer, and what the part's process published of it at its last pass (published, " + + "unchanged, or why not), with what the composition cannot read. Read-only.", Input: schema(map[string]string{"node": "the machine", "module": "the module"}, "node", "module"), Run: func(a map[string]any) (any, error) { return p.MembershipOf(grants.Answer(a, "node"), grants.Answer(a, "module")) @@ -77,11 +77,13 @@ func toolsOf(p *grants.Part) []stdio.Tool { "they differ: users added, users dropped (and of those, gone from placement), each user's grant differences, the " + "assignments left out, and every input the bus does not carry, named. Read-only.", Run: func(map[string]any) (any, error) { return p.UserList() }}, - {Name: grants.ToolComposeNow, Description: "Compose again now (ADR 0547 clause 6), as the part does on every change it hears: the " + - "user list, the memberships and the assertions, compared with what the bus loaded and holds; writes its own state only " + - "where a value changed (one revision when an input changed, none when not) and says user-list-changed and " + - "membership-changed for each. In shadow it publishes, asserts and grants nothing.", - Run: func(map[string]any) (any, error) { return p.Compose(), nil }}, + {Name: grants.ToolComposeNow, Description: "Compose again now (ADR 0547 clause 6), as the part does on every change it hears: asks " + + "the part's process (which alone holds the grant to publish a membership) to compose the memberships, the user list and " + + "the assertions, publish each membership that differs from the one last published, and keep what changed in its own " + + "state (one revision when an input changed, none when not), saying user-list-changed and membership-changed. Answers " + + "the ask and the last pass; grants_user_list shows the pass that answers it. The user list, assertions and grants stay " + + "in shadow.", + Run: func(a map[string]any) (any, error) { return p.AskCompose(stdio.CallerOf(a)) }}, {Name: grants.ToolAssert, Description: "One module's buckets, logs and streams (ADR 0547 clause 6), composed from its manifest at " + "the indexer's trunk with their limits, beside what the bus holds (through the bus module's nats_streams). In shadow it " + "asserts nothing, and says so.", diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index 10333d1..2936325 100644 --- a/modules/mesh-grants/go.mod +++ b/modules/mesh-grants/go.mod @@ -2,11 +2,13 @@ module mesh-grants go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025205458-b5f8604d037a +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025213320-881da7f59995 + github.com/nats-io/nats.go v1.54.0 +) require ( github.com/klauspost/compress v1.20.0 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect diff --git a/modules/mesh-grants/go.sum b/modules/mesh-grants/go.sum index f3f40fa..ae27ab6 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025205458-b5f8604d037a h1:yqCMZJSVRZ7VoBiZ9iIcYhQKn0BDhOJIRwZ430auLA0= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025205458-b5f8604d037a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025213320-881da7f59995 h1:jY8ufarHcRyVCc2snyMk9eTmMkM/pnLC1cAqb4/IC5A= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025213320-881da7f59995/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/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 7971898..259165f 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -35,6 +35,39 @@ type fakeBus struct { calls []string events []string writes []string + // The assignments stream: each subject's last membership and sequence, every publish, and whether it is refused. + stream map[string][]byte + seqs map[string]uint64 + seq uint64 + publishes []string + refusePub bool +} + +func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + subject := "mesh.assignment." + node + "." + module + v, ok := b.stream[subject] + return v, b.seqs[subject], ok, nil +} + +func (b *fakeBus) PublishMembership(node, module string, body []byte, after uint64) error { + b.mu.Lock() + defer b.mu.Unlock() + subject := "mesh.assignment." + node + "." + module + if b.refusePub { + return errors.New("Permissions Violation for Publish to " + subject) + } + if b.seqs[subject] != after { + return fmt.Errorf("wrong last sequence for %s: %d", subject, b.seqs[subject]) + } + if b.stream == nil { + b.stream, b.seqs = map[string][]byte{}, map[string]uint64{} + } + b.seq++ + b.stream[subject], b.seqs[subject] = body, b.seq + b.publishes = append(b.publishes, subject) + return nil } func newFakeBus() *fakeBus { @@ -187,7 +220,7 @@ func fixture() *fakeBus { } func partOn(b *fakeBus) *grants.Part { - return &grants.Part{Read: b, Call: b, Store: ownStore{b}, Emit: b, Now: func() time.Time { return time.Date(2026, 10, 25, 22, 0, 0, 0, time.UTC) }} + return &grants.Part{Read: b, Call: b, Store: ownStore{b}, Emit: b, Publish: b, Now: func() time.Time { return time.Date(2026, 10, 25, 22, 0, 0, 0, time.UTC) }} } // shadowOf is the list the part composes from b's records now. @@ -254,10 +287,12 @@ func TestGrants(t *testing.T) { } }) - t.Run("no timer, no publish, no create", func(t *testing.T) { + t.Run("no timer, no create, a publish only of memberships", func(t *testing.T) { forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc", - ".Publish(", "PublishMsg", "CreateStream", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue", - "CreateObjectStore", "mesh.assignment", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("} + "PublishMsg", "CreateStream", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue", + "CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("} + // The one publish: a membership, on its own subject, in publish.go. + publishOnly := filepath.Join("cmd", "mesh-grants", "publish.go") files := 0 for _, root := range []string{"cmd", "internal"} { err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { @@ -266,9 +301,15 @@ func TestGrants(t *testing.T) { } files++ src, err := os.ReadFile(path) + if path != publishOnly && (strings.Contains(string(src), ".Publish(") || strings.Contains(string(src), "mesh.assignment")) { + t.Errorf("%s publishes: only publish.go publishes, and only memberships", path) + } + if path == publishOnly && strings.Count(string(src), ".Publish(") != 1 { + t.Errorf("publish.go publishes other than one membership") + } for _, f := range forbidden { if strings.Contains(string(src), f) { - t.Errorf("%s holds %s: in shadow the part holds no timer and publishes or creates nothing", path, f) + t.Errorf("%s holds %s: the part holds no timer and creates or asserts nothing", path, f) } } return err @@ -293,6 +334,15 @@ func TestGrants(t *testing.T) { []string{"a.mod-x", "a.nats", "b.mod-y", "controller", "node.a", "node.b"}, "enrolment tokens"}, {"a manifest missing is a gap and its user is left out", func(b *fakeBus) { delete(b.states[grants.ReadModules], "mod-y") }, []string{"a.mod-x", "a.nats", "controller", "node.a", "node.b"}, "manifest of mod-y"}, + {"a want's built manifest is composed, not the trunk's", func(b *fakeBus) { + built := strings.Replace(xManifest, `"history":3}]`, `"history":3},{"name":"newer"}]`, 1) + digest := delivery.ManifestDigest([]byte(built)) + b.put(grants.ReadManifests, digest, delivery.Manifest{V: delivery.Version, Module: "mod-x", JSON: built}) + fp := "src2:" + strings.Repeat("cd", 32) + b.put(grants.ReadWants, "want.a.mod-x", delivery.Want{V: delivery.Version, FP: fp, Composed: "sha256:" + strings.Repeat("ef", 32)}) + key, _ := delivery.BuilderBuildKey("mod-x", fp) + b.put(grants.ReadBuilds, key, map[string]any{"v": delivery.Version, "state": "done", "manifest": digest}) + }, []string{"a.mod-x", "a.nats", "b.mod-y", "controller", "node.a", "node.b"}, "enrolment tokens"}, {"placement refused is a gap; the machines come from the haves", func(b *fakeBus) { b.refused[grants.ReadAssignments] = true }, []string{"controller", "node.a", "node.b"}, grants.ReadAssignments}, } { @@ -311,6 +361,12 @@ func TestGrants(t *testing.T) { if !gapNamed(append(in.Gaps, left...), row.gap) { t.Errorf("gaps %v name no %q", in.Gaps, row.gap) } + if strings.HasPrefix(row.name, "a want's") { + x := users[slices.IndexFunc(users, func(u grants.User) bool { return u.User == "a.mod-x" })] + if !slices.Contains(x.Publish, "$KV.mod-x_newer.>") { + t.Errorf("a.mod-x is not granted the state its wanted build declares: %v", x.Publish) + } + } }) } b := fixture() @@ -372,14 +428,21 @@ func TestGrants(t *testing.T) { if !first.Written || first.Revision != 1 || !first.UserList.Compare.Equal() || first.UserList.Users != 6 { t.Fatalf("first composition = %+v", first) } - if again := p.Compose(); again.Written || again.Revision != 1 || again.MembershipsWritten != 0 || again.AssertionsWritten != 0 { - t.Errorf("a composition with no input changed wrote: %+v", again) + if len(first.Published) != 3 || len(b.publishes) != 3 { + t.Errorf("the first composition published %v, want the three assignments' memberships", first.Published) + } + // The second pass says each membership unchanged (published at the first); the third writes nothing at all. + if again := p.Compose(); again.Written || again.Revision != 1 || len(again.Published) != 0 || again.AssertionsWritten != 0 { + t.Errorf("a composition with no input changed wrote or published: %+v", again) + } + if third := p.Compose(); third.Written || third.MembershipsWritten != 0 || len(third.Published) != 0 || len(b.publishes) != 3 { + t.Errorf("a membership equal to the one last published was published again, or a state written: %+v", third) } // mod-x placed on b too: b.mod-x is added, and a.mod-x's membership moves with b's, since mod-x now runs on two // machines and answers only on its machines' subjects. b.assign("b", "mod-x") moved := p.Compose() - if !moved.Written || moved.Revision != 2 || moved.MembershipsWritten != 2 || moved.UserList.Compare.Equal() || + if !moved.Written || moved.Revision != 2 || !slices.Equal(moved.Published, []string{"a.mod-x", "b.mod-x"}) || moved.UserList.Compare.Equal() || !slices.Equal(moved.UserList.Compare.Added, []string{"b.mod-x"}) { t.Errorf("a composition after placement moved = %+v", moved) } @@ -405,7 +468,7 @@ func TestGrants(t *testing.T) { } ms, _ := p.MembershipOf("a", "mod-x") - if mo, _ := ms.(map[string]any); mo["published"] != grants.PublishedUnread || mo["composed"] == nil || mo["kept"] == nil { + if mo, _ := ms.(map[string]any); !strings.Contains(fmt.Sprint(mo["published"]), "last_pass") || mo["composed"] == nil { t.Errorf("membership = %v", ms) } gr, _ := p.Grants("mod-x") @@ -464,7 +527,12 @@ func TestGrants(t *testing.T) { } }) - t.Run("nothing published or asserted", func(t *testing.T) { + t.Run("only memberships published, nothing asserted", func(t *testing.T) { + for _, s := range b.publishes { + if !strings.HasPrefix(s, "mesh.assignment.") { + t.Errorf("published on %s", s) + } + } for _, w := range b.writes { if !slices.Contains(grants.States, w) { t.Errorf("wrote %s, which is not the part's own state", w) @@ -483,6 +551,31 @@ func TestGrants(t *testing.T) { }) }) + t.Run("memberships: carried, refused, then published at the next event", func(t *testing.T) { + b := fixture() + b.loaded = shadowOf(t, b) + b.refusePub = true + p := partOn(b) + refused := p.Compose() + if len(refused.Published) != 0 || len(refused.NotPublished) != 3 || !strings.Contains(refused.NotPublished[0], "Permissions Violation") { + t.Errorf("a refused publish = %+v: said, and nothing published", refused) + } + raw, _, _, _ := b.ownGet(grants.StateMemberships, "a.mod-x") + if !strings.Contains(string(raw), "tried again at the next event") { + t.Errorf("the refusal is not kept: %s", raw) + } + b.refusePub = false + b.stream = map[string][]byte{"mesh.assignment.a.nats": []byte(`{"node":"a","module":"nats","mesh":["10.0.0.1"],"serves":[],"emits":"x","tools":"y"}`)} + b.seqs = map[string]uint64{"mesh.assignment.a.nats": 9} + next := p.Compose() + if !slices.Equal(next.Published, []string{"a.mod-x", "a.nats", "b.mod-y"}) { + t.Errorf("the next event published %v", next.Published) + } + if !strings.Contains(string(b.stream["mesh.assignment.a.nats"]), "10.0.0.1") { + t.Errorf("the private network of the membership last published is not carried: %s", b.stream["mesh.assignment.a.nats"]) + } + }) + t.Run("a refused read or write never stops it", func(t *testing.T) { b := fixture() b.loaded = shadowOf(t, b) diff --git a/modules/mesh-grants/internal/grants/follow.go b/modules/mesh-grants/internal/grants/follow.go index 90a8312..9c6e85f 100644 --- a/modules/mesh-grants/internal/grants/follow.go +++ b/modules/mesh-grants/internal/grants/follow.go @@ -15,8 +15,9 @@ type Watcher interface { } // Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved -// config values, and the node agents' haves (a machine's first have, and the bus node's loaded list). -var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave} +// config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's +// wants (the manifest each machine is wanted to run). +var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants} // Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot // open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect). @@ -24,10 +25,13 @@ type Follower struct { Part *Part Watch Watcher Logf func(string, ...any) + // Asked watches the part's own ask to compose now (a tool's compose-now), waking at each; nil watches none. + Asked func(wake func()) error wake chan struct{} once sync.Once mu sync.Mutex + asked bool watched map[string]bool refused map[string]string have map[string]string @@ -100,6 +104,13 @@ func (f *Follower) heard(state, key string, value []byte) bool { // open opens every watch not open yet; one refused is said and left for the next wake. func (f *Follower) open() { + if f.Asked != nil && !f.asked { + if err := f.Asked(f.Wake); err != nil { + f.logf("a tool's ask to compose now cannot be heard yet (%v): tried again at the next change", err) + } else { + f.asked = true + } + } for _, state := range Followed { f.mu.Lock() open := f.watched[state] @@ -141,10 +152,16 @@ func (f *Follower) Run(ctx context.Context) { return case <-f.wake: f.open() - res := f.Part.Compose() + res := f.Part.Pass() for _, e := range res.Errors { f.logf("%s", e) } + if len(res.Published) > 0 { + f.logf("published %d memberships: %v", len(res.Published), res.Published) + } + if len(res.NotPublished) > 0 { + f.logf("%d memberships not published, tried again at the next event; the first: %s", len(res.NotPublished), res.NotPublished[0]) + } if res.Written { f.logf("the shadow list moved to revision %d: %d users, %d differing from the loaded list", res.Revision, res.UserList.Users, len(res.UserList.Compare.Differing())) diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 4297bc4..5e65329 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -8,6 +8,7 @@ package grants import ( "encoding/json" "fmt" + "slices" "sort" "strings" "sync" @@ -25,10 +26,12 @@ const ( ReadManifests = "mesh-delivery-indexer.manifests" ReadHave = "node-engine.have" ReadResolved = "mesh-config.resolved" + ReadWants = "mesh-delivery-assigner.wants" + ReadBuilds = "mesh-delivery-builder.builds" ) // Reads is every state the part reads, in the manifest's order. -var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave} +var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds} // The bus module's tools the part calls, as its manifest's invokes name them: read-only, each. const ( @@ -97,6 +100,9 @@ type Inputs struct { Manifests map[string]catalogue.Manifest // Digests is each module's manifest digest, by module. Digests map[string]string + // Running is the manifest a machine is wanted to run, by `.`: the one the builder built at the + // fingerprint the assigner's want names. An assignment with none is composed from the trunk's manifest. + Running map[string]catalogue.Manifest // Values is each resolved config value, by `//`. Values map[string]any // BusNode is the machine running the bus's module, and Loaded what its node agent says it loaded. @@ -166,7 +172,8 @@ func each(r Reader, state string, gaps *[]Gap) map[string][]byte { // ReadInputs reads what the shadow composes from. It never fails: what it cannot read is a gap. func ReadInputs(r Reader, cache *Manifests) Inputs { in := Inputs{Assigned: map[string][]string{}, Manifests: map[string]catalogue.Manifest{}, Digests: map[string]string{}, - Values: map[string]any{}, Gaps: append([]Gap(nil), standingGaps...)} + Running: map[string]catalogue.Manifest{}, + Values: map[string]any{}, Gaps: append([]Gap(nil), standingGaps...)} nodes := map[string]bool{} for _, raw := range each(r, ReadAssignments, &in.Gaps) { var a assignment @@ -251,6 +258,45 @@ func ReadInputs(r Reader, cache *Manifests) Inputs { Why: "assigned, and the indexer holds no manifest of it at the trunk (a module registered from outside the indexed repositories): composed with no authority"}) } } + // What each machine is wanted to run: the want's fingerprint, the builder's build of it, the manifest it built. + var fromTrunk []string + for key, raw := range each(r, ReadWants, &in.Gaps) { + node, module, err := delivery.ParseWantKey(key) + if err != nil || !slices.Contains(in.Assigned[node], module) { + continue + } + w, err := delivery.DecodeWant(raw) + if err != nil { + fromTrunk = append(fromTrunk, node+"."+module) + continue + } + if w.Removed || w.FP == "" { + continue + } + bk, err := delivery.BuilderBuildKey(module, w.FP) + if err != nil { + continue + } + braw, found, err := r.Get(ReadBuilds, bk) + if err != nil || !found { + fromTrunk = append(fromTrunk, node+"."+module) + continue + } + b, err := delivery.DecodeBuilderBuild(braw) + if err != nil || b.Manifest == "" { + fromTrunk = append(fromTrunk, node+"."+module) + continue + } + if m, err := cache.get(r, b.Manifest); err == nil { + in.Running[node+"."+module] = m + } else { + fromTrunk = append(fromTrunk, node+"."+module) + } + } + if len(fromTrunk) > 0 { + sort.Strings(fromTrunk) + in.Gaps = append(in.Gaps, Gap{Input: "wanted manifests", Why: "no build record names the manifest at the wanted fingerprint, so the trunk's is used: " + strings.Join(fromTrunk, ", ")}) + } for _, raw := range each(r, ReadResolved, &in.Gaps) { var v resolved if json.Unmarshal(raw, &v) != nil || v.Node == "" || v.Module == "" || v.Key == "" || len(v.Value) == 0 { diff --git a/modules/mesh-grants/internal/grants/memberships.go b/modules/mesh-grants/internal/grants/memberships.go index 4dfdddf..6f1691f 100644 --- a/modules/mesh-grants/internal/grants/memberships.go +++ b/modules/mesh-grants/internal/grants/memberships.go @@ -4,28 +4,40 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "fmt" + "reflect" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/membership" ) -// PublishedUnread is why the membership as published is not compared: the bus grants each module the membership of its -// own assignment alone, so no other part reads another's from the assignments stream. -const PublishedUnread = "not readable from here: the bus grants each module its own membership alone (and the controller has published none since release 6 removed the push, issue 1945 note 44)" +// Memberships are published for real from this milestone (the operator, 2026-10-25 22:31): since release 6 nobody +// publishes them, so a module placed since cannot serve its seat or reach a new state. The user list, the assertions and +// the grants stay in shadow until take-over. -// MembershipGaps are what a membership carries that the shadow cannot compose from the bus. +// MembershipGaps are what a membership carries that the part cannot compose from the bus. var MembershipGaps = []Gap{ - {Input: "membership: private network and received contributions", Why: "kept in each node's composition, which only that node reads: composed without them"}, + {Input: "membership: private network and received contributions", Why: "kept in each node's composition, which only that node reads: carried over from the membership last published, none for a first one"}, {Input: "membership: root-free machines", Why: "judged by the controller on the bus at its composition, not kept: a verified sender is composed as from no root-free machine"}, } -// Membership is one module's membership on one node, as composed now, and its digest. +// Publisher publishes a membership on the assignments stream (membership.Subject), last per subject, and reads the one +// last published there. A refusal (the grant not yet on the bus) is an error the part says and tries again at the next +// event. +type Publisher interface { + LastMembership(node, module string) (body []byte, seq uint64, found bool, err error) + PublishMembership(node, module string, body []byte, after uint64) error +} + +// Membership is one module's membership on one node, as composed now, and what became of its publishing. type Membership struct { Node string `json:"node"` Module string `json:"module"` Digest string `json:"digest"` Membership membership.Membership `json:"membership"` - Published string `json:"published"` + // Published says what became of it at the last pass: "published", "unchanged" (equal to the one last published), + // or why it was not. + Published string `json:"published"` } // Memberships composes every assignment's membership from the records (buscompose.MembershipFor, as the controller's @@ -36,11 +48,65 @@ func Memberships(r buscompose.Records) map[string]Membership { for node, declared := range r.Assigned { for _, d := range declared { m := buscompose.MembershipFor(node, d, where) - raw, _ := json.Marshal(m) - sum := sha256.Sum256(raw) - out[node+"."+d.Module] = Membership{Node: node, Module: d.Module, Digest: "sha256:" + hex.EncodeToString(sum[:]), - Membership: m, Published: PublishedUnread} + out[node+"."+d.Module] = Membership{Node: node, Module: d.Module, Digest: digestOf(m), Membership: m} } } return out } + +func digestOf(m membership.Membership) string { + raw, _ := json.Marshal(m) + sum := sha256.Sum256(raw) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// Published words of a membership's last pass. +const ( + MembershipPublished = "published" + MembershipUnchanged = "unchanged" +) + +// publish publishes m when it differs from the membership last published for its assignment, carrying that one's private +// network and received contributions; it answers m as published (or why not) and whether it was published now. +func publish(p Publisher, m Membership) (Membership, bool) { + if p == nil { + m.Published = "not published: no publisher here (only the part's process publishes)" + return m, false + } + was, seq, found, err := p.LastMembership(m.Node, m.Module) + if err != nil { + m.Published = "not published: the membership last published cannot be read: " + err.Error() + return m, false + } + if found { + var last membership.Membership + if json.Unmarshal(was, &last) == nil { + m.Membership.Mesh, m.Membership.Receives = last.Mesh, last.Receives + m.Digest = digestOf(m.Membership) + } + } + body, err := json.Marshal(m.Membership) + if err != nil { + m.Published = "not published: " + err.Error() + return m, false + } + if found && sameMembership(was, body) { + m.Published = MembershipUnchanged + return m, false + } + if err := p.PublishMembership(m.Node, m.Module, body, seq); err != nil { + m.Published = fmt.Sprintf("not published (tried again at the next event): %v", err) + return m, false + } + m.Published = MembershipPublished + return m, true +} + +// sameMembership says two memberships say the same, whatever the order of their fields. Pure. +func sameMembership(was, now []byte) bool { + var a, b any + if json.Unmarshal(was, &a) != nil || json.Unmarshal(now, &b) != nil { + return false + } + return reflect.DeepEqual(a, b) +} diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index a7ade17..d43d3d5 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -22,6 +22,9 @@ const ( StateTakeover = "takeover" KeyList = "list" KeySwitch = "switch" + // KeyPass is the last pass's result, and KeyAsked a tool's ask to compose again now, both in StateUserList. + KeyPass = "pass" + KeyAsked = "asked" ) // States is every state the part keeps. @@ -82,14 +85,16 @@ type Result struct { Written bool `json:"written"` Memberships int `json:"memberships"` MembershipsWritten int `json:"memberships_written"` + Published []string `json:"published,omitempty"` + NotPublished []string `json:"not_published,omitempty"` Assertions int `json:"assertions"` AssertionsWritten int `json:"assertions_written"` Errors []string `json:"errors,omitempty"` PublishedOrAsserted string `json:"published_or_asserted"` } -// NothingTakesEffect is what every composition says it did on the bus beyond its own state. -const NothingTakesEffect = "nothing: in shadow the grants part writes only its own state (no membership, assertion, grant or user list the bus node reads)" +// NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships. +const NothingTakesEffect = "memberships only (on the assignments stream, each when it differs from the one last published); the user list, assertions and grants stay in shadow until take-over" // Part is the grants part: it reads, calls and writes through these, and nothing else. type Part struct { @@ -97,7 +102,9 @@ type Part struct { Call Caller Store Store Emit Emitter - Now func() time.Time + // Publish publishes memberships: the process's connection; nil in the tools, which ask the process instead. + Publish Publisher + Now func() time.Time cache Manifests mu sync.Mutex @@ -155,11 +162,17 @@ func (p *Part) Compose() Result { ms := Memberships(recs) res.Memberships = len(ms) for _, key := range sortedKeys(ms) { - m := ms[key] + m, now := publish(p.Publish, ms[key]) + switch { + case now: + res.Published = append(res.Published, key) + p.say(&res, EventMembership, map[string]any{"node": m.Node, "module": m.Module, "digest": m.Digest, "published": true}) + case m.Published != MembershipUnchanged: + res.NotPublished = append(res.NotPublished, key+": "+m.Published) + } data, _ := json.Marshal(m) if p.changed(&res, StateMemberships, key, data) { res.MembershipsWritten++ - p.say(&res, EventMembership, map[string]any{"node": m.Node, "module": m.Module, "digest": m.Digest, "mode": Shadow}) } } held, heldErr := HeldStreams(p.Call) diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index 848a497..94beb82 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -39,14 +39,21 @@ func Records(in Inputs) (buscompose.Records, []Gap) { } for _, node := range in.Nodes { out.Nodes = append(out.Nodes, node) + manifestOf := func(module string) (catalogue.Manifest, bool) { + if m, ok := in.Running[node+"."+module]; ok { + return m, true + } + m, ok := in.Manifests[module] + return m, ok + } var here []catalogue.Manifest for _, module := range in.Assigned[node] { - if m, ok := in.Manifests[module]; ok { + if m, ok := manifestOf(module); ok { here = append(here, m) } } for _, module := range in.Assigned[node] { - m, known := in.Manifests[module] + m, known := manifestOf(module) if !known { gaps = append(gaps, Gap{Input: module + " on " + node, Why: "left out of the shadow list: no manifest to derive what it may say from"}) continue diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index 6bad0ba..249a8a3 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/modules/mesh-grants/internal/grants/tools.go @@ -59,7 +59,8 @@ func (p *Part) Grants(module string) (any, error) { return map[string]any{"writes": WritesNoGrant, "controller": out, "count": len(out)}, nil } -// MembershipOf answers the tool membership: one module on one node as composed now, as last kept, and as published. +// MembershipOf answers the tool membership: one module on one node as composed now, and as last published by the +// part's process (its state memberships says what became of each at the last pass). func (p *Part) MembershipOf(node, module string) (any, error) { if node == "" || module == "" { return nil, errors.New("membership names a node and a module") @@ -67,7 +68,7 @@ func (p *Part) MembershipOf(node, module string) (any, error) { in := ReadInputs(p.Read, &p.cache) recs, left := Records(in) m, ok := Memberships(recs)[node+"."+module] - out := map[string]any{"node": node, "module": module, "published": PublishedUnread, "gaps": MembershipGaps} + out := map[string]any{"node": node, "module": module, "gaps": MembershipGaps} if ok { out["composed"] = m } else { @@ -80,9 +81,36 @@ func (p *Part) MembershipOf(node, module string) (any, error) { out["composed"] = why } if raw, rev, found, err := p.Store.Get(StateMemberships, node+"."+module); err != nil { - out["kept"] = "not readable: " + err.Error() + out["published"] = "not readable: " + err.Error() } else if found { - out["kept"] = map[string]any{"revision": rev, "value": json.RawMessage(raw)} + out["published"] = map[string]any{"revision": rev, "last_pass": json.RawMessage(raw)} + } else { + out["published"] = "no pass of the part's process has composed it yet" + } + return out, nil +} + +// Pass composes and keeps the result as the last pass, for the tools to read. +func (p *Part) Pass() Result { + res := p.Compose() + data, _ := json.Marshal(res) + if _, err := p.Store.Put(StateUserList, KeyPass, data); err != nil { + res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be written: %v", StateUserList, KeyPass, err)) + } + return res +} + +// AskCompose is the tool compose-now: it asks the part's process to compose and publish now (only the process holds the +// grant to publish a membership), and answers the last pass, whose revision moves when the process answered. +func (p *Part) AskCompose(caller string) (any, error) { + ask, _ := json.Marshal(map[string]any{"by": caller, "at": p.now()}) + rev, err := p.Store.Put(StateUserList, KeyAsked, ask) + if err != nil { + return nil, fmt.Errorf("the ask could not be written: %w", err) + } + out := map[string]any{"asked": rev, "says": "the part's process composes and publishes at this ask; grants_user_list's last pass says what it published"} + if raw, prev, found, err := p.Store.Get(StateUserList, KeyPass); err == nil && found { + out["last_pass"] = map[string]any{"revision": prev, "result": json.RawMessage(raw)} } return out, nil } @@ -102,6 +130,9 @@ func (p *Part) UserList() (any, error) { return nil, fmt.Errorf("its own user list cannot be read: %w", err) } out["revision"], out["written"] = rev, st + if raw, prev, found, err := p.Store.Get(StateUserList, KeyPass); err == nil && found { + out["last_pass"] = map[string]any{"revision": prev, "result": json.RawMessage(raw)} + } if st.BusNode != "" { if raw, found, err := p.Read.Get(ReadHave, st.BusNode+"."+BusModule); err == nil && found { if h, err := delivery.DecodeHave(raw); err == nil { diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 1f35100..20526d3 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -6,6 +6,8 @@ "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.grants", + "mesh-delivery-assigner.wants", + "mesh-delivery-builder.builds", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "node-engine.have" @@ -29,10 +31,22 @@ "grants_hand_back" ], "state": [ - { "name": "assertions", "history": 5 }, - { "name": "memberships", "history": 5 }, - { "name": "takeover", "history": 20 }, - { "name": "user-list", "history": 20 } + { + "name": "assertions", + "history": 5 + }, + { + "name": "memberships", + "history": 5 + }, + { + "name": "takeover", + "history": 20 + }, + { + "name": "user-list", + "history": 20 + } ], "own-secrets": { "broker": "${dir:state}/broker"