From 5e0c04c3cfbc0a718690330a0f5ac4e2e6cec114 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 12:15:39 +0100 Subject: [PATCH] CT10: meshcatalogclient follows the catalog half per watched branch, a first view then each catalog-revision of that branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The view of a branch is named mesh-catalog.catalog-revisions.; two branches give two views. A revision heard and not held asks revision once more, said in the log. Builds: design 47 G §358, §196, §479, §495 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../go/clients/meshcatalogclient/catalog.go | 227 ++++++++++++++++++ .../meshcatalogclient/catalog_view_test.go | 147 ++++++++++++ 2 files changed, 374 insertions(+) create mode 100644 contracts/go/clients/meshcatalogclient/catalog.go create mode 100644 contracts/go/clients/meshcatalogclient/catalog_view_test.go diff --git a/contracts/go/clients/meshcatalogclient/catalog.go b/contracts/go/clients/meshcatalogclient/catalog.go new file mode 100644 index 00000000..95d46cf5 --- /dev/null +++ b/contracts/go/clients/meshcatalogclient/catalog.go @@ -0,0 +1,227 @@ +package meshcatalogclient + +import ( + "context" + "encoding/json" + "fmt" + "maps" + "strconv" + "sync" + "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/keys" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// RevisionVerb is the seat verb a part takes its first view of the catalog half by, as its manifest's invokes names it. +const RevisionVerb = "seat:" + seat.Seat + "." + seat.VerbRevision + +// The kinds a catalog graph revision is of. +const ( + KindTrunk = "trunk" + KindCandidate = "candidate" +) + +// Half is one catalog graph revision whole: the revision with its entries by module. +type Half struct { + Revision seat.CatalogRevision + Entries map[string]seat.CatalogEntry +} + +// ViewName is how the catalog half's revision is named in a part's first-view status (views.Status.View, WaitsFor). +func ViewName(revision int64) string { return strconv.FormatInt(revision, 10) } + +// catalogCreated stands for the view's Created in its follower: a revision view is never made again, so one fixed time, +// never zero, lets the follower tell a revision heard from one it holds. +var catalogCreated = time.Unix(0, 0).UTC() + +// Catalog is a part's view of the catalog half of one watched branch (design 47 G §358, §479): a first view of the +// branch's newest trunk revision asked of the seat's verb `revision`, then kept from its event `catalog-revision` of that +// branch, built on go/views' Follower, so the runtime's tools `_first_views` and `_first_view_again` read +// and steer it as every first view. The view is named `mesh-catalog.catalog-revisions.`, so a part that watches two +// branches makes two Catalogs, two views. A `catalog-revision` the view missed (a revision heard that it does not hold) +// asks `revision` once more, said in the log (§196, ADR 0615 §5), and is waited for, said in the status and past +// views.WaitsBound raised as view-waits. +type Catalog struct { + Follower *views.Follower + // Branch is the watched branch the view is of, as the caller names it. + Branch string + + call Call + mu sync.Mutex + // held is the half the view holds; paging the one being read, page by page, of one revision. + held *Half + paging *Half + // waits is the revision waited for, 0 when none. + waits int64 +} + +// NewCatalog is a Catalog of branch asking the seat through call. Nothing is asked before Start. +func NewCatalog(call Call, branch string) (*Catalog, error) { + if err := keys.CheckName("branch", branch); err != nil { + return nil, err + } + c := &Catalog{call: call, Branch: branch} + c.Follower = &views.Follower{Owner: seat.Seat, Tool: RevisionVerb, State: views.BranchState(seat.StateCatalogRevisions, branch), Ask: c.ask} + return c, nil +} + +// WireCatalog is a Catalog of branch adopted by a stdio part's views.Reader: the reader writes its conditions and its +// view-waits finding, reads it as `mesh-catalog.catalog-revisions.` (a key a module, its value the module's +// entry), lists and asks it again in its first-view tools, and at its Start subscribes the seat's event catalog-revision +// (the part's manifest consumes it) into Hear and asks the first view. changed, when not nil, is the Follower's Changed. +// The part declares the seat's verb revision in invokes. Call it once per watched branch. +func WireCatalog(r *views.Reader, call Call, branch string, changed func(module string, entry json.RawMessage, deleted bool, at time.Time)) (*Catalog, error) { + c, err := NewCatalog(call, branch) + if err != nil { + return nil, err + } + c.Follower.Changed = changed + if err := r.Adopt(c.Follower, seat.Seat, seat.EventCatalogRevision, c.Hear); err != nil { + return nil, err + } + return c, nil +} + +// WireCatalogProcess is WireCatalog for a process's views.ProcessReader: its picture +// (`mesh-catalog_catalog-revisions.`) reads the half as a read-only bucket, and the seat's event catalog-revision +// on the module's durable consumer goes to Hear. +func WireCatalogProcess(r *views.ProcessReader, call Call, branch string, changed func(module string, entry json.RawMessage, deleted bool, at time.Time)) (*Catalog, *views.Picture, error) { + c, err := NewCatalog(call, branch) + if err != nil { + return nil, nil, err + } + c.Follower.Changed = changed + p, err := r.Adopt(c.Follower, seat.Seat, seat.EventCatalogRevision, c.Hear) + if err != nil { + return nil, nil, err + } + return c, p, nil +} + +// Start asks the first view, the branch's newest trunk revision, until it is answered or ctx ends. +func (c *Catalog) Start(ctx context.Context) error { return c.Follower.Start(ctx) } + +// Ready is closed once the first view is answered. +func (c *Catalog) Ready() <-chan struct{} { return c.Follower.Ready() } + +// View is the half the view holds; false while it has none, when the part acts on nothing new. +func (c *Catalog) View() (Half, bool) { + c.mu.Lock() + defer c.mu.Unlock() + if c.held == nil { + return Half{}, false + } + return Half{Revision: c.held.Revision, Entries: maps.Clone(c.held.Entries)}, true +} + +// Entry is one module's entry in the half the view holds. +func (c *Catalog) Entry(module string) (seat.CatalogEntry, bool) { + c.mu.Lock() + defer c.mu.Unlock() + if c.held == nil { + return seat.CatalogEntry{}, false + } + e, ok := c.held.Entries[module] + return e, ok +} + +// Hear takes one `catalog-revision` event, as the part's durable consumer of the seat's events hands it. A candidate's +// revision and another branch's move nothing here; a trunk revision of this branch the view holds, or one older, is passed +// over; a newer one is a revision the view missed: waited for, said in the log, and asked again on its own goroutine, so +// the consumer is never held while the holder is away. Data that is no such event is an error. +func (c *Catalog) Hear(ctx context.Context, data []byte) error { + var ev seat.CatalogRevisionEvent + if err := json.Unmarshal(data, &ev); err != nil || ev.Revision <= 0 { + return fmt.Errorf("meshcatalogclient: not a %s event: %s", seat.EventCatalogRevision, clipBytes(data)) + } + if ev.Kind != KindTrunk || ev.Branch != keys.EscapeName(c.Branch) { + return nil + } + c.mu.Lock() + held := int64(0) + if c.held != nil { + held = c.held.Revision.Revision + } + if ev.Revision > c.waits { + c.waits = ev.Revision + } + c.mu.Unlock() + if ev.Revision <= held { + return nil + } + // The follower keeps the newest revision heard, so an ask that was running when it was heard is asked once more. + c.Follower.Apply(views.Event{Owner: seat.Seat, State: c.Follower.State, Revision: uint64(ev.Revision), Created: catalogCreated, Gap: true}) + c.Follower.WaitFor(ctx, ViewName(ev.Revision)) + if c.Follower.Logf != nil { + c.Follower.Logf("meshcatalogclient: catalog revision %d of %s said, the view holds %d: %s is asked once more", ev.Revision, c.Branch, held, RevisionVerb) + } + go func() { + _ = c.Follower.Gap(ctx, fmt.Sprintf("catalog revision %d of %s said, the view holds %d", ev.Revision, c.Branch, held)) + }() + return nil +} + +// ask is the Follower's Ask: one page of the branch's newest trunk revision, the next pages of the same revision; once +// whole, it is the half the view holds. +func (c *Catalog) ask(_ context.Context, after string) (views.Answer, error) { + in := seat.RevisionInput{After: after, Branch: keys.EscapeName(c.Branch)} + c.mu.Lock() + if after != "" && c.paging != nil { + in.Revision = c.paging.Revision.Revision + } + c.mu.Unlock() + a, err := askRevision(c.call, in) + if err != nil { + return views.Answer{}, err + } + if a.Revision.Kind != KindTrunk { + return views.Answer{}, fmt.Errorf("%s answered a %s revision for the trunk", RevisionVerb, a.Revision.Kind) + } + c.mu.Lock() + if after == "" || c.paging == nil { + c.paging = &Half{Revision: a.Revision, Entries: map[string]seat.CatalogEntry{}} + } else if c.paging.Revision.Revision != a.Revision.Revision { + was := c.paging.Revision.Revision + c.paging = nil + c.mu.Unlock() + return views.Answer{}, fmt.Errorf("%s answered revision %d for a page of revision %d", RevisionVerb, a.Revision.Revision, was) + } + values := map[string]json.RawMessage{} + for _, e := range a.Entries { + c.paging.Entries[e.Module] = e + raw, _ := json.Marshal(e) + values[e.Module] = raw + } + rev := a.Revision.Revision + holds := false + if a.Next == "" { + c.held, c.paging = c.paging, nil + if c.waits <= rev { + c.waits, holds = 0, true + } + } + c.mu.Unlock() + if holds { + c.Follower.Holds(ViewName(rev)) + } + return views.Answer{V: 1, Owner: seat.Seat, State: c.Follower.State, Revision: uint64(rev), Created: catalogCreated, Values: values, Next: a.Next, At: time.Now()}, nil +} + +func askRevision(call Call, in seat.RevisionInput) (seat.RevisionAnswer, error) { + raw, err := call(RevisionVerb, in) + if err != nil { + return seat.RevisionAnswer{}, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return seat.RevisionAnswer{}, fmt.Errorf("%s answered: %s", RevisionVerb, said) + } + var a seat.RevisionAnswer + if err := json.Unmarshal(answer, &a); err != nil || a.Revision.Revision <= 0 { + return seat.RevisionAnswer{}, fmt.Errorf("%s answered no revision: %s", RevisionVerb, clipBytes(answer)) + } + return a, nil +} diff --git a/contracts/go/clients/meshcatalogclient/catalog_view_test.go b/contracts/go/clients/meshcatalogclient/catalog_view_test.go new file mode 100644 index 00000000..6077fede --- /dev/null +++ b/contracts/go/clients/meshcatalogclient/catalog_view_test.go @@ -0,0 +1,147 @@ +package meshcatalogclient + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// fakeRevisions answers the verb revision per branch from the newest revision each branch holds, and counts each ask. +type fakeRevisions struct { + mu sync.Mutex + newest map[string]int64 // by escaped branch + asks map[string]int + branches []string +} + +func (f *fakeRevisions) call(key string, body any) (json.RawMessage, error) { + in := body.(seat.RevisionInput) + f.mu.Lock() + defer f.mu.Unlock() + rev, ok := f.newest[in.Branch] + if !ok { + return nil, fmt.Errorf("asked for the branch %q", in.Branch) + } + f.asks[in.Branch]++ + return json.Marshal(seat.RevisionAnswer{ + Revision: seat.CatalogRevision{Revision: rev, Kind: KindTrunk, Branch: in.Branch}, + Entries: []seat.CatalogEntry{{Module: fmt.Sprintf("m-of-%s-%d", in.Branch, rev)}}, + }) +} + +func (f *fakeRevisions) set(branch string, rev int64) { + f.mu.Lock() + defer f.mu.Unlock() + f.newest[keys.EscapeName(branch)] = rev +} + +func (f *fakeRevisions) asked(branch string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.asks[keys.EscapeName(branch)] +} + +func revisionEvent(branch, kind string, rev int64) []byte { + b, _ := json.Marshal(seat.CatalogRevisionEvent{V: 1, Branch: keys.EscapeName(branch), Kind: kind, Revision: rev}) + return b +} + +// CT10 (design 47 G §358, §479, §196): the catalog half is a view per watched branch, named by its caller: two branches +// give two views, each named `mesh-catalog.catalog-revisions.`, each following only its own branch's +// `catalog-revision`; a revision the view missed asks `revision` once more, said in the log. +func TestCatalogIsAViewPerWatchedBranch(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + f := &fakeRevisions{newest: map[string]int64{"main": 3, keys.EscapeName("release/1"): 7}, asks: map[string]int{}} + var logMu sync.Mutex + var logged []string + logf := func(format string, args ...any) { + logMu.Lock() + defer logMu.Unlock() + logged = append(logged, fmt.Sprintf(format, args...)) + } + main, err := NewCatalog(f.call, "main") + if err != nil { + t.Fatal(err) + } + rel, err := NewCatalog(f.call, "release/1") + if err != nil { + t.Fatal(err) + } + for _, c := range []*Catalog{main, rel} { + c.Follower.Wait = time.Nanosecond + c.Follower.Logf = logf + } + if main.Follower.State == rel.Follower.State || main.Follower.State != "catalog-revisions.main" || rel.Follower.State != views.BranchState(seat.StateCatalogRevisions, "release/1") { + t.Fatalf("views are named %q and %q", main.Follower.State, rel.Follower.State) + } + for _, c := range []*Catalog{main, rel} { + if err := c.Start(ctx); err != nil { + t.Fatal(err) + } + <-c.Ready() + } + if h, ok := main.View(); !ok || h.Revision.Revision != 3 || len(h.Entries) != 1 { + t.Fatalf("main holds %+v, %v", h, ok) + } + if h, ok := rel.View(); !ok || h.Revision.Revision != 7 { + t.Fatalf("release/1 holds %+v, %v", h, ok) + } + + // An event of the other branch, a candidate's and one the view holds ask nothing. + for _, data := range [][]byte{revisionEvent("release/1", KindTrunk, 8), revisionEvent("main", KindCandidate, 4), revisionEvent("main", KindTrunk, 3)} { + if err := main.Hear(ctx, data); err != nil { + t.Fatal(err) + } + } + if n := f.asked("main"); n != 1 { + t.Errorf("main asked %d times, want only its first view", n) + } + + // A missed revision asks `revision` once more, and is said. + f.set("main", 4) + if err := main.Hear(ctx, revisionEvent("main", KindTrunk, 4)); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(5 * time.Second) + for { + if h, _ := main.View(); h.Revision.Revision == 4 { + break + } + if time.Now().After(deadline) { + t.Fatal("main never held revision 4") + } + time.Sleep(time.Millisecond) + } + if n := f.asked("release/1"); n != 1 { + t.Errorf("release/1 asked %d times after main's revision", n) + } + logMu.Lock() + said := false + for _, l := range logged { + if strings.Contains(l, "catalog revision 4 of main said") { + said = true + } + } + logMu.Unlock() + if !said { + t.Errorf("the gap was not said in the log: %q", logged) + } + + for _, data := range []string{`{}`, `not json`} { + if err := main.Hear(ctx, []byte(data)); err == nil { + t.Errorf("Hear(%s) took it", data) + } + } + if _, err := NewCatalog(f.call, ""); err == nil { + t.Error("a Catalog of no branch was made") + } +}