diff --git a/internal/inventory/mirrors.go b/internal/inventory/mirrors.go index 5d878a0f..c33db1c0 100644 --- a/internal/inventory/mirrors.go +++ b/internal/inventory/mirrors.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "time" ) @@ -25,14 +26,27 @@ type AssignmentMirrored struct { } // DeclaredConfigMirrored is one registered module's config declarations as the declared-config mirror holds -// them: the module, and its manifest's `settings` exactly as the manifest holds them (each key's kind, -// default and why today), carried whole so a declaration that gains fields (ADR 0348 clause 4) is mirrored -// without a change here. An empty object when the module declares none: registered, declaring nothing. +// them: the module, and each key its manifest's `settings` declares with that declaration's listed fields. +// An empty object when the module declares none: registered, declaring nothing. type DeclaredConfigMirrored struct { - Module string `json:"module"` - Settings json.RawMessage `json:"settings"` + Module string `json:"module"` + Settings map[string]DeclaredMirrored `json:"settings"` } +// DeclaredMirrored is one key's declaration as the mirror holds it: **its listed fields and no other** +// (ADR 0289 §12). Today a declaration is kind, default and why (catalogue.SettingDeclaration); a field the +// manifest gains reaches the bus only once it is added here, deliberately, and a test fails until it is +// listed or refused (ADR 0348 clause 4's fields arrive with package K1). A default is never a secret: the +// catalogue refuses a default for a key whose name says secret. +type DeclaredMirrored struct { + Kind string `json:"kind"` + Default json.RawMessage `json:"default"` + Why string `json:"why"` +} + +// DeclaredMirroredFields are the fields of a declaration the mirror carries, by their manifest names. +var DeclaredMirroredFields = []string{"kind", "default", "why"} + // AssignmentsMirror is every assignment's mirrored value, keyed `.`. func (i *Inventory) AssignmentsMirror(ctx context.Context) (map[string][]byte, error) { rows, err := i.store.Pool().Query(ctx, @@ -74,11 +88,19 @@ func (i *Inventory) DeclaredConfigMirror(ctx context.Context) (map[string][]byte if err := rows.Scan(&d.Module, &raw); err != nil { return nil, err } - settings, err := canonical(raw) - if err != nil { - return nil, err + var declared map[string]DeclaredMirrored + if err := json.Unmarshal(raw, &declared); err != nil { + return nil, fmt.Errorf("the declarations of %s cannot be read: %w", d.Module, err) + } + d.Settings = map[string]DeclaredMirrored{} + for key, decl := range declared { + if len(decl.Default) > 0 { + if decl.Default, err = canonical(decl.Default); err != nil { + return nil, err + } + } + d.Settings[key] = decl } - d.Settings = settings value, err := json.Marshal(d) if err != nil { return nil, err diff --git a/internal/inventory/record_mirrors_test.go b/internal/inventory/record_mirrors_test.go new file mode 100644 index 00000000..db2e9ba2 --- /dev/null +++ b/internal/inventory/record_mirrors_test.go @@ -0,0 +1,41 @@ +package inventory + +import ( + "reflect" + "slices" + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/catalogue" +) + +// jsonFields is the names a struct's fields are encoded under. +func jsonFields(t reflect.Type) []string { + var out []string + for i := 0; i < t.NumField(); i++ { + name, _, _ := strings.Cut(t.Field(i).Tag.Get("json"), ",") + if name != "" && name != "-" { + out = append(out, name) + } + } + return out +} + +// **A declaration reaches the bus with its listed fields and no other** (ADR 0289 §12): the mirror's value +// carries exactly DeclaredMirroredFields, and every field a manifest's declaration has is on that list. A +// field a declaration gains (package K1 widens it, ADR 0348 clause 4) fails here until it is listed — read, +// judged not a secret and carried — so nothing reaches the declared-config mirror unreviewed. +func TestTheDeclaredConfigMirrorCarriesOnlyItsListedFields(t *testing.T) { + if got := jsonFields(reflect.TypeOf(DeclaredMirrored{})); !slices.Equal(got, DeclaredMirroredFields) { + t.Errorf("the mirrored declaration carries %v, its list is %v", got, DeclaredMirroredFields) + } + for _, field := range jsonFields(reflect.TypeOf(catalogue.SettingDeclaration{})) { + if !slices.Contains(DeclaredMirroredFields, field) { + t.Errorf("a declaration has the field %q, which the declared-config mirror does not list: list it "+ + "in DeclaredMirrored (and say why it is no secret), or say here why it stays off the bus", field) + } + } + if got := jsonFields(reflect.TypeOf(DeclaredConfigMirrored{})); !slices.Equal(got, []string{"module", "settings"}) { + t.Errorf("the mirrored module carries %v", got) + } +} diff --git a/internal/recordmirror/mirror.go b/internal/recordmirror/mirror.go index ed9b7a48..3e6032e5 100644 --- a/internal/recordmirror/mirror.go +++ b/internal/recordmirror/mirror.go @@ -3,10 +3,14 @@ // 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. The writer keeps the last value it wrote for each key, filled from the bucket on its -// first pass, and compares byte for byte: a pass over unchanged records writes nothing and says nothing -// (ADR 0289 §1). Each write names the revision it last saw (compare-and-set), so a former lease holder that -// wrote in between is noticed, never overwritten blind. +// 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 @@ -83,6 +87,9 @@ type Announcer interface { // 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 @@ -90,14 +97,13 @@ type Writer struct { Announce Announcer // Epoch is the lease's epoch the events carry; nil carries none. Epoch func() uint64 - // Holds says this controller still holds the lease; nil is always. A write refused by compare-and-set - // is made again only while it does (design 54 §2). + // 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 - filled bool + known map[string]kept // LostEvents counts the events refused twice since the writer was made. LostEvents int } @@ -119,10 +125,13 @@ type Result struct { // 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.filled { - if err := w.fill(ctx); err != nil { - return r, fmt.Errorf("reading the %s mirror: %w", w.Mirror.Name, err) - } + 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)) @@ -141,6 +150,9 @@ func (w *Writer) Sync(ctx context.Context, want map[string][]byte) (Result, erro 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)) @@ -166,6 +178,9 @@ func (w *Writer) Sync(ctx context.Context, want map[string][]byte) (Result, erro 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. @@ -266,7 +281,6 @@ func (w *Writer) fill(ctx context.Context) error { return ctx.Err() case e := <-watch.Updates(): if e == nil { - w.filled = true return nil } w.known[e.Key()] = kept{value: append([]byte(nil), e.Value()...), revision: e.Revision()} @@ -274,7 +288,7 @@ func (w *Writer) fill(ctx context.Context) error { } } -func (w *Writer) holds() bool { return w.Holds == nil || w.Holds() } +func (w *Writer) holds() bool { return w.Holds != nil && w.Holds() } func (w *Writer) now() time.Time { if w.Now == nil { diff --git a/internal/recordmirror/mirror_store_test.go b/internal/recordmirror/mirror_store_test.go index c0601a7c..11e8de47 100644 --- a/internal/recordmirror/mirror_store_test.go +++ b/internal/recordmirror/mirror_store_test.go @@ -140,7 +140,7 @@ func TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent(t *tes if err != nil { t.Fatal(err) } - if want := `{"module":"ui","settings":{"font-size":{"default":16,"kind":"preference","why":"readable on the desktop"}}}`; string(got.Value()) != want { + 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) } diff --git a/internal/recordmirror/mirror_test.go b/internal/recordmirror/mirror_test.go index 41f0c2ac..31f67594 100644 --- a/internal/recordmirror/mirror_test.go +++ b/internal/recordmirror/mirror_test.go @@ -109,7 +109,7 @@ func aWriter(t *testing.T, js *broker.JetStream, api jetstream.JetStream, m Mirr t.Fatal(err) } return &Writer{Mirror: m, KV: kv, Announce: link.OverNATS{Conn: js.Conn(), JS: js.Context()}, - Epoch: func() uint64 { return 7 }} + Epoch: func() uint64 { return 7 }, Holds: func() bool { return true }} } func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) { @@ -242,6 +242,76 @@ func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) { } } +// **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 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 ||