Archived
Feature 57 (config and provisioning): release mesh-controller, the reviewed release branch merged with main, on top of main
This commit is contained in:
@@ -0,0 +1,348 @@
|
||||
// Package recordmirror writes the controller's record mirrors (novox/hq ADR 0289, design 54): a bucket per kind
|
||||
// of record, one key per record, written from the store only when the record changed, and one event on the
|
||||
// controller's seat per write.
|
||||
//
|
||||
// **The store stays the record.** A mirror is built from what the store committed, so a change rolled back
|
||||
// is never written. Every pass reads what the bucket holds and compares it byte for byte with what the store
|
||||
// says: a pass over unchanged records writes nothing and says nothing (ADR 0289 §1), and a value somebody
|
||||
// else wrote behind the writer's back (a former lease holder) differs from the store and is put right on
|
||||
// the next pass. Each write names the revision it read (compare-and-set), so a write in between is noticed,
|
||||
// never overwritten blind.
|
||||
//
|
||||
// **Only the lease holder writes** (ADR 0289 §9, design 54 §2): a pass begins by asking whether this
|
||||
// controller holds the lease, and asks again before every write; without it, it writes nothing.
|
||||
//
|
||||
// **The bucket is the authority, the event a notice** (design 54 §3): the event is sent after the write
|
||||
// succeeded, so there is no event for a value the bucket does not hold. A refused event is tried once more
|
||||
// and then counted as lost; a watcher that must not miss a change watches the bucket.
|
||||
//
|
||||
// What this package does not do yet is design 54 §4's audit (probe D17 and its healer H6), which compares a
|
||||
// mirror with its store behind the writer's back; novox/hq issue 419 carries it.
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// A Mirror is one kind of record the controller keeps on the bus: its local name, its bucket and the event
|
||||
// each write raises on the controller's seat.
|
||||
type Mirror struct {
|
||||
Name string
|
||||
Bucket string
|
||||
Event string
|
||||
}
|
||||
|
||||
// The mirrors this package writes (novox/hq design 57, package F3). The assignments mirror is design 54's,
|
||||
// keyed `<node>.<module>`; the declared config is each registered module's config declarations as its
|
||||
// manifest holds them, keyed by the module, which mesh-config resolves a module's config values from
|
||||
// (ADR 0348 clauses 2 and 4).
|
||||
var (
|
||||
Assignments = Mirror{Name: broker.MirrorAssignments, Bucket: broker.AssignmentsBucket,
|
||||
Event: link.KeyAssignmentChanged}
|
||||
DeclaredConfig = Mirror{Name: broker.MirrorDeclaredConfig, Bucket: broker.DeclaredConfigBucket,
|
||||
Event: link.KeyDeclaredConfigChanged}
|
||||
// Predefined is the list of predefined keys the controller kept (ADR 0348 §7), one key `list`, which
|
||||
// mesh-config publishes as its own (design 57 K3).
|
||||
Predefined = Mirror{Name: broker.MirrorPredefined, Bucket: broker.PredefinedBucket,
|
||||
Event: link.KeyPredefinedChanged}
|
||||
)
|
||||
|
||||
// Mirrors is every mirror this package writes, in the order they are written.
|
||||
func Mirrors() []Mirror { return []Mirror{Assignments, DeclaredConfig, Predefined} }
|
||||
|
||||
// The changes an event says.
|
||||
const (
|
||||
Added = "added"
|
||||
Changed = "changed"
|
||||
Removed = "removed"
|
||||
)
|
||||
|
||||
// Event is what one write says on the controller's seat (design 54 §3): which mirror and key, what kind of
|
||||
// change, the names of the top-level fields that changed, the bucket's revision of the write, the lease's
|
||||
// epoch and when. Never the value: a watcher reads it from the bucket at that revision or later.
|
||||
type Event struct {
|
||||
Mirror string `json:"mirror"`
|
||||
Key string `json:"key"`
|
||||
Change string `json:"change"`
|
||||
Fields []string `json:"fields,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Epoch uint64 `json:"epoch,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Announcer says an event on a seat; link.OverNATS is one, and so is link.Bus.
|
||||
type Announcer interface {
|
||||
PublishSeatEvent(ctx context.Context, seat, event string, body []byte) error
|
||||
}
|
||||
|
||||
// MarkerTTL is how long the marker a purge leaves is kept (design 54 §2): a day, so the markers of removed
|
||||
// records do not pile up in a bucket that has no age.
|
||||
const MarkerTTL = 24 * time.Hour
|
||||
|
||||
// ErrNotHeld is a pass refused because this controller does not hold the lease: it wrote nothing.
|
||||
var ErrNotHeld = errors.New("this controller does not hold the lease, so it writes no mirror")
|
||||
|
||||
// Writer writes one mirror. Not safe for use by two goroutines at once: one controller's one loop owns it.
|
||||
type Writer struct {
|
||||
Mirror Mirror
|
||||
KV jetstream.KeyValue
|
||||
Announce Announcer
|
||||
// Epoch is the lease's epoch the events carry; nil carries none.
|
||||
Epoch func() uint64
|
||||
// Holds says this controller holds the lease. Asked before every pass and every write; a writer
|
||||
// without it writes nothing (design 54 §2). Required: nil holds nothing.
|
||||
Holds func() bool
|
||||
// Now is the clock the events are dated by; nil is time.Now.
|
||||
Now func() time.Time
|
||||
|
||||
known map[string]kept
|
||||
// LostEvents counts the events refused twice since the writer was made.
|
||||
LostEvents int
|
||||
}
|
||||
|
||||
type kept struct {
|
||||
value []byte
|
||||
revision uint64
|
||||
}
|
||||
|
||||
// Result is what one pass did: keys written, keys removed, and keys refused with why.
|
||||
type Result struct {
|
||||
Written int
|
||||
Removed int
|
||||
Refused int
|
||||
}
|
||||
|
||||
// Sync makes the bucket hold want, one key per record: writes each value that differs from what the key
|
||||
// holds, purges each key no record is left for, and raises one event per write. A key the bus cannot hold
|
||||
// is refused alone, and the others are written; the error then names it.
|
||||
func (w *Writer) Sync(ctx context.Context, want map[string][]byte) (Result, error) {
|
||||
var r Result
|
||||
if !w.holds() {
|
||||
return r, ErrNotHeld
|
||||
}
|
||||
// What the bucket holds now, every pass: a value written behind this writer's back differs from the
|
||||
// store's and is written again below.
|
||||
if err := w.fill(ctx); err != nil {
|
||||
return r, fmt.Errorf("reading the %s mirror: %w", w.Mirror.Name, err)
|
||||
}
|
||||
var problems []error
|
||||
keys := make([]string, 0, len(want))
|
||||
for k := range want {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
for _, key := range keys {
|
||||
value := want[key]
|
||||
before, had := w.known[key]
|
||||
if had && bytes.Equal(before.value, value) {
|
||||
continue
|
||||
}
|
||||
if err := validKey(key); err != nil {
|
||||
r.Refused++
|
||||
problems = append(problems, err)
|
||||
continue
|
||||
}
|
||||
if !w.holds() {
|
||||
return r, errors.Join(append(problems, ErrNotHeld)...)
|
||||
}
|
||||
revision, wrote, err := w.put(ctx, key, value, before, had)
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Errorf("writing %s: %w", key, err))
|
||||
continue
|
||||
}
|
||||
w.known[key] = kept{value: append([]byte(nil), value...), revision: revision}
|
||||
if !wrote {
|
||||
continue
|
||||
}
|
||||
r.Written++
|
||||
change, fields := Added, []string(nil)
|
||||
if had {
|
||||
change, fields = Changed, changedFields(before.value, value)
|
||||
}
|
||||
w.say(ctx, key, change, fields, revision)
|
||||
}
|
||||
gone := make([]string, 0)
|
||||
for key := range w.known {
|
||||
if _, ok := want[key]; !ok {
|
||||
gone = append(gone, key)
|
||||
}
|
||||
}
|
||||
sort.Strings(gone)
|
||||
for _, key := range gone {
|
||||
k := w.known[key]
|
||||
if !w.holds() {
|
||||
return r, errors.Join(append(problems, ErrNotHeld)...)
|
||||
}
|
||||
err := w.KV.Purge(ctx, key, jetstream.LastRevision(k.revision), jetstream.PurgeTTL(MarkerTTL))
|
||||
if refusedByRevision(err) {
|
||||
// Read it again: purged already, or moved by another writer.
|
||||
if e, gerr := w.KV.Get(ctx, key); errors.Is(gerr, jetstream.ErrKeyNotFound) {
|
||||
delete(w.known, key)
|
||||
continue
|
||||
} else if gerr == nil && w.holds() {
|
||||
err = w.KV.Purge(ctx, key, jetstream.LastRevision(e.Revision()), jetstream.PurgeTTL(MarkerTTL))
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Errorf("removing %s: %w", key, err))
|
||||
continue
|
||||
}
|
||||
delete(w.known, key)
|
||||
r.Removed++
|
||||
revision := uint64(0)
|
||||
if status, err := w.KV.Status(ctx); err == nil {
|
||||
// The purge's own sequence is the stream's last one only when nothing wrote in between, which no
|
||||
// other writer may; said as the bucket's revision after it.
|
||||
if s, ok := status.(interface{ StreamInfo() *jetstream.StreamInfo }); ok {
|
||||
revision = s.StreamInfo().State.LastSeq
|
||||
}
|
||||
}
|
||||
w.say(ctx, key, Removed, nil, revision)
|
||||
}
|
||||
return r, errors.Join(problems...)
|
||||
}
|
||||
|
||||
// put writes one value by compare-and-set and answers its revision, and whether this call wrote it (false
|
||||
// when the bucket already held it).
|
||||
func (w *Writer) put(ctx context.Context, key string, value []byte, before kept, had bool) (uint64, bool, error) {
|
||||
var (
|
||||
revision uint64
|
||||
err error
|
||||
)
|
||||
if had {
|
||||
revision, err = w.KV.Update(ctx, key, value, before.revision)
|
||||
} else {
|
||||
revision, err = w.KV.Create(ctx, key, value)
|
||||
}
|
||||
if !refusedByRevision(err) {
|
||||
return revision, err == nil, err
|
||||
}
|
||||
// Refused: another writer in between, or this writer's own write whose acknowledgement was lost.
|
||||
e, gerr := w.KV.Get(ctx, key)
|
||||
switch {
|
||||
case errors.Is(gerr, jetstream.ErrKeyNotFound):
|
||||
if !w.holds() {
|
||||
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
|
||||
}
|
||||
revision, err = w.KV.Create(ctx, key, value)
|
||||
return revision, err == nil, err
|
||||
case gerr != nil:
|
||||
return 0, false, gerr
|
||||
case bytes.Equal(e.Value(), value):
|
||||
return e.Revision(), false, nil
|
||||
case !w.holds():
|
||||
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
|
||||
}
|
||||
revision, err = w.KV.Update(ctx, key, value, e.Revision())
|
||||
return revision, err == nil, err
|
||||
}
|
||||
|
||||
// say raises one write's event, tried once more when refused and then counted as lost.
|
||||
func (w *Writer) say(ctx context.Context, key, change string, fields []string, revision uint64) {
|
||||
if w.Announce == nil {
|
||||
return
|
||||
}
|
||||
e := Event{Mirror: w.Mirror.Name, Key: key, Change: change, Fields: fields, Revision: revision, At: w.now().UTC()}
|
||||
if w.Epoch != nil {
|
||||
e.Epoch = w.Epoch()
|
||||
}
|
||||
body, err := json.Marshal(e)
|
||||
if err != nil {
|
||||
w.LostEvents++
|
||||
return
|
||||
}
|
||||
for try := 0; try < 2; try++ {
|
||||
if err = w.Announce.PublishSeatEvent(ctx, broker.ControllerSeat, w.Mirror.Event, body); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
w.LostEvents++
|
||||
}
|
||||
|
||||
// fill reads what the bucket holds now, so a controller started again writes only what differs.
|
||||
func (w *Writer) fill(ctx context.Context) error {
|
||||
w.known = map[string]kept{}
|
||||
watch, err := w.KV.WatchAll(ctx, jetstream.IgnoreDeletes())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = watch.Stop() }()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case e := <-watch.Updates():
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
w.known[e.Key()] = kept{value: append([]byte(nil), e.Value()...), revision: e.Revision()}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Writer) holds() bool { return w.Holds != nil && w.Holds() }
|
||||
|
||||
func (w *Writer) now() time.Time {
|
||||
if w.Now == nil {
|
||||
return time.Now()
|
||||
}
|
||||
return w.Now()
|
||||
}
|
||||
|
||||
// refusedByRevision is the server refusing a compare-and-set: the key is not at the revision named.
|
||||
func refusedByRevision(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, jetstream.ErrKeyExists) {
|
||||
return true
|
||||
}
|
||||
var apiErr *jetstream.APIError
|
||||
return errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
|
||||
}
|
||||
|
||||
// keyChars is what a bucket's key may be made of.
|
||||
var keyChars = regexp.MustCompile(`^[-/_=.a-zA-Z0-9]+$`)
|
||||
|
||||
// validKey refuses a key the bus cannot hold, naming it.
|
||||
func validKey(key string) error {
|
||||
if key == "" || !keyChars.MatchString(key) || key[0] == '.' || key[len(key)-1] == '.' {
|
||||
return fmt.Errorf("%q cannot be a key of a bucket: letters, digits, '-', '_', '=', '/' and '.', not at "+
|
||||
"either end", key)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// changedFields names the top-level fields whose values differ between two encoded objects, sorted; nil
|
||||
// when either is not an object.
|
||||
func changedFields(before, after []byte) []string {
|
||||
var a, b map[string]json.RawMessage
|
||||
if json.Unmarshal(before, &a) != nil || json.Unmarshal(after, &b) != nil || a == nil || b == nil {
|
||||
return nil
|
||||
}
|
||||
var out []string
|
||||
for k, v := range b {
|
||||
if old, ok := a[k]; !ok || !bytes.Equal(old, v) {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
for k := range a {
|
||||
if _, ok := b[k]; !ok {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// **Design 57's F3, end to end** (novox/hq issue 652): from the controller's store to a real bus. An
|
||||
// assignment changes exactly one key of the assignments mirror and raises exactly one `assignment-changed`;
|
||||
// a registration changes exactly one key of the declared-config mirror and raises exactly one
|
||||
// `declared-config-changed`; and the same records written again change nothing and say nothing — a module
|
||||
// registered again unchanged included, although the store's registration time moved.
|
||||
func TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent(t *testing.T) {
|
||||
inv := inventory.ForTest(t)
|
||||
ctx := t.Context()
|
||||
js, api := aMirrorBus(t)
|
||||
assignments := aWriter(t, js, api, Assignments)
|
||||
declared := aWriter(t, js, api, DeclaredConfig)
|
||||
heardAssignments := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
|
||||
heardDeclared := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+DeclaredConfig.Event)
|
||||
|
||||
pass := func() (Result, Result) {
|
||||
t.Helper()
|
||||
a, err := inv.AssignmentsMirror(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, err := assignments.Sync(ctx, a)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d, err := inv.DeclaredConfigMirror(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rd, err := declared.Sync(ctx, d)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return ra, rd
|
||||
}
|
||||
|
||||
if _, err := inv.AddNode(ctx, "anchor"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ui := catalogue.Manifest{Module: "ui", Version: "1", Settings: map[string]catalogue.SettingDeclaration{
|
||||
"font-size": {Kind: catalogue.KindPreference, Default: 14, Why: "readable on the desktop"}}}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pass()
|
||||
assignmentWrites, declaredWrites := writes(t, api, Assignments.Bucket), writes(t, api, DeclaredConfig.Bucket)
|
||||
assignmentsSaid, declaredSaid := len(settle(heardAssignments, 0)), len(settle(heardDeclared, 1))
|
||||
|
||||
// **A registration**: a module new to the catalogue.
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "notes", Version: "1"}, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, rd := pass()
|
||||
if ra != (Result{}) || rd != (Result{Written: 1}) {
|
||||
t.Errorf("a registration wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
|
||||
t.Errorf("a registration changed %d keys of the declared config", got)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 0 {
|
||||
t.Errorf("a registration changed %d keys of the assignments", got)
|
||||
}
|
||||
said := settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Added {
|
||||
t.Errorf("a registration raised %+v", said)
|
||||
}
|
||||
declaredWrites, declaredSaid = writes(t, api, DeclaredConfig.Bucket), declaredSaid+len(said)
|
||||
|
||||
// **An assignment.**
|
||||
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, rd = pass()
|
||||
if ra != (Result{Written: 1}) || rd != (Result{}) {
|
||||
t.Errorf("an assignment wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 1 {
|
||||
t.Errorf("an assignment changed %d keys of the assignments", got)
|
||||
}
|
||||
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
|
||||
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Added {
|
||||
t.Errorf("an assignment raised %+v", said)
|
||||
}
|
||||
assignmentWrites, assignmentsSaid = writes(t, api, Assignments.Bucket), assignmentsSaid+len(said)
|
||||
e, err := assignments.KV.Get(ctx, "anchor.ui")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var value inventory.AssignmentMirrored
|
||||
if err := json.Unmarshal(e.Value(), &value); err != nil || value.Node != "anchor" || value.Module != "ui" ||
|
||||
value.Assigned.IsZero() {
|
||||
t.Errorf("the assignment is mirrored as %s (%v)", e.Value(), err)
|
||||
}
|
||||
|
||||
// **The same records written again change nothing**: an assignment made again, and a module registered
|
||||
// again with the manifest it had.
|
||||
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, rd := pass(); ra != (Result{}) || rd != (Result{}) {
|
||||
t.Errorf("unchanged records wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if writes(t, api, Assignments.Bucket) != assignmentWrites || writes(t, api, DeclaredConfig.Bucket) != declaredWrites {
|
||||
t.Error("unchanged records changed a key")
|
||||
}
|
||||
if a, d := settle(heardAssignments, assignmentsSaid+1), settle(heardDeclared, declaredSaid+1); len(a) != assignmentsSaid ||
|
||||
len(d) != declaredSaid {
|
||||
t.Errorf("unchanged records raised %+v and %+v", a[assignmentsSaid:], d[declaredSaid:])
|
||||
}
|
||||
|
||||
// **A declaration changed by a registration**: one key, one event, naming the field that moved.
|
||||
ui.Version = "2"
|
||||
ui.Settings["font-size"] = catalogue.SettingDeclaration{Kind: catalogue.KindPreference, Default: 16,
|
||||
Why: "readable on the desktop"}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, rd := pass(); ra != (Result{}) || rd != (Result{Written: 1}) {
|
||||
t.Errorf("a changed declaration wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "ui" || said[0].Change != Changed || strings.Join(said[0].Fields, ",") != "settings" {
|
||||
t.Errorf("a changed declaration raised %+v", said)
|
||||
}
|
||||
got, err := declared.KV.Get(ctx, "ui")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := `{"module":"ui","settings":{"font-size":{"kind":"preference","default":16,"why":"readable on the desktop"}}}`; string(got.Value()) != want {
|
||||
t.Errorf("the declaration is mirrored as %s, want %s", got.Value(), want)
|
||||
}
|
||||
|
||||
// **An unassignment** removes its key, said once.
|
||||
if err := inv.Unassign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, _ := pass(); ra != (Result{Removed: 1}) {
|
||||
t.Errorf("an unassignment did %+v", ra)
|
||||
}
|
||||
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
|
||||
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Removed {
|
||||
t.Errorf("an unassignment raised %+v", said)
|
||||
}
|
||||
|
||||
// **A module no longer registered** has its key removed, said once as `removed` (ADR 0356 §4).
|
||||
if err := inv.ForgetModule(ctx, "notes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
declaredWrites = writes(t, api, DeclaredConfig.Bucket)
|
||||
declaredSaid = len(heardDeclared.all())
|
||||
if _, rd := pass(); rd != (Result{Removed: 1}) {
|
||||
t.Errorf("a module forgotten did %+v", rd)
|
||||
}
|
||||
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
|
||||
t.Errorf("a module forgotten changed %d keys of the declared config", got)
|
||||
}
|
||||
if _, err := declared.KV.Get(ctx, "notes"); err == nil {
|
||||
t.Error("a module forgotten keeps its key")
|
||||
}
|
||||
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Removed {
|
||||
t.Errorf("a module forgotten raised %+v", said)
|
||||
}
|
||||
}
|
||||
|
||||
// **The list of predefined keys kept, on the bus** (design 57 K3, issue 662): mesh-config publishes the list the
|
||||
// controller kept, so the controller mirrors it. No list kept writes no key; a list kept changes the one key and
|
||||
// raises one `predefined-changed`; an older build's list heard later is not kept and changes nothing; a newer one
|
||||
// changes the key again, naming the fields that moved.
|
||||
func TestNatsTheListOfPredefinedKeysKeptChangesTheMirrorsOneKey(t *testing.T) {
|
||||
inv := inventory.ForTest(t)
|
||||
ctx := t.Context()
|
||||
js, api := aMirrorBus(t)
|
||||
predefined := aWriter(t, js, api, Predefined)
|
||||
heard := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Predefined.Event)
|
||||
pass := func() Result {
|
||||
t.Helper()
|
||||
want, err := inv.PredefinedMirror(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r, err := predefined.Sync(ctx, want)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return r
|
||||
}
|
||||
keep := func(generation int, keys catalogue.PredefinedList) {
|
||||
t.Helper()
|
||||
if _, err := inv.KeepPredefined(ctx, inventory.PredefinedRecord{Keys: keys, Repository: "mesh-catalog",
|
||||
Commit: strings.Repeat("a", 39) + string(rune('0'+generation)), Generation: generation,
|
||||
Build: "b" + string(rune('0'+generation))}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
six, seventy := 6.0, 72.0
|
||||
fontSize := catalogue.PredefinedKey{Type: catalogue.ConfigNumber, Description: "text size, in points",
|
||||
Minimum: &six, Maximum: &seventy}
|
||||
|
||||
if r := pass(); r != (Result{}) {
|
||||
t.Errorf("no list kept wrote %+v", r)
|
||||
}
|
||||
keep(5, catalogue.PredefinedList{"font-size": fontSize})
|
||||
if r := pass(); r != (Result{Written: 1}) {
|
||||
t.Errorf("a list kept wrote %+v", r)
|
||||
}
|
||||
said := settle(heard, 1)
|
||||
if len(said) != 1 || said[0].Key != inventory.PredefinedMirrorKey || said[0].Change != Added {
|
||||
t.Errorf("a list kept raised %+v", said)
|
||||
}
|
||||
e, err := predefined.KV.Get(ctx, inventory.PredefinedMirrorKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var value inventory.PredefinedMirrored
|
||||
if err := json.Unmarshal(e.Value(), &value); err != nil || value.Generation != 5 ||
|
||||
string(value.Keys) != `{"font-size":{"type":"number","description":"text size, in points","minimum":6,"maximum":72}}` {
|
||||
t.Errorf("the list is mirrored as %s (%v)", e.Value(), err)
|
||||
}
|
||||
|
||||
// An older build's list heard later is not kept, so nothing moves.
|
||||
keep(3, catalogue.PredefinedList{})
|
||||
if r := pass(); r != (Result{}) {
|
||||
t.Errorf("an older list wrote %+v", r)
|
||||
}
|
||||
if got := settle(heard, 2); len(got) != 1 {
|
||||
t.Errorf("an older list raised %+v", got[1:])
|
||||
}
|
||||
|
||||
// A newer one is: one write, naming what moved.
|
||||
keep(7, catalogue.PredefinedList{"font-size": fontSize, "theme": {Type: catalogue.ConfigString,
|
||||
Description: "the desktop's colour theme"}})
|
||||
if r := pass(); r != (Result{Written: 1}) {
|
||||
t.Errorf("a newer list wrote %+v", r)
|
||||
}
|
||||
said = settle(heard, 2)[1:]
|
||||
if len(said) != 1 || said[0].Change != Changed || strings.Join(said[0].Fields, ",") != "build,commit,generation,keys" {
|
||||
t.Errorf("a newer list raised %+v", said)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,356 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// A record mirror against a real bus (novox/hq design 54 §2, §3; ADR 0289): a change of a record changes
|
||||
// exactly one key and raises exactly one event; a pass over unchanged records writes nothing and says
|
||||
// nothing; a record gone is a purge said as `removed`.
|
||||
|
||||
// aMirrorBus is a bus of this test alone with the mesh's streams and the controller's buckets, as the
|
||||
// controller asserts them at its start, and a counter of the writes made to one bucket.
|
||||
func aMirrorBus(t *testing.T) (*broker.JetStream, jetstream.JetStream) {
|
||||
t.Helper()
|
||||
js, err := broker.Dial(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(js.Close)
|
||||
if err := broker.AssertMeshStreams(js); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := js.EnsureControllerBuckets(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return js, api
|
||||
}
|
||||
|
||||
// heard counts what is said on one subject, each body kept.
|
||||
type heard struct {
|
||||
mu sync.Mutex
|
||||
bodies []Event
|
||||
}
|
||||
|
||||
func (h *heard) all() []Event {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return append([]Event(nil), h.bodies...)
|
||||
}
|
||||
|
||||
func listen(t *testing.T, conn *nats.Conn, subject string) *heard {
|
||||
t.Helper()
|
||||
h := &heard{}
|
||||
sub, err := conn.Subscribe(subject, func(m *nats.Msg) {
|
||||
var e Event
|
||||
if err := json.Unmarshal(m.Data, &e); err != nil {
|
||||
t.Errorf("an event that is not one: %s", m.Data)
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.bodies = append(h.bodies, e)
|
||||
h.mu.Unlock()
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
if err := conn.Flush(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// writes is how many messages the bucket's stream has taken: every put and every purge is one.
|
||||
func writes(t *testing.T, api jetstream.JetStream, bucket string) uint64 {
|
||||
t.Helper()
|
||||
s, err := api.Stream(t.Context(), "KV_"+bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
info, err := s.Info(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return info.State.LastSeq
|
||||
}
|
||||
|
||||
// settle waits until n events are heard or a second passed, and answers what was heard.
|
||||
func settle(h *heard, n int) []Event {
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for time.Now().Before(deadline) && len(h.all()) < n {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
// And a moment more, so an event too many is heard too.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
return h.all()
|
||||
}
|
||||
|
||||
func aWriter(t *testing.T, js *broker.JetStream, api jetstream.JetStream, m Mirror) *Writer {
|
||||
t.Helper()
|
||||
kv, err := api.KeyValue(t.Context(), m.Bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return &Writer{Mirror: m, KV: kv, Announce: link.OverNATS{Conn: js.Conn(), JS: js.Context()},
|
||||
Epoch: func() uint64 { return 7 }, Holds: func() bool { return true }}
|
||||
}
|
||||
|
||||
func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) {
|
||||
for _, m := range []Mirror{Assignments, DeclaredConfig} {
|
||||
t.Run(m.Name, func(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+m.Event)
|
||||
w := aWriter(t, js, api, m)
|
||||
ctx := t.Context()
|
||||
|
||||
first := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
|
||||
if _, err := w.Sync(ctx, first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, m.Bucket)
|
||||
heardBefore := len(settle(events, 1))
|
||||
|
||||
// One record changes, one is added: two keys, two events, nothing else.
|
||||
second := map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)}
|
||||
r, err := w.Sync(ctx, second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Written != 2 || r.Removed != 0 {
|
||||
t.Errorf("a pass over one changed and one new record wrote %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 2 {
|
||||
t.Errorf("the bucket took %d writes for two changed records", got)
|
||||
}
|
||||
said := settle(events, heardBefore+2)[heardBefore:]
|
||||
if len(said) != 2 {
|
||||
t.Fatalf("two changes raised %d events: %+v", len(said), said)
|
||||
}
|
||||
changes := map[string]Event{}
|
||||
for _, e := range said {
|
||||
changes[e.Key] = e
|
||||
}
|
||||
if e := changes["anchor.gitea"]; e.Change != Changed || !slices.Equal(e.Fields, []string{"a"}) ||
|
||||
e.Mirror != m.Name || e.Epoch != 7 || e.Revision == 0 {
|
||||
t.Errorf("the change of anchor.gitea was said as %+v", e)
|
||||
}
|
||||
if e := changes["anchor.forgejo"]; e.Change != Added {
|
||||
t.Errorf("the new anchor.forgejo was said as %+v", e)
|
||||
}
|
||||
|
||||
// **An unchanged pass writes nothing**, and says nothing (ADR 0289 §1).
|
||||
heardBefore = len(events.all())
|
||||
before = writes(t, api, m.Bucket)
|
||||
r, err = w.Sync(ctx, second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{}) {
|
||||
t.Errorf("a pass over unchanged records did %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 0 {
|
||||
t.Errorf("a pass over unchanged records wrote the bucket %d times", got)
|
||||
}
|
||||
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 0 {
|
||||
t.Errorf("a pass over unchanged records raised %+v", said)
|
||||
}
|
||||
|
||||
// **A record gone is a purge**, one write and one `removed`.
|
||||
before = writes(t, api, m.Bucket)
|
||||
r, err = w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Removed != 1 || r.Written != 0 {
|
||||
t.Errorf("a record gone did %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 1 {
|
||||
t.Errorf("a record gone wrote the bucket %d times", got)
|
||||
}
|
||||
if _, err := w.KV.Get(ctx, "anchor.forgejo"); !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
t.Errorf("the key of a record gone is still read: %v", err)
|
||||
}
|
||||
said = settle(events, heardBefore+1)[heardBefore:]
|
||||
if len(said) != 1 || said[0].Change != Removed || said[0].Key != "anchor.forgejo" {
|
||||
t.Errorf("a record gone was said as %+v", said)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// **A controller started again writes nothing for what the bucket already holds** (design 54 §2: the writer's
|
||||
// memory is filled from the bucket), and a key nobody holds a record for is removed.
|
||||
func TestNatsAWriterStartedAgainWritesOnlyWhatDiffers(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
first := aWriter(t, js, api, Assignments)
|
||||
held := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`), "anchor.forgejo": []byte(`{"a":1}`)}
|
||||
if _, err := first.Sync(ctx, held); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
again := aWriter(t, js, api, Assignments)
|
||||
r, err := again.Sync(ctx, held)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{}) || writes(t, api, Assignments.Bucket) != before {
|
||||
t.Errorf("a writer started again over unchanged records did %+v", r)
|
||||
}
|
||||
}
|
||||
|
||||
// **Another writer in between is no lost value** (design 54 §2, compare-and-set): a write refused because the
|
||||
// key moved reads it again, takes the revision when it already holds the value, and otherwise writes again.
|
||||
func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
if _, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Behind its back, as a former lease holder would.
|
||||
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Written != 1 {
|
||||
t.Errorf("a refused write was not made again: %+v", r)
|
||||
}
|
||||
e, err := w.KV.Get(ctx, "anchor.gitea")
|
||||
if err != nil || string(e.Value()) != `{"a":2}` {
|
||||
t.Fatalf("the bucket holds %v, %v", e, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A value written behind the writer's back is put right on the next pass** (ADR 0289 §10): a former lease
|
||||
// holder's write differs from the store, and the store's value is written again, said as one change.
|
||||
func TestNatsAForeignWriteIsCorrectedOnTheNextPass(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
store := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
|
||||
if _, err := w.Sync(ctx, store); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
heardBefore := len(settle(events, 1))
|
||||
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r, err := w.Sync(ctx, store)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{Written: 1}) {
|
||||
t.Errorf("a pass after a foreign write did %+v", r)
|
||||
}
|
||||
e, err := w.KV.Get(ctx, "anchor.gitea")
|
||||
if err != nil || string(e.Value()) != `{"a":1}` {
|
||||
t.Fatalf("the bucket still holds %v, %v", e, err)
|
||||
}
|
||||
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 1 || said[0].Change != Changed {
|
||||
t.Errorf("the correction was said as %+v", said)
|
||||
}
|
||||
// And put right, it stays: the next pass writes nothing.
|
||||
if r, err := w.Sync(ctx, store); err != nil || r != (Result{}) {
|
||||
t.Errorf("a pass after the correction did %+v, %v", r, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A writer without the lease writes nothing** (ADR 0289 §9, design 54 §2): not a new key, not a change, not
|
||||
// a removal, and it says so.
|
||||
func TestNatsAWriterWithoutTheLeaseWritesNothing(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
held := aWriter(t, js, api, Assignments)
|
||||
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
for _, holds := range []func() bool{nil, func() bool { return false }} {
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = holds
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
|
||||
t.Errorf("a writer without the lease did %+v, %v", r, err)
|
||||
}
|
||||
r, err = w.Sync(ctx, map[string][]byte{})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
|
||||
t.Errorf("a writer without the lease removed: %+v, %v", r, err)
|
||||
}
|
||||
}
|
||||
// And one that loses it mid-pass stops before its next write.
|
||||
asked := 0
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = func() bool { asked++; return asked <= 2 }
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.a": []byte(`{}`), "anchor.b": []byte(`{}`), "anchor.gitea": []byte(`{"a":1}`)})
|
||||
if !errors.Is(err, ErrNotHeld) || r.Written != 1 {
|
||||
t.Errorf("a writer that lost the lease mid-pass did %+v, %v", r, err)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - before; got != 1 {
|
||||
t.Errorf("writers without the lease wrote the bucket %d times", got)
|
||||
}
|
||||
}
|
||||
|
||||
// **A removal asks for the lease too** (ADR 0289 §9): a writer that held the lease when its pass began and lost
|
||||
// it before its first removal purges nothing, and says it does not hold the lease.
|
||||
func TestNatsARemovalWithoutTheLeaseRemovesNothing(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
held := aWriter(t, js, api, Assignments)
|
||||
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
asked := 0
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = func() bool { asked++; return asked == 1 }
|
||||
r, err := w.Sync(ctx, map[string][]byte{})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) || asked != 2 {
|
||||
t.Errorf("a removal without the lease did %+v, %v (the lease asked %d times)", r, err, asked)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - before; got != 0 {
|
||||
t.Errorf("a removal without the lease wrote the bucket %d times", got)
|
||||
}
|
||||
if _, err := w.KV.Get(ctx, "anchor.gitea"); err != nil {
|
||||
t.Errorf("the key was removed without the lease: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A key the bus cannot hold is refused alone**, and the others are written.
|
||||
func TestAKeyTheBusCannotHoldIsRefusedAlone(t *testing.T) {
|
||||
if validKey("anchor.gitea") != nil || validKey("bad key") == nil || validKey(".lead") == nil ||
|
||||
validKey("") == nil || validKey("trail.") == nil {
|
||||
t.Error("the keys a bucket may hold are not told apart")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheFieldsThatChangedAreNamed(t *testing.T) {
|
||||
got := changedFields([]byte(`{"a":1,"b":{"x":1},"c":"same"}`), []byte(`{"a":2,"b":{"x":1},"d":true,"c":"same"}`))
|
||||
if !slices.Equal(got, []string{"a", "d"}) {
|
||||
t.Errorf("the fields that changed are %v", got)
|
||||
}
|
||||
if got := changedFields(nil, []byte(`{"a":1}`)); got != nil {
|
||||
t.Errorf("an added key names fields %v", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
type silent struct{}
|
||||
|
||||
func (silent) PublishSeatEvent(context.Context, string, string, []byte) error { return nil }
|
||||
|
||||
// **The provisioner's plan reaches its machine's client, and no other** (design 57 P5, issue 671): on a real
|
||||
// server composed from the grants, the controller writes each assignment with the plan of a module handed to the
|
||||
// provisioner into the assignments mirror; g14's client reads and watches its own machine's plan, reads shanks's
|
||||
// (who binds a provider, ADR 0373), and the server refuses it any write of the mirror.
|
||||
func TestNatsTheProvisionerClientReadsEveryMachinesPlanAndWritesNone(t *testing.T) {
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
accounts, err := broker.ComposeAccounts([]broker.Principal{
|
||||
{Kind: broker.KindController, PasswordHash: string(hash)},
|
||||
{Kind: broker.KindProvisioner, Node: "g14", Module: "mesh-provisioner-client", PasswordHash: string(hash)},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
conf := filepath.Join(t.TempDir(), "bus.conf")
|
||||
if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
|
||||
`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatalf("the composed accounts do not parse: %v", err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go s.Start()
|
||||
if !s.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
defer s.Shutdown()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
refusals := make(chan error, 16)
|
||||
connect := func(user string) *nats.Conn {
|
||||
t.Helper()
|
||||
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
|
||||
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
return nc
|
||||
}
|
||||
|
||||
// The controller writes the mirror from what the store says, the plan among it.
|
||||
control, err := jetstream.New(connect("controller"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kv, err := control.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.AssignmentsBucket, History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
plan := inventory.Provisioning{
|
||||
Templates: inventory.TemplatesArchive{Source: "artifacts/dunst/files.tar.gz",
|
||||
Digest: "sha256:" + strings.Repeat("cd", 32)},
|
||||
Files: []inventory.ProvisionedFile{{ID: "dunstrc", Path: "/home/nox/.config/dunst/dunstrc",
|
||||
Template: "dunstrc", OwnerAtRuntime: "owned", Takes: "reload-on"}},
|
||||
Placement: map[string]string{"machine:account-home": "/home/nox"},
|
||||
}
|
||||
at := time.Date(2026, 10, 14, 20, 0, 0, 0, time.UTC)
|
||||
values, err := inventory.AssignmentValues([]inventory.AssignmentMirrored{
|
||||
{Node: "g14", Module: "dunst", Assigned: at}, {Node: "shanks", Module: "dunst", Assigned: at}},
|
||||
map[string]inventory.Provisioning{"g14.dunst": plan, "shanks.dunst": plan})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w := &Writer{Mirror: Assignments, KV: kv, Announce: silent{}, Holds: func() bool { return true }}
|
||||
if r, err := w.Sync(ctx, values); err != nil || r.Written != 2 {
|
||||
t.Fatalf("the controller wrote %+v (%v)", r, err)
|
||||
}
|
||||
|
||||
client, err := jetstream.New(connect("provisioner.g14"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mirror, err := client.KeyValue(ctx, broker.AssignmentsBucket)
|
||||
if err != nil {
|
||||
t.Fatalf("the client cannot bind to the assignments mirror: %v", err)
|
||||
}
|
||||
watch, err := mirror.Watch(ctx, "g14.>")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() { _ = watch.Stop() }()
|
||||
select {
|
||||
case e := <-watch.Updates():
|
||||
var got inventory.AssignmentMirrored
|
||||
if e == nil || json.Unmarshal(e.Value(), &got) != nil || got.Provisioning == nil ||
|
||||
got.Provisioning.Files[0].Template != "dunstrc" ||
|
||||
got.Provisioning.Placement["machine:account-home"] != "/home/nox" {
|
||||
t.Fatalf("the client's watch delivered %v first", e)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the client's watch of its own machine's plans delivered nothing")
|
||||
}
|
||||
|
||||
refused := func(what string, act func(context.Context) error) {
|
||||
t.Helper()
|
||||
for len(refusals) > 0 {
|
||||
<-refusals
|
||||
}
|
||||
short, done := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer done()
|
||||
if err := act(short); err == nil {
|
||||
t.Errorf("the client may %s", what)
|
||||
}
|
||||
select {
|
||||
case err := <-refusals:
|
||||
if !errors.Is(err, nats.ErrPermissionViolation) {
|
||||
t.Errorf("%s: the server said %v, want a permissions violation", what, err)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Errorf("%s: the server did not refuse it", what)
|
||||
}
|
||||
}
|
||||
// Another machine's plan is read, never written: a provider here is bound by consumers on any machine, in their
|
||||
// plans, which its machine's client reads to fill ${consumers:<provision>} (novox/hq ADR 0373, issue 679).
|
||||
if _, err := mirror.Get(ctx, "shanks.dunst"); err != nil {
|
||||
t.Errorf("another machine's plan, which binds its consumers to a provider here: %v", err)
|
||||
}
|
||||
refused("write another machine's plan", func(c context.Context) error {
|
||||
_, err := mirror.PutString(c, "shanks.dunst", "{}")
|
||||
return err
|
||||
})
|
||||
refused("write its own machine's plan", func(c context.Context) error {
|
||||
_, err := mirror.PutString(c, "g14.dunst", "{}")
|
||||
return err
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"go/ast"
|
||||
"go/parser"
|
||||
"go/token"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// mayNameAMirror are the files of the controller that may name a record mirror's bucket outside their tests:
|
||||
// where the buckets are defined, asserted and granted (internal/broker), the list a reader names in `reads`
|
||||
// (internal/catalogue/state.go), and each mirror's own writer (this package and the loop that runs it). The
|
||||
// audit of a mirror against its store (ADR 0289 §4, issue 419) joins this list when it is built.
|
||||
var mayNameAMirror = []string{
|
||||
"cmd/mesh-controller/record_mirrors.go",
|
||||
"internal/broker/controller_buckets.go",
|
||||
// The grant of a machine's own keys of the assignments mirror to its provisioner client (design 57 P3).
|
||||
"internal/broker/provisioner.go",
|
||||
"internal/broker/writers.go",
|
||||
"internal/catalogue/state.go",
|
||||
}
|
||||
|
||||
// The packages whose names stand for a mirror, by import path, and the names in each.
|
||||
const (
|
||||
brokerPath = "github.com/novox/mesh-controller/internal/broker"
|
||||
recordmirrorPath = "github.com/novox/mesh-controller/internal/recordmirror"
|
||||
)
|
||||
|
||||
var mirrorNames = map[string][]string{
|
||||
brokerPath: {"AssignmentsBucket", "DeclaredConfigBucket", "MirrorAssignments", "MirrorDeclaredConfig",
|
||||
"ControllerMirrors"},
|
||||
recordmirrorPath: {"Assignments", "DeclaredConfig", "Mirrors"},
|
||||
}
|
||||
|
||||
// **Nothing in the controller reads a mirror to decide anything** (ADR 0289 §19 as ADR 0356 §2 changes it): no
|
||||
// file of the controller's packages outside its tests names a mirror's bucket but the audit and a mirror's own
|
||||
// writer. A reader elsewhere fails here, naming the file and what it named.
|
||||
//
|
||||
// **What it catches**: a name that stands for a mirror, qualified by the package it comes from under any
|
||||
// local name the file imports it as (`broker.AssignmentsBucket`, `b.AssignmentsBucket`); the same names bare
|
||||
// in a file of package broker or package recordmirror; and a string literal holding a mirror's bucket name.
|
||||
// **What it does not catch**: a bucket name built at run time, such as `BucketName(ControllerSeat,
|
||||
// "assignments")` or literals concatenated into one, and a mirror reached through a value passed in from an
|
||||
// allowed file. Those are deliberate evasions rather than ordinary code; a reviewer judges them.
|
||||
func TestOnlyTheAuditAndAMirrorsWriterReadAMirror(t *testing.T) {
|
||||
var buckets []string
|
||||
for _, m := range broker.ControllerMirrors() {
|
||||
buckets = append(buckets, broker.BucketName(broker.ControllerSeat, m))
|
||||
}
|
||||
root := filepath.Join("..", "..")
|
||||
found := mirrorReaders(t, root, buckets)
|
||||
if len(found) > 0 {
|
||||
t.Fatalf("a package of the controller reads a record mirror, which only the audit and the mirror's "+
|
||||
"writer may (ADR 0289 §19, ADR 0356 §2):\n %s", strings.Join(found, "\n "))
|
||||
}
|
||||
}
|
||||
|
||||
// **The reader test sees what it says it sees**: in a planted tree, each planted reader is named, and an allowed
|
||||
// file and a selector on a value are not.
|
||||
func TestTheReaderTestFindsPlantedReaders(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
plant := func(path, src string) {
|
||||
full := filepath.Join(root, path)
|
||||
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(full, []byte(src), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
plant("cmd/mesh-controller/qualified.go", "package main\n\nimport \""+brokerPath+"\"\n\nvar a = broker.AssignmentsBucket\n")
|
||||
plant("internal/x/aliased.go", "package x\n\nimport b \""+brokerPath+"\"\n\nvar a = b.DeclaredConfigBucket\n")
|
||||
plant("internal/broker/zz_reader.go", "package broker\n\nfunc zzR() string { return AssignmentsBucket }\n")
|
||||
plant("internal/inventory/literal.go", "package inventory\n\nvar a = \"mesh-controller_declared-config\"\n")
|
||||
plant("internal/y/writer.go", "package y\n\nimport \""+recordmirrorPath+"\"\n\nvar a = recordmirror.Assignments\n")
|
||||
// Allowed: a mirror's definition and its writer.
|
||||
plant("internal/broker/controller_buckets.go", "package broker\n\nvar AssignmentsBucket = \"x\"\n")
|
||||
plant("internal/broker/other.go", "package broker\n\nfunc other() string { return \"nothing\" }\n")
|
||||
// A selector on a value (v.AssignmentsBucket) is no use of the mirror, but declaring a field of a mirror's
|
||||
// name in package broker is named, on the safe side; and a bare use inside a call under a selector is found.
|
||||
plant("internal/broker/field.go", "package broker\n\ntype s struct{ AssignmentsBucket string }\n\nfunc f(v s) string { return v.AssignmentsBucket }\n")
|
||||
plant("internal/broker/nested.go", "package broker\n\nfunc g() int { return len(h(DeclaredConfigBucket).x) }\n")
|
||||
found := mirrorReaders(t, root, []string{"mesh-controller_assignments", "mesh-controller_declared-config"})
|
||||
want := []string{
|
||||
"cmd/mesh-controller/qualified.go: broker.AssignmentsBucket",
|
||||
"internal/broker/field.go: AssignmentsBucket",
|
||||
"internal/broker/nested.go: DeclaredConfigBucket",
|
||||
"internal/broker/zz_reader.go: AssignmentsBucket",
|
||||
"internal/inventory/literal.go: \"mesh-controller_declared-config\"",
|
||||
"internal/x/aliased.go: b.DeclaredConfigBucket",
|
||||
"internal/y/writer.go: recordmirror.Assignments",
|
||||
}
|
||||
slices.Sort(found)
|
||||
if !slices.Equal(found, want) {
|
||||
t.Errorf("the planted readers were found as\n %s\nwant\n %s", strings.Join(found, "\n "),
|
||||
strings.Join(want, "\n "))
|
||||
}
|
||||
}
|
||||
|
||||
// mirrorReaders is every place under root's cmd and internal, outside tests and the allowed files, that names a
|
||||
// mirror.
|
||||
func mirrorReaders(t *testing.T, root string, buckets []string) []string {
|
||||
t.Helper()
|
||||
var found []string
|
||||
for _, dir := range []string{"cmd", "internal"} {
|
||||
err := filepath.WalkDir(filepath.Join(root, dir), func(path string, e fs.DirEntry, err error) error {
|
||||
if err != nil || e.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
|
||||
return err
|
||||
}
|
||||
rel, err := filepath.Rel(root, path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rel = filepath.ToSlash(rel)
|
||||
if strings.HasPrefix(rel, "internal/recordmirror/") || slices.Contains(mayNameAMirror, rel) {
|
||||
return nil
|
||||
}
|
||||
f, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Each import's local name, mapped to the package it stands for.
|
||||
imported := map[string]string{}
|
||||
for _, imp := range f.Imports {
|
||||
p, _ := strconv.Unquote(imp.Path.Value)
|
||||
if _, ok := mirrorNames[p]; !ok {
|
||||
continue
|
||||
}
|
||||
local := filepath.Base(p)
|
||||
if imp.Name != nil {
|
||||
local = imp.Name.Name
|
||||
}
|
||||
imported[local] = p
|
||||
}
|
||||
// A file of the package that defines the names may use them bare.
|
||||
var bare []string
|
||||
switch f.Name.Name {
|
||||
case "broker":
|
||||
bare = mirrorNames[brokerPath]
|
||||
case "recordmirror":
|
||||
bare = mirrorNames[recordmirrorPath]
|
||||
}
|
||||
ast.Inspect(f, func(n ast.Node) bool {
|
||||
switch n := n.(type) {
|
||||
case *ast.SelectorExpr:
|
||||
pkg, ok := n.X.(*ast.Ident)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
if p, imp := imported[pkg.Name]; imp && slices.Contains(mirrorNames[p], n.Sel.Name) {
|
||||
found = append(found, rel+": "+pkg.Name+"."+n.Sel.Name)
|
||||
}
|
||||
// Neither a selector's own name nor its package is a bare use.
|
||||
return false
|
||||
case *ast.Ident:
|
||||
if slices.Contains(bare, n.Name) {
|
||||
found = append(found, rel+": "+n.Name)
|
||||
}
|
||||
case *ast.BasicLit:
|
||||
for _, b := range buckets {
|
||||
if n.Kind == token.STRING && strings.Contains(n.Value, b) {
|
||||
found = append(found, rel+": "+n.Value)
|
||||
}
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
return found
|
||||
}
|
||||
Reference in New Issue
Block a user