Merge pull request 'The bus node writes the newest signed bus user list by itself, without a want (ADR 0547 bridge, issue 1945)' (#143) from fix/1945-user-list-follows into main

This commit was merged in pull request #143.
This commit is contained in:
2026-10-25 21:16:06 +00:00
11 changed files with 987 additions and 30 deletions
+12
View File
@@ -391,6 +391,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
}
var ag *agent.Agent
kept := keptOnce(opts, signer)
// The bus's user list, written alone from the newest composition of the bus module (ADR 0547 §8): on every
// node, and a node whose compositions carry no list writes nothing.
users := userListFollower(opts, session, composedBucket, trustedKeys(opts, signer), say)
go func() {
if err := users.Run(linked); err != nil && linked.Err() == nil {
say("the bus's user list is not followed on this link: " + err.Error())
}
}()
env := agent.Env{
Node: node,
Wants: bucketWatch{session, delivery.AssignerWantsBucket},
@@ -421,6 +429,10 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
actx, done := applyContext(session.WithLink(ctx))
res := applyComposition(actx, opts, c, others, sched, say)
done()
// An apply of the bus module may write an older list than the newest composition holds: looked at again.
if module, _ := users.State(); module != "" && module == c.Module {
users.Again()
}
afterApply()
sayFacts(ctx, ag)
return res
+120
View File
@@ -0,0 +1,120 @@
package main
import (
"context"
"crypto/ed25519"
"encoding/json"
"errors"
"fmt"
"os"
"git.novox.be/novox/mesh-sdk/go/declaration"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/link"
"github.com/novox/mesh-host/internal/store"
"github.com/novox/mesh-host/internal/system"
"github.com/novox/mesh-host/internal/userlist"
)
// The bridge of novox/hq ADR 0547 §8, wired: the bus node's engine writes the bus's user list alone from the newest
// composition of the bus module, under the apply's two locks as a filled file is written (filling.go), and says what
// it wrote in the bus module's have.
// 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,
say link.Announce) *userlist.Follower {
return userlist.New(userlist.Env{
Node: session.Membership.Node,
Composed: bucketWatch{session, composedBucket},
Trusted: trusted,
Resolve: func(ctx context.Context, raw json.RawMessage) ([]byte, error) {
return resolveUserList(session.WithLink(ctx), raw)
},
OnDisk: userListOnDisk,
Write: func(ctx context.Context, module string, raw json.RawMessage, body []byte) (string, bool, error) {
return writeUserListAlone(session.WithLink(ctx), opts, raw, body, say)
},
Report: func(ctx context.Context, module string, w userlist.Written) {
session.NoteUserList(ctx, module, w)
},
Say: announceOr(say),
})
}
// userListFile is a user-list resource as the file it declares.
func userListFile(raw []byte) (*declaration.File, error) {
r, err := declaration.ParseResource(raw)
if err != nil {
return nil, fmt.Errorf("the user list is not a resource: %w", err)
}
f, ok := r.(*declaration.File)
if !ok {
return nil, fmt.Errorf("the user list is a %s, not a file", r.Kind())
}
return f, nil
}
// resolveUserList is the list a user-list resource holds: its content, or the object its digest names in this node's
// file bucket, fetched and verified (apply.ResolveFileReferences, ADR 0486 §1).
func resolveUserList(ctx context.Context, raw []byte) ([]byte, error) {
f, err := userListFile(raw)
if err != nil {
return nil, err
}
d := &declaration.Declaration{Version: declaration.Version, Resources: []declaration.Resource{f}}
if err := apply.ResolveFileReferences(ctx, d, link.FilesFrom(ctx)); err != nil {
return nil, fmt.Errorf("the user list could not be had by its digest: %w", err)
}
return []byte(d.Resources[0].(*declaration.File).Content), nil
}
// userListOnDisk is what the user list's file holds on this machine now; nil when there is none.
func userListOnDisk(raw json.RawMessage) ([]byte, error) {
f, err := userListFile(raw)
if err != nil {
return nil, err
}
body, err := os.ReadFile(f.Path)
if errors.Is(err, os.ErrNotExist) {
return nil, nil
}
return body, err
}
// writeUserListAlone writes one user-list resource alone (apply.WriteAlone) with body as its content, the node's state
// read and saved under the apply's locks, and answers the digest of the file on the machine after.
func writeUserListAlone(ctx context.Context, opts options, raw []byte, body []byte, say link.Announce) (string, bool, error) {
declared, err := userListFile(raw)
if err != nil {
return "", false, err
}
file := *declared
file.Content, file.Digest, file.Size = string(body), "", 0
applying.Lock()
defer applying.Unlock()
unlock, err := store.Lock(opts.state, nil)
if err != nil {
return "", false, err
}
defer unlock()
known, err := store.Load(opts.state)
if err != nil {
return "", false, err
}
built, err := system.For(builtFor)
if err != nil {
return "", false, err
}
out, updated, writeErr := apply.WriteAlone(ctx, built, &file, known, apply.ExecRunner, announceOr(say))
// Saved whichever way it went, as an apply's: what was written is on the machine either way.
if err := store.Save(opts.state, updated); err != nil {
writeErr = errors.Join(writeErr, errors.New("written, and the node's state could not be saved: "+err.Error()))
}
digest := ""
if body, err := os.ReadFile(file.Path); err == nil {
digest = userlist.Digest(body)
}
return digest, writeErr == nil && out.Action != "unchanged", writeErr
}
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
golang.org/x/crypto v0.57.0
+4
View File
@@ -1,5 +1,9 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f h1:lUEyNPPF3w4Q0LIgcUUD6ofkKVhCF0n7dVjMoixQmXo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204213-f63fa86c7456 h1:He2o9vSKjgKbWOCOgOJY0WlaDh1n8RcGhvGY/dwXsBQ=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204213-f63fa86c7456/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e h1:k/1NjjW5HuuuHW2ldyWTvjp+1u3fN0RO18Q57Iup6Dk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025204730-0be0ffd6159e/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/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+67
View File
@@ -0,0 +1,67 @@
package apply
import (
"context"
"fmt"
"git.novox.be/novox/mesh-sdk/go/declaration"
"github.com/novox/mesh-host/internal/store"
"github.com/novox/mesh-host/internal/system"
)
// WriteAlone writes one file a module's composition declares, through the apply's own file step, and records it as an
// apply would; nothing else of the module is applied, and nothing that reads the file is moved (novox/hq ADR 0547 §8).
//
// **For a file whose reader takes the change itself.** The bus's user list is the one it is for: the bus module's
// entrypoint watches the file and signals the server to reload in place, so the list is in force without a want of the
// bus, which nobody writes (ADR 0539 §15), and without the container being touched. A file that a service or a container
// must be moved to read is not written alone: it goes through the module's apply, which moves what reads it.
//
// The file is written only when its content differs from what is there (the file step's own rule), read back, and
// recorded under the module whose resource it is, so the module's next apply finds it as its own and writes nothing.
// The caller holds the apply's lock. log is said the path, the action and the digest, never the content.
func WriteAlone(ctx context.Context, sys system.System, file *declaration.File, known store.State, run Runner,
log func(string)) (Outcome, store.State, error) {
if log == nil {
log = func(string) {}
}
if file.Into != "" || file.CreateOnce || file.Sealed != "" || file.VaultSealed != nil || len(file.Secrets) > 0 {
return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("%s is not a whole file of plain content, "+
"and only such a file is written alone; nothing was written", file.Path)}
}
if file.IsReference() {
return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("%s is given by digest %s and was not "+
"resolved to its content; it is not written empty", file.Path, file.Digest)}
}
if held, isHeld := known.HeldAt(file.ID); isHeld {
return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("%s is held as found for a module not "+
"yet taken (%s), so nothing was written", file.Path, held.Why)}
}
was, _ := known.Find(file.ID)
if was.ID == "" {
// Only over a file the mesh wrote: a file this node holds no record of is the module's apply's to take, with
// the original kept, never this write's. A record of the path under another id (the declaration before the
// switch named it so) is the one written to, so no second record of the file is left for an apply to remove.
rec, ok := known.At(string(declaration.TypeFile), file.Path)
if !ok {
return Outcome{}, known, &Error{Resource: file.ID, Err: fmt.Errorf("this node holds no record of %s; it "+
"is written by the module's apply first, never alone", file.Path)}
}
renamed := *file
renamed.ID = rec.ID
file, was = &renamed, rec
}
in := inputs{declares: map[string]string{file.Identity(): declaredDigest(file)}, known: &known,
groups: groupsWanted([]declaration.Resource{file}), handed: Handed{}, dirsBefore: map[string]bool{},
files: map[string]bool{file.Path: true}}
outcome, err := applyOne(ctx, sys, file, run, map[string]bool{}, in, was, nil, nil)
if err != nil {
log(fmt.Sprintf(" failed %s (%s), alone: %v", file.ID, file.Path, err))
return outcome, known, &Error{Resource: file.ID, Err: err}
}
recordKeeping(&known, file, outcome, was)
if moved(outcome) {
log(fmt.Sprintf(" %s %s (%s), alone: %s", outcome.Action, file.ID, file.Path, outcome.wrote))
}
return outcome, known, nil
}
+90
View File
@@ -0,0 +1,90 @@
package apply
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/declaration"
"github.com/novox/mesh-host/internal/store"
)
// Defends novox/hq ADR 0547 §8: the bus node writes the bus's user list alone, with no want, through the apply's own
// file step: only when it differs, mode 0600, read back and recorded as the module's, and nothing else moves.
func TestAFileWrittenAloneMovesNothingElse(t *testing.T) {
const was, now = "user a\n", "user a\nuser b\n"
for _, c := range []struct {
name, fields, before string
recorded bool
recordedAs string // the id the record has, when not the composition's
action, refused string
after string
}{
{name: "no record of it", fields: `"content":%q,"mode":"0600"`, before: was, refused: "holds no record", after: was},
{name: "changed", fields: `"content":%q,"mode":"0600"`, before: was, recorded: true, action: "updated", after: now},
{name: "the same", fields: `"content":%q,"mode":"0600"`, before: now, recorded: true, action: "unchanged", after: now},
{name: "recorded under the id from before the switch", fields: `"content":%q,"mode":"0600"`, before: was,
recorded: true, recordedAs: "bus-users", action: "updated", after: now},
{name: "absent with a record", fields: `"content":%q,"mode":"0600"`, recorded: true, action: "created", after: now},
{name: "written into", fields: `"content":%q,"into":"block","owner-at-runtime":"managed"`, before: was,
recorded: true, refused: "only such a file", after: was},
{name: "sealed", fields: `"sealed":"c2VhbGVk"`, before: was, recorded: true,
refused: "only such a file", after: was},
} {
t.Run(c.name, func(t *testing.T) {
path := filepath.Join(t.TempDir(), "accounts.conf")
if c.before != "" {
if err := os.WriteFile(path, []byte(c.before), 0o600); err != nil {
t.Fatal(err)
}
}
fields := strings.Replace(c.fields, "%q", fmt.Sprintf("%q", now), 1)
d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"resources":[{"id":"nats.bus-users","module":"nats",
"type":"file","path":%q,`+fields+`}]}`, path))
file := d.Resources[0].(*declaration.File)
known := store.State{}
id := file.ID
if c.recordedAs != "" {
id = c.recordedAs
}
if c.recorded {
known.Record(store.Applied{ID: id, Type: string(declaration.TypeFile), Module: "nats", Target: path,
Origin: store.OriginDeclared, Wrote: digestOf(c.before)})
}
calls := 0
counting := func(context.Context, string, ...string) (string, error) { calls++; return "", nil }
out, after, err := WriteAlone(context.Background(), archHost(t), file, known, counting, nil)
if c.refused != "" {
if err == nil || !strings.Contains(err.Error(), c.refused) {
t.Fatalf("not refused with %q: %v", c.refused, err)
}
} else if err != nil {
t.Fatal(err)
} else if out.Action != c.action {
t.Errorf("action %q, want %q", out.Action, c.action)
}
if got := readText(t, path); got != c.after {
t.Errorf("file holds %q, want %q", got, c.after)
}
if calls != 0 {
t.Errorf("%d commands were run: a file written alone moves nothing", calls)
}
if c.refused != "" {
return
}
if m := modeAt(t, path); m != 0o600 {
t.Errorf("mode %o", m)
}
rec, ok := after.Find(id)
if !ok || rec.Wrote != digestOf(now) || rec.Target != path {
t.Errorf("record %+v (found %v), want it to say %s was written", rec, ok, digestOf(now))
}
if n := len(after.Resources); n != 1 {
t.Errorf("%d records of one file", n)
}
})
}
}
+29 -29
View File
@@ -52,6 +52,24 @@ func TestAWriteOfAPathNotFilledIsRefused(t *testing.T) {
if _, err := os.Stat(other); err == nil {
t.Errorf("%s was written", other)
}
t.Run("a filled path two resources declare", func(t *testing.T) {
path := filepath.Join(t.TempDir(), "hosts")
if err := os.WriteFile(path, []byte(workstationHosts), 0o644); err != nil {
t.Fatal(err)
}
d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"filled_files":[%q],"resources":[
{"id":"a.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.1 ace\n","owner-at-runtime":"managed"},
{"id":"b.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.2 g14\n","owner-at-runtime":"managed"}
]}`, path, path, path))
_, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("6.6.6.6 evil\n"),
noServices, nil, nil, nil)
if err == nil || !strings.Contains(err.Error(), "2 resources declare") {
t.Fatalf("a path two resources declare was not refused: %v", err)
}
if got := readText(t, path); got != workstationHosts || len(state.Resources) != 0 {
t.Errorf("something was written or recorded: %q %+v", got, state.Resources)
}
})
}
// Each ownership is written by the write verb exactly as an apply of the declared file with the same content: the
@@ -112,6 +130,17 @@ func TestAFilledFileIsWrittenAsItsOwnershipSays(t *testing.T) {
}
})
}
t.Run("a filled file that carried a secret is not world-readable", func(t *testing.T) {
path := filepath.Join(t.TempDir(), "token.env")
d := filledDecl(t, path, `"content":"TOKEN=${secret:token}\n","secrets":{"token":"c2VhbGVk"},"owner-at-runtime":"owned"`, "")
if _, _, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("TOKEN=abc\n"),
noServices, nil, nil, nil); err != nil {
t.Fatal(err)
}
if m := modeAt(t, path); m != 0o600 {
t.Errorf("a filled file that carried a secret is mode %o", m)
}
})
}
func contentOf(t *testing.T, fields string) string {
@@ -136,17 +165,6 @@ func modeAt(t *testing.T, path string) os.FileMode {
// A file that carried a secret keeps the mode a secret is written with: the client filled the value, and the file
// is no more readable for it.
func TestAFilledFileThatCarriedASecretIsNotWorldReadable(t *testing.T) {
path := filepath.Join(t.TempDir(), "token.env")
d := filledDecl(t, path, `"content":"TOKEN=${secret:token}\n","secrets":{"token":"c2VhbGVk"},"owner-at-runtime":"owned"`, "")
if _, _, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("TOKEN=abc\n"),
noServices, nil, nil, nil); err != nil {
t.Fatal(err)
}
if m := modeAt(t, path); m != 0o600 {
t.Errorf("a filled file that carried a secret is mode %o", m)
}
}
func TestAWriteUnderReloadOnReloadsItsProgramOnce(t *testing.T) {
path := filepath.Join(t.TempDir(), "daemon.conf")
@@ -251,21 +269,3 @@ func count(lines []string, prefix string) int {
// Two resources declaring one filled path — two modules' blocks in one file — are refused: the request carries the
// content only, so it cannot say whose block it is, and nothing is written.
func TestAFilledPathTwoResourcesDeclareIsRefused(t *testing.T) {
path := filepath.Join(t.TempDir(), "hosts")
if err := os.WriteFile(path, []byte(workstationHosts), 0o644); err != nil {
t.Fatal(err)
}
d := parseTrusted(t, fmt.Sprintf(`{"declaration":1,"filled_files":[%q],"resources":[
{"id":"a.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.1 ace\n","owner-at-runtime":"managed"},
{"id":"b.hosts","type":"file","path":%q,"into":"block","content":"10.42.0.2 g14\n","owner-at-runtime":"managed"}
]}`, path, path, path))
_, state, err := WriteFilled(context.Background(), archHost(t), d, store.State{}, path, []byte("6.6.6.6 evil\n"),
noServices, nil, nil, nil)
if err == nil || !strings.Contains(err.Error(), "2 resources declare") {
t.Fatalf("a path two resources declare was not refused: %v", err)
}
if got := readText(t, path); got != workstationHosts || len(state.Resources) != 0 {
t.Errorf("something was written or recorded: %q %+v", got, state.Resources)
}
}
+24
View File
@@ -47,6 +47,9 @@ type haveState struct {
// said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted).
written map[string]string
removed map[string]string
// userList is what the bus node's engine last wrote of the bus's user list, by the module whose have says it (novox/hq
// ADR 0547 §8); kept apart from the module's fact, which every apply replaces.
userList map[string]delivery.UserListWritten
// unjudged is each module whose composition moved (an apply began, or the fact names another build) and that no look
// has judged since it stopped moving: its health is not yet judged, never the last look's of what ran before.
unjudged map[string]bool
@@ -308,6 +311,9 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
h.Phase = delivery.PhaseFailed
}
h.Health = s.healthOf(module, f)
if w, ok := s.userList[module]; ok {
h.UserList = &w
}
key, err := delivery.HaveKey(node, module)
if err != nil {
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
@@ -466,6 +472,24 @@ func (q *Session) NoteModule(ctx context.Context, f ModuleFact) {
q.writeHaves(ctx, bus)
}
// NoteUserList takes what the bus node's engine wrote of the bus's user list, and writes it in the bus module's have
// where it differs from the bucket's (novox/hq ADR 0547 §8). A module this engine has said no have of yet carries it
// from its first have on.
func (q *Session) NoteUserList(ctx context.Context, module string, w delivery.UserListWritten) {
q.init()
s := &q.haves
s.mu.Lock()
if s.userList == nil {
s.userList = map[string]delivery.UserListWritten{}
}
s.userList[module] = w
s.mu.Unlock()
q.mu.Lock()
bus := q.bus
q.mu.Unlock()
q.writeHaves(ctx, bus)
}
// NoteRemoved takes that a module's own resources were removed, and writes its have with the phase removed once (issue
// 1369): the key is never deleted.
func (q *Session) NoteRemoved(ctx context.Context, module, build string) {
+12
View File
@@ -484,6 +484,18 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
t.Fatalf("a failing check after the start window: %+v", got)
}
// The bus's user list written alone rides on the module's have, and an apply after it keeps it (ADR 0547 §8).
w := delivery.UserListWritten{Source: "n1.web." + fp, Revision: 4, Wanted: dig, Digest: dig, Users: 3,
At: time.Date(2026, 10, 25, 21, 0, 0, 0, time.UTC)}
q.NoteUserList(context.Background(), "web", w)
if got := l.have(t, "n1.web"); got.UserList == nil || got.UserList.Revision != 4 || got.FP != fp {
t.Fatalf("the user list is not in the have: %+v", got)
}
q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig})
if got := l.have(t, "n1.web"); got.UserList == nil || got.UserList.Users != 3 {
t.Fatalf("an apply lost the user list: %+v", got)
}
// A module held for a failed backup says so and what it keeps running; removed is written once, over its key.
q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig, Phase: string(delivery.PhaseHeld), Detail: "the repository is locked"})
if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseHeld || got.Detail == "" {
+337
View File
@@ -0,0 +1,337 @@
// Package userlist is the bridge of novox/hq ADR 0547 §8: the node agent of the node holding mesh-broker applies the
// bus's user list from the newest composition of the bus module the controller writes for its node, alone and with no
// want, so a new account is in force on the bus without anybody's hand.
//
// **Why it exists.** The list reached the bus only through a want of the bus module, and ADR 0539 §15 bars the assigner
// from writing one: on 2026-10-25 it stood from 11:23 to 20:50 and every new account was refused until the operator put
// it back by hand (issue 1945). The bus module's entrypoint already reloads the server in place when the file changes;
// what was missing is a writer of the file that does not wait on a want.
//
// **It watches, it never polls.** It follows the node's composition bucket. Each composition is verified as the agent
// verifies it (signed by the controller's or the operator's key) before anything of it is read. Of those that carry the
// bus's user list (a file resource `bus-users`), the newest by revision is the source; its list is written alone
// (apply.WriteAlone) when it differs from the file on the machine, and what was written is reported in the bus
// module's have. It looks again after every apply of the bus module, since an apply of an older composition writes an
// older list, 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"
"slices"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/agent"
)
// ResourceID is the id the composer gives the bus's user list (mesh-sdk catalogue.BusUsersID); in a composition it may
// carry its module before it, `<module>.bus-users`.
const ResourceID = "bus-users"
// Written is what the follower last did: the source it took the list from, the list's digest there, the file's digest
// on the machine after the write, how many users the list holds, when, and why a write failed.
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, as the agent trusts them.
Trusted func() []ed25519.PublicKey
// Resolve is the list a user-list resource holds: its content, or the object its digest names, fetched and
// verified (the controller sends a file over 64 KiB by digest, ADR 0486 §1, and the live list is over it).
Resolve func(ctx context.Context, resource json.RawMessage) ([]byte, error)
// 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
// differs from the file on the machine, and answers the digest of the file there after it (`sha256:<hex>`) and
// whether it changed.
Write func(ctx context.Context, module string, resource json.RawMessage, body []byte) (digest string, changed bool, err error)
// Report says what was written, in the module's have.
Report func(ctx context.Context, module string, w Written)
Now func() time.Time
Say func(string)
}
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.
type Follower struct {
env Env
kick chan struct{}
mu sync.Mutex
srcs map[string]source // by composition key
// cur is the source last taken, and written whether its write went through.
cur source
written bool
state Written
mod string
said map[string]bool
}
// New is a follower for a node.
func New(env Env) *Follower {
if env.Now == nil {
env.Now = time.Now
}
if env.Say == nil {
env.Say = func(string) {}
}
return &Follower{env: env, kick: make(chan struct{}, 1), srcs: map[string]source{}, said: map[string]bool{}}
}
// 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.
func (f *Follower) Again() {
select {
case f.kick <- struct{}{}:
default:
}
}
// State is the module whose list the follower writes ("" when no composition of this node 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.
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
for {
again := false
select {
case <-ctx.Done():
return nil
case <-f.kick:
again = true
case u, ok := <-updates:
if !ok {
return fmt.Errorf("the watch of the compositions ended")
}
if u.Caught {
caught = true
} else {
f.take(u)
}
}
if caught {
f.settle(ctx, again)
}
}
}
// 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 newest source's list when it is newer than the one written, or the one written again when asked.
// A source never goes back: a key deleted, or a newer one refused, leaves the list written as it is, since an older
// list would take away accounts a newer one gave.
func (f *Follower) settle(ctx context.Context, again bool) {
f.mu.Lock()
newest := f.cur
for _, s := range f.srcs {
if s.rev > newest.rev {
newest = s
}
}
if newest.key == "" || (newest.rev == f.cur.rev && f.written && !again) {
f.mu.Unlock()
return
}
f.cur = newest
at := fmt.Sprintf("%s@%d", newest.key, newest.rev)
f.mu.Unlock()
w := Written{Source: newest.key, Revision: newest.rev, Wanted: newest.wanted, At: f.env.Now().UTC()}
var changed bool
body, err := f.env.Resolve(ctx, newest.resource)
if err == nil {
w.Users = Users(body)
err = f.guard(newest.resource, body)
if err != nil {
// Not written: what is on the machine stays, and the have says why, loudly.
if disk, derr := f.env.OnDisk(newest.resource); derr == nil && disk != nil {
w.Digest = Digest(disk)
}
}
}
if err == nil {
w.Digest, changed, err = f.env.Write(ctx, newest.module, newest.resource, body)
}
if err != nil {
w.Error = err.Error()
}
f.mu.Lock()
prev := f.state
f.mod, f.state = newest.module, w
f.written = err == nil
f.mu.Unlock()
if err != nil {
f.sayOnce(fmt.Sprintf("the bus's user list of %s could not be written: %v", at, err))
} else if changed {
f.env.Say(fmt.Sprintf("the bus's user list from %s is written alone (%d users, %s); the bus reloads it itself",
at, w.Users, short(w.Digest)))
}
// Reported when what it says changed: a look again that found the file as it was says nothing new.
if f.env.Report != nil && (changed || prev.Source != w.Source || prev.Revision != w.Revision ||
prev.Digest != w.Digest || prev.Error != w.Error) {
f.env.Report(ctx, newest.module, w)
}
}
// ControllerUser is the controller's own user in the list: a list without it would lock the controller out of the bus.
const ControllerUser = "controller"
// guard refuses a list that would lock the mesh out of its own bus: one without the controller's own user, or one that
// drops more than half the users the file on the machine holds. A list that shrinks that far is a composer that read
// half the mesh, not half the mesh gone; it waits for a person, and the bus keeps the list it has.
func (f *Follower) guard(resource json.RawMessage, body []byte) error {
names := Names(body)
if !slices.Contains(names, ControllerUser) {
return fmt.Errorf("refused: the list holds no user %q, and written it would lock the controller out of the "+
"bus; the list on the machine is kept", ControllerUser)
}
if f.env.OnDisk == nil {
return nil
}
disk, err := f.env.OnDisk(resource)
if err != nil || disk == nil {
return nil
}
if had := Users(disk); had > 0 && len(names)*2 < had {
return fmt.Errorf("refused: the list holds %d users and the one on the machine %d; a list that drops more "+
"than half is not written, and the list on the machine is kept", len(names), had)
}
return nil
}
func (f *Follower) sayOnceLocked(line string) {
if !f.said[line] {
f.said[line] = true
f.env.Say(line)
}
}
func (f *Follower) sayOnce(line string) {
f.mu.Lock()
said := f.said[line]
f.said[line] = true
f.mu.Unlock()
if !said {
f.env.Say(line)
}
}
// userList is the bus's user list among a composition's resources: the resource, its content, and its digest when it
// is given by digest rather than content.
func userList(resources []json.RawMessage) (json.RawMessage, string, string, bool) {
for _, raw := range resources {
var head struct {
ID string `json:"id"`
Type string `json:"type"`
Content string `json:"content"`
Digest string `json:"digest"`
}
if json.Unmarshal(raw, &head) != nil || head.Type != "file" {
continue
}
if head.ID == ResourceID || strings.HasSuffix(head.ID, "."+ResourceID) {
return raw, head.Content, head.Digest, true
}
}
return nil, "", "", false
}
// Digest is content's digest as the have says it, `sha256:<hex>`.
func Digest(content []byte) string {
sum := sha256.Sum256(content)
return "sha256:" + hex.EncodeToString(sum[:])
}
// userEntry is one user of the list as the composer writes it (and the bus module's own tools read it).
var userEntry = regexp.MustCompile(`\{\s*user:\s*"([^"]+)"`)
// Users is how many users a list holds.
func Users(content []byte) int { return len(userEntry.FindAllIndex(content, -1)) }
// Names is the users a list holds, as it names them.
func Names(content []byte) []string {
var out []string
for _, m := range userEntry.FindAllSubmatch(content, -1) {
out = append(out, string(m[1]))
}
return out
}
func short(d string) string {
d = strings.TrimPrefix(d, "sha256:")
if len(d) > 12 {
d = d[:12]
}
return d
}
+291
View File
@@ -0,0 +1,291 @@
package userlist
import (
"context"
"crypto/ed25519"
"encoding/json"
"errors"
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/agent"
)
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.
type feed struct {
ch chan agent.Update
rev uint64
first []agent.Update
}
func (f *feed) Watch(context.Context, string) (<-chan agent.Update, error) {
for _, u := range f.first {
f.ch <- u
}
f.ch <- agent.Update{Caught: true}
return f.ch, nil
}
func list(users ...string) string {
var b strings.Builder
b.WriteString("authorization {\n users = [\n")
for _, u := range users {
b.WriteString(` {user: "` + u + `", password: "$2a$11$x"}` + "\n")
}
b.WriteString(" ]\n}\n")
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,
"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,
"path": "/var/lib/nats-module/conf/accounts.conf", "content": content, "mode": "0600"}
if p.byDigest {
delete(r, "content")
r["digest"], r["size"] = Digest([]byte(content)), len(content)
}
users, _ := json.Marshal(r)
rs = append(rs, users)
}
c := delivery.Composed{Node: node, Module: module, FP: fp, Resources: rs}
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
}
type put struct {
module, fp, content string
stranger, deleted bool
byDigest bool
asNode, asModule string
}
// Defends novox/hq ADR 0547 §8: the bus node writes the newest signed list alone, only when it changed, and reports it.
func TestTheBusNodeWritesTheNewestSignedUserListAlone(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")
for _, c := range []struct {
name string
before, after []put
again bool
failWrite bool
onDisk string
wantWrites []string // contents written, in order
wantUsers int
wantErr string
wantRefused bool
}{
{name: "the list there is written once", before: []put{{module: "nats", fp: fpOf("a"), content: two}},
wantWrites: []string{two}, wantUsers: 3},
{name: "a newer revision is written, the older fingerprint is not",
before: []put{{module: "nats", fp: fpOf("a"), content: one}, {module: "nats", fp: fpOf("b"), content: two}},
after: []put{{module: "nats", fp: fpOf("a"), content: three}},
wantWrites: []string{two, three}, wantUsers: 4},
{name: "a composition signed by a stranger is no source",
before: []put{{module: "nats", fp: fpOf("a"), content: one}},
after: []put{{module: "nats", fp: fpOf("b"), content: three, stranger: true}},
wantWrites: []string{one}, wantUsers: 2},
{name: "a composition without the list is ignored", before: []put{{module: "web", fp: fpOf("a"), content: ""}}},
{name: "a deleted key is no source",
before: []put{{module: "nats", fp: fpOf("a"), content: one}, {module: "nats", fp: fpOf("b"), content: two}},
after: []put{{module: "nats", fp: fpOf("b"), content: "", deleted: true}, {module: "nats", fp: fpOf("a"), content: three}},
wantWrites: []string{two, three}, wantUsers: 4},
{name: "asked again, the same source is written again", before: []put{{module: "nats", fp: fpOf("a"), content: two}},
again: true, wantWrites: []string{two, two}, wantUsers: 3},
{name: "a list sent by digest counts the users of what it names",
before: []put{{module: "nats", fp: fpOf("a"), content: three, byDigest: true}},
wantWrites: []string{three}, wantUsers: 4},
{name: "another node's composition under this node's key is no source",
before: []put{{module: "nats", fp: fpOf("a"), content: two, asNode: "ace"}}},
{name: "another module's composition under the bus module's key is no source",
before: []put{{module: "nats", fp: fpOf("a"), content: two, asModule: "web"}}},
{name: "a list without the controller is refused and the file kept",
before: []put{{module: "nats", fp: fpOf("a"), content: list("a", "b")}},
onDisk: two, wantUsers: 2, wantErr: "holds no user \"controller\"", wantRefused: true},
{name: "a list that drops more than half is refused and the file kept",
before: []put{{module: "nats", fp: fpOf("a"), content: one}},
onDisk: five, wantUsers: 2, wantErr: "drops more than half", wantRefused: true},
{name: "a list that drops half is written", before: []put{{module: "nats", fp: fpOf("a"), content: two}},
onDisk: list("controller", "a", "b", "c", "d", "e"), wantWrites: []string{two}, wantUsers: 3},
{name: "a failed write is said and kept", before: []put{{module: "nats", fp: fpOf("a"), content: two}}, failWrite: true,
wantWrites: []string{two}, wantUsers: 3, wantErr: "disk full"},
} {
t.Run(c.name, func(t *testing.T) {
f := &feed{ch: make(chan agent.Update, 64)}
objects := map[string]string{}
var mu sync.Mutex
var writes []string
var reports []Written
reported := make(chan struct{}, 64)
fl := New(Env{Node: node, Composed: f,
Trusted: func() []ed25519.PublicKey { return []ed25519.PublicKey{controllerPub} },
Resolve: func(_ context.Context, r json.RawMessage) ([]byte, error) {
var res struct{ Content, Digest string }
_ = json.Unmarshal(r, &res)
if res.Digest != "" {
body, ok := objects[res.Digest]
if !ok {
return nil, errors.New("no object " + res.Digest)
}
return []byte(body), nil
}
return []byte(res.Content), nil
},
OnDisk: func(json.RawMessage) ([]byte, error) {
if c.onDisk == "" {
return nil, nil
}
return []byte(c.onDisk), nil
},
Write: func(_ context.Context, module string, r json.RawMessage, body []byte) (string, bool, error) {
mu.Lock()
writes = append(writes, string(body))
mu.Unlock()
if c.failWrite {
return "", false, errors.New("disk full")
}
return Digest(body), true, nil
},
Report: func(_ context.Context, module string, w Written) {
mu.Lock()
reports = append(reports, w)
mu.Unlock()
reported <- struct{}{}
},
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))
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() { done <- fl.Run(ctx) }()
for _, p := range c.after {
f.ch <- update(p)
}
if c.again {
// After the first write is reported, so the look again is not folded into it.
select {
case <-reported:
case <-time.After(5 * time.Second):
}
fl.Again()
}
deadline := time.After(5 * time.Second)
for {
mu.Lock()
n := len(writes)
mu.Unlock()
if n >= len(c.wantWrites) {
break
}
select {
case <-deadline:
t.Fatalf("writes %d, want %d", n, len(c.wantWrites))
case <-time.After(5 * time.Millisecond):
}
}
if c.wantRefused {
select {
case <-reported:
case <-time.After(5 * time.Second):
t.Fatal("a refused list was not reported")
}
}
time.Sleep(50 * time.Millisecond) // nothing more is written
cancel()
<-done
mu.Lock()
defer mu.Unlock()
if strings.Join(writes, "|") != strings.Join(c.wantWrites, "|") {
t.Fatalf("written %q, want %q", writes, c.wantWrites)
}
module, st := fl.State()
if c.wantRefused {
if !strings.Contains(st.Error, c.wantErr) || st.Digest != Digest([]byte(c.onDisk)) || st.Users != c.wantUsers {
t.Fatalf("refused list's state %+v", st)
}
return
}
if len(c.wantWrites) == 0 {
if module != "" || len(reports) != 0 {
t.Fatalf("a node with no list wrote or said something: %q %+v", module, reports)
}
return
}
if module != "nats" || st.Users != c.wantUsers || st.Wanted != Digest([]byte(c.wantWrites[len(c.wantWrites)-1])) {
t.Errorf("state %q %+v", module, st)
}
if c.wantErr != "" {
if !strings.Contains(st.Error, c.wantErr) || st.Digest != "" {
t.Errorf("a failed write is not kept: %+v", st)
}
} else if st.Digest != st.Wanted || st.Error != "" {
t.Errorf("written %+v", st)
}
if len(reports) == 0 || reports[len(reports)-1] != st {
t.Errorf("reports %+v, last state %+v", reports, st)
}
})
}
}