mesh/delivery merged: source.main moved to novox/mesh-host@d75c66dc7e27
mesh/delivery-group group fix/1945-user-list-follows merged: every member is merged
mesh/check pass 5e71ab6ae612 on d75c66dc7e27 | unit 303 run, 1077 kept, 5 skipped | integration 306 run, 1079 kept, 4 skipped | over bound: unit, inte…
B1: the controller sends the user list by digest (over 64 KiB), so its users are counted from the list as resolved, not from the composition's inline content, which is empty then. R1: a composition signed for another node or module than its key names is no source. R2: a list without the controller's own user, or one that drops more than half the users of the file on the machine, is not written; the have says why and the file on the machine is kept.
338 lines
11 KiB
Go
338 lines
11 KiB
Go
// 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
|
|
}
|