From d977f64ec9f84e7f0bd839f99187765a04324dbe Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 20:37:58 +0200 Subject: [PATCH 01/10] Read an empty source ref as the trunk and say a merge into another branch starts no delivery (issue 977, part A, issue 889) --- cmd/mesh-controller/batches.go | 2 +- cmd/mesh-controller/non_trunk_merge_test.go | 53 +++++++++++++++++++++ cmd/mesh-controller/upgrades.go | 34 ++++++++++--- 3 files changed, 81 insertions(+), 8 deletions(-) create mode 100644 cmd/mesh-controller/non_trunk_merge_test.go diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index 81a52afb..165bdb42 100644 --- a/cmd/mesh-controller/batches.go +++ b/cmd/mesh-controller/batches.go @@ -177,7 +177,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T repository, m.Base, m.Commit, already) return nil } - fmt.Printf("%s merged into %s (%.8s); nothing the mesh holds reads it\n", repository, m.Base, m.Commit) + fmt.Print(nothingReadsIt(repository, m)) return nil } // **Its delivery group's word, before the plans are held** (ADR 0306 decision 2): a slow answer holds up no walk. diff --git a/cmd/mesh-controller/non_trunk_merge_test.go b/cmd/mesh-controller/non_trunk_merge_test.go new file mode 100644 index 00000000..3e7c190d --- /dev/null +++ b/cmd/mesh-controller/non_trunk_merge_test.go @@ -0,0 +1,53 @@ +package main + +import ( + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// A merge into a branch other than the trunk starts no delivery (novox/hq ADR 0401 clause 3, issue 889): a +// module recorded with no ref follows the trunk, main, and no other branch. +func TestAMergeIntoANonTrunkBranchStartsNoDelivery(t *testing.T) { + for ref, name := range map[string]string{"": "noref", "main": "mainref"} { + t.Run(name, func(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + if err := open.inventory.RegisterModule(t.Context(), catalogue.Manifest{Module: "free", Version: "1"}, + inventory.Source{Repository: "novox/free", Seat: "git", Ref: ref, BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + feature := lone(repoMerge("free", "f1", t0)) + feature.Base = "feat/57-config-and-provisioning" + said := captured(t, func() error { return (following{open: open}).hearMerge(t.Context(), feature, t0) }) + if ws, bs := walks(t, open); len(ws) != 0 || len(bs) != 0 { + t.Fatalf("a merge into a feature branch: %d walk(s), %d batch(es)", len(ws), len(bs)) + } + if !strings.Contains(said, feature.Base) || !strings.Contains(said, "starts no delivery") { + t.Fatalf("the merge was not said to start no delivery: %q", said) + } + hear(t, open, lone(repoMerge("free", "m1", t0)), t0) + if ws, bs := walks(t, open); len(ws)+len(bs) != 1 { + t.Fatalf("a merge into main: %d walk(s), %d batch(es)", len(ws), len(bs)) + } + }) + } +} + +// A module recorded with an explicit branch keeps following that branch (the lead's settlement of issue 977 Q5). +func TestAModuleRecordedWithABranchStillFollowsIt(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + if err := open.inventory.RegisterModule(t.Context(), catalogue.Manifest{Module: "pinned", Version: "1"}, + inventory.Source{Repository: "novox/pinned", Seat: "git", Ref: "feat/x", BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + m := lone(repoMerge("pinned", "p1", t0)) + m.Base = "feat/x" + hear(t, open, m, t0) + if ws, bs := walks(t, open); len(ws)+len(bs) != 1 { + t.Fatalf("a merge into the branch a module follows: %d walk(s), %d batch(es)", len(ws), len(bs)) + } +} diff --git a/cmd/mesh-controller/upgrades.go b/cmd/mesh-controller/upgrades.go index 31626b80..ea9ec498 100644 --- a/cmd/mesh-controller/upgrades.go +++ b/cmd/mesh-controller/upgrades.go @@ -326,8 +326,7 @@ func movesOfMerge(ctx context.Context, inv *inventory.Inventory, m link.SourceMo "from it\n", m.Owner, m.Repo, m.Base, m.Commit, already) return nil, nil } - fmt.Printf("%s/%s merged into %s (%.8s); nothing the mesh holds reads it\n", - m.Owner, m.Repo, m.Base, m.Commit) + fmt.Print(nothingReadsIt(m.Owner+"/"+m.Repo, m)) return nil, nil } // Said, never silent (novox/hq 04-ISSUES/215): a module built from this repository that follows @@ -531,7 +530,7 @@ func whyMoved(e inventory.Entry, read []inventory.ReadRepository, m link.SourceM return "read whole: " + whole } for _, r := range read { - if r.Own || !sameRepository(r.Repository, m) || (r.Ref != "" && r.Ref != m.Base) { + if r.Own || !sameRepository(r.Repository, m) || !followsBase(r.Ref, m.Base) { continue } if len(r.Paths) > 0 { @@ -615,6 +614,17 @@ func retireDeleted(ctx context.Context, inv *inventory.Inventory, e inventory.En } } +// nothingReadsIt is what is said of a merge no module the mesh holds reads. **A merge into a branch other than the +// trunk is said to start no delivery** (novox/hq ADR 0401 clause 3, issue 889): every delivery is cut from a +// repository's trunk, and whoever merged into a feature branch should not look for a broken trigger. +func nothingReadsIt(repository string, m link.SourceMoved) string { + if m.Base != "" && m.Base != "main" { + return fmt.Sprintf("%s merged into %s (%.8s), not the trunk of %s; a merge into any other branch starts no "+ + "delivery\n", repository, m.Base, m.Commit, repository) + } + return fmt.Sprintf("%s merged into %s (%.8s); nothing the mesh holds reads it\n", repository, m.Base, m.Commit) +} + // sourceIs is whether a recorded source is the repository and branch a merge announced. A source on // the git seat is recorded as its path on the forge; one elsewhere as the URL it was cloned from. // An empty recorded ref is the repository's default branch, which is what a merge into the base @@ -623,8 +633,18 @@ func sourceIs(s inventory.Source, m link.SourceMoved) bool { if !sameRepository(s.Repository, m) { return false } - ref := followedBranch(s.Ref) - return ref == "" || ref == m.Base + return followsBase(s.Ref, m.Base) +} + +// followsBase is whether a recorded ref follows the branch a merge went into. **An empty ref is the +// trunk, main, and nothing else** (novox/hq ADR 0401 clause 3, issue 889): it used to match a merge +// into any branch, so a merge into a feature branch started a delivery. A ref recorded as a branch +// keeps following that branch. +func followsBase(ref, base string) bool { + if ref = strings.TrimSpace(followedBranch(ref)); ref == "" { + ref = "main" + } + return ref == base } // commitRef is a ref that names a commit rather than a branch: what `build --ref ` asks for. @@ -663,7 +683,7 @@ func sameRepository(repository string, m link.SourceMoved) bool { // read, as before. func readsFrom(read []inventory.ReadRepository, m link.SourceMoved) bool { for _, r := range read { - if r.Own || !sameRepository(r.Repository, m) || (r.Ref != "" && r.Ref != m.Base) { + if r.Own || !sameRepository(r.Repository, m) || !followsBase(r.Ref, m.Base) { continue } if len(r.Paths) == 0 || len(m.Paths) == 0 || m.PathsTruncated { @@ -757,7 +777,7 @@ func unchangedOfMerge(m link.SourceMoved, entries []inventory.Entry, // branch, as a context beside its own. func packagesSaid(read []inventory.ReadRepository, m link.SourceMoved) bool { for _, r := range read { - if !r.Own && sameRepository(r.Repository, m) && (r.Ref == "" || r.Ref == m.Base) && len(r.Paths) > 0 { + if !r.Own && sameRepository(r.Repository, m) && followsBase(r.Ref, m.Base) && len(r.Paths) > 0 { return true } } -- 2.54.0 From 14fe2ad7362b3617897c2696aeb6ac227f4b2de5 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 20:41:57 +0200 Subject: [PATCH 02/10] Split the unit level from the integration level outside cmd, builder, link, catalogue, inventory; merge-check.sh times and bounds the unit level (issue 895) --- Makefile | 8 +- internal/artifacts/live_registry_test.go | 2 + internal/artifacts/tagged_integration_test.go | 72 +++++ internal/artifacts/tagged_test.go | 62 ----- internal/broker/callers_live_test.go | 2 + internal/broker/config_owner_live_test.go | 2 + internal/broker/consumer_from_now_test.go | 2 + internal/broker/consumer_upgrade_test.go | 2 + .../controller_buckets_integration_test.go | 261 ++++++++++++++++++ internal/broker/controller_buckets_test.go | 248 ----------------- internal/broker/jetstream_test.go | 2 + internal/broker/log_integration_test.go | 69 +++++ internal/broker/log_test.go | 58 ---- internal/broker/modstreams_live_test.go | 2 + internal/broker/msgttl_live_test.go | 2 + internal/broker/prov_applied_live_test.go | 2 + internal/broker/provisioner_live_test.go | 2 + .../broker/provisioner_server_live_test.go | 2 + internal/broker/raise_live_test.go | 2 + internal/broker/snapshot_live_test.go | 2 + internal/broker/state_integration_test.go | 55 ++++ internal/broker/state_test.go | 46 --- .../vault_calls_live_integration_test.go | 130 +++++++++ internal/broker/vault_calls_live_test.go | 122 -------- internal/broker/worker_type_change_test.go | 2 + internal/conditions/bus_test.go | 2 + internal/conditions/store.go | 27 +- internal/conditions/store_test.go | 4 +- internal/identity/authority_test.go | 2 + internal/identity/identity_test.go | 2 + internal/lease/lease_test.go | 2 + internal/licences/licences_test.go | 2 + internal/licences/refresh_test.go | 2 + internal/licences/submitrefresh_test.go | 2 + internal/lint/mergecheck_unit_level_test.go | 76 +++++ .../recordmirror/mirror_integration_test.go | 248 +++++++++++++++++ internal/recordmirror/mirror_store_test.go | 2 + internal/recordmirror/mirror_test.go | 231 ---------------- .../recordmirror/provisioning_live_test.go | 2 + internal/store/live_test.go | 2 + internal/testbus/testbus_integration_test.go | 45 +++ internal/testbus/testbus_test.go | 36 --- merge-check.sh | 92 +++++- testdata/store-tests | 1 + 44 files changed, 1113 insertions(+), 826 deletions(-) create mode 100644 internal/artifacts/tagged_integration_test.go create mode 100644 internal/broker/controller_buckets_integration_test.go create mode 100644 internal/broker/log_integration_test.go create mode 100644 internal/broker/state_integration_test.go create mode 100644 internal/broker/vault_calls_live_integration_test.go create mode 100644 internal/lint/mergecheck_unit_level_test.go create mode 100644 internal/recordmirror/mirror_integration_test.go create mode 100644 internal/testbus/testbus_integration_test.go diff --git a/Makefile b/Makefile index 4522ff06..fce59ecf 100644 --- a/Makefile +++ b/Makefile @@ -110,14 +110,18 @@ proxy-image: # after formatting never ran through it. The script is the one list of steps; this target only gives it # a database (MESH_TEST_POSTGRES, exported above). # +# **Two levels** (issue 895): the unit level, `go test ./...` untagged, timed and held to a bound; and the integration +# level, `-tags integration`, the tests that start a bus, a store or another process. Each is said with its time. +# # **Packages in parallel, under the race detector, each test on a bus of its own** (internal/testbus), # as the script runs them: a bus per test, of the release the mesh runs, makes the suite the same in any # order, and the timeout bounds a hang to a failure with a stack, never a stalled gate. check: postgres @sh merge-check.sh ; status=$$? ; docker rm -f $(PG_CONTAINER) >/dev/null 2>&1 || true ; exit $$status -# Without a database the store's tests skip rather than fail, so this is the honest subset and not -# the gate. +# The unit level alone (untagged: no bus, no store, no other process), which a developer runs continuously. The tests +# that need a bus or the store carry `//go:build integration` and run in `check`, not here; without a database the +# store's tests would skip rather than fail anyway. test: go test -timeout 15m ./... diff --git a/internal/artifacts/live_registry_test.go b/internal/artifacts/live_registry_test.go index c88b37f6..1bbf32f2 100644 --- a/internal/artifacts/live_registry_test.go +++ b/internal/artifacts/live_registry_test.go @@ -1,3 +1,5 @@ +//go:build integration + package artifacts import ( diff --git a/internal/artifacts/tagged_integration_test.go b/internal/artifacts/tagged_integration_test.go new file mode 100644 index 00000000..0595fa9c --- /dev/null +++ b/internal/artifacts/tagged_integration_test.go @@ -0,0 +1,72 @@ +//go:build integration + +package artifacts + +import ( + "context" + "errors" + "fmt" + "os" + "os/exec" + "strings" + "testing" + "time" +) + +func TestLiveADocumentPutUnderATagIsReadBackAndOnlyTheNewestIsKept(t *testing.T) { + address := os.Getenv("MESH_TEST_REGISTRY") + container := os.Getenv("MESH_TEST_REGISTRY_CONTAINER") + if address == "" || container == "" { + t.Skip("no MESH_TEST_REGISTRY / MESH_TEST_REGISTRY_CONTAINER; see the collector test's comment for the registry to raise") + } + ctx := context.Background() + store := Store{Address: address} + repository := fmt.Sprintf("facts-live-%d", time.Now().UnixNano()) + + if _, _, err := store.GetTagged(ctx, repository, "latest"); !errors.Is(err, ErrNoTag) { + t.Fatalf("nothing was put and the read said %v, not that nothing is there", err) + } + first := []byte(`{"facts":1,"taken":"first"}`) + firstDigest, err := store.PutTagged(ctx, repository, "latest", "application/vnd.novox.mesh.facts.v1+json", first) + if err != nil { + t.Fatal(err) + } + second := []byte(`{"facts":1,"taken":"second"}`) + secondDigest, err := store.PutTagged(ctx, repository, "latest", "application/vnd.novox.mesh.facts.v1+json", second) + if err != nil { + t.Fatal(err) + } + got, digest, err := store.GetTagged(ctx, repository, "latest") + if err != nil { + t.Fatal(err) + } + if string(got) != string(second) || digest != secondDigest { + t.Fatalf("read back %q (%s), not the newest put %q (%s)", got, digest, second, secondDigest) + } + + // The collector, as the store's nightly step runs it — with no `--delete-untagged`: the newest kept, + // the replaced one taken because its manifest was let go of. + out, err := exec.Command("docker", "exec", container, "registry", "garbage-collect", + "/etc/docker/registry/config.yml").CombinedOutput() + if err != nil { + t.Fatalf("the collector did not run: %v\n%s", err, out) + } + if got, _, err := store.GetTagged(ctx, repository, "latest"); err != nil || string(got) != string(second) { + t.Fatalf("after collection the newest reads %q, %v", got, err) + } + // On the store's disk, not as the running server answers: it caches blob descriptors in memory. + onDisk := func(digest string) bool { + hex := strings.TrimPrefix(digest, "sha256:") + path := "/var/lib/registry/docker/registry/v2/blobs/sha256/" + hex[:2] + "/" + hex + "/data" + return exec.Command("docker", "exec", container, "test", "-f", path).Run() == nil + } + if onDisk(firstDigest) { + t.Errorf("the replaced snapshot %s is still in the store after collection; nothing would ever take it", firstDigest) + } + if !onDisk(secondDigest) { + t.Errorf("the collector took the newest snapshot %s", secondDigest) + } + if !strings.HasPrefix(secondDigest, "sha256:") { + t.Errorf("the digest said %q", secondDigest) + } +} diff --git a/internal/artifacts/tagged_test.go b/internal/artifacts/tagged_test.go index 341553c5..6eba0bd2 100644 --- a/internal/artifacts/tagged_test.go +++ b/internal/artifacts/tagged_test.go @@ -2,13 +2,8 @@ package artifacts import ( "context" - "errors" - "fmt" - "os" - "os/exec" "strings" "testing" - "time" ) // The facts snapshot is put under a tag and read back by it (novox/hq to-be 45 §9). @@ -17,63 +12,6 @@ import ( // that a manifest put by tag is read by tag, that the layer comes back whole and checked, and that the // snapshot a newer one replaced is the collector's to take while the newest is kept. Raised as the // collector test above says, with MESH_TEST_REGISTRY and MESH_TEST_REGISTRY_CONTAINER. -func TestLiveADocumentPutUnderATagIsReadBackAndOnlyTheNewestIsKept(t *testing.T) { - address := os.Getenv("MESH_TEST_REGISTRY") - container := os.Getenv("MESH_TEST_REGISTRY_CONTAINER") - if address == "" || container == "" { - t.Skip("no MESH_TEST_REGISTRY / MESH_TEST_REGISTRY_CONTAINER; see the collector test's comment for the registry to raise") - } - ctx := context.Background() - store := Store{Address: address} - repository := fmt.Sprintf("facts-live-%d", time.Now().UnixNano()) - - if _, _, err := store.GetTagged(ctx, repository, "latest"); !errors.Is(err, ErrNoTag) { - t.Fatalf("nothing was put and the read said %v, not that nothing is there", err) - } - first := []byte(`{"facts":1,"taken":"first"}`) - firstDigest, err := store.PutTagged(ctx, repository, "latest", "application/vnd.novox.mesh.facts.v1+json", first) - if err != nil { - t.Fatal(err) - } - second := []byte(`{"facts":1,"taken":"second"}`) - secondDigest, err := store.PutTagged(ctx, repository, "latest", "application/vnd.novox.mesh.facts.v1+json", second) - if err != nil { - t.Fatal(err) - } - got, digest, err := store.GetTagged(ctx, repository, "latest") - if err != nil { - t.Fatal(err) - } - if string(got) != string(second) || digest != secondDigest { - t.Fatalf("read back %q (%s), not the newest put %q (%s)", got, digest, second, secondDigest) - } - - // The collector, as the store's nightly step runs it — with no `--delete-untagged`: the newest kept, - // the replaced one taken because its manifest was let go of. - out, err := exec.Command("docker", "exec", container, "registry", "garbage-collect", - "/etc/docker/registry/config.yml").CombinedOutput() - if err != nil { - t.Fatalf("the collector did not run: %v\n%s", err, out) - } - if got, _, err := store.GetTagged(ctx, repository, "latest"); err != nil || string(got) != string(second) { - t.Fatalf("after collection the newest reads %q, %v", got, err) - } - // On the store's disk, not as the running server answers: it caches blob descriptors in memory. - onDisk := func(digest string) bool { - hex := strings.TrimPrefix(digest, "sha256:") - path := "/var/lib/registry/docker/registry/v2/blobs/sha256/" + hex[:2] + "/" + hex + "/data" - return exec.Command("docker", "exec", container, "test", "-f", path).Run() == nil - } - if onDisk(firstDigest) { - t.Errorf("the replaced snapshot %s is still in the store after collection; nothing would ever take it", firstDigest) - } - if !onDisk(secondDigest) { - t.Errorf("the collector took the newest snapshot %s", secondDigest) - } - if !strings.HasPrefix(secondDigest, "sha256:") { - t.Errorf("the digest said %q", secondDigest) - } -} // A store with no address is said, not dialled. func TestADocumentWithNowhereToGoIsRefusedByName(t *testing.T) { diff --git a/internal/broker/callers_live_test.go b/internal/broker/callers_live_test.go index cacb0424..8d36d37f 100644 --- a/internal/broker/callers_live_test.go +++ b/internal/broker/callers_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/config_owner_live_test.go b/internal/broker/config_owner_live_test.go index 6d364778..b5f80422 100644 --- a/internal/broker/config_owner_live_test.go +++ b/internal/broker/config_owner_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/consumer_from_now_test.go b/internal/broker/consumer_from_now_test.go index 3500603c..fa64d770 100644 --- a/internal/broker/consumer_from_now_test.go +++ b/internal/broker/consumer_from_now_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/consumer_upgrade_test.go b/internal/broker/consumer_upgrade_test.go index 5cc0215a..1c7419e1 100644 --- a/internal/broker/consumer_upgrade_test.go +++ b/internal/broker/consumer_upgrade_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/controller_buckets_integration_test.go b/internal/broker/controller_buckets_integration_test.go new file mode 100644 index 00000000..106bb3b9 --- /dev/null +++ b/internal/broker/controller_buckets_integration_test.go @@ -0,0 +1,261 @@ +//go:build integration + +package broker + +import ( + "context" + "slices" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/testbus" +) + +func TestWhatTheControllerAskedIsBounded(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + kv, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + bucket, err := kv.KeyValue(ctx, AskedBucket) + if err != nil { + t.Fatal(err) + } + status, err := bucket.Status(ctx) + if err != nil { + t.Fatal(err) + } + info := status.(*jetstream.KeyValueBucketStatus).StreamInfo() + if status.History() != 1 || status.TTL() != AskedKeptFor || info.Config.MaxBytes <= 0 || info.Config.MaxBytes > 64<<20 { + t.Errorf("history %d, age %s, bytes %d", status.History(), status.TTL(), info.Config.MaxBytes) + } +} + +func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) { + controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(controller.Publish, "$KV."+MergeWindowBucket+".>") { + t.Error("the controller may not write the merge window's bucket") + } + if !IsControllerBucket(MergeWindowBucket) || !slices.Contains(ControllerBuckets(), MergeWindowBucket) { + t.Error("the merge window's bucket is not asserted as the controller's own") + } + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.KeyValue(ctx, MergeWindowBucket) + if err != nil { + t.Fatal(err) + } + w, err := kv.WatchAll(ctx) + if err != nil { + t.Fatal(err) + } + defer w.Stop() + if _, err := kv.Create(ctx, "plan-1.quiet", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { + t.Fatal(err) + } + for { + select { + case e := <-w.Updates(): + if e != nil && e.Key() == "plan-1.quiet" && e.Operation() != jetstream.KeyValuePut { + return + } + case <-ctx.Done(): + t.Fatal("no watcher heard the key expire") + } + } +} + +func TestTheRolloutsSilentKeysExpireAsAPurgeAndAReportsDeleteIsNot(t *testing.T) { + controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(controller.Publish, "$KV."+RolloutSilentBucket+".>") { + t.Error("the controller may not write the rollout's silent keys' bucket") + } + if !IsControllerBucket(RolloutSilentBucket) || !slices.Contains(ControllerBuckets(), RolloutSilentBucket) { + t.Error("the rollout's silent keys' bucket is not asserted as the controller's own") + } + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.KeyValue(ctx, RolloutSilentBucket) + if err != nil { + t.Fatal(err) + } + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + t.Fatal(err) + } + defer w.Stop() + if _, err := kv.Create(ctx, "plan-1.0.ace", []byte("x"), jetstream.KeyTTL(time.Hour)); err != nil { + t.Fatal(err) + } + if err := kv.Delete(ctx, "plan-1.0.ace"); err != nil { + t.Fatal(err) + } + if _, err := kv.Create(ctx, "plan-1.0.g14", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { + t.Fatal(err) + } + for { + select { + case e := <-w.Updates(): + if e == nil || e.Operation() == jetstream.KeyValuePut { + continue + } + if e.Key() == "plan-1.0.ace" && e.Operation() != jetstream.KeyValueDelete { + t.Fatalf("a report's delete was heard as %v", e.Operation()) + } + if e.Key() == "plan-1.0.g14" { + if e.Operation() != jetstream.KeyValuePurge { + t.Fatalf("an expiry was heard as %v, not a purge", e.Operation()) + } + return + } + case <-ctx.Done(): + t.Fatal("no watcher heard the key expire") + } + } +} + +func TestTheSilentKeysBucketAgesAndAnOlderOneIsBroughtToIt(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + // As the first build made it: no age. + if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: RolloutSilentBucket, History: 1, + MaxValueSize: 4 << 10, MaxBytes: 8 << 20, Storage: jetstream.FileStorage, LimitMarkerTTL: RolloutSilentMarkerTTL}); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatalf("a bucket made without the age is not brought to it: %v", err) + } + kv, err := api.KeyValue(ctx, RolloutSilentBucket) + if err != nil { + t.Fatal(err) + } + status, err := kv.Status(ctx) + if err != nil { + t.Fatal(err) + } + if status.TTL() != RolloutSilentKeptFor { + t.Errorf("the bucket ages at %v; want %v", status.TTL(), RolloutSilentKeptFor) + } +} + +func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + if err := AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + if err := AssertMeshConsumers(js); err != nil { + t.Fatal(err) + } + for _, c := range MeshConsumers() { + if c.Stream == EventsStream && c.Name == MergesConsumerName { + if c.Resettable != "" || c.MaxDeliver != 0 || len(c.Filters) != 1 || c.Filters[0] != ControllerFollows[3] { + t.Errorf("the merges consumer is %+v; want one filter, no max-deliver, not resettable", c) + } + } + if c.Stream == EventsStream && c.Name == ControllerName { + for _, f := range c.Filters { + if f == ControllerFollows[3] { + t.Errorf("the events consumer still follows the merges, which the merges consumer takes") + } + } + } + } + if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil { + t.Fatal(err) + } + if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil { + t.Fatal(err) + } + merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) + if err != nil || merges.NumPending != 1 { + t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err) + } + if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil { + t.Fatal(err) + } + merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName) + if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 { + t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err) + } +} + +func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + if err := AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + old := &nats.ConsumerConfig{Durable: ControllerName, AckPolicy: nats.AckExplicitPolicy, AckWait: 30 * time.Second, + MaxDeliver: 5, MaxAckPending: 1, DeliverSubject: "_DELIVER.controller.EVENTS", DeliverPolicy: nats.DeliverNewPolicy, + FilterSubjects: append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName))} + if _, err := js.Context().AddConsumer(EventsStream, old); err != nil { + t.Fatal(err) + } + if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil { + t.Fatal(err) + } + if err := AssertMeshConsumers(js); err != nil { + t.Fatal(err) + } + merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) + if err != nil || merges.NumPending != 1 { + t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err) + } +} diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 4b337dc1..7b9d9f9b 100644 --- a/internal/broker/controller_buckets_test.go +++ b/internal/broker/controller_buckets_test.go @@ -1,16 +1,9 @@ package broker import ( - "context" - "github.com/nats-io/nats.go" "slices" "strings" "testing" - "time" - - "github.com/nats-io/nats.go/jetstream" - - "github.com/novox/mesh-controller/internal/testbus" ) // **The controller may write every bucket it writes** (novox/hq to-be 45 §1, issue 269). Writing a @@ -100,34 +93,6 @@ func TestTheWatchedSignalsMayBeSaidAndHeard(t *testing.T) { // The controller's record of what it asked the operator is bounded (correctness review of 2026-10-08): one // value a key, a month's age, and a size it cannot outgrow. -func TestWhatTheControllerAskedIsBounded(t *testing.T) { - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatal(err) - } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - kv, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - bucket, err := kv.KeyValue(ctx, AskedBucket) - if err != nil { - t.Fatal(err) - } - status, err := bucket.Status(ctx) - if err != nil { - t.Fatal(err) - } - info := status.(*jetstream.KeyValueBucketStatus).StreamInfo() - if status.History() != 1 || status.TTL() != AskedKeptFor || info.Config.MaxBytes <= 0 || info.Config.MaxBytes > 64<<20 { - t.Errorf("history %d, age %s, bytes %d", status.History(), status.TTL(), info.Config.MaxBytes) - } -} // **A listed mirror is no undeclared bucket** (ADR 0356 §3, ADR 0201 §16): the controller's list of its mirrors // stands for their declaration, so the report of state nothing declares never names one. @@ -273,232 +238,19 @@ func TestTheControllerMayAskMeshConfigInItsOwnName(t *testing.T) { // **The merge window is a bucket whose keys expire** (novox/hq ADR 0407 decision 1, issue 973): a key made with a // time to live goes when it is up, and a watcher on the bucket hears it go. The controller may write it, and it is // no undeclared state. -func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) { - controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) - if err != nil { - t.Fatal(err) - } - if !slices.Contains(controller.Publish, "$KV."+MergeWindowBucket+".>") { - t.Error("the controller may not write the merge window's bucket") - } - if !IsControllerBucket(MergeWindowBucket) || !slices.Contains(ControllerBuckets(), MergeWindowBucket) { - t.Error("the merge window's bucket is not asserted as the controller's own") - } - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatal(err) - } - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - kv, err := api.KeyValue(ctx, MergeWindowBucket) - if err != nil { - t.Fatal(err) - } - w, err := kv.WatchAll(ctx) - if err != nil { - t.Fatal(err) - } - defer w.Stop() - if _, err := kv.Create(ctx, "plan-1.quiet", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { - t.Fatal(err) - } - for { - select { - case e := <-w.Updates(): - if e != nil && e.Key() == "plan-1.quiet" && e.Operation() != jetstream.KeyValuePut { - return - } - case <-ctx.Done(): - t.Fatal("no watcher heard the key expire") - } - } -} // **A rollout's wait for a machine is a bucket whose keys expire** (novox/hq ADR 0406 §5, issue 974): a key made with a // time to live goes when it is up, and a watcher hears it go as a purge; a key a report deleted is a delete. The // controller may write it, and it is no undeclared state. -func TestTheRolloutsSilentKeysExpireAsAPurgeAndAReportsDeleteIsNot(t *testing.T) { - controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) - if err != nil { - t.Fatal(err) - } - if !slices.Contains(controller.Publish, "$KV."+RolloutSilentBucket+".>") { - t.Error("the controller may not write the rollout's silent keys' bucket") - } - if !IsControllerBucket(RolloutSilentBucket) || !slices.Contains(ControllerBuckets(), RolloutSilentBucket) { - t.Error("the rollout's silent keys' bucket is not asserted as the controller's own") - } - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatal(err) - } - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - kv, err := api.KeyValue(ctx, RolloutSilentBucket) - if err != nil { - t.Fatal(err) - } - w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) - if err != nil { - t.Fatal(err) - } - defer w.Stop() - if _, err := kv.Create(ctx, "plan-1.0.ace", []byte("x"), jetstream.KeyTTL(time.Hour)); err != nil { - t.Fatal(err) - } - if err := kv.Delete(ctx, "plan-1.0.ace"); err != nil { - t.Fatal(err) - } - if _, err := kv.Create(ctx, "plan-1.0.g14", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { - t.Fatal(err) - } - for { - select { - case e := <-w.Updates(): - if e == nil || e.Operation() == jetstream.KeyValuePut { - continue - } - if e.Key() == "plan-1.0.ace" && e.Operation() != jetstream.KeyValueDelete { - t.Fatalf("a report's delete was heard as %v", e.Operation()) - } - if e.Key() == "plan-1.0.g14" { - if e.Operation() != jetstream.KeyValuePurge { - t.Fatalf("an expiry was heard as %v, not a purge", e.Operation()) - } - return - } - case <-ctx.Done(): - t.Fatal("no watcher heard the key expire") - } - } -} // **The silent keys' bucket ages**, or its delete markers would fill it (issue 974, review 1): every report deletes a key, // the delete leaves a marker, and with one value per key and no age the markers stay for ever. A bucket made before the // age was set is brought to it by the next start, which the server accepts as an update of an existing bucket. -func TestTheSilentKeysBucketAgesAndAnOlderOneIsBroughtToIt(t *testing.T) { - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - // As the first build made it: no age. - if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: RolloutSilentBucket, History: 1, - MaxValueSize: 4 << 10, MaxBytes: 8 << 20, Storage: jetstream.FileStorage, LimitMarkerTTL: RolloutSilentMarkerTTL}); err != nil { - t.Fatal(err) - } - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatalf("a bucket made without the age is not brought to it: %v", err) - } - kv, err := api.KeyValue(ctx, RolloutSilentBucket) - if err != nil { - t.Fatal(err) - } - status, err := kv.Status(ctx) - if err != nil { - t.Fatal(err) - } - if status.TTL() != RolloutSilentKeptFor { - t.Errorf("the bucket ages at %v; want %v", status.TTL(), RolloutSilentKeptFor) - } -} // **The merges have a consumer of their own, which a reset of the events consumer leaves alone** (issue 973, novox/hq // ADR 0407 decision 3): healer H4 re-makes the events consumer from now, dropping what it holds pending; a merge // waiting in the merges consumer is not that consumer's, and nothing would catch it up. -func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testing.T) { - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - if err := AssertMeshStreams(js); err != nil { - t.Fatal(err) - } - if err := AssertMeshConsumers(js); err != nil { - t.Fatal(err) - } - for _, c := range MeshConsumers() { - if c.Stream == EventsStream && c.Name == MergesConsumerName { - if c.Resettable != "" || c.MaxDeliver != 0 || len(c.Filters) != 1 || c.Filters[0] != ControllerFollows[3] { - t.Errorf("the merges consumer is %+v; want one filter, no max-deliver, not resettable", c) - } - } - if c.Stream == EventsStream && c.Name == ControllerName { - for _, f := range c.Filters { - if f == ControllerFollows[3] { - t.Errorf("the events consumer still follows the merges, which the merges consumer takes") - } - } - } - } - if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil { - t.Fatal(err) - } - if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil { - t.Fatal(err) - } - merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) - if err != nil || merges.NumPending != 1 { - t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err) - } - if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil { - t.Fatal(err) - } - merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName) - if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 { - t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err) - } -} // **Delivering this change loses no merge** (review of issue 973): a merge published under the old filter of the // `controller` consumer and never acknowledged by it is delivered by the merges consumer once the assertion has // made it, because it starts where `controller` has acknowledged up to, not from now. -func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T) { - js, err := Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - defer js.Close() - if err := AssertMeshStreams(js); err != nil { - t.Fatal(err) - } - old := &nats.ConsumerConfig{Durable: ControllerName, AckPolicy: nats.AckExplicitPolicy, AckWait: 30 * time.Second, - MaxDeliver: 5, MaxAckPending: 1, DeliverSubject: "_DELIVER.controller.EVENTS", DeliverPolicy: nats.DeliverNewPolicy, - FilterSubjects: append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName))} - if _, err := js.Context().AddConsumer(EventsStream, old); err != nil { - t.Fatal(err) - } - if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil { - t.Fatal(err) - } - if err := AssertMeshConsumers(js); err != nil { - t.Fatal(err) - } - merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) - if err != nil || merges.NumPending != 1 { - t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err) - } -} diff --git a/internal/broker/jetstream_test.go b/internal/broker/jetstream_test.go index 5124696d..4cf84a33 100644 --- a/internal/broker/jetstream_test.go +++ b/internal/broker/jetstream_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/log_integration_test.go b/internal/broker/log_integration_test.go new file mode 100644 index 00000000..69c62f0b --- /dev/null +++ b/internal/broker/log_integration_test.go @@ -0,0 +1,69 @@ +//go:build integration + +package broker + +import ( + "context" + "slices" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +func TestALogIsAssertedInPlaceAndNeverRemoved(t *testing.T) { + bus := aLiveBus(t) + l := Log{Module: "logtest", Name: "changes", MaxMiB: 1} + if _, err := RaiseLogs(bus, []Log{l}); err != nil { + t.Fatalf("a real server refused a module's log: %v", err) + } + js, err := jetstream.New(bus.Conn()) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + ack, err := js.Publish(ctx, l.Subject()+".7", []byte(`{"op":"opened"}`)) + if err != nil { + t.Fatalf("an append to the log was refused: %v", err) + } + stream, err := js.Stream(ctx, l.Stream()) + if err != nil { + t.Fatal(err) + } + have := stream.CachedInfo().Config + want := l.Config() + if have.Storage != want.Storage || have.Retention != want.Retention || have.MaxAge != 0 || + have.MaxMsgsPerSubject != -1 || have.MaxBytes != want.MaxBytes || have.MaxMsgSize != want.MaxMsgSize || + have.Discard != jetstream.DiscardNew || !have.AllowDirect || !have.DenyDelete || !have.DenyPurge || + have.Replicas != 1 || !slices.Equal(have.Subjects, want.Subjects) { + t.Fatalf("the server holds the log as %+v", have) + } + if err := stream.DeleteMsg(ctx, ack.Sequence); err == nil { + t.Fatal("an entry was deleted from a log") + } + l.MaxMiB = 2 + if _, err := RaiseLogs(bus, []Log{l}); err != nil { + t.Fatalf("asserting the log again failed, so a restart would: %v", err) + } + info, err := stream.Info(ctx) + if err != nil { + t.Fatal(err) + } + if info.Config.MaxBytes != 2*1024*1024 { + t.Fatalf("the changed cap was not brought to match: %d", info.Config.MaxBytes) + } + if info.State.Msgs < 1 { + t.Fatal("asserting the log again lost what it held") + } + undeclared, err := RaiseLogs(bus, nil) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(undeclared, l.Stream()) { + t.Fatalf("a log nothing declares was not reported: %v", undeclared) + } + if _, err := js.Stream(ctx, l.Stream()); err != nil { + t.Fatalf("a log nothing declares is gone: %v", err) + } +} diff --git a/internal/broker/log_test.go b/internal/broker/log_test.go index 95043e1c..b1988cdd 100644 --- a/internal/broker/log_test.go +++ b/internal/broker/log_test.go @@ -1,7 +1,6 @@ package broker import ( - "context" "encoding/json" "go/ast" "go/parser" @@ -11,7 +10,6 @@ import ( "slices" "strings" "testing" - "time" "github.com/nats-io/nats.go/jetstream" ) @@ -240,59 +238,3 @@ func stringsIn(f *ast.File) []string { // Against a real server: a log is created as its configuration says, asserting it again keeps what // it holds, a changed cap is brought to match in place, an entry cannot be deleted from it, and one // nothing declares any more is reported and stays. -func TestALogIsAssertedInPlaceAndNeverRemoved(t *testing.T) { - bus := aLiveBus(t) - l := Log{Module: "logtest", Name: "changes", MaxMiB: 1} - if _, err := RaiseLogs(bus, []Log{l}); err != nil { - t.Fatalf("a real server refused a module's log: %v", err) - } - js, err := jetstream.New(bus.Conn()) - if err != nil { - t.Fatal(err) - } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - ack, err := js.Publish(ctx, l.Subject()+".7", []byte(`{"op":"opened"}`)) - if err != nil { - t.Fatalf("an append to the log was refused: %v", err) - } - stream, err := js.Stream(ctx, l.Stream()) - if err != nil { - t.Fatal(err) - } - have := stream.CachedInfo().Config - want := l.Config() - if have.Storage != want.Storage || have.Retention != want.Retention || have.MaxAge != 0 || - have.MaxMsgsPerSubject != -1 || have.MaxBytes != want.MaxBytes || have.MaxMsgSize != want.MaxMsgSize || - have.Discard != jetstream.DiscardNew || !have.AllowDirect || !have.DenyDelete || !have.DenyPurge || - have.Replicas != 1 || !slices.Equal(have.Subjects, want.Subjects) { - t.Fatalf("the server holds the log as %+v", have) - } - if err := stream.DeleteMsg(ctx, ack.Sequence); err == nil { - t.Fatal("an entry was deleted from a log") - } - l.MaxMiB = 2 - if _, err := RaiseLogs(bus, []Log{l}); err != nil { - t.Fatalf("asserting the log again failed, so a restart would: %v", err) - } - info, err := stream.Info(ctx) - if err != nil { - t.Fatal(err) - } - if info.Config.MaxBytes != 2*1024*1024 { - t.Fatalf("the changed cap was not brought to match: %d", info.Config.MaxBytes) - } - if info.State.Msgs < 1 { - t.Fatal("asserting the log again lost what it held") - } - undeclared, err := RaiseLogs(bus, nil) - if err != nil { - t.Fatal(err) - } - if !slices.Contains(undeclared, l.Stream()) { - t.Fatalf("a log nothing declares was not reported: %v", undeclared) - } - if _, err := js.Stream(ctx, l.Stream()); err != nil { - t.Fatalf("a log nothing declares is gone: %v", err) - } -} diff --git a/internal/broker/modstreams_live_test.go b/internal/broker/modstreams_live_test.go index 68f42df6..4532d913 100644 --- a/internal/broker/modstreams_live_test.go +++ b/internal/broker/modstreams_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/msgttl_live_test.go b/internal/broker/msgttl_live_test.go index 02589420..777d522d 100644 --- a/internal/broker/msgttl_live_test.go +++ b/internal/broker/msgttl_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/prov_applied_live_test.go b/internal/broker/prov_applied_live_test.go index f7c4bd6f..2b8069f7 100644 --- a/internal/broker/prov_applied_live_test.go +++ b/internal/broker/prov_applied_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/provisioner_live_test.go b/internal/broker/provisioner_live_test.go index c5fe6892..9c48880f 100644 --- a/internal/broker/provisioner_live_test.go +++ b/internal/broker/provisioner_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/provisioner_server_live_test.go b/internal/broker/provisioner_server_live_test.go index 95847566..554b1bb9 100644 --- a/internal/broker/provisioner_server_live_test.go +++ b/internal/broker/provisioner_server_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/raise_live_test.go b/internal/broker/raise_live_test.go index 1d550c70..fee13277 100644 --- a/internal/broker/raise_live_test.go +++ b/internal/broker/raise_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/snapshot_live_test.go b/internal/broker/snapshot_live_test.go index af9d33c5..528ee2d1 100644 --- a/internal/broker/snapshot_live_test.go +++ b/internal/broker/snapshot_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/broker/state_integration_test.go b/internal/broker/state_integration_test.go new file mode 100644 index 00000000..fa92d0cb --- /dev/null +++ b/internal/broker/state_integration_test.go @@ -0,0 +1,55 @@ +//go:build integration + +package broker + +import ( + "testing" + + "github.com/nats-io/nats.go" +) + +func TestABucketIsAssertedInPlace(t *testing.T) { + js := aLiveBus(t) + b := Bucket{Module: "statetest", Name: "servers"} + if _, err := RaiseBuckets(js, []Bucket{b}); err != nil { + t.Fatalf("a real server refused a module's bucket: %v", err) + } + kv, err := js.Context().KeyValue(b.Bucket()) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put("all.one", []byte(`{"kept":true}`)); err != nil { + t.Fatal(err) + } + b.History = 3 + if _, err := RaiseBuckets(js, []Bucket{b}); err != nil { + t.Fatalf("asserting the bucket again failed, so a restart would: %v", err) + } + got, err := kv.Get("all.one") + if err != nil || string(got.Value()) != `{"kept":true}` { + t.Fatalf("asserting again lost what the bucket held: %v %v", got, err) + } + status, err := kv.Status() + if err != nil { + t.Fatal(err) + } + if status.History() != 3 { + t.Fatalf("history is %d, the owner declared 3", status.History()) + } + if s, ok := status.(*nats.KeyValueBucketStatus); ok { + if c := s.StreamInfo().Config; c.MaxMsgSize != StateMaxValueBytes || c.MaxBytes != StateMaxBytes { + t.Fatalf("the mesh's caps are not on the bucket: value %d, bucket %d", c.MaxMsgSize, c.MaxBytes) + } + } +} + +func TestABucketIsAssertedOverAHeldConnection(t *testing.T) { + js := aLiveBus(t) + held := OnConn(js.Conn()) + if _, err := RaiseBuckets(held, []Bucket{{Module: "statetest", Name: "held"}}); err != nil { + t.Fatalf("asserting over a held connection failed: %v", err) + } + if _, err := js.Context().KeyValue("statetest_held"); err != nil { + t.Fatalf("the bucket is not there: %v", err) + } +} diff --git a/internal/broker/state_test.go b/internal/broker/state_test.go index 402cd317..8d4d7aeb 100644 --- a/internal/broker/state_test.go +++ b/internal/broker/state_test.go @@ -4,8 +4,6 @@ import ( "slices" "strings" "testing" - - "github.com/nats-io/nats.go" ) // The grants measured against a running server (novox/hq research 024): an owner reads and writes @@ -131,54 +129,10 @@ func TestRaisingStateReportsWhatNothingDeclares(t *testing.T) { // Against a real server: a bucket is created with the owner's options and the mesh's caps, // asserting it again changes nothing and keeps what it holds, and a changed option is brought to // match in place. -func TestABucketIsAssertedInPlace(t *testing.T) { - js := aLiveBus(t) - b := Bucket{Module: "statetest", Name: "servers"} - if _, err := RaiseBuckets(js, []Bucket{b}); err != nil { - t.Fatalf("a real server refused a module's bucket: %v", err) - } - kv, err := js.Context().KeyValue(b.Bucket()) - if err != nil { - t.Fatal(err) - } - if _, err := kv.Put("all.one", []byte(`{"kept":true}`)); err != nil { - t.Fatal(err) - } - b.History = 3 - if _, err := RaiseBuckets(js, []Bucket{b}); err != nil { - t.Fatalf("asserting the bucket again failed, so a restart would: %v", err) - } - got, err := kv.Get("all.one") - if err != nil || string(got.Value()) != `{"kept":true}` { - t.Fatalf("asserting again lost what the bucket held: %v %v", got, err) - } - status, err := kv.Status() - if err != nil { - t.Fatal(err) - } - if status.History() != 3 { - t.Fatalf("history is %d, the owner declared 3", status.History()) - } - if s, ok := status.(*nats.KeyValueBucketStatus); ok { - if c := s.StreamInfo().Config; c.MaxMsgSize != StateMaxValueBytes || c.MaxBytes != StateMaxBytes { - t.Fatalf("the mesh's caps are not on the bucket: value %d, bucket %d", c.MaxMsgSize, c.MaxBytes) - } - } -} // Against a real server: the handle over a connection the control plane already holds asserts a // bucket as Dial's does — what a push uses, so a module registered since the last start has its // bucket before its membership names it. -func TestABucketIsAssertedOverAHeldConnection(t *testing.T) { - js := aLiveBus(t) - held := OnConn(js.Conn()) - if _, err := RaiseBuckets(held, []Bucket{{Module: "statetest", Name: "held"}}); err != nil { - t.Fatalf("asserting over a held connection failed: %v", err) - } - if _, err := js.Context().KeyValue("statetest_held"); err != nil { - t.Fatalf("the bucket is not there: %v", err) - } -} // **State keyed by machine reaches its machine's key alone** (novox/hq ADR 0260): the owner's instance on // a machine writes and reads that key, and a bar granted a key reads that key — a direct get of its diff --git a/internal/broker/vault_calls_live_integration_test.go b/internal/broker/vault_calls_live_integration_test.go new file mode 100644 index 00000000..65cac495 --- /dev/null +++ b/internal/broker/vault_calls_live_integration_test.go @@ -0,0 +1,130 @@ +//go:build integration + +package broker + +import ( + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "golang.org/x/crypto/bcrypt" +) + +func TestOnARealBusTheVaultsAccountIsAnsweredByTheDeskAndTheProviderAndNothingElse(t *testing.T) { + users, err := Users(vaultRecords()) + if err != nil { + t.Fatal(err) + } + hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost) + if err != nil { + t.Fatal(err) + } + for i := range users { + users[i].PasswordHash = string(hash) + } + accounts, err := ComposeAccounts(users) + if err != nil { + t.Fatal(err) + } + conf := filepath.Join(t.TempDir(), "bus.conf") + if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+ + `"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatalf("the composed accounts do not parse: %v", err) + } + opts.NoLog, opts.NoSigs = true, true + s, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + defer s.Shutdown() + + // anchor's runtime serves the desk's prompt and plex's issuing tool, on every subject it is granted, and says + // which subject it heard. + holder, err := nats.Connect(s.ClientURL(), nats.UserInfo("anchor.node-tools", "pw")) + if err != nil { + t.Fatal(err) + } + defer holder.Close() + heard := make(chan string, 8) + answer := func(m *nats.Msg) { + heard <- m.Subject + _ = m.Respond([]byte(`{"result":{"sealed":"c2VhbGVkLXRvLXRoZS1jYWxs"}}`)) + } + for _, subject := range []string{"mesh.seat.node-launcher.call.secret.anchor.*", + "mesh.seat.node-launcher.tool.secret.anchor", "mesh.mod.plex.call.plex_issued_credential.anchor.*", + "mesh.mod.other.call.tool.anchor.*", "mesh.seat.node-launcher.call.menu.anchor.*"} { + sub, err := holder.Subscribe(subject, answer) + if err != nil { + t.Fatal(err) + } + _ = holder.Flush() + if !sub.IsValid() { + t.Fatalf("the desk's runtime may not hear %s", subject) + } + } + + refusals := make(chan error, 16) + vault, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.mesh-vault", "pw"), + nats.CustomInboxPrefix("_INBOX.novox.mesh-vault"), + nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err })) + if err != nil { + t.Fatalf("the vault's own account cannot connect: %v", err) + } + defer vault.Close() + token := CallerToken("novox.mesh-vault") + for _, subject := range []string{"mesh.seat.node-launcher.call.secret.anchor." + token, + "mesh.mod.plex.call.plex_issued_credential.anchor." + token} { + if _, err := vault.Request(subject, []byte(`{"seal_to":"a2V5LW9mLXRoZS1jYWxs"}`), 3*time.Second); err != nil { + t.Fatalf("the vault's call %s was not answered: %v", subject, err) + } + if got := <-heard; got != subject { + t.Fatalf("the holder heard %s, want %s", got, subject) + } + } + for _, subject := range []string{"mesh.seat.node-launcher.tool.secret.anchor", + "mesh.seat.node-launcher.call.secret.anchor.controller", + "mesh.seat.node-launcher.call.menu.anchor." + token, + "mesh.mod.other.call.tool.anchor." + token} { + if err := vault.Publish(subject, []byte(`{}`)); err != nil { + t.Fatal(err) + } + _ = vault.Flush() + select { + case err := <-refusals: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("%s: the server said %v, want a permissions violation", subject, err) + } + case <-time.After(3 * time.Second): + t.Errorf("the server did not refuse the vault %s", subject) + } + } + select { + case got := <-heard: + t.Errorf("a call the vault may not make reached the holder: %s", got) + case <-time.After(200 * time.Millisecond): + } + + // The controller is still answered on today's path, for the modules filled the old way (until R1). + controller, err := nats.Connect(s.ClientURL(), nats.UserInfo("controller", "pw"), + nats.CustomInboxPrefix("_INBOX.controller")) + if err != nil { + t.Fatal(err) + } + defer controller.Close() + if _, err := controller.Request("mesh.seat.node-launcher.tool.secret.anchor", []byte(`{}`), 3*time.Second); err != nil { + t.Errorf("the controller's call to the desk on today's path: %v", err) + } + <-heard +} diff --git a/internal/broker/vault_calls_live_test.go b/internal/broker/vault_calls_live_test.go index 7a769870..17a72747 100644 --- a/internal/broker/vault_calls_live_test.go +++ b/internal/broker/vault_calls_live_test.go @@ -1,16 +1,8 @@ package broker import ( - "errors" - "os" - "path/filepath" "strings" "testing" - "time" - - "github.com/nats-io/nats-server/v2/server" - "github.com/nats-io/nats.go" - "golang.org/x/crypto/bcrypt" ) // vaultRecords is a mesh of two machines: anchor, a desk whose runtime carries the desk's prompt (rofi holding @@ -111,117 +103,3 @@ func TestTheVaultAsksTheDeskAndTheIssuingToolsInItsOwnNameAlone(t *testing.T) { // **On a real server, as composed** (issue 723): the vault's account asks the desk's prompt and an issuing tool // and is answered, naming itself; the server refuses it any other call; and the controller is still answered on // today's path. That only sealed values cross is the vault's test (mesh-catalog, mesh-vault intake_test.go). -func TestOnARealBusTheVaultsAccountIsAnsweredByTheDeskAndTheProviderAndNothingElse(t *testing.T) { - users, err := Users(vaultRecords()) - if err != nil { - t.Fatal(err) - } - hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost) - if err != nil { - t.Fatal(err) - } - for i := range users { - users[i].PasswordHash = string(hash) - } - accounts, err := ComposeAccounts(users) - if err != nil { - t.Fatal(err) - } - conf := filepath.Join(t.TempDir(), "bus.conf") - if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+ - `"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil { - t.Fatal(err) - } - opts, err := server.ProcessConfigFile(conf) - if err != nil { - t.Fatalf("the composed accounts do not parse: %v", err) - } - opts.NoLog, opts.NoSigs = true, true - s, err := server.NewServer(opts) - if err != nil { - t.Fatal(err) - } - go s.Start() - if !s.ReadyForConnections(10 * time.Second) { - t.Fatal("the bus did not come up") - } - defer s.Shutdown() - - // anchor's runtime serves the desk's prompt and plex's issuing tool, on every subject it is granted, and says - // which subject it heard. - holder, err := nats.Connect(s.ClientURL(), nats.UserInfo("anchor.node-tools", "pw")) - if err != nil { - t.Fatal(err) - } - defer holder.Close() - heard := make(chan string, 8) - answer := func(m *nats.Msg) { - heard <- m.Subject - _ = m.Respond([]byte(`{"result":{"sealed":"c2VhbGVkLXRvLXRoZS1jYWxs"}}`)) - } - for _, subject := range []string{"mesh.seat.node-launcher.call.secret.anchor.*", - "mesh.seat.node-launcher.tool.secret.anchor", "mesh.mod.plex.call.plex_issued_credential.anchor.*", - "mesh.mod.other.call.tool.anchor.*", "mesh.seat.node-launcher.call.menu.anchor.*"} { - sub, err := holder.Subscribe(subject, answer) - if err != nil { - t.Fatal(err) - } - _ = holder.Flush() - if !sub.IsValid() { - t.Fatalf("the desk's runtime may not hear %s", subject) - } - } - - refusals := make(chan error, 16) - vault, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.mesh-vault", "pw"), - nats.CustomInboxPrefix("_INBOX.novox.mesh-vault"), - nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err })) - if err != nil { - t.Fatalf("the vault's own account cannot connect: %v", err) - } - defer vault.Close() - token := CallerToken("novox.mesh-vault") - for _, subject := range []string{"mesh.seat.node-launcher.call.secret.anchor." + token, - "mesh.mod.plex.call.plex_issued_credential.anchor." + token} { - if _, err := vault.Request(subject, []byte(`{"seal_to":"a2V5LW9mLXRoZS1jYWxs"}`), 3*time.Second); err != nil { - t.Fatalf("the vault's call %s was not answered: %v", subject, err) - } - if got := <-heard; got != subject { - t.Fatalf("the holder heard %s, want %s", got, subject) - } - } - for _, subject := range []string{"mesh.seat.node-launcher.tool.secret.anchor", - "mesh.seat.node-launcher.call.secret.anchor.controller", - "mesh.seat.node-launcher.call.menu.anchor." + token, - "mesh.mod.other.call.tool.anchor." + token} { - if err := vault.Publish(subject, []byte(`{}`)); err != nil { - t.Fatal(err) - } - _ = vault.Flush() - select { - case err := <-refusals: - if !errors.Is(err, nats.ErrPermissionViolation) { - t.Errorf("%s: the server said %v, want a permissions violation", subject, err) - } - case <-time.After(3 * time.Second): - t.Errorf("the server did not refuse the vault %s", subject) - } - } - select { - case got := <-heard: - t.Errorf("a call the vault may not make reached the holder: %s", got) - case <-time.After(200 * time.Millisecond): - } - - // The controller is still answered on today's path, for the modules filled the old way (until R1). - controller, err := nats.Connect(s.ClientURL(), nats.UserInfo("controller", "pw"), - nats.CustomInboxPrefix("_INBOX.controller")) - if err != nil { - t.Fatal(err) - } - defer controller.Close() - if _, err := controller.Request("mesh.seat.node-launcher.tool.secret.anchor", []byte(`{}`), 3*time.Second); err != nil { - t.Errorf("the controller's call to the desk on today's path: %v", err) - } - <-heard -} diff --git a/internal/broker/worker_type_change_test.go b/internal/broker/worker_type_change_test.go index 7ddd588d..f3bd7712 100644 --- a/internal/broker/worker_type_change_test.go +++ b/internal/broker/worker_type_change_test.go @@ -1,3 +1,5 @@ +//go:build integration + package broker import ( diff --git a/internal/conditions/bus_test.go b/internal/conditions/bus_test.go index 7f945c83..4526dd5b 100644 --- a/internal/conditions/bus_test.go +++ b/internal/conditions/bus_test.go @@ -1,3 +1,5 @@ +//go:build integration + package conditions import ( diff --git a/internal/conditions/store.go b/internal/conditions/store.go index 0311d60d..c9ba4fd9 100644 --- a/internal/conditions/store.go +++ b/internal/conditions/store.go @@ -53,14 +53,15 @@ const Seat = "mesh-controller" // Keeper raises, observes, silences and clears conditions, and says each transition. type Keeper struct { - store Backend - looks LookStore - history History - teller Teller - now func() time.Time - say func(format string, args ...any) - changed func() - epoch func() (uint64, error) + store Backend + looks LookStore + history History + teller Teller + now func() time.Time + say func(format string, args ...any) + changed func() + epoch func() (uint64, error) + tellBackoff time.Duration mu sync.Mutex // cleared is when each recently cleared condition cleared and how often it had been raised, so @@ -108,6 +109,9 @@ type Options struct { // Epoch is the controller lease this keeper writes under (to-be 45 §6): every write carries its // epoch, and none is made while it is not held. Nil writes with no epoch (a test, the memory store). Epoch func() (uint64, error) + // TellBackoff is the first wait before a transition the bus would not take is offered again; it doubles each + // time. Zero is 200 ms; a test sets it short so that it need not wait. + TellBackoff time.Duration } // TellFor is how long one transition is offered to the bus before it is said lost. @@ -117,7 +121,7 @@ var TellFor = 10 * time.Minute // that cleared just before this controller started and is raised again now is a reopening. func NewKeeper(ctx context.Context, o Options) *Keeper { k := &Keeper{store: o.Store, looks: o.Looks, history: o.History, teller: o.Teller, now: o.Now, say: o.Say, changed: o.Changed, - epoch: o.Epoch, + epoch: o.Epoch, tellBackoff: o.TellBackoff, cleared: map[string]clearing{}, reworded: map[string]bool{}, out: make(chan Event, 1024), drained: make(chan struct{})} if k.now == nil { k.now = time.Now @@ -713,7 +717,10 @@ func (k *Keeper) telling() { continue } deadline := time.Now().Add(TellFor) - wait := 200 * time.Millisecond + wait := k.tellBackoff + if wait <= 0 { + wait = 200 * time.Millisecond + } kept, said := k.history == nil, k.teller == nil for { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) diff --git a/internal/conditions/store_test.go b/internal/conditions/store_test.go index f1d49a28..a1f58869 100644 --- a/internal/conditions/store_test.go +++ b/internal/conditions/store_test.go @@ -370,12 +370,12 @@ func TestTheEventShapeIsTheContract(t *testing.T) { func TestATransitionIsOfferedAgainWhileTheBusIsAway(t *testing.T) { store, told, c := NewInMemory(), &Told{}, newClock() told.SetFail(errors.New("no responders")) - k := NewKeeper(t.Context(), Options{Store: store, History: store, Teller: told, Now: c.now}) + k := NewKeeper(t.Context(), Options{Store: store, History: store, Teller: told, Now: c.now, TellBackoff: 5 * time.Millisecond}) defer k.Close(context.Background()) if _, err := k.Observe(t.Context(), silent("ace")); err != nil { t.Fatal(err) } - time.Sleep(300 * time.Millisecond) + time.Sleep(30 * time.Millisecond) told.SetFail(nil) said := settled(t, told, 1) if said[0].Key != "machine.ace.silent" || k.Unsaid() != 0 { diff --git a/internal/identity/authority_test.go b/internal/identity/authority_test.go index d14cae06..c674316f 100644 --- a/internal/identity/authority_test.go +++ b/internal/identity/authority_test.go @@ -1,3 +1,5 @@ +//go:build integration + package identity import ( diff --git a/internal/identity/identity_test.go b/internal/identity/identity_test.go index 7aa15825..bacf1d62 100644 --- a/internal/identity/identity_test.go +++ b/internal/identity/identity_test.go @@ -1,3 +1,5 @@ +//go:build integration + package identity import ( diff --git a/internal/lease/lease_test.go b/internal/lease/lease_test.go index c242b572..e2fcf4d9 100644 --- a/internal/lease/lease_test.go +++ b/internal/lease/lease_test.go @@ -1,3 +1,5 @@ +//go:build integration + package lease import ( diff --git a/internal/licences/licences_test.go b/internal/licences/licences_test.go index c941e92c..e322cc5d 100644 --- a/internal/licences/licences_test.go +++ b/internal/licences/licences_test.go @@ -1,3 +1,5 @@ +//go:build integration + package licences import ( diff --git a/internal/licences/refresh_test.go b/internal/licences/refresh_test.go index b7600bbf..9fa0e68f 100644 --- a/internal/licences/refresh_test.go +++ b/internal/licences/refresh_test.go @@ -1,3 +1,5 @@ +//go:build integration + package licences import ( diff --git a/internal/licences/submitrefresh_test.go b/internal/licences/submitrefresh_test.go index 0e325679..302db468 100644 --- a/internal/licences/submitrefresh_test.go +++ b/internal/licences/submitrefresh_test.go @@ -1,3 +1,5 @@ +//go:build integration + package licences import ( diff --git a/internal/lint/mergecheck_unit_level_test.go b/internal/lint/mergecheck_unit_level_test.go new file mode 100644 index 00000000..b019ce54 --- /dev/null +++ b/internal/lint/mergecheck_unit_level_test.go @@ -0,0 +1,76 @@ +//go:build integration + +package lint + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +// judgeUnitRun runs merge-check.sh's judging of the unit level on a recorded `go test -v` output (issue 895), as it +// would judge the output of the untagged run: it starts the shell, so it is an integration test. +func judgeUnitRun(t *testing.T, recorded string, bound string) (string, error) { + t.Helper() + out := filepath.Join(t.TempDir(), "unit.out") + if err := os.WriteFile(out, []byte(recorded), 0o600); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, "sh", "merge-check.sh", "judge-unit-run", out, bound) + cmd.Dir = filepath.Join("..", "..") + said, err := cmd.CombinedOutput() + return string(said), err +} + +const recordedUnitRun = `=== RUN TestQuick +--- PASS: TestQuick (0.01s) +=== RUN TestSlowOne +--- PASS: TestSlowOne (2.50s) +=== RUN TestSlowTwo + --- PASS: TestSlowTwo/sub (1.00s) +--- PASS: TestSlowTwo (1.20s) +--- SKIP: TestNeedsAStore (0.00s) +PASS +ok example.test/internal/one 2.80s +=== RUN TestElsewhere +--- PASS: TestElsewhere (0.30s) +PASS +ok example.test/internal/two 0.35s +? example.test/internal/three [no test files] +` + +// **The unit level over its bound fails the check and names the slowest tests** (issue 895). +func TestAUnitRunOverItsBoundFailsAndNamesTheSlowestTests(t *testing.T) { + said, err := judgeUnitRun(t, recordedUnitRun, "1") + if err == nil { + t.Fatalf("a unit run of 2.80 s passed a bound of 1 s: %s", said) + } + for _, want := range []string{"5 tests (1 skipped)", "2.80 s (bound 1 s)", "TestSlowOne", "TestSlowTwo", "example.test/internal/one"} { + if !strings.Contains(said, want) { + t.Errorf("the verdict does not say %q: %s", want, said) + } + } + if strings.Index(said, "TestSlowOne") > strings.Index(said, "TestSlowTwo") { + t.Errorf("the slowest test is not named first: %s", said) + } + if strings.Contains(said, "TestSlowTwo/sub") { + t.Errorf("a subtest is named as if it were a test of its own: %s", said) + } +} + +// A unit run within its bound passes and says its count and time. +func TestAUnitRunWithinItsBoundPasses(t *testing.T) { + said, err := judgeUnitRun(t, recordedUnitRun, "3") + if err != nil { + t.Fatalf("a unit run of 2.80 s failed a bound of 3 s: %v: %s", err, said) + } + if !strings.Contains(said, "unit level: 5 tests (1 skipped), 2.80 s (bound 3 s)") { + t.Errorf("the level's count and time are not said: %s", said) + } +} diff --git a/internal/recordmirror/mirror_integration_test.go b/internal/recordmirror/mirror_integration_test.go new file mode 100644 index 00000000..b4a1bebd --- /dev/null +++ b/internal/recordmirror/mirror_integration_test.go @@ -0,0 +1,248 @@ +//go:build integration + +package recordmirror + +import ( + "errors" + "slices" + "testing" + + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/testbus" +) + +func aMirrorBus(t *testing.T) (*broker.JetStream, jetstream.JetStream) { + t.Helper() + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + return js, api +} + +func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) { + for _, m := range []Mirror{Assignments, DeclaredConfig} { + t.Run(m.Name, func(t *testing.T) { + js, api := aMirrorBus(t) + events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+m.Event) + w := aWriter(t, js, api, m) + ctx := t.Context() + + first := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)} + if _, err := w.Sync(ctx, first); err != nil { + t.Fatal(err) + } + before := writes(t, api, m.Bucket) + heardBefore := len(settle(events, 1)) + + // One record changes, one is added: two keys, two events, nothing else. + second := map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)} + r, err := w.Sync(ctx, second) + if err != nil { + t.Fatal(err) + } + if r.Written != 2 || r.Removed != 0 { + t.Errorf("a pass over one changed and one new record wrote %+v", r) + } + if got := writes(t, api, m.Bucket) - before; got != 2 { + t.Errorf("the bucket took %d writes for two changed records", got) + } + said := settle(events, heardBefore+2)[heardBefore:] + if len(said) != 2 { + t.Fatalf("two changes raised %d events: %+v", len(said), said) + } + changes := map[string]Event{} + for _, e := range said { + changes[e.Key] = e + } + if e := changes["anchor.gitea"]; e.Change != Changed || !slices.Equal(e.Fields, []string{"a"}) || + e.Mirror != m.Name || e.Epoch != 7 || e.Revision == 0 { + t.Errorf("the change of anchor.gitea was said as %+v", e) + } + if e := changes["anchor.forgejo"]; e.Change != Added { + t.Errorf("the new anchor.forgejo was said as %+v", e) + } + + // **An unchanged pass writes nothing**, and says nothing (ADR 0289 §1). + heardBefore = len(events.all()) + before = writes(t, api, m.Bucket) + r, err = w.Sync(ctx, second) + if err != nil { + t.Fatal(err) + } + if r != (Result{}) { + t.Errorf("a pass over unchanged records did %+v", r) + } + if got := writes(t, api, m.Bucket) - before; got != 0 { + t.Errorf("a pass over unchanged records wrote the bucket %d times", got) + } + if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 0 { + t.Errorf("a pass over unchanged records raised %+v", said) + } + + // **A record gone is a purge**, one write and one `removed`. + before = writes(t, api, m.Bucket) + r, err = w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)}) + if err != nil { + t.Fatal(err) + } + if r.Removed != 1 || r.Written != 0 { + t.Errorf("a record gone did %+v", r) + } + if got := writes(t, api, m.Bucket) - before; got != 1 { + t.Errorf("a record gone wrote the bucket %d times", got) + } + if _, err := w.KV.Get(ctx, "anchor.forgejo"); !errors.Is(err, jetstream.ErrKeyNotFound) { + t.Errorf("the key of a record gone is still read: %v", err) + } + said = settle(events, heardBefore+1)[heardBefore:] + if len(said) != 1 || said[0].Change != Removed || said[0].Key != "anchor.forgejo" { + t.Errorf("a record gone was said as %+v", said) + } + }) + } +} + +func TestNatsAWriterStartedAgainWritesOnlyWhatDiffers(t *testing.T) { + js, api := aMirrorBus(t) + ctx := t.Context() + first := aWriter(t, js, api, Assignments) + held := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`), "anchor.forgejo": []byte(`{"a":1}`)} + if _, err := first.Sync(ctx, held); err != nil { + t.Fatal(err) + } + before := writes(t, api, Assignments.Bucket) + again := aWriter(t, js, api, Assignments) + r, err := again.Sync(ctx, held) + if err != nil { + t.Fatal(err) + } + if r != (Result{}) || writes(t, api, Assignments.Bucket) != before { + t.Errorf("a writer started again over unchanged records did %+v", r) + } +} + +func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) { + js, api := aMirrorBus(t) + ctx := t.Context() + w := aWriter(t, js, api, Assignments) + if _, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil { + t.Fatal(err) + } + // Behind its back, as a former lease holder would. + if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil { + t.Fatal(err) + } + r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)}) + if err != nil { + t.Fatal(err) + } + if r.Written != 1 { + t.Errorf("a refused write was not made again: %+v", r) + } + e, err := w.KV.Get(ctx, "anchor.gitea") + if err != nil || string(e.Value()) != `{"a":2}` { + t.Fatalf("the bucket holds %v, %v", e, err) + } +} + +func TestNatsAForeignWriteIsCorrectedOnTheNextPass(t *testing.T) { + js, api := aMirrorBus(t) + ctx := t.Context() + events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event) + w := aWriter(t, js, api, Assignments) + store := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)} + if _, err := w.Sync(ctx, store); err != nil { + t.Fatal(err) + } + heardBefore := len(settle(events, 1)) + if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil { + t.Fatal(err) + } + r, err := w.Sync(ctx, store) + if err != nil { + t.Fatal(err) + } + if r != (Result{Written: 1}) { + t.Errorf("a pass after a foreign write did %+v", r) + } + e, err := w.KV.Get(ctx, "anchor.gitea") + if err != nil || string(e.Value()) != `{"a":1}` { + t.Fatalf("the bucket still holds %v, %v", e, err) + } + if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 1 || said[0].Change != Changed { + t.Errorf("the correction was said as %+v", said) + } + // And put right, it stays: the next pass writes nothing. + if r, err := w.Sync(ctx, store); err != nil || r != (Result{}) { + t.Errorf("a pass after the correction did %+v, %v", r, err) + } +} + +func TestNatsAWriterWithoutTheLeaseWritesNothing(t *testing.T) { + js, api := aMirrorBus(t) + ctx := t.Context() + held := aWriter(t, js, api, Assignments) + if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil { + t.Fatal(err) + } + before := writes(t, api, Assignments.Bucket) + for _, holds := range []func() bool{nil, func() bool { return false }} { + w := aWriter(t, js, api, Assignments) + w.Holds = holds + r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)}) + if !errors.Is(err, ErrNotHeld) || r != (Result{}) { + t.Errorf("a writer without the lease did %+v, %v", r, err) + } + r, err = w.Sync(ctx, map[string][]byte{}) + if !errors.Is(err, ErrNotHeld) || r != (Result{}) { + t.Errorf("a writer without the lease removed: %+v, %v", r, err) + } + } + // And one that loses it mid-pass stops before its next write. + asked := 0 + w := aWriter(t, js, api, Assignments) + w.Holds = func() bool { asked++; return asked <= 2 } + r, err := w.Sync(ctx, map[string][]byte{"anchor.a": []byte(`{}`), "anchor.b": []byte(`{}`), "anchor.gitea": []byte(`{"a":1}`)}) + if !errors.Is(err, ErrNotHeld) || r.Written != 1 { + t.Errorf("a writer that lost the lease mid-pass did %+v, %v", r, err) + } + if got := writes(t, api, Assignments.Bucket) - before; got != 1 { + t.Errorf("writers without the lease wrote the bucket %d times", got) + } +} + +func TestNatsARemovalWithoutTheLeaseRemovesNothing(t *testing.T) { + js, api := aMirrorBus(t) + ctx := t.Context() + held := aWriter(t, js, api, Assignments) + if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil { + t.Fatal(err) + } + before := writes(t, api, Assignments.Bucket) + asked := 0 + w := aWriter(t, js, api, Assignments) + w.Holds = func() bool { asked++; return asked == 1 } + r, err := w.Sync(ctx, map[string][]byte{}) + if !errors.Is(err, ErrNotHeld) || r != (Result{}) || asked != 2 { + t.Errorf("a removal without the lease did %+v, %v (the lease asked %d times)", r, err, asked) + } + if got := writes(t, api, Assignments.Bucket) - before; got != 0 { + t.Errorf("a removal without the lease wrote the bucket %d times", got) + } + if _, err := w.KV.Get(ctx, "anchor.gitea"); err != nil { + t.Errorf("the key was removed without the lease: %v", err) + } +} diff --git a/internal/recordmirror/mirror_store_test.go b/internal/recordmirror/mirror_store_test.go index 304f8e33..34d196df 100644 --- a/internal/recordmirror/mirror_store_test.go +++ b/internal/recordmirror/mirror_store_test.go @@ -1,3 +1,5 @@ +//go:build integration + package recordmirror import ( diff --git a/internal/recordmirror/mirror_test.go b/internal/recordmirror/mirror_test.go index 9d987e37..71cec884 100644 --- a/internal/recordmirror/mirror_test.go +++ b/internal/recordmirror/mirror_test.go @@ -2,7 +2,6 @@ package recordmirror import ( "encoding/json" - "errors" "slices" "sync" "testing" @@ -13,7 +12,6 @@ import ( "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/link" - "github.com/novox/mesh-controller/internal/testbus" ) // A record mirror against a real bus (novox/hq design 54 §2, §3; ADR 0289): a change of a record changes @@ -22,25 +20,6 @@ import ( // aMirrorBus is a bus of this test alone with the mesh's streams and the controller's buckets, as the // controller asserts them at its start, and a counter of the writes made to one bucket. -func aMirrorBus(t *testing.T) (*broker.JetStream, jetstream.JetStream) { - t.Helper() - js, err := broker.Dial(testbus.URL(t)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(js.Close) - if err := broker.AssertMeshStreams(js); err != nil { - t.Fatal(err) - } - if err := js.EnsureControllerBuckets(); err != nil { - t.Fatal(err) - } - api, err := jetstream.New(js.Conn()) - if err != nil { - t.Fatal(err) - } - return js, api -} // heard counts what is said on one subject, each body kept. type heard struct { @@ -112,230 +91,20 @@ func aWriter(t *testing.T, js *broker.JetStream, api jetstream.JetStream, m Mirr Epoch: func() uint64 { return 7 }, Holds: func() bool { return true }} } -func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) { - for _, m := range []Mirror{Assignments, DeclaredConfig} { - t.Run(m.Name, func(t *testing.T) { - js, api := aMirrorBus(t) - events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+m.Event) - w := aWriter(t, js, api, m) - ctx := t.Context() - - first := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)} - if _, err := w.Sync(ctx, first); err != nil { - t.Fatal(err) - } - before := writes(t, api, m.Bucket) - heardBefore := len(settle(events, 1)) - - // One record changes, one is added: two keys, two events, nothing else. - second := map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)} - r, err := w.Sync(ctx, second) - if err != nil { - t.Fatal(err) - } - if r.Written != 2 || r.Removed != 0 { - t.Errorf("a pass over one changed and one new record wrote %+v", r) - } - if got := writes(t, api, m.Bucket) - before; got != 2 { - t.Errorf("the bucket took %d writes for two changed records", got) - } - said := settle(events, heardBefore+2)[heardBefore:] - if len(said) != 2 { - t.Fatalf("two changes raised %d events: %+v", len(said), said) - } - changes := map[string]Event{} - for _, e := range said { - changes[e.Key] = e - } - if e := changes["anchor.gitea"]; e.Change != Changed || !slices.Equal(e.Fields, []string{"a"}) || - e.Mirror != m.Name || e.Epoch != 7 || e.Revision == 0 { - t.Errorf("the change of anchor.gitea was said as %+v", e) - } - if e := changes["anchor.forgejo"]; e.Change != Added { - t.Errorf("the new anchor.forgejo was said as %+v", e) - } - - // **An unchanged pass writes nothing**, and says nothing (ADR 0289 §1). - heardBefore = len(events.all()) - before = writes(t, api, m.Bucket) - r, err = w.Sync(ctx, second) - if err != nil { - t.Fatal(err) - } - if r != (Result{}) { - t.Errorf("a pass over unchanged records did %+v", r) - } - if got := writes(t, api, m.Bucket) - before; got != 0 { - t.Errorf("a pass over unchanged records wrote the bucket %d times", got) - } - if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 0 { - t.Errorf("a pass over unchanged records raised %+v", said) - } - - // **A record gone is a purge**, one write and one `removed`. - before = writes(t, api, m.Bucket) - r, err = w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)}) - if err != nil { - t.Fatal(err) - } - if r.Removed != 1 || r.Written != 0 { - t.Errorf("a record gone did %+v", r) - } - if got := writes(t, api, m.Bucket) - before; got != 1 { - t.Errorf("a record gone wrote the bucket %d times", got) - } - if _, err := w.KV.Get(ctx, "anchor.forgejo"); !errors.Is(err, jetstream.ErrKeyNotFound) { - t.Errorf("the key of a record gone is still read: %v", err) - } - said = settle(events, heardBefore+1)[heardBefore:] - if len(said) != 1 || said[0].Change != Removed || said[0].Key != "anchor.forgejo" { - t.Errorf("a record gone was said as %+v", said) - } - }) - } -} - // **A controller started again writes nothing for what the bucket already holds** (design 54 §2: the writer's // memory is filled from the bucket), and a key nobody holds a record for is removed. -func TestNatsAWriterStartedAgainWritesOnlyWhatDiffers(t *testing.T) { - js, api := aMirrorBus(t) - ctx := t.Context() - first := aWriter(t, js, api, Assignments) - held := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`), "anchor.forgejo": []byte(`{"a":1}`)} - if _, err := first.Sync(ctx, held); err != nil { - t.Fatal(err) - } - before := writes(t, api, Assignments.Bucket) - again := aWriter(t, js, api, Assignments) - r, err := again.Sync(ctx, held) - if err != nil { - t.Fatal(err) - } - if r != (Result{}) || writes(t, api, Assignments.Bucket) != before { - t.Errorf("a writer started again over unchanged records did %+v", r) - } -} // **Another writer in between is no lost value** (design 54 §2, compare-and-set): a write refused because the // key moved reads it again, takes the revision when it already holds the value, and otherwise writes again. -func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) { - js, api := aMirrorBus(t) - ctx := t.Context() - w := aWriter(t, js, api, Assignments) - if _, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil { - t.Fatal(err) - } - // Behind its back, as a former lease holder would. - if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil { - t.Fatal(err) - } - r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)}) - if err != nil { - t.Fatal(err) - } - if r.Written != 1 { - t.Errorf("a refused write was not made again: %+v", r) - } - e, err := w.KV.Get(ctx, "anchor.gitea") - if err != nil || string(e.Value()) != `{"a":2}` { - t.Fatalf("the bucket holds %v, %v", e, err) - } -} // **A value written behind the writer's back is put right on the next pass** (ADR 0289 §10): a former lease // holder's write differs from the store, and the store's value is written again, said as one change. -func TestNatsAForeignWriteIsCorrectedOnTheNextPass(t *testing.T) { - js, api := aMirrorBus(t) - ctx := t.Context() - events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event) - w := aWriter(t, js, api, Assignments) - store := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)} - if _, err := w.Sync(ctx, store); err != nil { - t.Fatal(err) - } - heardBefore := len(settle(events, 1)) - if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil { - t.Fatal(err) - } - r, err := w.Sync(ctx, store) - if err != nil { - t.Fatal(err) - } - if r != (Result{Written: 1}) { - t.Errorf("a pass after a foreign write did %+v", r) - } - e, err := w.KV.Get(ctx, "anchor.gitea") - if err != nil || string(e.Value()) != `{"a":1}` { - t.Fatalf("the bucket still holds %v, %v", e, err) - } - if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 1 || said[0].Change != Changed { - t.Errorf("the correction was said as %+v", said) - } - // And put right, it stays: the next pass writes nothing. - if r, err := w.Sync(ctx, store); err != nil || r != (Result{}) { - t.Errorf("a pass after the correction did %+v, %v", r, err) - } -} // **A writer without the lease writes nothing** (ADR 0289 §9, design 54 §2): not a new key, not a change, not // a removal, and it says so. -func TestNatsAWriterWithoutTheLeaseWritesNothing(t *testing.T) { - js, api := aMirrorBus(t) - ctx := t.Context() - held := aWriter(t, js, api, Assignments) - if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil { - t.Fatal(err) - } - before := writes(t, api, Assignments.Bucket) - for _, holds := range []func() bool{nil, func() bool { return false }} { - w := aWriter(t, js, api, Assignments) - w.Holds = holds - r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)}) - if !errors.Is(err, ErrNotHeld) || r != (Result{}) { - t.Errorf("a writer without the lease did %+v, %v", r, err) - } - r, err = w.Sync(ctx, map[string][]byte{}) - if !errors.Is(err, ErrNotHeld) || r != (Result{}) { - t.Errorf("a writer without the lease removed: %+v, %v", r, err) - } - } - // And one that loses it mid-pass stops before its next write. - asked := 0 - w := aWriter(t, js, api, Assignments) - w.Holds = func() bool { asked++; return asked <= 2 } - r, err := w.Sync(ctx, map[string][]byte{"anchor.a": []byte(`{}`), "anchor.b": []byte(`{}`), "anchor.gitea": []byte(`{"a":1}`)}) - if !errors.Is(err, ErrNotHeld) || r.Written != 1 { - t.Errorf("a writer that lost the lease mid-pass did %+v, %v", r, err) - } - if got := writes(t, api, Assignments.Bucket) - before; got != 1 { - t.Errorf("writers without the lease wrote the bucket %d times", got) - } -} // **A removal asks for the lease too** (ADR 0289 §9): a writer that held the lease when its pass began and lost // it before its first removal purges nothing, and says it does not hold the lease. -func TestNatsARemovalWithoutTheLeaseRemovesNothing(t *testing.T) { - js, api := aMirrorBus(t) - ctx := t.Context() - held := aWriter(t, js, api, Assignments) - if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil { - t.Fatal(err) - } - before := writes(t, api, Assignments.Bucket) - asked := 0 - w := aWriter(t, js, api, Assignments) - w.Holds = func() bool { asked++; return asked == 1 } - r, err := w.Sync(ctx, map[string][]byte{}) - if !errors.Is(err, ErrNotHeld) || r != (Result{}) || asked != 2 { - t.Errorf("a removal without the lease did %+v, %v (the lease asked %d times)", r, err, asked) - } - if got := writes(t, api, Assignments.Bucket) - before; got != 0 { - t.Errorf("a removal without the lease wrote the bucket %d times", got) - } - if _, err := w.KV.Get(ctx, "anchor.gitea"); err != nil { - t.Errorf("the key was removed without the lease: %v", err) - } -} // **A key the bus cannot hold is refused alone**, and the others are written. func TestAKeyTheBusCannotHoldIsRefusedAlone(t *testing.T) { diff --git a/internal/recordmirror/provisioning_live_test.go b/internal/recordmirror/provisioning_live_test.go index d80f4bc9..52f30171 100644 --- a/internal/recordmirror/provisioning_live_test.go +++ b/internal/recordmirror/provisioning_live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package recordmirror import ( diff --git a/internal/store/live_test.go b/internal/store/live_test.go index a03a03ae..9fc67ba4 100644 --- a/internal/store/live_test.go +++ b/internal/store/live_test.go @@ -1,3 +1,5 @@ +//go:build integration + package store import ( diff --git a/internal/testbus/testbus_integration_test.go b/internal/testbus/testbus_integration_test.go new file mode 100644 index 00000000..f189891f --- /dev/null +++ b/internal/testbus/testbus_integration_test.go @@ -0,0 +1,45 @@ +//go:build integration + +package testbus + +import ( + "strings" + "testing" + + "github.com/nats-io/nats.go" +) + +func TestEveryTestHasABusOfItsOwn(t *testing.T) { + seen := map[string]bool{} + for i := 0; i < 2; i++ { + t.Run("one", func(t *testing.T) { + url := URL(t) + if seen[url] { + t.Fatalf("two tests were handed %s", url) + } + seen[url] = true + conn, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + js, err := conn.JetStream() + if err != nil { + t.Fatal(err) + } + if _, err := js.StreamInfo("EVENTS"); err == nil { + t.Fatal("a stream an earlier test made is here") + } + if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.>"}}); err != nil { + t.Fatal(err) + } + }) + } + // Asked twice, one test reaches one bus. + if a, b := URL(t), URL(t); a != b { + t.Errorf("one test was handed two buses: %s and %s", a, b) + } + if v := strings.Count(Version, "."); v != 2 { + t.Errorf("the server's release reads %q", Version) + } +} diff --git a/internal/testbus/testbus_test.go b/internal/testbus/testbus_test.go index 5c261b9e..c851d459 100644 --- a/internal/testbus/testbus_test.go +++ b/internal/testbus/testbus_test.go @@ -9,8 +9,6 @@ import ( "encoding/json" - "github.com/nats-io/nats.go" - "github.com/novox/mesh-controller/internal/beside" ) @@ -59,37 +57,3 @@ func TestTheBusTheTestsRunIsTheBusTheMeshRuns(t *testing.T) { } // A test's bus is its own: what one test makes, the next never sees. -func TestEveryTestHasABusOfItsOwn(t *testing.T) { - seen := map[string]bool{} - for i := 0; i < 2; i++ { - t.Run("one", func(t *testing.T) { - url := URL(t) - if seen[url] { - t.Fatalf("two tests were handed %s", url) - } - seen[url] = true - conn, err := nats.Connect(url) - if err != nil { - t.Fatal(err) - } - defer conn.Close() - js, err := conn.JetStream() - if err != nil { - t.Fatal(err) - } - if _, err := js.StreamInfo("EVENTS"); err == nil { - t.Fatal("a stream an earlier test made is here") - } - if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.>"}}); err != nil { - t.Fatal(err) - } - }) - } - // Asked twice, one test reaches one bus. - if a, b := URL(t), URL(t); a != b { - t.Errorf("one test was handed two buses: %s and %s", a, b) - } - if v := strings.Count(Version, "."); v != 2 { - t.Errorf("the server's release reads %q", Version) - } -} diff --git a/merge-check.sh b/merge-check.sh index 239bf8ac..54e8f90c 100755 --- a/merge-check.sh +++ b/merge-check.sh @@ -9,15 +9,28 @@ # seat's first layer (`mesh/merge-gate`), run because the mesh's module graph builds `mesh-controller` # from this repository; it is not repeated here. This is the code's own quality and its suite: # -# 1. formatted and vetted; -# 2. the whole suite, packages in parallel, against the throwaway store the build seat raised at the -# release the mesh runs; every live test on a bus of its own (internal/testbus), so nothing is -# serialised. Under the race detector when the toolchain has a C compiler, and said when it has not. +# 1. formatted and vetted, with and without the build tag `integration`; +# 2. the unit level: `go test ./...` untagged and without the race detector, timed and held to a bound +# (unit_bound below): a test with no database, bus, running service or other process (novox/hq ADR 0391, +# how-we-build section 5). Over the bound the check fails and names the slowest tests (issue 895); +# 3. the integration level: the whole suite with `-tags integration`, packages in parallel, against the +# throwaway store the build seat raised at the release the mesh runs; every live test on a bus of its own +# (internal/testbus), so nothing is serialised. Under the race detector when the toolchain has a C +# compiler, and said when it has not. +# +# **A test that starts a bus, a store or another process, or composes the whole mesh, carries +# `//go:build integration`** as the first line of its file; a file that held both kinds is two files +# (`_integration_test.go` beside `_test.go`). Each level says its test count and its time. # # Exit 0 is a pass; anything else fails `mesh/repo-check` with the last line printed. set -eu export GOFLAGS=-mod=vendor GOPROXY=off +# **The unit level's bound, in seconds** (issue 895): the slowest package's own test time of the untagged run, the +# compile not counted. It is a few seconds while the tests of the last packages are still being split from the +# integration level; it becomes 1 s (how-we-build section 5: the whole unit suite under a second) when that is done. +unit_bound=5 + # judge_store_tests says each listed test of one package as `go test -v` reported it, and counts in $missing each # that did not pass: $1 the package, $2 go test's exit status, $3 the file holding its output, then the names. # A result is found anywhere in a line, not only at its start: when a test's last output leaves a line open, @@ -36,6 +49,40 @@ judge_store_tests() { missing=$((missing + 1)) fi } +# judge_unit_run says the unit level as `go test -v` reported it and fails when it was over the bound: $1 the file +# holding the output of the untagged `go test -v` of every package (packages one after another, each ending in its +# `ok