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.
This commit is contained in:
@@ -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 " +
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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{}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user