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.
This commit is contained in:
@@ -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
|
// Command mesh-grants is the grants part (novox/hq ADR 0547 clauses 5 to 8, issue 1945): a process on its own bus
|
||||||
// milestone: a process on its own bus credential that watches placement, the indexer's manifests, mesh-config's
|
// 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 the bus's user list, each assignment's
|
// resolved values and the node agents' haves, and on every change composes each assignment's membership, the bus's user
|
||||||
// membership and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk
|
// list and each manifest's buckets, logs and streams with the bus composer the controller uses (mesh-sdk buscompose).
|
||||||
// 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.
|
// **Memberships are published for real** (the operator, 2026-10-25 22:31): since release 6 nobody publishes them, so a
|
||||||
// It holds no timer, and never exits for a refused read or write: a refusal is said and tried again at the next event.
|
// 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
|
// **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
|
// 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).
|
// Started by the tool runner, with no credential of its own, it serves those tools (a process never serves, ADR 0436).
|
||||||
package main
|
package main
|
||||||
@@ -27,6 +34,7 @@ import (
|
|||||||
"github.com/nats-io/nats.go"
|
"github.com/nats-io/nats.go"
|
||||||
"github.com/nats-io/nats.go/jetstream"
|
"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/announce"
|
||||||
"git.novox.be/novox/mesh-sdk/go/process"
|
"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
|
return err
|
||||||
}
|
}
|
||||||
bus := newBusReader(ctx, js)
|
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
|
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)
|
f.Run(ctx)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -66,8 +66,8 @@ func toolsOf(p *grants.Part) []stdio.Tool {
|
|||||||
Input: schema(map[string]string{"module": "one module"}),
|
Input: schema(map[string]string{"module": "one module"}),
|
||||||
Run: func(a map[string]any) (any, error) { return p.Grants(grants.Answer(a, "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 " +
|
{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 " +
|
"bus's records with the bus composer, and what the part's process published of it at its last pass (published, " +
|
||||||
"grants each module its own alone), with what the composition cannot read. Read-only.",
|
"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"),
|
Input: schema(map[string]string{"node": "the machine", "module": "the module"}, "node", "module"),
|
||||||
Run: func(a map[string]any) (any, error) {
|
Run: func(a map[string]any) (any, error) {
|
||||||
return p.MembershipOf(grants.Answer(a, "node"), grants.Answer(a, "module"))
|
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 " +
|
"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.",
|
"assignments left out, and every input the bus does not carry, named. Read-only.",
|
||||||
Run: func(map[string]any) (any, error) { return p.UserList() }},
|
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 " +
|
{Name: grants.ToolComposeNow, Description: "Compose again now (ADR 0547 clause 6), as the part does on every change it hears: asks " +
|
||||||
"user list, the memberships and the assertions, compared with what the bus loaded and holds; writes its own state only " +
|
"the part's process (which alone holds the grant to publish a membership) to compose the memberships, the user list and " +
|
||||||
"where a value changed (one revision when an input changed, none when not) and says user-list-changed and " +
|
"the assertions, publish each membership that differs from the one last published, and keep what changed in its own " +
|
||||||
"membership-changed for each. In shadow it publishes, asserts and grants nothing.",
|
"state (one revision when an input changed, none when not), saying user-list-changed and membership-changed. Answers " +
|
||||||
Run: func(map[string]any) (any, error) { return p.Compose(), nil }},
|
"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 " +
|
{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 " +
|
"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.",
|
"asserts nothing, and says so.",
|
||||||
|
|||||||
@@ -2,11 +2,13 @@ module mesh-grants
|
|||||||
|
|
||||||
go 1.26.0
|
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 (
|
require (
|
||||||
github.com/klauspost/compress v1.20.0 // indirect
|
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/nkeys v0.4.16 // indirect
|
||||||
github.com/nats-io/nuid v1.0.1 // indirect
|
github.com/nats-io/nuid v1.0.1 // indirect
|
||||||
golang.org/x/crypto v0.57.0 // indirect
|
golang.org/x/crypto v0.57.0 // indirect
|
||||||
|
|||||||
@@ -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.20261025213320-881da7f59995 h1:jY8ufarHcRyVCc2snyMk9eTmMkM/pnLC1cAqb4/IC5A=
|
||||||
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/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||||
|
|||||||
@@ -35,6 +35,39 @@ type fakeBus struct {
|
|||||||
calls []string
|
calls []string
|
||||||
events []string
|
events []string
|
||||||
writes []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 {
|
func newFakeBus() *fakeBus {
|
||||||
@@ -187,7 +220,7 @@ func fixture() *fakeBus {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func partOn(b *fakeBus) *grants.Part {
|
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.
|
// 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",
|
forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc",
|
||||||
".Publish(", "PublishMsg", "CreateStream", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue",
|
"PublishMsg", "CreateStream", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue",
|
||||||
"CreateObjectStore", "mesh.assignment", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
|
"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
|
files := 0
|
||||||
for _, root := range []string{"cmd", "internal"} {
|
for _, root := range []string{"cmd", "internal"} {
|
||||||
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
||||||
@@ -266,9 +301,15 @@ func TestGrants(t *testing.T) {
|
|||||||
}
|
}
|
||||||
files++
|
files++
|
||||||
src, err := os.ReadFile(path)
|
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 {
|
for _, f := range forbidden {
|
||||||
if strings.Contains(string(src), f) {
|
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
|
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"},
|
[]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") },
|
{"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"},
|
[]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 },
|
{"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},
|
[]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) {
|
if !gapNamed(append(in.Gaps, left...), row.gap) {
|
||||||
t.Errorf("gaps %v name no %q", in.Gaps, 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()
|
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 {
|
if !first.Written || first.Revision != 1 || !first.UserList.Compare.Equal() || first.UserList.Users != 6 {
|
||||||
t.Fatalf("first composition = %+v", first)
|
t.Fatalf("first composition = %+v", first)
|
||||||
}
|
}
|
||||||
if again := p.Compose(); again.Written || again.Revision != 1 || again.MembershipsWritten != 0 || again.AssertionsWritten != 0 {
|
if len(first.Published) != 3 || len(b.publishes) != 3 {
|
||||||
t.Errorf("a composition with no input changed wrote: %+v", again)
|
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
|
// 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.
|
// machines and answers only on its machines' subjects.
|
||||||
b.assign("b", "mod-x")
|
b.assign("b", "mod-x")
|
||||||
moved := p.Compose()
|
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"}) {
|
!slices.Equal(moved.UserList.Compare.Added, []string{"b.mod-x"}) {
|
||||||
t.Errorf("a composition after placement moved = %+v", moved)
|
t.Errorf("a composition after placement moved = %+v", moved)
|
||||||
}
|
}
|
||||||
@@ -405,7 +468,7 @@ func TestGrants(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
ms, _ := p.MembershipOf("a", "mod-x")
|
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)
|
t.Errorf("membership = %v", ms)
|
||||||
}
|
}
|
||||||
gr, _ := p.Grants("mod-x")
|
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 {
|
for _, w := range b.writes {
|
||||||
if !slices.Contains(grants.States, w) {
|
if !slices.Contains(grants.States, w) {
|
||||||
t.Errorf("wrote %s, which is not the part's own state", 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) {
|
t.Run("a refused read or write never stops it", func(t *testing.T) {
|
||||||
b := fixture()
|
b := fixture()
|
||||||
b.loaded = shadowOf(t, b)
|
b.loaded = shadowOf(t, b)
|
||||||
|
|||||||
@@ -15,8 +15,9 @@ type Watcher interface {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
|
// 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).
|
// config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's
|
||||||
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave}
|
// 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
|
// 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).
|
// 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
|
Part *Part
|
||||||
Watch Watcher
|
Watch Watcher
|
||||||
Logf func(string, ...any)
|
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{}
|
wake chan struct{}
|
||||||
once sync.Once
|
once sync.Once
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
|
asked bool
|
||||||
watched map[string]bool
|
watched map[string]bool
|
||||||
refused map[string]string
|
refused map[string]string
|
||||||
have 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.
|
// open opens every watch not open yet; one refused is said and left for the next wake.
|
||||||
func (f *Follower) open() {
|
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 {
|
for _, state := range Followed {
|
||||||
f.mu.Lock()
|
f.mu.Lock()
|
||||||
open := f.watched[state]
|
open := f.watched[state]
|
||||||
@@ -141,10 +152,16 @@ func (f *Follower) Run(ctx context.Context) {
|
|||||||
return
|
return
|
||||||
case <-f.wake:
|
case <-f.wake:
|
||||||
f.open()
|
f.open()
|
||||||
res := f.Part.Compose()
|
res := f.Part.Pass()
|
||||||
for _, e := range res.Errors {
|
for _, e := range res.Errors {
|
||||||
f.logf("%s", e)
|
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 {
|
if res.Written {
|
||||||
f.logf("the shadow list moved to revision %d: %d users, %d differing from the loaded list",
|
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()))
|
res.Revision, res.UserList.Users, len(res.UserList.Compare.Differing()))
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ package grants
|
|||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"slices"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -25,10 +26,12 @@ const (
|
|||||||
ReadManifests = "mesh-delivery-indexer.manifests"
|
ReadManifests = "mesh-delivery-indexer.manifests"
|
||||||
ReadHave = "node-engine.have"
|
ReadHave = "node-engine.have"
|
||||||
ReadResolved = "mesh-config.resolved"
|
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.
|
// 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.
|
// The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
|
||||||
const (
|
const (
|
||||||
@@ -97,6 +100,9 @@ type Inputs struct {
|
|||||||
Manifests map[string]catalogue.Manifest
|
Manifests map[string]catalogue.Manifest
|
||||||
// Digests is each module's manifest digest, by module.
|
// Digests is each module's manifest digest, by module.
|
||||||
Digests map[string]string
|
Digests map[string]string
|
||||||
|
// Running is the manifest a machine is wanted to run, by `<node>.<module>`: 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 `<node>/<module>/<key>`.
|
// Values is each resolved config value, by `<node>/<module>/<key>`.
|
||||||
Values map[string]any
|
Values map[string]any
|
||||||
// BusNode is the machine running the bus's module, and Loaded what its node agent says it loaded.
|
// 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.
|
// ReadInputs reads what the shadow composes from. It never fails: what it cannot read is a gap.
|
||||||
func ReadInputs(r Reader, cache *Manifests) Inputs {
|
func ReadInputs(r Reader, cache *Manifests) Inputs {
|
||||||
in := Inputs{Assigned: map[string][]string{}, Manifests: map[string]catalogue.Manifest{}, Digests: map[string]string{},
|
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{}
|
nodes := map[string]bool{}
|
||||||
for _, raw := range each(r, ReadAssignments, &in.Gaps) {
|
for _, raw := range each(r, ReadAssignments, &in.Gaps) {
|
||||||
var a assignment
|
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"})
|
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) {
|
for _, raw := range each(r, ReadResolved, &in.Gaps) {
|
||||||
var v resolved
|
var v resolved
|
||||||
if json.Unmarshal(raw, &v) != nil || v.Node == "" || v.Module == "" || v.Key == "" || len(v.Value) == 0 {
|
if json.Unmarshal(raw, &v) != nil || v.Node == "" || v.Module == "" || v.Key == "" || len(v.Value) == 0 {
|
||||||
|
|||||||
@@ -4,28 +4,40 @@ import (
|
|||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"reflect"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
"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
|
// Memberships are published for real from this milestone (the operator, 2026-10-25 22:31): since release 6 nobody
|
||||||
// own assignment alone, so no other part reads another's from the assignments stream.
|
// publishes them, so a module placed since cannot serve its seat or reach a new state. The user list, the assertions and
|
||||||
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)"
|
// 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{
|
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"},
|
{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 {
|
type Membership struct {
|
||||||
Node string `json:"node"`
|
Node string `json:"node"`
|
||||||
Module string `json:"module"`
|
Module string `json:"module"`
|
||||||
Digest string `json:"digest"`
|
Digest string `json:"digest"`
|
||||||
Membership membership.Membership `json:"membership"`
|
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
|
// 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 node, declared := range r.Assigned {
|
||||||
for _, d := range declared {
|
for _, d := range declared {
|
||||||
m := buscompose.MembershipFor(node, d, where)
|
m := buscompose.MembershipFor(node, d, where)
|
||||||
raw, _ := json.Marshal(m)
|
out[node+"."+d.Module] = Membership{Node: node, Module: d.Module, Digest: digestOf(m), Membership: m}
|
||||||
sum := sha256.Sum256(raw)
|
|
||||||
out[node+"."+d.Module] = Membership{Node: node, Module: d.Module, Digest: "sha256:" + hex.EncodeToString(sum[:]),
|
|
||||||
Membership: m, Published: PublishedUnread}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return out
|
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)
|
||||||
|
}
|
||||||
|
|||||||
@@ -22,6 +22,9 @@ const (
|
|||||||
StateTakeover = "takeover"
|
StateTakeover = "takeover"
|
||||||
KeyList = "list"
|
KeyList = "list"
|
||||||
KeySwitch = "switch"
|
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.
|
// States is every state the part keeps.
|
||||||
@@ -82,14 +85,16 @@ type Result struct {
|
|||||||
Written bool `json:"written"`
|
Written bool `json:"written"`
|
||||||
Memberships int `json:"memberships"`
|
Memberships int `json:"memberships"`
|
||||||
MembershipsWritten int `json:"memberships_written"`
|
MembershipsWritten int `json:"memberships_written"`
|
||||||
|
Published []string `json:"published,omitempty"`
|
||||||
|
NotPublished []string `json:"not_published,omitempty"`
|
||||||
Assertions int `json:"assertions"`
|
Assertions int `json:"assertions"`
|
||||||
AssertionsWritten int `json:"assertions_written"`
|
AssertionsWritten int `json:"assertions_written"`
|
||||||
Errors []string `json:"errors,omitempty"`
|
Errors []string `json:"errors,omitempty"`
|
||||||
PublishedOrAsserted string `json:"published_or_asserted"`
|
PublishedOrAsserted string `json:"published_or_asserted"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// NothingTakesEffect is what every composition says it did on the bus beyond its own state.
|
// NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships.
|
||||||
const NothingTakesEffect = "nothing: in shadow the grants part writes only its own state (no membership, assertion, grant or user list the bus node reads)"
|
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.
|
// Part is the grants part: it reads, calls and writes through these, and nothing else.
|
||||||
type Part struct {
|
type Part struct {
|
||||||
@@ -97,7 +102,9 @@ type Part struct {
|
|||||||
Call Caller
|
Call Caller
|
||||||
Store Store
|
Store Store
|
||||||
Emit Emitter
|
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
|
cache Manifests
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
@@ -155,11 +162,17 @@ func (p *Part) Compose() Result {
|
|||||||
ms := Memberships(recs)
|
ms := Memberships(recs)
|
||||||
res.Memberships = len(ms)
|
res.Memberships = len(ms)
|
||||||
for _, key := range sortedKeys(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)
|
data, _ := json.Marshal(m)
|
||||||
if p.changed(&res, StateMemberships, key, data) {
|
if p.changed(&res, StateMemberships, key, data) {
|
||||||
res.MembershipsWritten++
|
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)
|
held, heldErr := HeldStreams(p.Call)
|
||||||
|
|||||||
@@ -39,14 +39,21 @@ func Records(in Inputs) (buscompose.Records, []Gap) {
|
|||||||
}
|
}
|
||||||
for _, node := range in.Nodes {
|
for _, node := range in.Nodes {
|
||||||
out.Nodes = append(out.Nodes, node)
|
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
|
var here []catalogue.Manifest
|
||||||
for _, module := range in.Assigned[node] {
|
for _, module := range in.Assigned[node] {
|
||||||
if m, ok := in.Manifests[module]; ok {
|
if m, ok := manifestOf(module); ok {
|
||||||
here = append(here, m)
|
here = append(here, m)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for _, module := range in.Assigned[node] {
|
for _, module := range in.Assigned[node] {
|
||||||
m, known := in.Manifests[module]
|
m, known := manifestOf(module)
|
||||||
if !known {
|
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"})
|
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
|
continue
|
||||||
|
|||||||
@@ -59,7 +59,8 @@ func (p *Part) Grants(module string) (any, error) {
|
|||||||
return map[string]any{"writes": WritesNoGrant, "controller": out, "count": len(out)}, nil
|
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) {
|
func (p *Part) MembershipOf(node, module string) (any, error) {
|
||||||
if node == "" || module == "" {
|
if node == "" || module == "" {
|
||||||
return nil, errors.New("membership names a node and a 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)
|
in := ReadInputs(p.Read, &p.cache)
|
||||||
recs, left := Records(in)
|
recs, left := Records(in)
|
||||||
m, ok := Memberships(recs)[node+"."+module]
|
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 {
|
if ok {
|
||||||
out["composed"] = m
|
out["composed"] = m
|
||||||
} else {
|
} else {
|
||||||
@@ -80,9 +81,36 @@ func (p *Part) MembershipOf(node, module string) (any, error) {
|
|||||||
out["composed"] = why
|
out["composed"] = why
|
||||||
}
|
}
|
||||||
if raw, rev, found, err := p.Store.Get(StateMemberships, node+"."+module); err != nil {
|
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 {
|
} 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
|
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)
|
return nil, fmt.Errorf("its own user list cannot be read: %w", err)
|
||||||
}
|
}
|
||||||
out["revision"], out["written"] = rev, st
|
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 st.BusNode != "" {
|
||||||
if raw, found, err := p.Read.Get(ReadHave, st.BusNode+"."+BusModule); err == nil && found {
|
if raw, found, err := p.Read.Get(ReadHave, st.BusNode+"."+BusModule); err == nil && found {
|
||||||
if h, err := delivery.DecodeHave(raw); err == nil {
|
if h, err := delivery.DecodeHave(raw); err == nil {
|
||||||
|
|||||||
@@ -6,6 +6,8 @@
|
|||||||
"mesh-config.resolved",
|
"mesh-config.resolved",
|
||||||
"mesh-controller.assignments",
|
"mesh-controller.assignments",
|
||||||
"mesh-controller.grants",
|
"mesh-controller.grants",
|
||||||
|
"mesh-delivery-assigner.wants",
|
||||||
|
"mesh-delivery-builder.builds",
|
||||||
"mesh-delivery-indexer.manifests",
|
"mesh-delivery-indexer.manifests",
|
||||||
"mesh-delivery-indexer.modules",
|
"mesh-delivery-indexer.modules",
|
||||||
"node-engine.have"
|
"node-engine.have"
|
||||||
@@ -29,10 +31,22 @@
|
|||||||
"grants_hand_back"
|
"grants_hand_back"
|
||||||
],
|
],
|
||||||
"state": [
|
"state": [
|
||||||
{ "name": "assertions", "history": 5 },
|
{
|
||||||
{ "name": "memberships", "history": 5 },
|
"name": "assertions",
|
||||||
{ "name": "takeover", "history": 20 },
|
"history": 5
|
||||||
{ "name": "user-list", "history": 20 }
|
},
|
||||||
|
{
|
||||||
|
"name": "memberships",
|
||||||
|
"history": 5
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"name": "takeover",
|
||||||
|
"history": 20
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"name": "user-list",
|
||||||
|
"history": 20
|
||||||
|
}
|
||||||
],
|
],
|
||||||
"own-secrets": {
|
"own-secrets": {
|
||||||
"broker": "${dir:state}/broker"
|
"broker": "${dir:state}/broker"
|
||||||
|
|||||||
Reference in New Issue
Block a user