CAT-D7: distribution's whole-module integration test against a real bus

The program had no test that runs it whole. This one starts run() on a nats-server of its own, with the registry
container faked on the loopback and node_haves and manifest-reads answered on their call subjects: an away node holds
both windows open, its return closes them, each closing said once on distribution.events, a direct write is then
refused, reads stay open, the health check answers through the program's port, and a waiter on published wakes on
none of the windows' events.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

modules/distribution/cmd/distribution/whole_integration_test.go

@@ -0,0 +1,220 @@
+//go:build integration
+
+package main
+
+// distribution's program whole against a real bus (CAT-D7; design 51 *Acceptance tests*, the holder's cases): its
+// run() as the node starts it, its broker account on a nats-server of this test alone, the registry container a fake
+// OCI store on the loopback, the status view's node_haves and the node-engine seat's manifest-reads answered on their
+// call subjects. An away node holds the old name open; its return (manifest-reads-changed) closes it, said once on
+// distribution.events, after which a direct write is refused naming publish; reads stay open; and a waiter on a seat's
+// `published` hears none of the windows' events.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	neturl "net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/testbus"
+
+	"distribution/internal/holder"
+)
+
+func TestDistributionWhole(t *testing.T) {
+	url := testbus.URL(t)
+	nc, err := nats.Connect(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, _ := jetstream.New(nc)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	for _, b := range []string{"distribution_windows", "distribution_findings"} {
+		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "DISTRIBUTION_EVENTS", Subjects: []string{"mesh.stream.distribution.event.>"}, Duplicates: time.Hour}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "novox_distribution", AckPolicy: jetstream.AckExplicitPolicy,
+		FilterSubjects: []string{"mesh.mod.mesh-delivery-view.event.delivered", "mesh.seat.node-engine.event.manifest-reads-changed.*"}}); err != nil {
+		t.Fatal(err)
+	}
+
+	// The status view and the node-engines: g14 away (its node-engine answers nothing) and running the old build.
+	var mu sync.Mutex
+	g14 := false
+	answer := func(m *nats.Msg, result any) {
+		b, _ := json.Marshal(map[string]any{"result": result, "node": "novox"})
+		_ = m.Respond(b)
+	}
+	if _, err := nc.Subscribe("mesh.mod.mesh-delivery-view.>", func(m *nats.Msg) {
+		mu.Lock()
+		defer mu.Unlock()
+		fp := "fp-old"
+		if g14 {
+			fp = "fp-new"
+		}
+		answer(m, map[string]any{"keys": map[string]any{
+			"ace.mesh-build-go": map[string]any{"fp": "fp-new"}, "g14.mesh-build-go": map[string]any{"fp": fp},
+			"ace.gitea": map[string]any{"fp": "x"}, "g14.gitea": map[string]any{"fp": "x"},
+		}})
+	}); err != nil {
+		t.Fatal(err)
+	}
+	engine := func(node string) {
+		for _, subj := range []string{"mesh.seat.node-engine.call.manifest-reads." + node + ".>", "mesh.seat.node-engine.tool.manifest-reads." + node} {
+			if _, err := nc.Subscribe(subj, func(m *nats.Msg) {
+				answer(m, map[string]any{"v": 1, "manifest-reads": []string{holder.NewNamesCapability}})
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}
+		_ = nc.Flush()
+	}
+	engine("ace") // g14 is away: nothing answers its node-engine, so it states nothing
+	// A waiter on a seat's published: it must hear none of the windows' events.
+	var woke []string
+	if _, err := nc.Subscribe("mesh.seat.container-image-registry.event.published", func(m *nats.Msg) { woke = append(woke, m.Subject) }); err != nil {
+		t.Fatal(err)
+	}
+	pub, _ := json.Marshal(holder.Publishing{Name: holder.PublishingKey, Modules: map[string][]string{"mesh-build-go": {"fp-new"}}})
+	windows, _ := js.KeyValue(ctx, "distribution_windows")
+	if _, err := windows.Put(ctx, holder.PublishingKey, pub); err != nil {
+		t.Fatal(err)
+	}
+
+	// The registry container on the loopback, and the program's port.
+	store := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/tags/list") {
+			_, _ = io.WriteString(w, `{"tags":[]}`)
+			return
+		}
+		if r.Method == http.MethodHead {
+			w.WriteHeader(http.StatusNotFound) // no tag published through the seat
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+	}))
+	defer store.Close()
+	l, _ := net.Listen("tcp", "127.0.0.1:0")
+	listen := l.Addr().String()
+	l.Close()
+	dir := t.TempDir()
+	acct, _ := json.Marshal(map[string]any{"url": url, "node": "novox", "module": "distribution", "user": "novox.distribution", "password": "unused"})
+	_ = os.WriteFile(filepath.Join(dir, "broker"), acct, 0o600)
+	env := map[string]string{"MESH_BROKER_FILE": filepath.Join(dir, "broker"), "MESH_REGISTRY_LOOPBACK": store.URL, "MESH_REGISTRY_LISTEN": listen}
+	runCtx, stop := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	go func() { done <- run(runCtx, func(k string) string { return env[k] }) }()
+	defer func() { stop(); <-done }()
+
+	window := func(name string) holder.Window {
+		var w holder.Window
+		if e, err := windows.Get(ctx, name); err == nil {
+			_ = json.Unmarshal(e.Value(), &w)
+		}
+		return w
+	}
+	eventually := func(what string, ok func() bool) {
+		t.Helper()
+		for end := time.Now().Add(20 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) {
+			if ok() {
+				return
+			}
+			select {
+			case err := <-done:
+				t.Fatalf("the program ended: %v", err)
+			default:
+			}
+		}
+		t.Fatalf("not within 20s: %s (old-name %+v, direct-writes %+v)", what, window(holder.OldNameWindow), window(holder.DirectWritesWindow))
+	}
+	write := func() int {
+		req, _ := http.NewRequest(http.MethodPut, "http://"+listen+"/v2/m/i/manifests/0.1.9", nil)
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return 0
+		}
+		resp.Body.Close()
+		return resp.StatusCode
+	}
+	eventually("the start's wake: g14 holds both windows", func() bool {
+		o, d := window(holder.OldNameWindow), window(holder.DirectWritesWindow)
+		return o.Open && fmt.Sprint(o.Holding) == "[g14]" && d.Open && fmt.Sprint(d.Holding) == "[g14]"
+	})
+	if got := write(); got != http.StatusCreated {
+		t.Errorf("a direct write while the window is open = %d, want it through", got)
+	}
+
+	// g14 returns with the new node-engine and the new build: its manifest-reads-changed closes both windows.
+	mu.Lock()
+	g14 = true
+	mu.Unlock()
+	engine("g14")
+	if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.manifest-reads-changed.g14", []byte(`{"v":1,"node":"g14","why":"returned"}`)); err != nil {
+		t.Fatal(err)
+	}
+	eventually("g14's return closes both windows", func() bool {
+		return !window(holder.OldNameWindow).Closed.IsZero() && !window(holder.DirectWritesWindow).Closed.IsZero()
+	})
+	// A delivered after the close says nothing again.
+	if _, err := js.Publish(ctx, "mesh.mod.mesh-delivery-view.event.delivered", []byte(`{}`)); err != nil {
+		t.Fatal(err)
+	}
+	eventually("a direct write refused naming publish once closed", func() bool { return write() == http.StatusForbidden })
+	resp, err := http.Get("http://" + listen + "/v2/m/i/tags/list")
+	if err != nil || resp.StatusCode != http.StatusOK {
+		t.Errorf("a read once the window closed: %v %v", resp, err)
+	}
+	if _, err := holder.Health(&holder.OCI{Base: mustURL("http://" + listen)}, nil); err != nil {
+		t.Errorf("the health check through the program's port: %v", err)
+	}
+	s, _ := js.Stream(ctx, "DISTRIBUTION_EVENTS")
+	eventually("the program heard every waking event, the delivered after the close included", func() bool {
+		c, err := js.Consumer(ctx, "EVENTS", "novox_distribution")
+		if err != nil {
+			return false
+		}
+		i, err := c.Info(ctx)
+		return err == nil && i.NumPending == 0 && i.NumAckPending == 0 && i.Delivered.Consumer >= 2
+	})
+	info, _ := s.Info(ctx)
+	if info.State.Msgs != 2 {
+		t.Errorf("distribution.events holds %d events, want old-name-closed and direct-writes-closed once each", info.State.Msgs)
+	}
+	for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} {
+		if _, err := s.GetLastMsgForSubject(ctx, "mesh.stream.distribution.event."+ev); err != nil {
+			t.Errorf("%s not said: %v", ev, err)
+		}
+	}
+	if len(woke) != 0 {
+		t.Errorf("a waiter on published woke on %v", woke)
+	}
+}
+
+func mustURL(s string) *neturl.URL {
+	u, err := neturl.Parse(s)
+	if err != nil {
+		panic(err)
+	}
+	return u
+}

modules/distribution/go.mod

@@ -8,9 +8,15 @@ require (
 )
 
 require (
+	github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
+	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/klauspost/compress v1.20.0 // indirect
+	github.com/minio/highwayhash v1.0.4 // indirect
+	github.com/nats-io/jwt/v2 v2.8.1 // indirect
+	github.com/nats-io/nats-server/v2 v2.11.17 // indirect
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	golang.org/x/crypto v0.57.0 // indirect
 	golang.org/x/sys v0.48.0 // indirect
+	golang.org/x/time v0.15.0 // indirect
 )

modules/distribution/go.sum

@@ -20,6 +20,7 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
 golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
 golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=