Compose the declared-packages mirror only when what it composes from changed (issue 1945)

The mirror composed every node in full on the record mirrors' 10 s tick; a
pass takes about 45 s on the control node, so the controller never rested
(130-175 % CPU, issue 1945 note 22). It now passes on a watch of the
compositions' digests and of the assignments, one pass at a time with the
wakes of a pass coalesced into one after it (at least 30 s apart), skips the
composition when the stores' Input digest is unchanged, and keeps a safety
pass every 10 minutes that composes nothing when nothing changed.

cmd/mesh-controller/declared_packages_integration_test.go

@@ -87,3 +87,29 @@ func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T)
 		}
 	}
 }
+
+// 1945/AC-mirror-5: the digest the declared-packages mirror skips an unchanged pass by is made from a real reading of the
+// stores, is the same over two readings of the same records, and moves when an assignment does.
+func TestTheInputDigestOfTheStoresIsStable(t *testing.T) {
+	open := aMesh(t)
+	ctx := t.Context()
+	register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{
+		{"id": "pkg-curl", "type": "package", "package": "curl"}}})
+	read := func() string {
+		in, err := readComposeInput(ctx, open)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return inputDigest(in)
+	}
+	first, second := read(), read()
+	if first == "" || first != second {
+		t.Fatalf("two readings of the same records gave the digests %q and %q", first, second)
+	}
+	if _, err := assign(ctx, open, "anchor", "viewer"); err != nil {
+		t.Fatalf("assign: %v", err)
+	}
+	if read() == first {
+		t.Errorf("an assignment left the digest of the stores' Input at %s", first)
+	}
+}

cmd/mesh-controller/record_mirrors.go

@@ -2,12 +2,15 @@ package main
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"sort"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/compose"
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -24,7 +27,8 @@ import (
 // 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.
+// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes
+// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945).
 
 // mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
 const mirrorEvery = 10 * time.Second
@@ -45,21 +49,31 @@ func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventor
 		return e
 	}
 	sources := map[string]func(context.Context) (map[string][]byte, error){
-		recordmirror.DeclaredConfig.Name:   inv.DeclaredConfigMirror,
+		recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
-		recordmirror.Predefined.Name:       inv.PredefinedMirror,
+		recordmirror.Predefined.Name:     inv.PredefinedMirror,
-		recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
 	}
