From 176a3d4fdf3b0bc4057289c71fcf2f1fc3f00276 Mon Sep 17 00:00:00 2001 From: mesh-admin Date: Wed, 14 Oct 2026 00:31:17 +0000 Subject: [PATCH] Write a record mirror only where its store differs, one event per write MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The writer behind the assignments and declared-config mirrors (hq issue 652, ADR 0289, design 54 §2-3): filled from the bucket on its first pass, it compares each value byte for byte, writes by compare-and-set, purges a key whose record is gone with a marker that lives a day, and says each write once after it succeeded. Co-Authored-By: Claude Opus 5.5 --- internal/recordmirror/mirror.go | 330 ++++++++++++++++++++++++++++++++ 1 file changed, 330 insertions(+) create mode 100644 internal/recordmirror/mirror.go diff --git a/internal/recordmirror/mirror.go b/internal/recordmirror/mirror.go new file mode 100644 index 00000000..ed9b7a48 --- /dev/null +++ b/internal/recordmirror/mirror.go @@ -0,0 +1,330 @@ +// 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. 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. +// +// **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 `.`; 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} +) + +// Mirrors is every mirror this package writes, in the order they are written. +func Mirrors() []Mirror { return []Mirror{Assignments, DeclaredConfig} } + +// 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 + +// 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 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 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 + // 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.filled { + 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 + } + 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] + 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 { + w.filled = true + 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 +}