From 98f4ff6163adf61085438267241dd1992db0b920 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 11:34:15 +0100 Subject: [PATCH] Read the bus's user-list file's place from what the node runs, not from the controller's compositions The bus module on the bus node runs from what its agent holds (its own composition, or the declaration kept before the switch: every novox have read 2026-11-02 names its own composition or none, and nats names a build alone), while composed- was written by the controller, which ended (issue 1126). The follower now looks at the agent's record of what runs whenever it changed, writes again only when the file it names moved, and trusts no composition signer. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- cmd/mesh-host/nodeagent.go | 52 ++---- cmd/mesh-host/nodeagent_integration_test.go | 16 +- cmd/mesh-host/userlist.go | 10 +- internal/userlist/userlist.go | 182 ++++++++++---------- internal/userlist/userlist_test.go | 118 ++++--------- 5 files changed, 143 insertions(+), 235 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 5b55123..c9a140a 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -3,7 +3,6 @@ package main import ( "context" "crypto/ed25519" - "encoding/base64" "encoding/json" "errors" "fmt" @@ -375,42 +374,6 @@ func resourcesOf(cs []delivery.Composed) []declaration.Resource { return out } -// operatorKeysFile is the file beside the node's state that holds the operator's public keys, one ed25519 key in base64 a -// line (a line that starts with # is a note). The switch's last declaration writes it as a file resource, beside the -// controller's key the node was enrolled with (novox/hq ADR 0539 §3, §11); a node adopted later has it at its adoption. -const operatorKeysFile = "operator-keys" - -// trustedKeys are the keys a composition in the node's composition bucket may be signed by: the one given (none for the -// agent; the enrolment's signer for the user-list follower, which reads only where the bus's user-list file is from -// such a composition, never its list) and the operator's, read from operatorKeysFile each time it is asked, so a key -// that arrives while the agent runs is trusted at the next composition. A file that cannot be read trusts the one given -// alone, and says so. -func trustedKeys(opts options, controllerNow func() ed25519.PublicKey) func() []ed25519.PublicKey { - return func() []ed25519.PublicKey { - keys := []ed25519.PublicKey{} - if controller := controllerNow(); len(controller) == ed25519.PublicKeySize { - keys = append(keys, controller) - } - raw, err := os.ReadFile(filepath.Join(filepath.Dir(opts.state), operatorKeysFile)) - if err != nil { - return keys - } - for _, line := range strings.Split(string(raw), "\n") { - line = strings.TrimSpace(line) - if line == "" || strings.HasPrefix(line, "#") { - continue - } - key, err := base64.StdEncoding.DecodeString(line) - if err != nil || len(key) != ed25519.PublicKeySize { - fmt.Fprintf(os.Stderr, "mesh-host: a line of %s is not an ed25519 public key in base64, and is not trusted\n", operatorKeysFile) - continue - } - keys = append(keys, ed25519.PublicKey(key)) - } - return keys - } -} - // keptOnce reads the kept declaration once and splits it by module: its resources (the node's own are those that say // none, ADR 0539 §9) and its filled paths. Once the read is retired (keptRetired) nothing is read. func keptOnce(opts options, signer ed25519.PublicKey) func() (map[string][]json.RawMessage, map[string][]string) { @@ -511,10 +474,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus } } // The bus's user list, the grants part's alone (ADR 0547 §8, clause 11; issue 1126), written into the file the - // bus module's composition in the node's bucket declares: on every node, and a node whose compositions carry no - // list writes nothing. Those compositions were signed with the enrolment's signer and nobody writes them now; they - // are read for the file's place alone, never for a list, until the place is read elsewhere (milestone B17). - users := userListFollower(opts, session, composedBucket, trustedKeys(opts, func() ed25519.PublicKey { return signer }), say, func() { + // bus module this node runs declares, as its agent holds what runs (its own composition, or the declaration kept + // before the switch): never read from the node's composition bucket, which the controller wrote. On every node; + // a node that runs no list writes nothing. Looked at again whenever what runs changed (RunsChanged, Settled). + users := userListFollower(opts, session, func() []delivery.Composed { + if now := current.get(); now != nil { + return now.Everything() + } + return nil + }, say, func() { // This engine wrote the list, so the bus reloads its accounts and answers no JetStream request for a few // seconds: what failed in them is tried again once that is over, not at its next wait. time.AfterFunc(reloadSettles, func() { @@ -558,12 +526,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus Settled: func(module string) { knockFilling(module+" settled", say) judgeWhatRunsNow(opts.state, signer) + users.RunsChanged() }, // And once the haves are read at the start, and whenever a module is taken as running without an apply, so // a refusal from the engine's start does not wait for some module to settle (issue 924). RunsChanged: func(why string) { knockFilling(why, say) judgeWhatRunsNow(opts.state, signer) + users.RunsChanged() }, Say: say, // A failure of the bus not answering for a moment is tried again: 2 s, 5 s, 15 s, 60 s, then every 5 min. diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index cf471e4..a00bef5 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -5,7 +5,6 @@ package main import ( "context" "crypto/ed25519" - "encoding/base64" "encoding/json" "os" "path/filepath" @@ -45,7 +44,6 @@ type itBed struct { opts options session *link.Session controll ed25519.PrivateKey - operator ed25519.PrivateKey said []string // refusals are the bus's asynchronous errors on the node's connection: each permissions violation it met. refusals []string @@ -137,20 +135,12 @@ func newItBedAs(t *testing.T, principal buscompose.Principal) *itBed { t.Fatal(err) } _, b.controll, _ = ed25519.GenerateKey(nil) - _, b.operator, _ = ed25519.GenerateKey(nil) return b } -// start runs the node agent under a session on the bus, trusting the controller's key, and the operator's when asked. -func (b *itBed) start(trustOperator bool) (stop func()) { +// start runs the node agent under a session on the bus. +func (b *itBed) start() (stop func()) { b.t.Helper() - if trustOperator { - pub := b.operator.Public().(ed25519.PublicKey) - if err := os.WriteFile(filepath.Join(filepath.Dir(b.opts.state), operatorKeysFile), - []byte("# the operator\n"+base64.StdEncoding.EncodeToString(pub)+"\n"), 0o600); err != nil { - b.t.Fatal(err) - } - } b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second, Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }} b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode}) @@ -493,7 +483,7 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { t.Fatal(err) } - stop := b.start(false) + stop := b.start() // Not placed yet: the agent applies the controller's compositions, of which there are none. The catalogue's seed then // places the node, and the agent begins again composing its own (ADR 0549 §4), with no restart of the engine. time.Sleep(300 * time.Millisecond) diff --git a/cmd/mesh-host/userlist.go b/cmd/mesh-host/userlist.go index bf79c2f..54cb6f1 100644 --- a/cmd/mesh-host/userlist.go +++ b/cmd/mesh-host/userlist.go @@ -2,7 +2,6 @@ package main import ( "context" - "crypto/ed25519" "encoding/json" "errors" "fmt" @@ -26,13 +25,12 @@ import ( // userListFollower is the follower of the bus's user list for one link: the node's composition bucket, the keys the // agent trusts, the write alone, and the have. -func userListFollower(opts options, session *link.Session, composedBucket string, trusted func() []ed25519.PublicKey, +func userListFollower(opts options, session *link.Session, running func() []delivery.Composed, say link.Announce, wrote func()) *userlist.Follower { return userlist.New(userlist.Env{ - Node: session.Membership.Node, - Composed: bucketWatch{session, composedBucket}, - Trusted: trusted, - OnDisk: userListOnDisk, + Node: session.Membership.Node, + Running: running, + OnDisk: userListOnDisk, Write: func(ctx context.Context, module string, raw json.RawMessage, body []byte) (string, bool, error) { digest, changed, err := writeUserListAlone(session.WithLink(ctx), opts, raw, body, say) if changed && wrote != nil { diff --git a/internal/userlist/userlist.go b/internal/userlist/userlist.go index 668f473..b818729 100644 --- a/internal/userlist/userlist.go +++ b/internal/userlist/userlist.go @@ -13,23 +13,26 @@ // or is deleted is said, the key last heard stands, and the list loaded stays; with no key ever heard nothing is // written. The list a composition carries is never written. // -// **Where it writes.** The file is the user-list resource (`bus-users`) of the bus module's composition in the node's -// composition bucket, verified as the agent verified it before anything of it is read: what tells the bus node it is -// one, and the file's place. The have of that module says what was written, which list it follows and this node's key, -// which the grants part seals to. +// **Where it writes.** The file is the user-list resource (`bus-users`) of the bus module's composition as this node runs +// it (Env.Running: the agent's own record of what runs, its own composition or the declaration kept before the switch), +// never one read from a composition bucket: the controller that wrote `composed-` ended (issue 1126), and the bus +// module on the bus node runs from what the agent holds. That is what tells the bus node it is one, and the file's +// place. The have of that module says what was written, which list it follows and this node's key, which the grants +// part seals to. // -// **It watches, it never polls**, and looks again after every apply of the bus module and when asked (Again). It reads no -// want and writes none, moves no fingerprint, and touches nothing of the bus module but its user list. +// **It watches, it never polls**: the grants part's key is watched, and what runs is looked at again when the agent says +// it changed, after every apply of the bus module and when asked (Again). It reads no want and writes none, moves no +// fingerprint, and touches nothing of the bus module but its user list. package userlist import ( "context" - "crypto/ed25519" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "regexp" + "sort" "strings" "sync" "time" @@ -49,10 +52,10 @@ type Written = delivery.UserListWritten // Env is everything the follower needs of the node. type Env struct { - Node string - Composed agent.Watcher - // Trusted is the keys a composition may be signed by: one any other key signed declares no file. - Trusted func() []ed25519.PublicKey + Node string + // Running is the compositions this node runs now, as its agent holds them (agent.Agent.Everything): where the user + // list's file is is read from these alone (issue 1126). Nil runs nothing. + Running func() []delivery.Composed // OnDisk is what the resource's file holds on the machine now; nil when there is none. OnDisk func(resource json.RawMessage) ([]byte, error) // Write writes the user list's file resource of a module's composition alone, with body as its content, when it @@ -79,19 +82,18 @@ type Env struct { } type source struct { - key string module string - rev uint64 resource json.RawMessage - wanted string } -// Follower follows the node's compositions for the bus's user list. +// Follower follows the grants part's user list into the file the bus module this node runs declares. type Follower struct { env Env kick chan struct{} - mu sync.Mutex - srcs map[string]source // by composition key + // looks asks a look at what runs (RunsChanged): written only when the file it names moved. + looks chan struct{} + mu sync.Mutex + srcs map[string]source // by module, of what runs now // cur is the source last taken, and written whether its write went through. cur source written bool @@ -116,7 +118,8 @@ func New(env Env) *Follower { if env.Backoff == nil { env.Backoff = func(int) time.Duration { return time.Minute } } - f := &Follower{env: env, kick: make(chan struct{}, 1), srcs: map[string]source{}, said: map[string]bool{}} + f := &Follower{env: env, kick: make(chan struct{}, 1), looks: make(chan struct{}, 1), srcs: map[string]source{}, + said: map[string]bool{}} if env.Grants != nil && env.Kept != nil { if raw, rev := env.Kept(); raw != nil { if g, err := delivery.DecodeGrantsUserList(raw); err == nil { @@ -127,8 +130,8 @@ func New(env Env) *Follower { return f } -// Again asks the follower to look at the file again, though no composition changed: after an apply of the bus module, -// and when a person asks. It never blocks. +// Again asks the follower to look at what runs and the file again: when what runs changed, after an apply of the bus +// module, and when a person asks. It never blocks. func (f *Follower) Again() { select { case f.kick <- struct{}{}: @@ -136,21 +139,25 @@ func (f *Follower) Again() { } } -// State is the module whose list the follower writes ("" when no composition of this node carries one) and what it -// last did. +// RunsChanged asks the follower to look at what this node runs again: the list is written again only when the file the +// bus module declares moved. It never blocks. +func (f *Follower) RunsChanged() { + select { + case f.looks <- struct{}{}: + default: + } +} + +// State is the module whose list the follower writes ("" when nothing this node runs carries one) and what it last did. func (f *Follower) State() (string, Written) { f.mu.Lock() defer f.mu.Unlock() return f.mod, f.state } -// Run follows the compositions until ctx ends. +// Run follows the grants part's list until ctx ends. func (f *Follower) Run(ctx context.Context) error { - updates, err := f.env.Composed.Watch(ctx, f.env.Node+".>") - if err != nil { - return fmt.Errorf("the user list cannot be followed: %w", err) - } - caught := false + f.RunsChanged() // the first look at what runs, at once // The grants part's key, on the node whose compositions carry the list: the bus node alone is granted it (mesh-sdk // buscompose.GrantsUserListFollows). Until its watch caught up nothing is written, so a new follower (one per link) // never writes a list older than the key; a watch refused or ended is tried again on the backoff, and meanwhile the @@ -185,17 +192,11 @@ func (f *Follower) Run(ctx context.Context) error { return nil case <-f.kick: again = true + f.look() + case <-f.looks: + f.look() case <-retry: retry = nil - case u, ok := <-updates: - if !ok { - return fmt.Errorf("the watch of the compositions ended") - } - if u.Caught { - caught = true - } else { - f.take(u) - } case u, ok := <-grants: if !ok { failed("its watch ended") @@ -210,9 +211,6 @@ func (f *Follower) Run(ctx context.Context) error { } f.takeGrants(u) } - if !caught { - continue - } open() if opened && !grantsCaught && grants != nil { continue // the switch is not heard yet: nothing is written before it is @@ -226,11 +224,37 @@ func (f *Follower) Run(ctx context.Context) error { } } -// busNode says some composition of this node carries the bus's user list. +// busNode says something this node runs carries the bus's user list, or did. func (f *Follower) busNode() bool { f.mu.Lock() defer f.mu.Unlock() - return f.cur.key != "" || len(f.srcs) > 0 + return f.cur.module != "" || len(f.srcs) > 0 +} + +// look reads what this node runs now for the bus's user list: each module whose composition carries it. +func (f *Follower) look() { + var running []delivery.Composed + if f.env.Running != nil { + running = f.env.Running() + } + srcs := map[string]source{} + for _, c := range running { + if c.Node != "" && c.Node != f.env.Node { + continue + } + raw, _, _, ok := userList(c.Resources) + if !ok { + continue + } + module := c.Module + if module == "" { + module = agent.NodeModule + } + srcs[module] = source{module: module, resource: raw} + } + f.mu.Lock() + f.srcs = srcs + f.mu.Unlock() } // takeGrants keeps the grants part's key as heard. A key deleted, or a value it cannot read, is said and changes nothing: @@ -272,51 +296,6 @@ func (f *Follower) key() string { return f.env.Key() } -// take keeps a composition key's change when it carries the bus's user list, and forgets a key deleted. -func (f *Follower) take(u agent.Update) { - node, module, _, isNode, err := delivery.ParseComposedKey(u.Key) - if err != nil || node != f.env.Node || isNode { - return - } - f.mu.Lock() - defer f.mu.Unlock() - if u.Deleted { - delete(f.srcs, u.Key) - return - } - c, err := delivery.DecodeComposed(u.Value) - if err != nil { - delete(f.srcs, u.Key) - return - } - var trusted []ed25519.PublicKey - if f.env.Trusted != nil { - trusted = f.env.Trusted() - } - if _, err := delivery.VerifyComposed(c, trusted); err != nil { - // The agent says it is refused; a refused composition is never a source of the list. - delete(f.srcs, u.Key) - return - } - if c.Node != f.env.Node || c.Module != module { - // Signed, and for another node or module than its key says: a re-put of another composition is no source. - delete(f.srcs, u.Key) - f.sayOnceLocked(fmt.Sprintf("the composition under %s is %s's of %s, and is no source of the bus's user list", - u.Key, c.Node, c.Module)) - return - } - raw, content, digest, ok := userList(c.Resources) - if !ok { - delete(f.srcs, u.Key) - return - } - wanted := digest - if wanted == "" { - wanted = Digest([]byte(content)) - } - f.srcs[u.Key] = source{key: u.Key, module: module, rev: u.Revision, resource: raw, wanted: wanted} -} - // settle writes the grants part's list: when its key moved, or when asked. func (f *Follower) settle(ctx context.Context, again bool) { f.settleGrants(ctx, again) } @@ -337,18 +316,26 @@ func (f *Follower) settleGrants(ctx context.Context, again bool) { g = *f.grants } rev := f.grantsRev + // What runs now: the module written before while it still carries the list, else the first by name; what was + // written before when nothing runs that carries it now (a look between two applies). file := f.cur - for _, s := range f.srcs { - if s.rev > file.rev { - file = s + if s, ok := f.srcs[f.cur.module]; ok { + file = s + } else if len(f.srcs) > 0 { + names := make([]string, 0, len(f.srcs)) + for m := range f.srcs { + names = append(names, m) } + sort.Strings(names) + file = f.srcs[names[0]] } - if file.key == "" { + if file.module == "" { f.mu.Unlock() - return // no composition of this node declares the file: not the bus node + return // nothing this node runs declares the file: not the bus node } + moved := file.module != f.cur.module || pathOf(file.resource) != pathOf(f.cur.resource) f.cur = file - if !again && f.state.Source != "" && f.state.Revision == rev && f.written { + if !again && !moved && f.state.Source != "" && f.state.Revision == rev && f.written { f.mu.Unlock() return } @@ -522,6 +509,15 @@ func userList(resources []json.RawMessage) (json.RawMessage, string, string, boo return nil, "", "", false } +// pathOf is where a file resource says its file is; "" when it says none. +func pathOf(raw json.RawMessage) string { + var head struct { + Path string `json:"path"` + } + _ = json.Unmarshal(raw, &head) + return head.Path +} + // Digest is content's digest as the have says it, `sha256:`. func Digest(content []byte) string { sum := sha256.Sum256(content) diff --git a/internal/userlist/userlist_test.go b/internal/userlist/userlist_test.go index 33153da..a891016 100644 --- a/internal/userlist/userlist_test.go +++ b/internal/userlist/userlist_test.go @@ -2,7 +2,6 @@ package userlist import ( "context" - "crypto/ed25519" "encoding/json" "errors" "strings" @@ -17,17 +16,6 @@ import ( const node = "novox" -func key() (ed25519.PublicKey, ed25519.PrivateKey) { - pub, k, err := ed25519.GenerateKey(nil) - if err != nil { - panic(err) - } - return pub, k -} - -var controllerPub, controllerKey = key() -var _, strangerKey = key() - func fpOf(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) } // feed is a watch the test writes to: the keys there before the watch, Caught, then each put. watching is closed once @@ -69,19 +57,14 @@ func list(users ...string) string { return b.String() } -// comp is a composition of module at fp holding the user list content (none when empty), signed by signer; by digest -// when byDigest, as the controller sends a file over 64 KiB; for another node or module than its key when asked. -func comp(t *testing.T, module, fp, content string, signer ed25519.PrivateKey, p put) (string, []byte) { - t.Helper() - owner := module - if p.asModule != "" { - owner = p.asModule - } - server, _ := json.Marshal(map[string]any{"id": owner + ".server-conf", "type": "file", "module": owner, +// comp is a composition of module at fp as this node runs it, holding the user list content (none when empty); by +// digest when byDigest, as a file over 64 KiB is given; for another node than this one when asked. +func comp(module, fp, content string, p put) delivery.Composed { + server, _ := json.Marshal(map[string]any{"id": module + ".server-conf", "type": "file", "module": module, "path": "/var/lib/nats-module/conf/nats.conf", "content": "include accounts.conf\n"}) rs := []json.RawMessage{server} if content != "" { - r := map[string]any{"id": owner + ".bus-users", "type": "file", "module": owner, + r := map[string]any{"id": module + ".bus-users", "type": "file", "module": module, "path": "/var/lib/nats-module/conf/accounts.conf", "content": content, "mode": "0600"} if p.byDigest { delete(r, "content") @@ -94,22 +77,7 @@ func comp(t *testing.T, module, fp, content string, signer ed25519.PrivateKey, p if p.asNode != "" { c.Node = p.asNode } - if p.asModule != "" { - c.Module = p.asModule - } - k, err := delivery.ComposedKey(node, module, fp) - if err != nil { - t.Fatal(err) - } - signed, err := delivery.SignComposed(c, signer) - if err != nil { - t.Fatal(err) - } - raw, err := delivery.EncodeComposed(signed) - if err != nil { - t.Fatal(err) - } - return k, raw + return c } // gput is the grants part's key as the bus node hears it at the watch's start (ADR 0547 clauses 7, 8): which list to @@ -125,12 +93,12 @@ type gput struct { type put struct { module, fp, content string - stranger, deleted bool byDigest bool - asNode, asModule string + asNode string } -// Defends novox/hq ADR 0547 §8: the bus node writes the newest signed list alone, only when it changed, and reports it. +// Defends novox/hq ADR 0547 §8: the bus node writes the grants part's list alone, only when it changed, and reports it, +// into the file the bus module it runs declares (issue 1126, B17: what runs, never a composition bucket). func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { one, two, three := list("controller", "a"), list("controller", "a", "b"), list("controller", "a", "b", "c") five := list("controller", "a", "b", "c", "d") @@ -152,13 +120,14 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { // wantSource and wantFollows are what the last have says it took the list from and follows. wantSource, wantFollows string }{ - // Issue 1126, A6 (ADR 0547 clauses 7, 8, 11): the grants part's list alone, written into the file the bus module's - // composition declares; the list a composition carries is never written, whatever becomes of the grants key. - {name: "the grants part's list is written into the file a composition declares", + // Issue 1126, A6 (ADR 0547 clauses 7, 8, 11): the grants part's list alone, written into the file the bus module + // this node runs declares; the list a composition carries is never written, whatever becomes of the grants key. + // B17: the file's place is read from what runs, as the agent holds it, never from a composition bucket. + {name: "the grants part's list is written into the file the bus module it runs declares", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowGrants, content: three}}, wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants}, - {name: "a key that says follow the controller still writes the grants part's list", - before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{follow: delivery.FollowController, content: three}}, + {name: "a bus module it runs by digest still names the file", + before: []put{{module: "nats", fp: fpOf("a"), content: two, byDigest: true}}, grants: []gput{{follow: delivery.FollowGrants, content: three}}, wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants}, {name: "a newer composition's list is not written", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, @@ -167,8 +136,8 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { wantWrites: []string{three}, wantUsers: 4, wantSource: delivery.GrantsUserListSource, wantFollows: delivery.FollowGrants}, {name: "a composition without the list is ignored", before: []put{{module: "web", fp: fpOf("a"), content: ""}}, grants: []gput{{follow: delivery.FollowGrants, content: three}}}, - {name: "a composition signed by a stranger declares no file", - before: []put{{module: "nats", fp: fpOf("a"), content: two, stranger: true}}, grants: []gput{{follow: delivery.FollowGrants, content: three}}}, + {name: "another node's composition declares no file", + before: []put{{module: "nats", fp: fpOf("a"), content: two, asNode: "ace"}}, grants: []gput{{follow: delivery.FollowGrants, content: three}}}, {name: "an unreadable grants key with none heard before writes nothing, never the composition's list", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, grants: []gput{{unreadable: true}}, onDisk: two, wantErr: "key is not heard", wantRefused: true, wantSource: delivery.GrantsUserListSource}, @@ -201,9 +170,8 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { onDisk: two, wantErr: "sealed to another key", wantRefused: true, wantSource: delivery.GrantsUserListSource}, } { t.Run(c.name, func(t *testing.T) { - f := &feed{ch: make(chan agent.Update, 64), watching: make(chan struct{})} - objects := map[string]string{} var mu sync.Mutex + var running []delivery.Composed var writes []string var reports []Written reported := make(chan struct{}, 64) @@ -257,7 +225,12 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { gf.first = append(gf.first, u) } } - fl := New(Env{Node: node, Composed: f, Grants: gf, + fl := New(Env{Node: node, Grants: gf, + Running: func() []delivery.Composed { + mu.Lock() + defer mu.Unlock() + return append([]delivery.Composed(nil), running...) + }, Kept: func() ([]byte, uint64) { return keptRaw, 9 }, Keep: func([]byte, uint64) {}, Backoff: func(int) time.Duration { return time.Hour }, @@ -271,8 +244,7 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { out, err := seal.Open(b, nodeKey.PrivateKey) return []byte(out), err }, - Key: func() string { return nodeKey.PublicKey }, - Trusted: func() []ed25519.PublicKey { return []ed25519.PublicKey{controllerPub} }, + Key: func() string { return nodeKey.PublicKey }, OnDisk: func(json.RawMessage) ([]byte, error) { if c.onDisk == "" { return nil, nil @@ -296,30 +268,14 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { }, Now: func() time.Time { return time.Date(2026, 10, 25, 21, 50, 0, 0, time.UTC) }, }) - update := func(p put) agent.Update { - f.rev++ - signer := controllerKey - if p.stranger { - signer = strangerKey - } - k, raw := comp(t, p.module, p.fp, p.content, signer, p) - if p.byDigest { - objects[Digest([]byte(p.content))] = p.content - } - if p.deleted { - return agent.Update{Key: k, Revision: f.rev, Deleted: true} - } - return agent.Update{Key: k, Value: raw, Revision: f.rev} - } for _, p := range c.before { - f.first = append(f.first, update(p)) + running = append(running, comp(p.module, p.fp, p.content, p)) } ctx, cancel := context.WithCancel(context.Background()) defer cancel() done := make(chan error, 1) go func() { done <- fl.Run(ctx) }() - <-f.watching - // A later put after the first list is written: the first write waits for the grants part's key to catch up. + // What runs changes after the first list is written: the first write waits for the grants part's key. if len(c.after) > 0 { for until := time.Now().Add(5 * time.Second); time.Now().Before(until); time.Sleep(5 * time.Millisecond) { mu.Lock() @@ -331,7 +287,10 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { } } for _, p := range c.after { - f.ch <- update(p) + mu.Lock() + running = []delivery.Composed{comp(p.module, p.fp, p.content, p)} + mu.Unlock() + fl.RunsChanged() } if len(grantsAfter) > 0 { for until := time.Now().Add(5 * time.Second); ; time.Sleep(5 * time.Millisecond) { @@ -384,23 +343,18 @@ func TestTheBusNodeWritesTheGrantsPartsUserListAlone(t *testing.T) { if c.wantUnwatched != (st.Unwatched != "") { t.Fatalf("the have says unwatched %q", st.Unwatched) } - // While the grants part's list is followed, an apply of the controller's composition writes not its list. - if _, raw := comp(t, "nats", fpOf("a"), two, controllerKey, put{}); c.wantFollows != "" { - cmp, err := delivery.DecodeComposed(raw) - if err != nil { - t.Fatal(err) - } + // While the grants part's list is followed, an apply of the bus module's composition writes not its list. + if cmp := comp("nats", fpOf("a"), two, put{}); c.wantFollows != "" { held, err := fl.Hold(cmp) if err != nil { t.Fatal(err) } - _, _, _, hasList := userList(held.Resources) - if want := c.wantFollows == delivery.FollowController; hasList != want || len(held.Resources) < 1 { - t.Fatalf("an apply while following %s holds the controller's list: %v", c.wantFollows, hasList) + if _, _, _, hasList := userList(held.Resources); hasList || len(held.Resources) < 1 { + t.Fatalf("an apply while following %s holds the composition's list: %v", c.wantFollows, hasList) } // A list on the machine that cannot be read stops the apply; it is never left out to be removed. if c.wantFollows == delivery.FollowGrants { - readErr := New(Env{Node: node, Composed: f, OnDisk: func(json.RawMessage) ([]byte, error) { + readErr := New(Env{Node: node, OnDisk: func(json.RawMessage) ([]byte, error) { return nil, errors.New("permission denied") }}) readErr.grants = &delivery.GrantsUserList{Follow: delivery.FollowGrants}