+	packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds)
+	held := false
 	writers := map[string]*recordmirror.Writer{}
 	failing := map[string]string{}
 	tick := time.NewTicker(mirrorEvery)
 	defer tick.Stop()
 	events := &assignmentEvents{}
 	for {
-		if holds() {
+		now := holds()
+		if now && !held {
+			// The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change.
+			packages.poke()
+		}
+		held = now
+		if now {
 			// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
 			// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
 			keepingAssignments(ctx, inv, announce, epoch, events, failing)
 			for _, m := range recordmirror.Mirrors() {
+				if sources[m.Name] == nil {
+					continue // the declared packages, kept on a change (keepingDeclaredPackages)
+				}
 				why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
 				if why != failing[m.Name] {
 					if why != "" {
@@ -179,35 +193,261 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
 	events.known = now
 }
 
-// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a
+// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
-// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does
-// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that
-// could not be read.
 func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
 	return func(ctx context.Context) (map[string][]byte, error) {
 		in, err := inputFor(ctx, open)
 		if err != nil {
 			return nil, err
 		}
-		// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
+		return declaredPackagesFrom(ctx, in)
-		asked := in.Input
+	}
-		asked.Nodes = append([]compose.Node(nil), in.Nodes...)
+}
-		for i := range asked.Nodes {
+
-			asked.Nodes[i].ReadsModules = true
+// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes
-		}
+// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key,
-		out := map[string][]byte{}
+// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read.
-		for _, n := range in.sortedNodes() {
+func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
-			d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
+	// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
-			if compose.IsUnresolvable(err) {
+	asked := in.Input
-				continue
+	asked.Nodes = append([]compose.Node(nil), in.Nodes...)
+	for i := range asked.Nodes {
+		asked.Nodes[i].ReadsModules = true
+	}
+	out := map[string][]byte{}
+	for _, n := range in.sortedNodes() {
+		d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
+		if compose.IsUnresolvable(err) {
+			continue
+		}
+		if err != nil {
+			return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
+		}
+		if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in
+// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every
+// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the
+// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the
+// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last
+// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass
+// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a
+// change no watched key says, and with an unchanged digest composes nothing either.
+const (
+	packagesSafetyEvery = 10 * time.Minute
+	packagesGap         = 30 * time.Second
+)
+
+// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the
+// starts of two passes.
+type packagesTiming struct{ safety, gap time.Duration }
+
+// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes
+// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for.
+type packagesKeeper struct {
+	read    func(ctx context.Context) (*meshInput, error)
+	compose func(ctx context.Context, in *meshInput) (map[string][]byte, error)
+	sync    func(ctx context.Context, want map[string][]byte) error
+	holds   func() bool
+	wake    chan struct{}
+	// last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the
+	// next pass composes.
+	last string
+}
+
+func newPackagesKeeper(read func(context.Context) (*meshInput, error),
+	compose func(context.Context, *meshInput) (map[string][]byte, error),
+	sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper {
+	return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)}
+}
+
+// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one.
+func (k *packagesKeeper) poke() {
+	select {
+	case k.wake <- struct{}{}:
+	default:
+	}
+}
+
+// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored.
+func (k *packagesKeeper) pass(ctx context.Context) error {
+	if !k.holds() {
+		k.last = ""
+		return nil
+	}
+	in, err := k.read(ctx)
+	if err != nil {
+		k.last = ""
+		return fmt.Errorf("the store cannot be read: %w", err)
+	}
+	digest := inputDigest(in)
+	if digest != "" && digest == k.last {
+		return nil
+	}
+	want, err := k.compose(ctx, in)
+	if err != nil {
+		k.last = ""
+		return err
+	}
+	if err := k.sync(ctx, want); err != nil {
+		k.last = ""
+		return err
+	}
+	k.last = digest
+	return nil
+}
+
+// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed,
+// and an empty line when one succeeds again.
+func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) {
+	safety := time.NewTicker(at.safety)
+	defer safety.Stop()
+	failing := ""
+	var began time.Time
+	for {
+		// The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own.
+		select {
+		case <-k.wake:
+		default:
+		}
+		began = time.Now()
+		why := ""
+		if err := k.pass(ctx); err != nil {
+			why = err.Error()
+		}
+		if why != failing {
+			said(why)
+			failing = why
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-k.wake:
+		case <-safety.C:
+		}
+		if rest := at.gap - time.Since(began); rest > 0 {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(rest):
 			}
+		}
+	}
+}
+
+// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same
+// for the same Input. Empty when it cannot be made, and then the pass composes.
+func inputDigest(in *meshInput) string {
+	refused := map[string]string{}
+	for node, err := range in.refused {
+		if err != nil {
+			refused[node] = err.Error()
+		}
+	}
+	bus := ""
+	if in.busRefused != nil {
+		bus = in.busRefused.Error()
+	}
+	body, err := json.Marshal(struct {
+		Input      compose.Input
+		Refused    map[string]string
+		BusRefused string
+	}{in.Input, refused, bus})
+	if err != nil {
+		return ""
+	}
+	sum := sha256.Sum256(body)
+	return hex.EncodeToString(sum[:])
+}
+
+// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions'
+// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with.
+func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer,
+	epoch func() uint64, holds func() bool) *packagesKeeper {
+	m := recordmirror.DeclaredPackages
+	var w *recordmirror.Writer
+	read := func(ctx context.Context) (*meshInput, error) {
+		reading, cancel := context.WithTimeout(ctx, 2*time.Minute)
+		defer cancel()
+		return readComposeInput(reading, open)
+	}
+	sync := func(ctx context.Context, want map[string][]byte) error {
+		writing, cancel := context.WithTimeout(ctx, 30*time.Second)
+		defer cancel()
+		if w == nil {
+			kv, err := api.KeyValue(writing, m.Bucket)
 			if err != nil {
-				return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
+				return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err)
+			}
+			w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
+		}
+		lost := w.LostEvents
+		r, err := w.Sync(writing, 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)
+		}
+		return err
+	}
+	k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds)
+	for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} {
+		go wakeOnBucket(ctx, api, bucket, k.poke)
+	}
+	go k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) {
+		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)
+		}
+	})
+	return k
+}
+
+// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is
+// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap.
+func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) {
+	for {
+		if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil {
+			fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(time.Minute):
+		}
+	}
+}
+
+// watchBucketOnce watches the bucket until its watch ends or ctx does.
+func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error {
+	kv, err := api.KeyValue(ctx, bucket)
+	if err != nil {
+		return err
+	}
+	watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly())
+	if err != nil {
+		return err
+	}
+	defer func() { _ = watch.Stop() }()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case e, ok := <-watch.Updates():
+			if !ok {
+				return fmt.Errorf("the watch ended")
 			}
