mesh-grants: every node's engine writes its own compositions and the named readers read them all
ADR 0593 §6, §7: the grants part creates composed-<node> per node, grants each engine the writes of its own bucket and a direct get of the indexer's candidates for compose-candidate, and grants mesh-host.composed read-only, with the stream listing and creation advisory a per-node read needs, to the provisioner, the assigner and the view alone; memberships issue it per node. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -9,6 +9,7 @@ import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -31,9 +32,12 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
|
||||
Assigned: map[string][]buscompose.Declared{
|
||||
"novox": {{Module: "nats", SnapshotsTheBus: true},
|
||||
{Module: buscompose.GrantsModule, State: []buscompose.Bucket{{Module: buscompose.GrantsModule, Name: "user-list"}}}},
|
||||
"ace": {{Module: "plex"}},
|
||||
"ace": {{Module: "plex", Reads: []string{delivery.ComposedRead}},
|
||||
{Module: "mesh-delivery-view", Reads: []string{delivery.ComposedRead}}},
|
||||
},
|
||||
}
|
||||
// The node-engine's state as the part's records carry it: every node's compositions (ADR 0593 §6).
|
||||
records.Engine = &buscompose.Declared{Module: buscompose.EngineModule, State: buscompose.EngineComposedBuckets(records.Nodes)}
|
||||
users, err := buscompose.Users(records)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -212,4 +216,44 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
|
||||
if _, err := other.GetLastMsg("KV_"+buscompose.GrantsUserListBucket, "$KV."+buscompose.GrantsUserListBucket+"."+buscompose.GrantsUserListKey, nats.DirectGet()); err == nil {
|
||||
t.Errorf("another node read the key: %v", err)
|
||||
}
|
||||
// ADR 0593 §6 (D14): the grants part creates each node's compositions; ace's engine writes its own and not novox's;
|
||||
// the view, a named reader, finds every node's bucket by listing and reads it, writing nothing; plex reads none.
|
||||
for _, node := range records.Nodes {
|
||||
cfg := buscompose.BucketStreamConfig(buscompose.EngineComposedBuckets([]string{node})[0])
|
||||
if _, err := grants.CreateStream(ctx, cfg); err != nil {
|
||||
t.Fatalf("the grants part creates %s: %v", cfg.Name, err)
|
||||
}
|
||||
}
|
||||
aceEngine, _ := dial("node.ace").JetStream(nats.MaxWait(2 * time.Second))
|
||||
if _, err := aceEngine.Publish("$KV.composed-ace.ace.node", []byte(`{}`)); err != nil {
|
||||
t.Errorf("ace's engine cannot write its own compositions: %v", err)
|
||||
}
|
||||
if _, err := aceEngine.Publish("$KV.composed-novox.novox.node", []byte(`{}`)); err == nil {
|
||||
t.Error("ace's engine wrote novox's compositions")
|
||||
}
|
||||
view, err := jetstream.New(dial("ace.mesh-delivery-view"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var found []string
|
||||
for name := range view.KeyValueStoreNames(ctx).Name() {
|
||||
found = append(found, name)
|
||||
}
|
||||
if !slices.Contains(found, "composed-ace") || !slices.Contains(found, "composed-novox") {
|
||||
t.Errorf("the view finds the buckets %v", found)
|
||||
}
|
||||
composed, err := view.KeyValue(ctx, "composed-ace")
|
||||
if err != nil {
|
||||
t.Fatalf("the view binds ace's compositions: %v", err)
|
||||
}
|
||||
if _, err := composed.Get(ctx, "ace.node"); err != nil {
|
||||
t.Errorf("the view reads ace's own composition: %v", err)
|
||||
}
|
||||
if _, err := composed.Put(ctx, "ace.node", []byte(`{}`)); err == nil {
|
||||
t.Error("the view wrote a composition")
|
||||
}
|
||||
plex, _ := dial("ace.plex").JetStream(nats.MaxWait(2 * time.Second))
|
||||
if _, err := plex.GetLastMsg("KV_composed-ace", "$KV.composed-ace.ace.node", nats.DirectGet()); err == nil {
|
||||
t.Error("plex, not a named reader, read a composition")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1692,10 +1692,22 @@ func TestGrants(t *testing.T) {
|
||||
if !slices.Contains(b.creates, "KV_"+buscompose.EngineActsBucket) {
|
||||
t.Fatalf("the node-engine's acts were not created: %v", b.creates)
|
||||
}
|
||||
// ADR 0593 §6: and every node's bucket of compositions, composed-<node>, keeping five revisions, which its engine
|
||||
// writes and never creates.
|
||||
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
|
||||
if recs.Engine == nil || len(recs.Engine.State) != 1 || recs.Engine.State[0].Bucket() != buscompose.EngineActsBucket {
|
||||
if recs.Engine == nil || len(recs.Engine.State) != 1+len(recs.Nodes) || recs.Engine.State[0].Bucket() != buscompose.EngineActsBucket {
|
||||
t.Fatalf("the records carry the node-engine as %+v", recs.Engine)
|
||||
}
|
||||
for _, node := range recs.Nodes {
|
||||
bucket, _ := delivery.ComposedBucketFor(node)
|
||||
if !slices.Contains(b.creates, "KV_"+bucket) {
|
||||
t.Errorf("the compositions of %s were not created: %v", node, b.creates)
|
||||
}
|
||||
engine := catalogue.Manifest{Module: buscompose.EngineModule}
|
||||
if cfg := grants.Configs(engine, recs.Nodes)["KV_"+bucket]; cfg.MaxMsgsPerSubject != delivery.ComposedHistory {
|
||||
t.Errorf("the compositions of %s keep %d revisions, want %d", node, cfg.MaxMsgsPerSubject, delivery.ComposedHistory)
|
||||
}
|
||||
}
|
||||
for node, mods := range recs.Assigned {
|
||||
for _, d := range mods {
|
||||
if d.Module == buscompose.EngineModule {
|
||||
|
||||
@@ -48,6 +48,7 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
|
||||
// The creates the grants part makes from it (issue 1790), as subtests of the one test (the test budget).
|
||||
t.Run("only declared streams are created", theGrantsPartCreatesOnlyDeclaredStreams)
|
||||
t.Run("a declared stream is created as the controller asserted it", declaredStreamsAreCreatedAsTheControllerAssertsThem)
|
||||
t.Run("every node's compositions are read by their named readers alone and written by their node", everyNodesCompositions)
|
||||
// ADR 0567 §3, §9: a node reads every key of the seat's placement and its candidates (it composes with the whole
|
||||
// proposal laid over), and none of its open proposals; it writes neither.
|
||||
reads := strings.Join(NodeComposeReads("laptop"), " ")
|
||||
@@ -881,3 +882,52 @@ func healthChecks(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ADR 0593 §6 (D14): a named reader of delivery.ComposedRead reads every node's composed-<node>, read-only, with the
|
||||
// listing and the creation advisory that find them, and its membership issues the family by its prefix; a module not
|
||||
// named is refused it; each node's engine writes its own node's bucket alone.
|
||||
func everyNodesCompositions(t *testing.T) {
|
||||
records := Records{Nodes: []string{"one", "two"}, Assigned: map[string][]Declared{
|
||||
"one": {{Module: "mesh-delivery-view", Reads: []string{delivery.ComposedRead}},
|
||||
{Module: "plex", Reads: []string{delivery.ComposedRead}}},
|
||||
}}
|
||||
users, err := Users(records)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
perms := map[string]Permissions{}
|
||||
for _, u := range users {
|
||||
p, err := PermissionsFor(u)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
perms[u.Username()] = p
|
||||
}
|
||||
view, plex, one := perms["one.mesh-delivery-view"], perms["one.plex"], perms["node.one"]
|
||||
for _, node := range []string{"one", "two"} {
|
||||
bucket := "composed-" + node
|
||||
if !MayPublish(view, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+node+".node") ||
|
||||
!MayPublish(view, "$JS.API.CONSUMER.CREATE.KV_"+bucket+".x.$KV."+bucket+".>") {
|
||||
t.Errorf("the view does not read %s", bucket)
|
||||
}
|
||||
if MayPublish(view, "$KV."+bucket+"."+node+".node") || MayPublish(plex, "$JS.API.DIRECT.GET.KV_"+bucket+".x") {
|
||||
t.Errorf("the view writes %s, or a module not named reads it", bucket)
|
||||
}
|
||||
if MayPublish(one, "$KV."+bucket+"."+node+".node") != (node == "one") {
|
||||
t.Errorf("node one writes %s: %v", bucket, node != "one")
|
||||
}
|
||||
}
|
||||
if !MayPublish(view, "$JS.API.STREAM.NAMES") || !MaySubscribe(view, "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_composed-three") ||
|
||||
MayPublish(plex, "$JS.API.STREAM.NAMES") {
|
||||
t.Error("the view cannot find a node's bucket, or a module not named lists the streams")
|
||||
}
|
||||
if MayPublish(view, "$JS.API.DIRECT.GET.KV_mesh-host_composed.x") {
|
||||
t.Error("the read is granted as one bucket mesh-host_composed")
|
||||
}
|
||||
m := MembershipFor("one", Declared{Module: "mesh-delivery-view", Reads: []string{delivery.ComposedRead}}, Placements{})
|
||||
if !slices.ContainsFunc(m.State, func(s StateIssued) bool {
|
||||
return s.Name == delivery.ComposedRead && s.PerNode && s.Bucket == delivery.ComposedBucketPrefix && !s.Writes
|
||||
}) {
|
||||
t.Errorf("the view's membership issues %+v", m.State)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,13 +6,16 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The compositions' buckets (novox/hq ADR 0539 clauses 2 and 4): one per node, `composed-<node>`, holding the
|
||||
// compositions the ended controller made for that node, which only that node's node-engine reads. Nothing writes them
|
||||
// since the controller ended (ADR 0590); a node composes its own modules (ADR 0547 §2).
|
||||
// The compositions' buckets (novox/hq ADR 0539 clauses 2 and 4, as ADR 0593 §6 changes them): one per node,
|
||||
// `composed-<node>`, which that node's engine alone writes, unsigned, with every composition it makes, and which the
|
||||
// grants part creates (EngineComposedBuckets). The provisioner, the assigner and the view read every node's
|
||||
// (delivery.ComposedRead, a named-reader state); no engine applies a value read from it.
|
||||
|
||||
// ComposedReads is what a node's node-engine publishes to follow its wants and compositions (ADR 0539 clauses 1, 2 and
|
||||
// 9): every key of its own composition bucket, and its own wants (`want.<node>.<module>` and `want.<node>.node`) in the
|
||||
// assigner's bucket; nothing of another node's, nor the bucket of digests.
|
||||
// ComposedReads is what a node's node-engine publishes to keep and follow its compositions and wants (ADR 0593 §6, ADR
|
||||
// 0539 clause 9): the writes of its own composition bucket's keys (`<node>.<module>.<fp>` and `<node>.node`, all under
|
||||
// `<node>.`, so the bucket's whole) and their read, its own wants (`want.<node>.<module>` and `want.<node>.node`) in the
|
||||
// assigner's bucket, and the indexer's candidates by one key's direct get (ADR 0593 §7: `compose-candidate` reads the
|
||||
// check's candidate itself; no listing and no watch); nothing of another node's, never the bucket's creation.
|
||||
func ComposedReads(node string) ([]string, error) {
|
||||
bucket, err := delivery.ComposedBucketFor(node)
|
||||
if err != nil {
|
||||
@@ -21,9 +24,50 @@ func ComposedReads(node string) ([]string, error) {
|
||||
if !safeSubject.MatchString(node) {
|
||||
return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node)
|
||||
}
|
||||
return append(readGrants(bucket, ""), readGrants(delivery.AssignerWantsBucket, "want."+node+".>")...), nil
|
||||
out := append([]string{"$KV." + bucket + ".>"}, readGrants(bucket, "")...)
|
||||
out = append(out, readGrants(delivery.AssignerWantsBucket, "want."+node+".>")...)
|
||||
return append(out, CandidateReads...), nil
|
||||
}
|
||||
|
||||
// CandidateReads is what a node's engine is granted of the indexer's candidates to answer `compose-candidate` (ADR 0593
|
||||
// §7): binding the bucket and a direct get of one key; no consumer, so it neither lists nor watches them.
|
||||
var CandidateReads = []string{
|
||||
"$JS.API.DIRECT.GET.KV_" + delivery.CandidatesBucket + ".>",
|
||||
"$JS.API.STREAM.INFO.KV_" + delivery.CandidatesBucket,
|
||||
}
|
||||
|
||||
// EngineComposedBuckets are the composition buckets of every node, as the node-engine's state (ADR 0593 §6): the grants
|
||||
// part creates each, create-only, with the history ADR 0539 clause 4 keeps (delivery.ComposedHistory), as it creates a
|
||||
// module's declared state. A node whose name a bucket cannot take is left out (its engine is refused its writes too).
|
||||
func EngineComposedBuckets(nodes []string) []Bucket {
|
||||
var out []Bucket
|
||||
for _, node := range sortedCopy(nodes) {
|
||||
bucket, err := delivery.ComposedBucketFor(node)
|
||||
if err != nil || !safeSubject.MatchString(node) {
|
||||
continue
|
||||
}
|
||||
out = append(out, Bucket{Module: EngineModule, Name: delivery.ComposedState, Whole: bucket, History: delivery.ComposedHistory})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// composedFamily is every node's composition bucket, for a reader of delivery.ComposedRead.
|
||||
func composedFamily(nodes []string) []string {
|
||||
var out []string
|
||||
for _, b := range EngineComposedBuckets(nodes) {
|
||||
out = append(out, b.Bucket())
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// What a reader of a per-node state is granted beside each bucket's read (mesh-sdk process.FamilyListSubject and
|
||||
// FamilyCreatedAdvisory, ADR 0593 §6): the listing of stream names, by which it finds every node's bucket, and the
|
||||
// creation advisory of any stream, by which it binds a node's bucket made after it started. Both say a name only.
|
||||
const (
|
||||
familyListSubject = "$JS.API.STREAM.NAMES"
|
||||
familyCreatedAdvisory = "$JS.EVENT.ADVISORY.STREAM.CREATED.*"
|
||||
)
|
||||
|
||||
// VerbsTheNodeAgentAsks are the seat verbs a node's node-engine calls of its own machine while it applies a want (ADR
|
||||
// 0539 clause 16, ADR 0429 §14): its backup holder backs a data module up before it moves, and says when it was. Asked
|
||||
// of its own machine only, `mesh.seat.<seat>.tool.<verb>.<node>`.
|
||||
|
||||
@@ -106,6 +106,9 @@ type Principal struct {
|
||||
// Username and inbox are <node>.<module>.<process id>.
|
||||
Process string
|
||||
|
||||
// Nodes are every node the records know, for a read of a state kept one bucket per node (delivery.ComposedRead, ADR
|
||||
// 0593 §6): set only where the principal or a module it carries reads one.
|
||||
Nodes []string
|
||||
// FilesFor are the nodes whose declaration-file buckets (novox/hq ADR 0486 §2) this principal may remove, for the bus
|
||||
// module's tools removing the ended controller's leftovers alone (ADR 0590 clause 6): every node of the mesh, one
|
||||
// bucket each.
|
||||
@@ -921,7 +924,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
// 5. Its state, and the state of others it reads (novox/hq ADR 0201): every one read and
|
||||
// watched, its own written too.
|
||||
kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State, PerMachine: p.PerMachine, Reads: p.Reads,
|
||||
KeyedReads: p.KeyedReads}
|
||||
KeyedReads: p.KeyedReads, Nodes: p.Nodes}
|
||||
pub = append(pub, stateGrants(kept)...)
|
||||
// And it hears each of those buckets created, so one not on the bus at its start is bound when it appears
|
||||
// (novox/hq issue 1790; mesh-sdk process.AwaitBucket).
|
||||
@@ -1022,7 +1025,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
// each assignment is issued, as it keeps each module's events under that module's own name.
|
||||
for _, d := range p.Carries {
|
||||
pub = append(pub, stateGrants(stateAccess{Module: d.Module, Node: p.Node, Keeps: stateNames(d.State),
|
||||
PerMachine: perMachineNames(d.State), Reads: d.Reads, KeyedReads: d.KeyedReads})...)
|
||||
PerMachine: perMachineNames(d.State), Reads: d.Reads, KeyedReads: d.KeyedReads, Nodes: p.Nodes})...)
|
||||
}
|
||||
// **And it keeps the logs of the modules it carries** (novox/hq ADR 0297): each module's own, the
|
||||
// union over them. That one module's code does not append to another's log through it is the
|
||||
|
||||
@@ -40,6 +40,9 @@ type Bucket struct {
|
||||
// KeyTTL says the bucket allows a time to live per key (novox/hq ADR 0433 §1): its owner writes each key with its
|
||||
// own, and the server writes a delete marker, kept KeyTTLMarkerTTL, when it ends.
|
||||
KeyTTL bool
|
||||
// Whole is the bucket's name where it is not BucketName(Module, Name): one node's bucket of a state kept one bucket per
|
||||
// node, the node-engine's compositions `composed-<node>` (ADR 0593 §6, EngineComposedBuckets).
|
||||
Whole string
|
||||
}
|
||||
|
||||
// KeyTTLMarkerTTL is how long the marker an expired key leaves is kept: long enough for a watcher to hear it.
|
||||
@@ -50,7 +53,12 @@ const KeyTTLMarkerTTL = time.Minute
|
||||
func BucketName(module, name string) string { return module + "_" + name }
|
||||
|
||||
// Bucket is this bucket's name on the bus.
|
||||
func (b Bucket) Bucket() string { return BucketName(b.Module, b.Name) }
|
||||
func (b Bucket) Bucket() string {
|
||||
if b.Whole != "" {
|
||||
return b.Whole
|
||||
}
|
||||
return BucketName(b.Module, b.Name)
|
||||
}
|
||||
|
||||
// Why is carried into the server's description of the bucket, so somebody reading the server's
|
||||
// own state finds whose it is and why it is kept.
|
||||
@@ -154,6 +162,14 @@ func stateGrants(a stateAccess) []string {
|
||||
out = append(out, announce.Subject(a.Module))
|
||||
}
|
||||
for _, r := range a.Reads {
|
||||
if r == delivery.ComposedRead {
|
||||
// Every node's compositions, read-only (ADR 0593 §6), with the listing that finds their buckets.
|
||||
for _, bucket := range composedFamily(a.Nodes) {
|
||||
read(bucket, "")
|
||||
}
|
||||
out = append(out, familyListSubject)
|
||||
continue
|
||||
}
|
||||
if bucket, ok := bucketOfRead(r); ok {
|
||||
if NodeKeyedReads[r] {
|
||||
if machineKey != "" {
|
||||
@@ -208,6 +224,14 @@ func stateCreatedHeard(a stateAccess) []string {
|
||||
add(BucketName(a.Module, name))
|
||||
}
|
||||
for _, r := range a.Reads {
|
||||
if r == delivery.ComposedRead {
|
||||
// A node's bucket made after the reader began is heard by any stream's creation (ADR 0593 §6).
|
||||
if !seen[familyCreatedAdvisory] {
|
||||
seen[familyCreatedAdvisory] = true
|
||||
out = append(out, familyCreatedAdvisory)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if bucket, ok := bucketOfRead(r); ok {
|
||||
add(bucket)
|
||||
}
|
||||
@@ -233,6 +257,8 @@ type stateAccess struct {
|
||||
Keeps, PerMachine []string
|
||||
Reads []string
|
||||
KeyedReads []KeyedRead
|
||||
// Nodes are every node, whose bucket of a per-node state (delivery.ComposedRead) a read of it reaches.
|
||||
Nodes []string
|
||||
}
|
||||
|
||||
// perMachineNames is the local names of a module's buckets keyed by machine.
|
||||
@@ -257,6 +283,12 @@ func stateIssuedFor(d Declared, node string) []StateIssued {
|
||||
out = append(out, issued)
|
||||
}
|
||||
for _, r := range d.Reads {
|
||||
if r == delivery.ComposedRead {
|
||||
// One bucket per node, issued as the family's prefix (ADR 0593 §6): the runtime reads every node's bucket as one
|
||||
// state, a node's bucket made later included.
|
||||
out = append(out, StateIssued{Name: r, Bucket: delivery.ComposedBucketPrefix, PerNode: true})
|
||||
continue
|
||||
}
|
||||
if bucket, ok := bucketOfRead(r); ok {
|
||||
out = append(out, StateIssued{Name: r, Bucket: bucket})
|
||||
}
|
||||
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -97,7 +97,9 @@ var NamedReaders = map[string][]string{VaultBusAccounts: {GrantsModule}, GrantsA
|
||||
// machine principal (NodeLoginsReads), never through `reads`.
|
||||
delivery.NodeLoginsRead: {GrantsModule},
|
||||
// ADR 0587 §2: the open enrolments are read by their holder and, as a named reader, by the grants part.
|
||||
delivery.EnrolmentsRead: {GrantsModule}}
|
||||
delivery.EnrolmentsRead: {GrantsModule},
|
||||
// ADR 0593 §6: every node's compositions are read by the provisioner, the assigner and the view alone.
|
||||
delivery.ComposedRead: delivery.ComposedReaders}
|
||||
|
||||
// ReaderMay says whether module may read the state read (`<owner>.<name>`) by the named readers: always, for a state
|
||||
// that names none.
|
||||
|
||||
@@ -313,6 +313,17 @@ func Users(r Records) ([]Principal, error) {
|
||||
}
|
||||
seen[name] = string(p.Kind)
|
||||
}
|
||||
// A reader of every node's compositions (ADR 0593 §6) reaches each node's bucket: the nodes are given to the principal
|
||||
// that reads them, itself or through a module it carries, and to no other.
|
||||
for i := range out {
|
||||
reads := slices.Clone(out[i].Reads)
|
||||
for _, d := range out[i].Carries {
|
||||
reads = append(reads, d.Reads...)
|
||||
}
|
||||
if slices.Contains(reads, delivery.ComposedRead) {
|
||||
out[i].Nodes = sortedCopy(r.Nodes)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -74,9 +74,9 @@ const (
|
||||
|
||||
// Configs are the streams a manifest declares, as the controller asserts them at its start (mesh-sdk buscompose), by
|
||||
// stream name. Pure.
|
||||
func Configs(m catalogue.Manifest) map[string]jetstream.StreamConfig {
|
||||
func Configs(m catalogue.Manifest, nodes []string) map[string]jetstream.StreamConfig {
|
||||
out := map[string]jetstream.StreamConfig{}
|
||||
for _, b := range bucketsOf(m) {
|
||||
for _, b := range bucketsFor(m, nodes) {
|
||||
c := buscompose.BucketStreamConfig(b)
|
||||
out[c.Name] = c
|
||||
}
|
||||
@@ -116,8 +116,8 @@ func (c *creations) keep(stream, digest string) {
|
||||
// createMissing creates each item of a the bus lacks, with the manifest's limits, and says what became of it. It never
|
||||
// touches an item the bus holds, and does nothing when what the bus holds was not read. force asks again a stream
|
||||
// created at this manifest and gone since (a hand assert). It answers the streams created.
|
||||
func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifest, force bool) []string {
|
||||
cfgs := Configs(m)
|
||||
func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifest, nodes []string, force bool) []string {
|
||||
cfgs := Configs(m, nodes)
|
||||
var made []string
|
||||
a.Says = CreatesMissing
|
||||
for i := range a.Items {
|
||||
@@ -163,9 +163,9 @@ func notCreated(a Assertion) []string {
|
||||
}
|
||||
|
||||
// Composed is a module's buckets, logs and streams as an assertion would ask for them. Pure.
|
||||
func Composed(m catalogue.Manifest) []Item {
|
||||
func Composed(m catalogue.Manifest, nodes []string) []Item {
|
||||
var out []Item
|
||||
for _, b := range bucketsOf(m) {
|
||||
for _, b := range bucketsFor(m, nodes) {
|
||||
per := b.History
|
||||
if per < 1 {
|
||||
per = 1
|
||||
@@ -213,8 +213,8 @@ func HeldStreams(c Caller) (map[string]Limits, error) {
|
||||
}
|
||||
|
||||
// Assert composes a module's items and compares each with what the bus holds; held nil is not read. It asserts nothing.
|
||||
func Assert(module, digest string, m catalogue.Manifest, held map[string]Limits, heldErr error) Assertion {
|
||||
a := Assertion{Module: module, Manifest: digest, Items: Composed(m), Says: ShadowAsserts}
|
||||
func Assert(module, digest string, m catalogue.Manifest, nodes []string, held map[string]Limits, heldErr error) Assertion {
|
||||
a := Assertion{Module: module, Manifest: digest, Items: Composed(m, nodes), Says: ShadowAsserts}
|
||||
for i := range a.Items {
|
||||
it := &a.Items[i]
|
||||
if heldErr != nil {
|
||||
|
||||
@@ -369,10 +369,10 @@ func (p *Part) Compose() Result {
|
||||
if !assigned(in, module) {
|
||||
continue
|
||||
}
|
||||
a := Assert(module, in.Digests[module], in.Manifests[module], held, heldErr)
|
||||
a := Assert(module, in.Digests[module], in.Manifests[module], in.Nodes, held, heldErr)
|
||||
// Created only where the bus was read and lacks it, never changed where it holds it (issue 1790).
|
||||
if p.Create != nil && heldErr == nil {
|
||||
if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], forced == module); len(made) > 0 {
|
||||
if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], in.Nodes, forced == module); len(made) > 0 {
|
||||
res.Asserted = append(res.Asserted, fmt.Sprintf("%s at %s: created %s", module, a.Manifest, strings.Join(made, ", ")))
|
||||
p.say(&res, EventAsserted, map[string]any{"module": module, "manifest": a.Manifest, "created": made})
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
|
||||
// The node-engine's own manifest, placed on every node (ADR 0563): its state, logs and streams alone, created by this
|
||||
// part; it adds no principal, since each node's engine speaks as the node's account.
|
||||
if m, ok := in.Manifests[buscompose.EngineModule]; ok {
|
||||
out.Engine = &buscompose.Declared{Module: m.Module, State: bucketsOf(m), Logs: logsOf(m), Streams: streamsOf(m)}
|
||||
out.Engine = &buscompose.Declared{Module: m.Module, State: bucketsFor(m, in.Nodes), Logs: logsOf(m), Streams: streamsOf(m)}
|
||||
}
|
||||
return out, gaps
|
||||
}
|
||||
@@ -233,6 +233,18 @@ func bucketsOf(m catalogue.Manifest) []buscompose.Bucket {
|
||||
return out
|
||||
}
|
||||
|
||||
// bucketsFor is a manifest's buckets as this part creates them: bucketsOf, and for the node-engine every node's bucket of
|
||||
// compositions, `composed-<node>`, in place of a state it names so (novox/hq ADR 0593 §6): the engine writes its own node's
|
||||
// and never creates it.
|
||||
func bucketsFor(m catalogue.Manifest, nodes []string) []buscompose.Bucket {
|
||||
out := bucketsOf(m)
|
||||
if m.Module != buscompose.EngineModule {
|
||||
return out
|
||||
}
|
||||
out = slices.DeleteFunc(out, func(b buscompose.Bucket) bool { return b.Name == delivery.ComposedState })
|
||||
return append(out, buscompose.EngineComposedBuckets(nodes)...)
|
||||
}
|
||||
|
||||
func logsOf(m catalogue.Manifest) []buscompose.Log {
|
||||
var out []buscompose.Log
|
||||
for _, l := range m.Logs {
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
"mesh-grants/internal/buscompose"
|
||||
)
|
||||
|
||||
// The part's tools, as its manifest lists them. The part claims no seat: the seat mesh-grants served the lease write
|
||||
@@ -271,7 +272,11 @@ func (p *Part) AssertOne(module string) (any, error) {
|
||||
return nil, err
|
||||
}
|
||||
held, heldErr := HeldStreams(p.Call)
|
||||
a := Assert(module, v.Manifest, m, held, heldErr)
|
||||
var nodes []string
|
||||
if module == buscompose.EngineModule {
|
||||
nodes = ReadInputs(p.Read, &p.cache).Nodes // every node's bucket of compositions (ADR 0593 §6)
|
||||
}
|
||||
a := Assert(module, v.Manifest, m, nodes, held, heldErr)
|
||||
out := map[string]any{"assertion": a}
|
||||
missing := 0
|
||||
for _, it := range a.Items {
|
||||
|
||||
Reference in New Issue
Block a user