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.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.Predefined.Name: inv.PredefinedMirror,
|
||||
recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
|
||||
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
|
||||
recordmirror.Predefined.Name: inv.PredefinedMirror,
|
||||
}
|
||||
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
|
||||
// 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.
|
||||
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
|
||||
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.
|
||||
asked := in.Input
|
||||
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
|
||||
return declaredPackagesFrom(ctx, in)
|
||||
}
|
||||
}
|
||||
|
||||
// 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,
|
||||
// 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 declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
|
||||
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
|
||||
asked := in.Input
|
||||
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 {
|
||||
return nil, err
|
||||
if e != nil {
|
||||
wake()
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user