-			if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
+			if e != nil {
-				return nil, err
+				wake()
 			}
 		}
-		return out, nil
 	}
 }

cmd/mesh-controller/record_mirrors_test.go

@@ -0,0 +1,191 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/compose"
+)
+
+// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
+// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).
+
+// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
+func scaled() packagesTiming {
+	return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
+}
+
+const scaledMinute = 60 * time.Millisecond
+
+// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
+type fakeMesh struct {
+	mu       sync.Mutex
+	assigned map[string][]string
+	composed atomic.Int64 // node compositions
+	synced   atomic.Int64
+	lastSync map[string][]byte
+	// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
+	block   chan struct{}
+	entered chan struct{}
+}
+
+func newFakeMesh(nodes int) *fakeMesh {
+	f := &fakeMesh{assigned: map[string][]string{}}
+	for i := 0; i < nodes; i++ {
+		f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
+	}
+	return f
+}
+
+func (f *fakeMesh) assign(node, module string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
+}
+
+func (f *fakeMesh) keeper() *packagesKeeper {
+	read := func(context.Context) (*meshInput, error) {
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
+		for _, n := range nodesOfFake(f.assigned) {
+			in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
+		}
+		return &meshInput{Input: in}, nil
+	}
+	composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
+		f.mu.Lock()
+		block, entered := f.block, f.entered
+		f.block = nil
+		f.mu.Unlock()
+		if block != nil {
+			close(entered)
+			<-block
+		}
+		out := map[string][]byte{}
+		for _, n := range in.Nodes {
+			f.composed.Add(1)
+			out[n.Name] = []byte(fmt.Sprint(n.Assigned))
+		}
+		return out, nil
+	}
+	syncAll := func(_ context.Context, want map[string][]byte) error {
+		f.mu.Lock()
+		f.lastSync = want
+		f.mu.Unlock()
+		f.synced.Add(1)
+		return nil
+	}
+	return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
+}
+
+func nodesOfFake(m map[string][]string) []string {
+	out := make([]string, 0, len(m))
+	for k := range m {
+		out = append(out, k)
+	}
+	for i := range out {
+		for j := i + 1; j < len(out); j++ {
+			if out[j] < out[i] {
+				out[i], out[j] = out[j], out[i]
+			}
+		}
+	}
+	return out
+}
+
+func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		k.run(ctx, scaled(), func(string) {})
+	}()
+	t.Cleanup(func() {
+		cancel()
+		<-done
+	})
+	return cancel
+}
+
+func eventually(t *testing.T, what string, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for !cond() {
+		if time.Now().After(deadline) {
+			t.Fatalf("not within a second: %s", what)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
+// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
+func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
+	const nodes = 5
+	f := newFakeMesh(nodes)
+	runKeeper(t, f.keeper())
+	eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
+	start := f.composed.Load()
+	time.Sleep(scaledMinute)
+	perMinute := f.composed.Load() - start
+	t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
+	if perMinute != 0 {
+		t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
+	}
+}
+
+// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
+// never two at once and never one per change.
+func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
+	f := newFakeMesh(1)
+	release := make(chan struct{})
+	f.block, f.entered = release, make(chan struct{})
+	k := f.keeper()
+	runKeeper(t, k)
+	<-f.entered
+	for i := 0; i < 50; i++ {
+		f.assign("node0", fmt.Sprintf("m%d", i))
+		k.poke()
+	}
+	close(release)
+	eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
+	time.Sleep(2*scaled().gap + 20*time.Millisecond)
+	if got := f.synced.Load(); got != 2 {
+		t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
+	}
+}
+
+// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
+func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
+	f := newFakeMesh(2)
+	k := f.keeper()
+	runKeeper(t, k)
+	eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
+	f.assign("node1", "viewer")
+	k.poke()
+	eventually(t, "the change mirrored", func() bool {
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		return string(f.lastSync["node1"]) == "[base viewer]"
+	})
+}
+
+// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
+func TestTheInputDigestFollowsTheInput(t *testing.T) {
+	of := func(assigned ...string) *meshInput {
+		return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
+	}
+	a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
+	if a == "" || a != b {
+		t.Errorf("the same input gave the digests %q and %q", a, b)
+	}
+	if a == c {
+		t.Errorf("an added assignment left the digest at %q", a)
+	}
+}