From b4af9cdb7a2e37c296f5ef8c1aa02a8a65a77646 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 22:52:00 +0100 Subject: [PATCH] mesh-grants publishes only a first or an additive membership, and holds no lock across the bus (review of 1ff5049) Composed without a root-free judgement, a membership would drop a kind's verified-sender and the operator's channel approvals would stop being honoured; any loss now stays a shadow difference, named. A tool no longer waits on a pass's round trip. compose-now says the controller serves no verb to write a module's grant again. --- modules/mesh-grants/cmd/mesh-grants/tools.go | 9 +- modules/mesh-grants/grants_test.go | 128 ++++++++++++++---- modules/mesh-grants/internal/grants/inputs.go | 10 +- .../internal/grants/memberships.go | 83 ++++++++++++ modules/mesh-grants/internal/grants/part.go | 19 ++- modules/mesh-grants/internal/grants/tools.go | 8 +- 6 files changed, 223 insertions(+), 34 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 0ab9305..00e1aa5 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -67,7 +67,8 @@ func toolsOf(p *grants.Part) []stdio.Tool { 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, 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.", + "unchanged, or why not: a shadow difference names what publishing would lose), 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")) @@ -81,8 +82,10 @@ func toolsOf(p *grants.Part) []stdio.Tool { "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.", + "the ask and the last pass; grants_user_list shows the pass that answers it. A membership is published only when it is " + + "the first or only adds to the last one; any loss stays a shadow difference, named. The user list, assertions and " + + "grants stay in shadow. It cannot ask the controller to write a module's grant again: the controller serves no such " + + "verb, and the answer says so.", 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 " + diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 259165f..418dfc6 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -41,6 +41,8 @@ type fakeBus struct { seq uint64 publishes []string refusePub bool + // hold, when set, keeps every nats_users call waiting until it is closed: a pass waiting on the bus. + hold chan struct{} } func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) { @@ -111,6 +113,12 @@ func (b *fakeBus) Get(state, key string) ([]byte, bool, error) { } func (b *fakeBus) Call(key string, body any) (json.RawMessage, error) { + b.mu.Lock() + hold := b.hold + b.mu.Unlock() + if hold != nil && key == grants.CallUsers { + <-hold + } b.mu.Lock() defer b.mu.Unlock() b.calls = append(b.calls, key) @@ -438,11 +446,11 @@ func TestGrants(t *testing.T) { 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. + // mod-x placed on b too: b.mod-x is added and its first membership published; a.mod-x's would lose the module's + // plain subject (mod-x now runs on two machines), a loss, so it stays a shadow difference. b.assign("b", "mod-x") moved := p.Compose() - if !moved.Written || moved.Revision != 2 || !slices.Equal(moved.Published, []string{"a.mod-x", "b.mod-x"}) || moved.UserList.Compare.Equal() || + if !moved.Written || moved.Revision != 2 || !slices.Equal(moved.Published, []string{"b.mod-x"}) || len(moved.NotPublished) != 1 || !strings.HasPrefix(moved.NotPublished[0], "a.mod-x: "+grants.ShadowDifference) || moved.UserList.Compare.Equal() || !slices.Equal(moved.UserList.Compare.Added, []string{"b.mod-x"}) { t.Errorf("a composition after placement moved = %+v", moved) } @@ -551,29 +559,101 @@ 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) + t.Run("memberships: a first or an additive one published, a loss never", func(t *testing.T) { + const key = "a.mod-x" + composed := func(b *fakeBus) map[string]any { + recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{})) + raw, _ := json.Marshal(grants.Memberships(recs)[key].Membership) + var m map[string]any + _ = json.Unmarshal(raw, &m) + return m } - 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"]) + for _, row := range []struct { + name string + last func(map[string]any) map[string]any + published bool + names string + }{ + {"a first membership is published", nil, true, ""}, + {"an additive change is published, the private network carried", func(m map[string]any) map[string]any { + delete(m, "state") + m["mesh"] = []any{"10.0.0.1"} + return m + }, true, ""}, + {"a change that drops verified-sender is not published, and named", func(m map[string]any) map[string]any { + m["seat-traffic"] = map[string]any{"kinds": []any{map[string]any{"seat": "channel", "kind": "telegram", "capabilities": []any{"verified-sender"}}}} + return m + }, false, "verified-sender"}, + {"a change that drops any other field is not published", func(m map[string]any) map[string]any { + m["serves"] = append(m["serves"].([]any), map[string]any{"subject": "mesh.mod.mod-x.tool.gone"}) + return m + }, false, "tool.gone"}, + } { + t.Run(row.name, func(t *testing.T) { + b := fixture() + b.loaded = shadowOf(t, b) + if row.last != nil { + raw, _ := json.Marshal(row.last(composed(b))) + b.stream = map[string][]byte{"mesh.assignment." + key: raw} + b.seqs = map[string]uint64{"mesh.assignment." + key: 4} + } + res := partOn(b).Compose() + if got := slices.Contains(res.Published, key); got != row.published { + t.Fatalf("published %v, not published %v", res.Published, res.NotPublished) + } + if row.published { + if row.last != nil && !strings.Contains(string(b.stream["mesh.assignment."+key]), "10.0.0.1") { + t.Errorf("the private network of the last membership is not carried: %s", b.stream["mesh.assignment."+key]) + } + return + } + said := strings.Join(res.NotPublished, "\n") + if !strings.Contains(said, grants.ShadowDifference) || !strings.Contains(said, row.names) { + t.Errorf("the loss is not named: %s", said) + } + ms, _ := partOn(b).MembershipOf("a", "mod-x") + if !strings.Contains(fmt.Sprint(ms), row.names) { + t.Errorf("grants_membership does not name what would be lost: %v", ms) + } + }) } + t.Run("a refused publish: no exit, said, 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) + } + b.refusePub = false + if next := p.Compose(); !slices.Equal(next.Published, []string{"a.mod-x", "a.nats", "b.mod-y"}) { + t.Errorf("the next event published %v", next.Published) + } + }) + t.Run("a tool answers while a pass waits on the bus", func(t *testing.T) { + b := fixture() + b.loaded = shadowOf(t, b) + b.hold = make(chan struct{}) + p := partOn(b) + done := make(chan struct{}) + go func() { p.Compose(); close(done) }() + answered := make(chan struct{}) + go func() { + _, _ = p.UserList() + _, _ = p.MembershipOf("a", "mod-x") + _, _ = p.Grants("") + _ = p.Last() + close(answered) + }() + select { + case <-answered: + case <-time.After(10 * time.Second): + t.Error("a tool waited on the pass's round trip") + } + close(b.hold) + <-done + }) }) t.Run("a refused read or write never stops it", func(t *testing.T) { diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 5e65329..238639e 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -121,9 +121,11 @@ type Manifests struct { } func (c *Manifests) get(r Reader, digest string) (catalogue.Manifest, error) { + // The lock guards the map alone, never a read on the bus: a tool never waits on a pass's round trip. c.mu.Lock() - defer c.mu.Unlock() - if m, ok := c.parsed[digest]; ok { + m, ok := c.parsed[digest] + c.mu.Unlock() + if ok { return m, nil } raw, found, err := r.Get(ReadManifests, digest) @@ -137,10 +139,12 @@ func (c *Manifests) get(r Reader, digest string) (catalogue.Manifest, error) { if err != nil { return catalogue.Manifest{}, err } - m, err := catalogue.ParseManifest([]byte(v.JSON)) + m, err = catalogue.ParseManifest([]byte(v.JSON)) if err != nil { return catalogue.Manifest{}, fmt.Errorf("the manifest %s is refused by mesh-sdk's reader: %w", digest, err) } + c.mu.Lock() + defer c.mu.Unlock() if c.parsed == nil { c.parsed = map[string]catalogue.Manifest{} } diff --git a/modules/mesh-grants/internal/grants/memberships.go b/modules/mesh-grants/internal/grants/memberships.go index 6f1691f..6291542 100644 --- a/modules/mesh-grants/internal/grants/memberships.go +++ b/modules/mesh-grants/internal/grants/memberships.go @@ -6,6 +6,9 @@ import ( "encoding/json" "fmt" "reflect" + "slices" + "sort" + "strings" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/membership" @@ -94,6 +97,15 @@ func publish(p Publisher, m Membership) (Membership, bool) { m.Published = MembershipUnchanged return m, false } + // Only a first membership, or one that only adds to the last, is published (the lead's decision, review of 1ff5049): + // composed with no root-free judgement, a kind's verified-sender would be dropped, and the operator's approvals by + // a channel would no longer be honoured. Any loss stays a shadow difference, named. + if found { + if lost := Losses(was, body); len(lost) > 0 { + m.Published = ShadowDifference + strings.Join(lost, "; ") + 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 @@ -110,3 +122,74 @@ func sameMembership(was, now []byte) bool { } return reflect.DeepEqual(a, b) } + +// ShadowDifference opens what a membership's pass says when it was not published because it would lose something. +const ShadowDifference = "not published, a shadow difference: publishing would lose " + +// Losses names everything the last membership holds that now drops or weakens: a field gone, a value changed, a list +// element (a capability, a kind, a served address, a seat, a state, a stream) with no element in now that holds all of it. +// Empty means now holds all the last one did, and maybe more. Pure. +func Losses(last, now []byte) []string { + var a, b any + if json.Unmarshal(last, &a) != nil { + return nil + } + if json.Unmarshal(now, &b) != nil { + return []string{"the new membership is not JSON"} + } + return losses("", a, b) +} + +func losses(path string, last, now any) []string { + switch l := last.(type) { + case map[string]any: + n, ok := now.(map[string]any) + if !ok { + return []string{fmt.Sprintf("%s: %s", name(path), brief(last))} + } + keys := make([]string, 0, len(l)) + for k := range l { + keys = append(keys, k) + } + sort.Strings(keys) + var out []string + for _, k := range keys { + nv, ok := n[k] + if !ok { + out = append(out, fmt.Sprintf("%s: %s", name(path+"."+k), brief(l[k]))) + continue + } + out = append(out, losses(path+"."+k, l[k], nv)...) + } + return out + case []any: + n, _ := now.([]any) + var out []string + for i, e := range l { + if !slices.ContainsFunc(n, func(c any) bool { return len(losses("", e, c)) == 0 }) { + out = append(out, fmt.Sprintf("%s[%d]: %s", name(path), i, brief(e))) + } + } + return out + } + if !reflect.DeepEqual(last, now) { + return []string{fmt.Sprintf("%s: %s, now %s", name(path), brief(last), brief(now))} + } + return nil +} + +func name(path string) string { + if path == "" { + return "the membership" + } + return strings.TrimPrefix(path, ".") +} + +// brief is a value in JSON, cut at 200 bytes. +func brief(v any) string { + raw, _ := json.Marshal(v) + if len(raw) > 200 { + return string(raw[:200]) + "…" + } + return string(raw) +} diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index d43d3d5..4fc5361 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -94,7 +94,7 @@ type Result struct { } // 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" +const NothingTakesEffect = "memberships only (on the assignments stream, a first one or one that only adds to the last published; a loss stays a shadow difference); 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 { @@ -108,6 +108,7 @@ type Part struct { cache Manifests mu sync.Mutex + last Result } func (p *Part) now() time.Time { @@ -121,9 +122,14 @@ func (p *Part) now() time.Time { // writes each of its own keys whose value changed, saying each change. It never stops at a refusal: a read refused is // a gap, a write refused is said in Errors, and the next event composes again. func (p *Part) Compose() Result { - p.mu.Lock() - defer p.mu.Unlock() + // No lock is held across a round trip on the bus: the reads, calls, publishes and writes below take none, and the + // lock is taken only to keep the result, so a tool answers while a pass waits on the bus. res := Result{PublishedOrAsserted: NothingTakesEffect} + defer func() { + p.mu.Lock() + p.last = res + p.mu.Unlock() + }() in := ReadInputs(p.Read, &p.cache) recs, left := Records(in) st := UserListState{V: 1, Mode: Shadow, LeftOut: left, Loaded: in.Loaded, BusNode: in.BusNode, @@ -196,6 +202,13 @@ func (p *Part) Compose() Result { return res } +// Last is the result of the last pass this part made, kept under the lock alone. +func (p *Part) Last() Result { + p.mu.Lock() + defer p.mu.Unlock() + return p.last +} + func assigned(in Inputs, module string) bool { for _, ms := range in.Assigned { for _, m := range ms { diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index 249a8a3..adee678 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/modules/mesh-grants/internal/grants/tools.go @@ -100,6 +100,11 @@ func (p *Part) Pass() Result { return res } +// NoGrantAgain is what compose-now says of the controller's grants: the controller serves no verb that writes a module's +// grant again (its issue replaces the account, which is not that), so a grant write it lost stays lost until it composes +// again; the part writes grants only from ADR 0547's third milestone. +const NoGrantAgain = "missing: a verb to ask the controller to write a module's grant again (mesh-controller serves none; its issue replaces the bus account, which is not that), so compose-now cannot ask it" + // 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) { @@ -108,7 +113,8 @@ func (p *Part) AskCompose(caller string) (any, error) { 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"} + 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", + "controller_grants": NoGrantAgain} 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)} }