Keep the assignments and each module's declared config on the bus

mesh-config resolves a module's config values per node from which module runs
where and what each registered module declares, and both lived only in the
controller's store (hq issue 652, design 57 F3). The controller now keeps two
record mirrors, mesh-controller_assignments (<node>.<module>) and
mesh-controller_declared-config (<module>), built from the store by the lease
holder every ten seconds and written only where a value differs, by
compare-and-set, with one assignment-changed or declared-config-changed per
write and a purge said as removed (ADR 0289, design 54 §2-3). A module reads
them through `reads` as mesh-controller.<mirror>; the writers table keeps every
other writer out. Neither holds a secret: an assignment names its node and
module, a declaration is what the public manifest declares.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

cmd/mesh-controller/push.go

@@ -264,6 +264,9 @@ func serve(ctx context.Context) (err error) {
 	// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
 	// controller holds the lease, read by the build seat from the artifact store.
 	go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
+	// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
+	// declared config on the bus, written from the store only when a record changed, by the lease holder.
+	go mirroringRecords(ctx, inv, bus.Conn, bus)
 	// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
 	// to-be 45 §6).
 	link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })

cmd/mesh-controller/record_mirrors.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-controller/internal/inventory"
+	"github.com/novox/mesh-controller/internal/recordmirror"
+)
+
+// The controller's record mirrors on the bus (novox/hq ADR 0289, design 54, design 57 F3): the assignments,
+// and each registered module's declared config, which mesh-config resolves a module's config values from.
+//
+// **Built from the store, on a tick of their own, by the lease holder alone.** A record changes through a
+// verb that runs as a command of its own, a reconcile, a merge's registration or a pending assignment made,
+// so no one call site sees every change; what every change has in common is that it is committed to the
+// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
+// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
+// mirror is therefore at most mirrorEvery behind its store.
+
+// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
+const mirrorEvery = 10 * time.Second
+
+// mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops.
+func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
+	api, err := jetstream.New(conn)
+	if err != nil {
+		fmt.Printf("the record mirrors are not kept: %v\n", err)
+		return
+	}
+	holds := func() bool {
+		epoch, err := theLease.epoch(ctx)
+		return err == nil && epoch > 0
+	}
+	epoch := func() uint64 {
+		e, _ := theLease.epoch(ctx)
+		return e
+	}
+	sources := map[string]func(context.Context) (map[string][]byte, error){
+		recordmirror.Assignments.Name:    inv.AssignmentsMirror,
+		recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
+	}
+	writers := map[string]*recordmirror.Writer{}
+	failing := map[string]string{}
+	tick := time.NewTicker(mirrorEvery)
+	defer tick.Stop()
+	for {
+		if holds() {
+			for _, m := range recordmirror.Mirrors() {
+				why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
+				if why != failing[m.Name] {
+					if why != "" {
+						fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
+					} else {
+						fmt.Printf("the %s mirror is kept again\n", m.Name)
+					}
+					failing[m.Name] = why
+				}
+			}
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+		}
+	}
+}
+
+// mirrorOnce brings one mirror to what the store holds, and answers why it could not; empty when it could.
+func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
+	source func(context.Context) (map[string][]byte, error), announce recordmirror.Announcer, epoch func() uint64,
+	holds func() bool) string {
+	reading, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	w := writers[m.Name]
+	if w == nil {
+		kv, err := api.KeyValue(reading, m.Bucket)
+		if err != nil {
+			return fmt.Sprintf("the bucket %s cannot be read: %v", m.Bucket, err)
+		}
+		w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
+		writers[m.Name] = w
+	}
+	want, err := source(reading)
+	if err != nil {
+		return fmt.Sprintf("the store cannot be read: %v", err)
+	}
+	lost := w.LostEvents
+	r, err := w.Sync(reading, want)
+	if r.Written+r.Removed > 0 {
+		fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
+	}
+	if w.LostEvents > lost {
+		fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
+			m.Name, w.LostEvents-lost)
+	}
+	if err != nil {
+		return err.Error()
+	}
+	return ""
+}