CT10: meshcatalogclient follows the catalog half per watched branch, a first view then each catalog-revision of that branch
The view of a branch is named mesh-catalog.catalog-revisions.<branch>; 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
This commit is contained in:
@@ -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 `<module>_first_views` and `<module>_first_view_again` read
|
||||
// and steer it as every first view. The view is named `mesh-catalog.catalog-revisions.<branch>`, 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.<branch>` (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.<branch>`) 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
|
||||
}
|
||||
@@ -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.<branch>`, 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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user