CT10: meshcatalogclient follows the registrations, a first view then each registration-changed

The assigner and the forge module's merge switch on their first registrations (design 47 G §353), so the seat's client carries the typed follower on go/views: a first view asked of the verb registrations, a newer version heard asked again until the picture holds it, an answer read before the change asked once more. Builds: design 47 G §353, §245 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-11 11:43:53 +01:00
parent 5b0f155116
commit 496989ccfb
4 changed files with 369 additions and 6 deletions
@@ -6,3 +6,12 @@ require (
 	git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
 )
+
+require (
+	github.com/klauspost/compress v1.20.0 // indirect
+	github.com/nats-io/nats.go v1.54.0 // 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
+)
@@ -1,12 +1,26 @@
-github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
+github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
+github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
+github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
+github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
+github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
 github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
-github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
+github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
+github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
+github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
+github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
 github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
-github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
+github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
+github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
+github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
 github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
-github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
-golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
 golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
@@ -0,0 +1,200 @@
+package meshcatalogclient
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"time"
+
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/views"
+)
+
+// RegistrationsVerb is the seat verb a part takes its first view of the registrations by, as its manifest's invokes
+// names it.
+const RegistrationsVerb = "seat:" + seat.Seat + "." + seat.VerbRegistrations
+
+// Call asks one seat verb over the bus and answers what the holder said, whatever wraps it (process.Conn.Call, or
+// stdio.Ask through the tool runner).
+type Call func(key string, body any) (json.RawMessage, error)
+
+// registrationsCreated stands for the picture's Created in its follower: the registrations are followed by the seat's
+// event, never by an owner's views family, so one fixed time, never zero.
+var registrationsCreated = time.Unix(0, 0).UTC()
+
+// Registrations is a part's typed follower of the registrations (design 47 G §353): a first view asked of the seat's verb
+// `registrations`, then each `registration-changed` heard, built on go/views' Follower, so the runtime's tools
+// `<module>_first_views` and `<module>_first_view_again` read and steer it as every first view. Its picture is keyed by
+// module, each value the module's newest registration. Ready is closed at the first view: a part that switches on its
+// first registrations waits for it.
+type Registrations struct {
+	Follower *views.Follower
+
+	call Call
+}
+
+// NewRegistrations is a Registrations asking the seat through call. Nothing is asked before Start.
+func NewRegistrations(call Call) *Registrations {
+	r := &Registrations{call: call}
+	r.Follower = &views.Follower{Owner: seat.Seat, Tool: RegistrationsVerb, State: seat.StateRegistrations, Ask: r.ask}
+	return r
+}
+
+// WireRegistrations is a Registrations adopted by a stdio part's views.Reader: the reader writes its conditions, reads it
+// as `mesh-catalog.registrations`, lists and asks it again in its first-view tools, and at its Start subscribes the
+// seat's event registration-changed (the part's manifest consumes it) into Hear and asks the first view. changed, when
+// not nil, is told of each module whose registration the picture changes, after the reader's own. The part declares the
+// seat's verb registrations in invokes.
+func WireRegistrations(r *views.Reader, call Call, changed func(reg seat.Registration, gone bool)) (*Registrations, error) {
+	g := NewRegistrations(call)
+	g.Follower.Changed = typedChanged(changed)
+	if err := r.Adopt(g.Follower, seat.Seat, seat.EventRegistrationChanged, g.Hear); err != nil {
+		return nil, err
+	}
+	return g, nil
+}
+
+// WireRegistrationsProcess is WireRegistrations for a process's views.ProcessReader: its picture
+// (`mesh-catalog_registrations`) reads the registrations as a read-only bucket, and the seat's event
+// registration-changed on the module's durable consumer goes to Hear.
+func WireRegistrationsProcess(r *views.ProcessReader, call Call, changed func(reg seat.Registration, gone bool)) (*Registrations, *views.Picture, error) {
+	g := NewRegistrations(call)
+	g.Follower.Changed = typedChanged(changed)
+	p, err := r.Adopt(g.Follower, seat.Seat, seat.EventRegistrationChanged, g.Hear)
+	if err != nil {
+		return nil, nil, err
+	}
+	return g, p, nil
+}
+
+func typedChanged(changed func(seat.Registration, bool)) func(string, json.RawMessage, bool, time.Time) {
+	if changed == nil {
+		return nil
+	}
+	return func(module string, value json.RawMessage, gone bool, _ time.Time) {
+		reg := seat.Registration{Module: module}
+		if !gone {
+			_ = json.Unmarshal(value, &reg)
+		}
+		changed(reg, gone)
+	}
+}
+
+// Start asks the first view until it is answered or ctx ends.
+func (r *Registrations) Start(ctx context.Context) error { return r.Follower.Start(ctx) }
+
+// Ready is closed once the first view is answered.
+func (r *Registrations) Ready() <-chan struct{} { return r.Follower.Ready() }
+
+// Registration is one module's newest registration in the picture.
+func (r *Registrations) Registration(module string) (seat.Registration, bool) {
+	raw, ok := r.Follower.Get(module)
+	if !ok {
+		return seat.Registration{}, false
+	}
+	var reg seat.Registration
+	if err := json.Unmarshal(raw, &reg); err != nil {
+		return seat.Registration{}, false
+	}
+	return reg, true
+}
+
+// All is every registration in the picture, by module name.
+func (r *Registrations) All() []seat.Registration {
+	values := r.Follower.Values()
+	out := make([]seat.Registration, 0, len(values))
+	for _, raw := range values {
+		var reg seat.Registration
+		if json.Unmarshal(raw, &reg) == nil {
+			out = append(out, reg)
+		}
+	}
+	slices.SortFunc(out, func(a, b seat.Registration) int {
+		if a.Module < b.Module {
+			return -1
+		}
+		if a.Module > b.Module {
+			return 1
+		}
+		return 0
+	})
+	return out
+}
+
+// Hear takes one `registration-changed` event, as the part's durable consumer of the seat's events hands it. A version
+// the picture holds, or one older, is passed over; a newer one asks the first view again on its own goroutine, so the
+// consumer is never held while the holder is away, and again until the picture holds that version or ctx ends (an ask
+// already running, the first view among them, may have been answered before the change was written). Data that is no such event is an error.
+func (r *Registrations) Hear(ctx context.Context, data []byte) error {
+	var ev seat.RegistrationChangedEvent
+	if err := json.Unmarshal(data, &ev); err != nil || ev.Module == "" || ev.Version <= 0 {
+		return fmt.Errorf("meshcatalogclient: not a %s event: %s", seat.EventRegistrationChanged, clipBytes(data))
+	}
+	if r.holds(ev.Module, ev.Version) {
+		return nil
+	}
+	go r.catchUp(ctx, ev.Module, ev.Version)
+	return nil
+}
+
+func (r *Registrations) holds(module string, version int64) bool {
+	reg, ok := r.Registration(module)
+	return ok && reg.Version >= version
+}
+
+// catchUp asks the first view again until the picture holds version of module, pausing between asks as the follower's
+// Wait says (doubling to a minute), or until ctx ends.
+func (r *Registrations) catchUp(ctx context.Context, module string, version int64) {
+	wait := r.Follower.Wait
+	if wait == 0 {
+		wait = 5 * time.Second
+	}
+	for !r.holds(module, version) {
+		if err := r.Follower.Gap(ctx, fmt.Sprintf("registration of %s at version %d said", module, version)); err != nil {
+			return
+		}
+		if r.holds(module, version) {
+			return
+		}
+		p, cancel := context.WithTimeout(ctx, wait)
+		<-p.Done()
+		cancel()
+		if ctx.Err() != nil {
+			return
+		}
+		if wait *= 2; wait > time.Minute {
+			wait = time.Minute
+		}
+	}
+}
+
+// ask is the Follower's Ask: every registration, whole in one answer.
+func (r *Registrations) ask(_ context.Context, _ string) (views.Answer, error) {
+	raw, err := r.call(RegistrationsVerb, seat.RegistrationsInput{})
+	if err != nil {
+		return views.Answer{}, err
+	}
+	answer, said, ok := buscalls.Unwrap(raw)
+	if !ok {
+		return views.Answer{}, fmt.Errorf("%s answered: %s", RegistrationsVerb, said)
+	}
+	var a seat.RegistrationsAnswer
+	if err := json.Unmarshal(answer, &a); err != nil || a.Registrations == nil {
+		return views.Answer{}, fmt.Errorf("%s answered no registrations: %s", RegistrationsVerb, clipBytes(answer))
+	}
+	values := make(map[string]json.RawMessage, len(a.Registrations))
+	for _, reg := range a.Registrations {
+		v, _ := json.Marshal(reg)
+		values[reg.Module] = v
+	}
+	return views.Answer{V: 1, Owner: seat.Seat, State: seat.StateRegistrations, Created: registrationsCreated, Values: values, At: time.Now()}, nil
+}
+
+func clipBytes(b []byte) string {
+	if len(b) > 200 {
+		return string(b[:200]) + "…"
+	}
+	return string(b)
+}
@@ -0,0 +1,140 @@
+package meshcatalogclient
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+)
+
+// fakeRegistrations answers the verb registrations from the versions it keeps, and counts each ask.
+type fakeRegistrations struct {
+	mu       sync.Mutex
+	versions map[string]int64
+	asks     int
+	// stale answers the next ask from these versions, once: an answer read before a change was written.
+	stale map[string]int64
+}
+
+func (f *fakeRegistrations) call(key string, body any) (json.RawMessage, error) {
+	if key != RegistrationsVerb {
+		return nil, fmt.Errorf("asked %s", key)
+	}
+	if in := body.(seat.RegistrationsInput); in.Module != "" || in.History {
+		return nil, fmt.Errorf("asked %+v", in)
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.asks++
+	versions := f.versions
+	if f.stale != nil {
+		versions, f.stale = f.stale, nil
+	}
+	a := seat.RegistrationsAnswer{Registrations: []seat.Registration{}}
+	for m, v := range versions {
+		a.Registrations = append(a.Registrations, seat.Registration{Module: m, Version: v, Origin: "novox/" + m, Path: "module.json", Branches: []string{"main"}})
+	}
+	return json.Marshal(a)
+}
+
+func (f *fakeRegistrations) set(module string, version int64) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.versions[module] = version
+}
+
+func (f *fakeRegistrations) asked() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.asks
+}
+
+func changedEvent(module string, version int64) []byte {
+	b, _ := json.Marshal(seat.RegistrationChangedEvent{V: 1, Module: module, Version: version, Moved: []string{"branches"}, Warrant: "ask-1"})
+	return b
+}
+
+// CT10 (design 47 G §353): a part following the registrations takes a first view of the verb registrations, then each
+// registration-changed: a newer version is asked and held, one it holds asks nothing, a change heard while an answer
+// read before it was given asks again, and what is no such event is refused.
+func TestRegistrationsAreAFirstViewThenTheirEvents(t *testing.T) {
+	changed := make(chan seat.Registration, 16)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	next := func() seat.Registration {
+		t.Helper()
+		select {
+		case reg := <-changed:
+			return reg
+		case <-ctx.Done():
+			t.Fatal("no registration changed")
+			return seat.Registration{}
+		}
+	}
+	f := &fakeRegistrations{versions: map[string]int64{"nats": 1, "zsh": 3}}
+	r := NewRegistrations(f.call)
+	r.Follower.Wait = time.Nanosecond
+	r.Follower.Changed = typedChanged(func(reg seat.Registration, gone bool) {
+		if !gone {
+			changed <- reg
+		}
+	})
+	select {
+	case <-r.Ready():
+		t.Fatal("ready before its first view")
+	default:
+	}
+	if err := r.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	<-r.Ready()
+	if reg, ok := r.Registration("nats"); !ok || reg.Version != 1 || reg.Origin != "novox/nats" {
+		t.Fatalf("first view holds nats as %+v, %v", reg, ok)
+	}
+	if all := r.All(); len(all) != 2 || all[0].Module != "nats" || all[1].Module != "zsh" {
+		t.Fatalf("All = %+v", all)
+	}
+	for range 2 {
+		next()
+	}
+
+	if err := r.Hear(ctx, changedEvent("zsh", 3)); err != nil {
+		t.Fatal(err)
+	}
+	if n := f.asked(); n != 1 {
+		t.Errorf("a version held asked the first view again: %d asks", n)
+	}
+
+	f.set("nats", 2)
+	if err := r.Hear(ctx, changedEvent("nats", 2)); err != nil {
+		t.Fatal(err)
+	}
+	if reg := next(); reg.Module != "nats" || reg.Version != 2 {
+		t.Fatalf("after registration-changed: %+v", reg)
+	}
+
+	// The answer to the ask the event starts was read before the change was written: it is asked once more.
+	f.mu.Lock()
+	f.stale = map[string]int64{"nats": 2, "zsh": 3}
+	f.mu.Unlock()
+	f.set("zsh", 4)
+	if err := r.Hear(ctx, changedEvent("zsh", 4)); err != nil {
+		t.Fatal(err)
+	}
+	if reg := next(); reg.Module != "zsh" || reg.Version != 4 {
+		t.Fatalf("after a stale answer: %+v", reg)
+	}
+	if n := f.asked(); n != 4 {
+		t.Errorf("asks = %d, want 4 (first view, nats, a stale answer and the one after)", n)
+	}
+
+	for _, data := range []string{`{}`, `not json`, `{"module":"nats"}`} {
+		if err := r.Hear(ctx, []byte(data)); err == nil {
+			t.Errorf("Hear(%s) took it", data)
+		}
+	}
+}