From 69367aa786427326d48acf270186a28d5f9a33ee Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 01:01:00 +0000 Subject: [PATCH] Correct a mirror written behind the writer's back, and write none without the lease MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of mesh-controller#315 (hq issue 652): the writer trusted the bucket as it read it once, so a former lease holder's write stayed for good; it wrote without the lease, asking only after a refused compare-and-set; and the declared config carried a manifest's declarations whole, so a field the manifest gains would reach the bus unreviewed (ADR 0289 §9, §10, §12). Every pass now reads the bucket and writes back the store's value where it differs, asks for the lease before the pass and before every write, and a declaration is mirrored as its listed fields (kind, default, why), with a test that fails on a declaration field the list does not name. Co-Authored-By: Claude Opus 5.5 --- internal/inventory/mirrors.go | 40 +++++++++--- internal/inventory/record_mirrors_test.go | 41 ++++++++++++ internal/recordmirror/mirror.go | 42 ++++++++----- internal/recordmirror/mirror_store_test.go | 2 +- internal/recordmirror/mirror_test.go | 72 +++++++++++++++++++++- 5 files changed, 172 insertions(+), 25 deletions(-) create mode 100644 internal/inventory/record_mirrors_test.go 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 ||