From 496989ccfb485488bf9a6c42d61d0409ad1d2adc Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:43:53 +0100 Subject: [PATCH] CT10: meshcatalogclient follows the registrations, a first view then each registration-changed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- contracts/go/clients/meshcatalogclient/go.mod | 9 + contracts/go/clients/meshcatalogclient/go.sum | 26 ++- .../meshcatalogclient/registrations.go | 200 ++++++++++++++++++ .../meshcatalogclient/registrations_test.go | 140 ++++++++++++ 4 files changed, 369 insertions(+), 6 deletions(-) create mode 100644 contracts/go/clients/meshcatalogclient/registrations.go create mode 100644 contracts/go/clients/meshcatalogclient/registrations_test.go diff --git a/contracts/go/clients/meshcatalogclient/go.mod b/contracts/go/clients/meshcatalogclient/go.mod index 450b54a2..433b7918 100644 --- a/contracts/go/clients/meshcatalogclient/go.mod +++ b/contracts/go/clients/meshcatalogclient/go.mod @@ -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 +) diff --git a/contracts/go/clients/meshcatalogclient/go.sum b/contracts/go/clients/meshcatalogclient/go.sum index 0daef11f..48737795 100644 --- a/contracts/go/clients/meshcatalogclient/go.sum +++ b/contracts/go/clients/meshcatalogclient/go.sum @@ -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= diff --git a/contracts/go/clients/meshcatalogclient/registrations.go b/contracts/go/clients/meshcatalogclient/registrations.go new file mode 100644 index 00000000..d58aaf0e --- /dev/null +++ b/contracts/go/clients/meshcatalogclient/registrations.go @@ -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 +// `_first_views` and `_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, ®) + } + 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, ®); 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, ®) == 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) +} diff --git a/contracts/go/clients/meshcatalogclient/registrations_test.go b/contracts/go/clients/meshcatalogclient/registrations_test.go new file mode 100644 index 00000000..66ff2546 --- /dev/null +++ b/contracts/go/clients/meshcatalogclient/registrations_test.go @@ -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) + } + } +}