Merge pull request 'The controller's check stops waiting on fixtures and fixed intervals (issue 591)' (#279) from fix/591-the-controller-check-runs-in-minutes into main

cmd/mesh-controller/acting.go

@@ -64,6 +64,11 @@ type actor struct {
 	// the highest epoch issued, and what was said of it; zero when it was not (S12).
 	reset     time.Time
 	resetSaid string
+	// renewEvery and poll, when set, are how often the serving controller renews the lease it holds and
+	// looks again at one another holds, in place of the lease's own: a test's, so two controllers hand the
+	// lease over in a second rather than in a renewal of five (novox/hq issue 591). A renewal is given
+	// the lease's own RenewEvery to answer however often it is made, so a slow bus does not lose it.
+	renewEvery, poll time.Duration
 }
 
 // theLease is this process's standing to the lease.
@@ -223,7 +228,7 @@ func (a *actor) serveUnderTheLease(ctx context.Context, inv *inventory.Inventory
 	}
 	say := func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
 	l, err := lease.Open(ctx, api, broker.LeaseBucket, lease.Options{Holder: holderOf(instance),
-		Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) {
+		RenewEvery: a.renewEvery, RenewWithin: lease.RenewEvery, Poll: a.poll, Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) {
 			a.mu.Lock()
 			defer a.mu.Unlock()
 			a.reset = time.Now()

cmd/mesh-controller/lease_test.go

@@ -55,8 +55,14 @@ func TestTwoControllersOneActs(t *testing.T) {
 	inv := inventory.ForTest(t)
 	ctx := t.Context()
 
+	// Each renews five times and looks again ten times as often as a serving controller does, so the
+	// handover and the loss below are seen in a second, not in a renewal of five (novox/hq issue 591): what
+	// is checked is the order of events, which does not change with the interval. Each renewal still has a
+	// serving controller's five seconds to answer (acting.go), so a slow bus does not lose A the lease.
+	const renewEvery, poll = lease.RenewEvery / 5, lease.Poll / 10
+
 	// Controller A takes the lease.
-	a := &actor{}
+	a := &actor{renewEvery: renewEvery, poll: poll}
 	aCtx, stopA := context.WithCancel(ctx)
 	defer stopA()
 	lostA, err := a.serveUnderTheLease(aCtx, inv, url)
@@ -69,7 +75,7 @@ func TestTwoControllersOneActs(t *testing.T) {
 	}
 
 	// Controller B starts while A holds it, and waits — acting on nothing meanwhile.
-	b := &actor{}
+	b := &actor{renewEvery: renewEvery, poll: poll}
 	bCtx, stopB := context.WithCancel(ctx)
 	defer stopB()
 	tookB := make(chan (<-chan struct{}), 1)
@@ -82,10 +88,11 @@ func TestTwoControllersOneActs(t *testing.T) {
 		}
 		tookB <- lost
 	}()
+	// B looks fifteen times, across one of A's renewals at least.
 	select {
 	case <-tookB:
 		t.Fatal("B took the lease while A held it")
-	case <-time.After(3 * time.Second):
+	case <-time.After(15 * poll):
 	}
 	if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) {
 		t.Fatalf("B, waiting, may act: %v", err)
@@ -125,7 +132,7 @@ func TestTwoControllersOneActs(t *testing.T) {
 	}
 	select {
 	case <-lostB:
-	case <-time.After(2 * lease.RenewEvery):
+	case <-time.After(2 * renewEvery):
 		t.Fatal("B was not told it lost the lease")
 	}
 	if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) {

cmd/mesh-controller/plans_follow.go

@@ -30,8 +30,9 @@ var maxPlanWait = link.LongestAnswer - planWaitStart
 // answered that the call is still running, and `calls` with its id says what came of it.
 const planWaitStart = 5 * time.Second
 
-// planWaitEvery is how often a wait reads the plan again.
-const planWaitEvery = time.Second
+// planWaitEvery is how often a wait reads the plan again. A variable so a test reads it more often
+// (novox/hq issue 591).
+var planWaitEvery = time.Second
 
 // The things a wait waits until.
 var planWaitUntil = []string{"moved", "tier", "done"}

cmd/mesh-controller/plans_follow_test.go

@@ -138,6 +138,12 @@ func TestThePlansVerbComposesAWait(t *testing.T) {
 func TestAPlanFollowedFromTheStoreAnswersWhenItMoves(t *testing.T) {
 	open := aMesh(t)
 	ctx := t.Context()
+	// Read every 50ms rather than every second, so the wait answers as soon as the plan moves and the bound
+	// below passes in a fraction of a second (novox/hq issue 591); the move still waits 1.5s, long enough
+	// for the wait to have read the plan before it.
+	was := planWaitEvery
+	planWaitEvery = 50 * time.Millisecond
+	t.Cleanup(func() { planWaitEvery = was })
 	now := time.Now().UTC()
 	walk := inventory.Plan{ID: "plan-229", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c229c229",
 		Created: now, State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
@@ -169,11 +175,11 @@ func TestAPlanFollowedFromTheStoreAnswersWhenItMoves(t *testing.T) {
 	if got.Came != "tier" || got.Plan.Tier != 1 || got.Plan.State != inventory.PlanRolling {
 		t.Fatalf("the wait answered %+v", got)
 	}
-	if waited := time.Since(began); waited > 10*time.Second {
+	if waited := time.Since(began); waited > 3*time.Second {
 		t.Errorf("it answered after %s, not when the plan moved", waited)
 	}
 	out = printed(t, func() error {
-		return plansCommand(ctx, []string{"wait", walk.ID, "--bound", "2"})
+		return plansCommand(ctx, []string{"wait", walk.ID, "--bound", "300ms"})
 	})
 	if err := json.Unmarshal([]byte(out), &got); err != nil || got.Came != "bound" || got.Until != "moved" {
 		t.Fatalf("a plan that did not move answered %+v (%v)", got, err)

cmd/mesh-controller/sayable_test.go

@@ -26,6 +26,14 @@ import (
 // shape fails the suite, naming its source. The keeper would say such a summary in words at run time;
 // this is where the producer is made to say it rightly in the first place.
 func TestMain(m *testing.M) {
+	// **A command line this suite runs ends when it is done** (novox/hq issue 591). Many tests run this test
+	// binary as a verb's command line, and a binary built with the race detector sleeps a second before it
+	// exits (GORACE's atexit_sleep_ms), which was more than the rest of each such test. The children still
+	// run under the race detector and still fail on a race they find; they only stop waiting at their end.
+	// Set before any child starts, and only where nobody chose GORACE.
+	if _, chosen := os.LookupEnv("GORACE"); !chosen {
+		_ = os.Setenv("GORACE", "atexit_sleep_ms=0")
+	}
 	// The process a mesh-cli test runs as a command line: it says the verb and the caller it was given, and
 	// ends (meshcli_test.go).
 	// The process a mesh-cli test runs as `secret accept`: it reads its value exactly as `secret accept` does

internal/identity/fortest.go

@@ -29,14 +29,13 @@ func ForTest(t *testing.T) *Identity {
 	if len(name) > 60 {
 		name = name[:60]
 	}
-	conn, err := pgx.Connect(t.Context(), admin)
+	// Created already migrated, a copy of a template migrated once (novox/hq issue 591); the Migrate
+	// below then finds nothing pending, which it does only if the copy's record matches every migration.
+	migrations, err := Migrations()
 	if err != nil {
-		t.Fatalf("cannot reach the test PostgreSQL: %v", err)
+		t.Fatal(err)
 	}
-	if _, err := conn.Exec(t.Context(), "create database "+name); err != nil {
-		t.Fatalf("cannot create %s: %v", name, err)
-	}
-	conn.Close(t.Context())
+	store.CreateMigratedForTest(t, admin, name, migrations)
 
 	cut := strings.LastIndex(admin, "/")
 	t.Setenv(store.Variable(Name), admin[:cut]+"/"+name+"?sslmode=disable")
@@ -57,10 +56,6 @@ func ForTest(t *testing.T) *Identity {
 	if err := ident.Ready(t.Context(), 20*time.Second); err != nil {
 		t.Fatal(err)
 	}
-	migrations, err := Migrations()
-	if err != nil {
-		t.Fatal(err)
-	}
 	if _, err := ident.store.Migrate(t.Context(), migrations); err != nil {
 		t.Fatal(err)
 	}

internal/inventory/bususers.go

@@ -6,6 +6,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"sync/atomic"
 
 	"github.com/jackc/pgx/v5"
 	"golang.org/x/crypto/bcrypt"
@@ -23,6 +24,25 @@ import (
 // access change would blank every module's password if the mesh kept nothing (design 25 §4, and the
 // migration beside this).
 
+// busPasswordCostForTest, when set, is the bcrypt cost a bus password is hashed at instead of the
+// library's default. Only ForTest sets it, to bcrypt's least (novox/hq issue 591): at the default
+// cost one hash takes most of a second under the race detector, and a hash at the least cost is the
+// same kind of hash, checked by the same comparison and by the bus the same way.
+var busPasswordCostForTest atomic.Int32
+
+// busPasswordCost is the bcrypt cost a bus password is hashed at: the library's default, but in a test.
+func busPasswordCost() int {
+	if c := busPasswordCostForTest.Load(); c != 0 {
+		return int(c)
+	}
+	return bcrypt.DefaultCost
+}
+
+// hashBusPassword is the hash a bus password is kept as, minted, recorded or seeded alike.
+func hashBusPassword(password string) ([]byte, error) {
+	return bcrypt.GenerateFromPassword([]byte(password), busPasswordCost())
+}
+
 // BusUser is one user of the bus, as the mesh records it.
 type BusUser struct {
 	Username string
@@ -67,7 +87,7 @@ func (i *Inventory) RecordBusPassword(ctx context.Context, u BusUser, password s
 	if password == "" {
 		return errors.New("a bus user needs a password")
 	}
-	hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
+	hash, err := hashBusPassword(password)
 	if err != nil {
 		return fmt.Errorf("cannot hash a bus password: %w", err)
 	}
@@ -101,7 +121,7 @@ func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, err
 	// The cost the server will pay on every connection. Left at the library's default rather than
 	// raised: a node reconnecting after a network blip pays it, and the mesh's own links reconnect
 	// far more often than a person logs in anywhere.
-	hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
+	hash, err := hashBusPassword(password)
 	if err != nil {
 		return "", fmt.Errorf("cannot hash a bus password: %w", err)
 	}
@@ -183,7 +203,7 @@ func (i *Inventory) SeedBusUser(ctx context.Context, u BusUser, password string)
 	if u.Username == "" || u.Kind == "" || password == "" {
 		return errors.New("a bus user needs a username, a kind and the credential it is using")
 	}
-	hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
+	hash, err := hashBusPassword(password)
 	if err != nil {
 		return fmt.Errorf("cannot hash a bus password: %w", err)
 	}

internal/inventory/bususers_test.go

@@ -158,3 +158,17 @@ func TestACredentialTheMeshDidNotMintIsRecordedOnceAndNotOverwritten(t *testing.
 		t.Fatal("a restart put the bootstrap credential back over a rotated one")
 	}
 }
+
+// Outside a test a bus password is hashed at bcrypt's default cost: only ForTest lowers it (novox/hq issue 591).
+func TestABusPasswordIsHashedAtTheDefaultCostOutsideATest(t *testing.T) {
+	was := busPasswordCostForTest.Load()
+	t.Cleanup(func() { busPasswordCostForTest.Store(was) })
+	busPasswordCostForTest.Store(0)
+	hash, err := hashBusPassword("a password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, err := bcrypt.Cost(hash); err != nil || got != bcrypt.DefaultCost {
+		t.Fatalf("a bus password is hashed at cost %d (%v), not bcrypt's default %d", got, err, bcrypt.DefaultCost)
+	}
+}

internal/inventory/fortest.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
+	"golang.org/x/crypto/bcrypt"
 
 	"github.com/novox/mesh-controller/internal/store"
 )
@@ -30,14 +31,14 @@ func ForTest(t *testing.T) *Inventory {
 		name = name[:60]
 	}
 
-	conn, err := pgx.Connect(t.Context(), admin)
+	// Created already migrated, a copy of a template migrated once (novox/hq issue 591); the Migrate
+	// below then finds nothing pending, which it does only if the copy's record matches every migration.
+	migrations, err := Migrations()
 	if err != nil {
-		t.Fatalf("cannot reach the test PostgreSQL: %v", err)
+		t.Fatal(err)
 	}
-	if _, err := conn.Exec(t.Context(), "create database "+name); err != nil {
-		t.Fatalf("cannot create %s: %v", name, err)
-	}
-	conn.Close(t.Context())
+	store.CreateMigratedForTest(t, admin, name, migrations)
+	busPasswordCostForTest.Store(int32(bcrypt.MinCost))
 
 	cut := strings.LastIndex(admin, "/")
 	t.Setenv(store.Variable(Name), admin[:cut]+"/"+name+"?sslmode=disable")
@@ -59,10 +60,6 @@ func ForTest(t *testing.T) *Inventory {
 		t.Fatal(err)
 	}
 
-	migrations, err := Migrations()
-	if err != nil {
-		t.Fatal(err)
-	}
 	if _, err := inv.store.Migrate(t.Context(), migrations); err != nil {
 		t.Fatal(err)
 	}

internal/lease/lease.go

@@ -75,6 +75,10 @@ type Options struct {
 	Holder Holder
 	// RenewEvery, Margin and Poll default to the package's.
 	RenewEvery, Margin, Poll time.Duration
+	// RenewWithin is how long one renewal may take before it counts as failed: RenewEvery when zero. Apart
+	// from it so a test that renews often still gives a slow bus a serving controller's time to answer
+	// (novox/hq issue 591).
+	RenewWithin time.Duration
 	// Floor is the highest epoch this mesh has issued, from the controller's own store. **An epoch is
 	// never issued twice**: a lease bucket raised again from nothing — a bus whose data was replaced —
 	// starts its revisions over, and every machine that heard epoch 57 would refuse the next controller
@@ -138,6 +142,9 @@ func Open(ctx context.Context, js jetstream.JetStream, bucket string, o Options)
 	if o.RenewEvery <= 0 {
 		o.RenewEvery = RenewEvery
 	}
+	if o.RenewWithin <= 0 {
+		o.RenewWithin = o.RenewEvery
+	}
 	if o.Margin <= 0 {
 		o.Margin = Margin
 	}
@@ -406,7 +413,7 @@ func (l *Lease) Renew(ctx context.Context) error {
 	if err != nil {
 		return l.lose(err)
 	}
-	renewing, cancel := context.WithTimeout(ctx, l.o.RenewEvery)
+	renewing, cancel := context.WithTimeout(ctx, l.o.RenewWithin)
 	defer cancel()
 	next, err := l.kv.Update(renewing, Key, value, revision)
 	if err != nil {

internal/licences/fortest.go

@@ -32,14 +32,13 @@ func ForTest(t *testing.T) *Licences {
 		name = name[:60]
 	}
 
-	conn, err := pgx.Connect(t.Context(), admin)
+	// Created already migrated, a copy of a template migrated once (novox/hq issue 591); the Migrate
+	// below then finds nothing pending, which it does only if the copy's record matches every migration.
+	migrations, err := Migrations()
 	if err != nil {
-		t.Fatalf("cannot reach the test PostgreSQL: %v", err)
+		t.Fatal(err)
 	}
-	if _, err := conn.Exec(t.Context(), "create database "+name); err != nil {
-		t.Fatalf("cannot create %s: %v", name, err)
-	}
-	conn.Close(t.Context())
+	store.CreateMigratedForTest(t, admin, name, migrations)
 
 	cut := strings.LastIndex(admin, "/")
 	t.Setenv(store.Variable(Name), admin[:cut]+"/"+name+"?sslmode=disable")
@@ -61,10 +60,6 @@ func ForTest(t *testing.T) *Licences {
 		t.Fatal(err)
 	}
 
-	migrations, err := Migrations()
-	if err != nil {
-		t.Fatal(err)
-	}
 	if _, err := held.store.Migrate(t.Context(), migrations); err != nil {
 		t.Fatal(err)
 	}

internal/link/queue.go

@@ -321,11 +321,23 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin
 		return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
 	}
 	defer func() { _ = sub.Unsubscribe() }()
-	for {
-		batch, err := sub.Fetch(500, nats.MaxWait(2*time.Second))
+	// **Read what is there, then stop** (novox/hq issue 591): the consumer says how many events it holds
+	// from the moment asked, and the reading ends once they are read. A fetch for a full batch waits out
+	// its whole wait when fewer are there, so every reading of the queue took two seconds and more.
+	info, err := sub.ConsumerInfo()
+	if err != nil {
+		return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
+	}
+	for remaining := info.NumPending; remaining > 0; {
+		batch, err := sub.Fetch(int(min(remaining, 500)), nats.MaxWait(2*time.Second))
 		if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) {
 			return heard, fmt.Errorf("reading %s: %w", subject, err)
 		}
+		// None in the whole wait: what was counted is gone from the stream since (its limits), not coming.
+		if len(batch) == 0 {
+			break
+		}
+		remaining -= uint64(len(batch))
 		for _, msg := range batch {
 			switch msg.Subject {
 			case BuildStartedOf(seat):
@@ -342,9 +354,6 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin
 				}
 			}
 		}
-		if len(batch) < 500 {
-			break
-		}
 		if ctx.Err() != nil {
 			return heard, ctx.Err()
 		}

internal/store/fortest.go

@@ -0,0 +1,138 @@
+package store
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// CreateMigratedForTest creates the database name on the test PostgreSQL that admin reaches, holding
+// every one of migrations already applied and recorded, as a fresh database migrated by Migrate would.
+//
+// **A copy of a template, migrated once** (novox/hq issue 591): a store test used to create an empty
+// database and run every migration into it, some 90 for the inventory, and with hundreds of store tests
+// that alone was most of the controller's check. The template is migrated by Migrate itself, once for
+// each set of migrations, and every test's database is copied from it by PostgreSQL. The caller still
+// runs Migrate on the copy, which finds nothing pending only if the copy's record matches the migrations
+// exactly, so a template that disagreed with them would fail the test rather than pass it unseen.
+//
+// The template is named after a digest of the migrations, so a changed or added migration is a new
+// template, never an old one reused. Packages tested at once in separate processes share one: it is
+// built under an advisory lock, and a template a crashed build left half done is not marked a template
+// and is built again. It takes a *testing.T, so nothing that is not a test can call it.
+func CreateMigratedForTest(t *testing.T, admin, name string, migrations []Migration) {
+	t.Helper()
+	template, err := templateFor(admin, migrations)
+	if err != nil {
+		t.Fatalf("cannot raise the migrated template: %v", err)
+	}
+	conn, err := pgx.Connect(t.Context(), admin)
+	if err != nil {
+		t.Fatalf("cannot reach the test PostgreSQL: %v", err)
+	}
+	defer conn.Close(context.Background())
+	// A copy is refused while anything is connected to its template; the template allows no connection
+	// once built, so only the builder's last connection closing can still be in the way.
+	for deadline := time.Now().Add(10 * time.Second); ; {
+		_, err = conn.Exec(t.Context(), "create database "+name+" template "+template)
+		var pgErr *pgconn.PgError
+		if err == nil || !errors.As(err, &pgErr) || pgErr.Code != "55006" || time.Now().After(deadline) {
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	if err != nil {
+		t.Fatalf("cannot create %s from %s: %v", name, template, err)
+	}
+}
+
+// templates are the templates this process has raised or found, by name.
+var templates sync.Map // name → *templateOnce
+
+type templateOnce struct {
+	once sync.Once
+	err  error
+}
+
+// templateFor is the name of the template holding migrations, raised if no process has yet.
+func templateFor(admin string, migrations []Migration) (string, error) {
+	digest := sha256.New()
+	for _, m := range migrations {
+		fmt.Fprintf(digest, "%d %s %s\n", m.Number, m.Name, m.Checksum)
+	}
+	sum := digest.Sum(nil)
+	name := "tpl_" + hex.EncodeToString(sum[:12])
+	held, _ := templates.LoadOrStore(name, &templateOnce{})
+	once := held.(*templateOnce)
+	once.once.Do(func() {
+		// Not the asking test's context: the template outlives that test, and is raised once for all.
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		defer cancel()
+		once.err = raiseTemplate(ctx, admin, name, int64(binary.BigEndian.Uint64(sum[12:20])), migrations)
+	})
+	return name, once.err
+}
+
+// raiseTemplate builds the template name under the advisory lock key, unless it stands built.
+func raiseTemplate(ctx context.Context, admin, name string, key int64, migrations []Migration) error {
+	conn, err := pgx.Connect(ctx, admin)
+	if err != nil {
+		return err
+	}
+	defer conn.Close(context.Background())
+	if _, err := conn.Exec(ctx, `select pg_advisory_lock($1)`, key); err != nil {
+		return err
+	}
+	defer func() { _, _ = conn.Exec(context.Background(), `select pg_advisory_unlock($1)`, key) }()
+
+	var built bool
+	if err := conn.QueryRow(ctx, `select exists (select 1 from pg_database where datname = $1 and datistemplate)`,
+		name).Scan(&built); err != nil {
+		return err
+	}
+	if built {
+		return nil
+	}
+	if _, err := conn.Exec(ctx, "drop database if exists "+name+" with (force)"); err != nil {
+		return err
+	}
+	if _, err := conn.Exec(ctx, "create database "+name); err != nil {
+		return err
+	}
+	cut := strings.LastIndex(admin, "/")
+	rest := ""
+	if q := strings.Index(admin[cut:], "?"); q >= 0 {
+		rest = admin[cut+q:]
+	}
+	pool, err := pgxpool.New(ctx, admin[:cut]+"/"+name+rest)
+	if err != nil {
+		return err
+	}
+	_, err = (&Store{context: "template", pool: pool}).Migrate(ctx, migrations)
+	pool.Close()
+	if err != nil {
+		return err
+	}
+	// Marked a template only once whole, and closed to connections, so a copy is never refused for one.
+	for _, sql := range []string{
+		"alter database " + name + " with allow_connections false",
+		"select pg_terminate_backend(pid) from pg_stat_activity where datname = '" + name + "'",
+		"alter database " + name + " with is_template true",
+	} {
+		if _, err := conn.Exec(ctx, sql); err != nil {
+			return err
+		}
+	}
+	return nil
+}