From 9c68e42c44b35c45ad3053cd3eadf64443b4a984 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 21:41:28 +0100 Subject: [PATCH] =?UTF-8?q?A-5=20(=C2=A7614):=20meshcatalogclient=20follow?= =?UTF-8?q?s=20the=20branches=20once=20for=20every=20part:=20registrations?= =?UTF-8?q?,=20the=20gate's=20source,=20views=20adopted=20and=20dropped,?= =?UTF-8?q?=20entries=20by=20fingerprint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §614, A-5 (G-20) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../go/clients/meshcatalogclient/branches.go | 504 ++++++++++++++++++ .../meshcatalogclient/branches_test.go | 224 ++++++++ contracts/go/clients/meshcatalogclient/go.mod | 1 + .../meshcatalogclient/registrations.go | 24 +- 4 files changed, 730 insertions(+), 23 deletions(-) create mode 100644 contracts/go/clients/meshcatalogclient/branches.go create mode 100644 contracts/go/clients/meshcatalogclient/branches_test.go diff --git a/contracts/go/clients/meshcatalogclient/branches.go b/contracts/go/clients/meshcatalogclient/branches.go new file mode 100644 index 00000000..aa7b79e8 --- /dev/null +++ b/contracts/go/clients/meshcatalogclient/branches.go @@ -0,0 +1,504 @@ +package meshcatalogclient + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "sync" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// SourceVerb is the gate's verb a part reads the branches the gate keeps a source revision of by, before the first +// registrations, as its manifest's invokes names it (design 47 G §614). +const SourceVerb = "seat:" + delivery.Seat + "." + delivery.VerbSource + +// EntryVerb is the seat verb a part reads an entry by its fingerprint by, as its manifest's invokes names it. +const EntryVerb = "seat:" + seat.Seat + "." + seat.VerbEntry + +// StateSourceBranches names the view of the branches the gate's source keeps in a part's reader (`mesh-delivery.source`). +const StateSourceBranches = "source" + +// Wiring is how Branches puts a view on its part's one reader and takes it off: ReaderWiring for a bundle's +// views.Reader, ProcessWiring for a process's views.ProcessReader. Adopt is the reader's Adopt, so a view adopted after +// the reader's Start takes its first view at once; Drop is its Drop, which clears every finding the view raised. +type Wiring interface { + Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) error + Drop(owner, state string) error +} + +type readerWiring struct{ r *views.Reader } + +func (w readerWiring) Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) error { + return w.r.Adopt(f, seat, event, hear) +} +func (w readerWiring) Drop(owner, state string) error { return w.r.Drop(owner + "." + state) } + +type processWiring struct{ r *views.ProcessReader } + +func (w processWiring) Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) error { + _, err := w.r.Adopt(f, seat, event, hear) + return err +} +func (w processWiring) Drop(owner, state string) error { return w.r.Drop(owner, state) } + +// ReaderWiring wires Branches on a bundle's reader. +func ReaderWiring(r *views.Reader) Wiring { return readerWiring{r} } + +// ProcessWiring wires Branches on a process's reader. +func ProcessWiring(r *views.ProcessReader) Wiring { return processWiring{r} } + +// SourceBranches is a part's follower of the branches the gate's `source` keeps (design 47 G §614): a first view of the +// verb `source` given no branch, then each `source-moved` heard. Its picture is keyed by the escaped branch name. +type SourceBranches struct { + Follower *views.Follower + + call Call +} + +// NewSourceBranches is a SourceBranches asking the gate through call. Nothing is asked before the reader starts it. +func NewSourceBranches(call Call) *SourceBranches { + s := &SourceBranches{call: call} + s.Follower = &views.Follower{Owner: delivery.Seat, Tool: SourceVerb, State: StateSourceBranches, Ask: s.ask} + return s +} + +// Branches is the escaped name of each branch the picture holds. +func (s *SourceBranches) Branches() []string { + out := make([]string, 0) + for b := range s.Follower.Values() { + out = append(out, b) + } + slices.Sort(out) + return out +} + +// Hear takes one `source-moved` event: a branch the picture does not hold asks the first view again until it does, on +// its own goroutine, so the consumer is never held. Data that is no such event is an error. +func (s *SourceBranches) Hear(ctx context.Context, data []byte) error { + var ev delivery.SourceMovedEvent + if err := json.Unmarshal(data, &ev); err != nil || ev.Branch == "" { + return fmt.Errorf("meshcatalogclient: not a %s event: %s", delivery.EventSourceMoved, clipBytes(data)) + } + held := func() bool { _, ok := s.Follower.Get(keys.EscapeName(ev.Branch)); return ok } + if held() { + return nil + } + go catchUp(ctx, s.Follower, held, fmt.Sprintf("source of %s moved", ev.Branch)) + return nil +} + +func (s *SourceBranches) ask(_ context.Context, _ string) (views.Answer, error) { + raw, err := s.call(SourceVerb, delivery.SourceInput{}) + if err != nil { + return views.Answer{}, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return views.Answer{}, fmt.Errorf("%s answered: %s", SourceVerb, said) + } + var a delivery.SourceAnswer + if err := json.Unmarshal(answer, &a); err != nil { + return views.Answer{}, fmt.Errorf("%s answered no branches: %s", SourceVerb, clipBytes(answer)) + } + if a.Error != "" { + return views.Answer{}, fmt.Errorf("%s answered: %s", SourceVerb, a.Error) + } + values := make(map[string]json.RawMessage, len(a.Branches)) + for _, b := range a.Branches { + values[b] = json.RawMessage(`{}`) + } + return views.Answer{V: 1, Owner: delivery.Seat, State: StateSourceBranches, Created: registrationsCreated, Values: values, At: time.Now()}, nil +} + +// catchUp asks f's first view again until held says it holds what was heard, pausing between asks as the follower's Wait +// says (doubling to a minute), or until ctx ends: an ask already running may have been answered before the change was +// written. +func catchUp(ctx context.Context, f *views.Follower, held func() bool, why string) { + wait := f.Wait + if wait == 0 { + wait = 5 * time.Second + } + for !held() { + if err := f.Gap(ctx, why); err != nil { + return + } + if held() { + return + } + p, cancel := context.WithTimeout(ctx, wait) + <-p.Done() + cancel() + if ctx.Err() != nil { + return + } + if wait *= 2; wait > time.Minute { + wait = time.Minute + } + } +} + +// Branches is the one branch follower of design 47 G §614, built once and used by every part that reads the catalog half: +// it follows the watched branches of the registrations (the first view of `registrations` at the start, then each +// `registration-changed`); before the first registration, the branches the gate's `source` answers (its first view at +// the start, then each `source-moved`), which it stops following at the first `registration-changed`; it adopts a Catalog +// view on the part's reader for each branch that comes and drops the view of each that goes, and keeps the entries by +// fingerprint a part asks of the verb `entry` (Entries). The part declares what §614 says in `consumes` and `invokes`. +type Branches struct { + // Registrations is the follower of the registrations, also the part's own reading of them. + Registrations *Registrations + // Entries is the cache of the entries read by fingerprint. + Entries *Entries + // Changed, when not nil, is told of each change to a branch's view (module, its entry, deleted), set before Wire. + Changed func(branch, module string, entry json.RawMessage, deleted bool, at time.Time) + // Logf, when not nil, says each view adopted and dropped. + Logf func(format string, args ...any) + + call Call + w Wiring + + mu sync.Mutex + source *SourceBranches + left []string // the source's branches when it was left, kept until the registrations hold one + catalogs map[string]*Catalog + order []string +} + +// NewBranches adopts the registrations and the gate's source on w, so the reader's Start asks them, and then adopts a +// Catalog per branch as the picture says. call asks the seat verbs. +func NewBranches(w Wiring, call Call) (*Branches, error) { + b := &Branches{call: call, w: w, catalogs: map[string]*Catalog{}, Entries: NewEntries(call)} + b.Registrations = NewRegistrations(call) + b.Registrations.Follower.Changed = func(string, json.RawMessage, bool, time.Time) { b.reconcile() } + b.source = NewSourceBranches(call) + b.source.Follower.Changed = func(string, json.RawMessage, bool, time.Time) { b.reconcile() } + if err := w.Adopt(b.Registrations.Follower, seat.Seat, seat.EventRegistrationChanged, b.hearRegistration); err != nil { + return nil, err + } + if err := w.Adopt(b.source.Follower, delivery.Seat, delivery.EventSourceMoved, b.source.Hear); err != nil { + return nil, err + } + return b, nil +} + +// FollowBranches is NewBranches on a bundle's reader. +func FollowBranches(r *views.Reader, call Call) (*Branches, error) { + return NewBranches(ReaderWiring(r), call) +} + +// FollowBranchesProcess is NewBranches on a process's reader. +func FollowBranchesProcess(r *views.ProcessReader, call Call) (*Branches, error) { + return NewBranches(ProcessWiring(r), call) +} + +func (b *Branches) logf(format string, args ...any) { + if b.Logf != nil { + b.Logf(format, args...) + } +} + +// hearRegistration is the handler of `registration-changed`: the first one ends the following of the gate's source. +func (b *Branches) hearRegistration(ctx context.Context, data []byte) error { + err := b.Registrations.Hear(ctx, data) + if err != nil { + return err + } + b.mu.Lock() + s := b.source + if s != nil { + b.left, b.source = s.Branches(), nil + } + b.mu.Unlock() + if s != nil { + if err := b.w.Drop(delivery.Seat, StateSourceBranches); err != nil { + b.logf("meshcatalogclient: the gate's source is not dropped: %v", err) + } else { + b.logf("meshcatalogclient: the first registration-changed is heard: the gate's source is followed no more") + } + b.reconcile() + } + return nil +} + +// Followed is the branches (as named) whose views are adopted, in the order they came. +func (b *Branches) Followed() []string { + b.mu.Lock() + defer b.mu.Unlock() + out := make([]string, 0, len(b.order)) + for _, e := range b.order { + out = append(out, b.catalogs[e].Branch) + } + return out +} + +// Catalog is the view of one branch, as named; false when it is not followed. +func (b *Branches) Catalog(branch string) (*Catalog, bool) { + b.mu.Lock() + defer b.mu.Unlock() + c, ok := b.catalogs[keys.EscapeName(branch)] + return c, ok +} + +// Resolve reads an entry by its fingerprint: from a branch's held half when one lists it, else from the entries kept +// by fingerprint, asked of `entry` once when none is kept. The state is `answered` for an entry at hand, `asked` while the +// ask is out, `not-found` once the holder said it holds none. +func (b *Branches) Resolve(ctx context.Context, fingerprint string) (seat.CatalogEntry, EntryState) { + b.mu.Lock() + cs := make([]*Catalog, 0, len(b.order)) + for _, e := range b.order { + cs = append(cs, b.catalogs[e]) + } + b.mu.Unlock() + for _, c := range cs { + if half, ok := c.View(); ok { + for _, e := range half.Entries { + if e.Fp == fingerprint { + return e, EntryAnswered + } + } + } + } + return b.Entries.Need(ctx, fingerprint) +} + +// Listed says whether a branch's held half lists fingerprint. +func (b *Branches) Listed(fingerprint string) bool { + b.mu.Lock() + cs := make([]*Catalog, 0, len(b.order)) + for _, e := range b.order { + cs = append(cs, b.catalogs[e]) + } + b.mu.Unlock() + for _, c := range cs { + if half, ok := c.View(); ok { + for _, e := range half.Entries { + if e.Fp == fingerprint { + return true + } + } + } + } + return false +} + +// reconcile adopts the view of each branch that is wanted and drops the view of each that is not: the registrations' +// branches, and the gate's source's while it is followed (and, once left, until the registrations hold a branch). +func (b *Branches) reconcile() { + b.mu.Lock() + defer b.mu.Unlock() + want := map[string]string{} // escaped -> as named + for _, reg := range b.Registrations.All() { + for _, br := range reg.Branches { + want[keys.EscapeName(br)] = br + } + } + var from []string + switch { + case b.source != nil: + from = b.source.Branches() + case len(want) == 0: + from = b.left + default: + b.left = nil + } + for _, e := range from { + if name, err := keys.UnescapeName(e); err == nil { + want[e] = name + } + } + for _, e := range slices.Sorted(mapsKeys(want)) { + if b.catalogs[e] != nil { + continue + } + c, err := NewCatalog(b.call, want[e]) + if err != nil { + b.logf("meshcatalogclient: the branch %q is not followed: %v", want[e], err) + continue + } + branch := want[e] + c.Follower.Changed = func(module string, entry json.RawMessage, deleted bool, at time.Time) { + if b.Changed != nil { + b.Changed(branch, module, entry, deleted, at) + } + } + if err := b.w.Adopt(c.Follower, seat.Seat, seat.EventCatalogRevision, c.Hear); err != nil { + b.logf("meshcatalogclient: the view of the branch %q is not adopted: %v", want[e], err) + continue + } + b.catalogs[e] = c + b.order = append(b.order, e) + b.logf("meshcatalogclient: the branch %q is followed", want[e]) + } + for _, e := range slices.Clone(b.order) { + if _, ok := want[e]; ok { + continue + } + c := b.catalogs[e] + if err := b.w.Drop(seat.Seat, c.Follower.State); err != nil { + b.logf("meshcatalogclient: the view of the branch %q is not dropped: %v", c.Branch, err) + } + delete(b.catalogs, e) + b.order = slices.DeleteFunc(b.order, func(o string) bool { return o == e }) + b.logf("meshcatalogclient: the branch %q is followed no more", c.Branch) + } +} + +func mapsKeys(m map[string]string) func(func(string) bool) { + return func(yield func(string) bool) { + for k := range m { + if !yield(k) { + return + } + } + } +} + +// EntryState is where an entry read by fingerprint stands. +type EntryState string + +// The states of an entry kept by fingerprint (design 47 G §614, An older build's entry). +const ( + EntryAsked EntryState = "asked" + EntryAnswered EntryState = "answered" + EntryNotFound EntryState = "not-found" +) + +// EntryRow is one entry kept by fingerprint: its state, since when, and the entry once answered. +type EntryRow struct { + Fingerprint string + State EntryState + Since time.Time + // Detail says why an ask is still out when its last try failed. + Detail string + Entry *seat.CatalogEntry +} + +// Entries keeps, beside a part's views, the entries by fingerprint it asked of `entry` (design 47 G §614): each is +// asked, answered or not found. The reader a part passes to a pure function answers from the views and these alone. +type Entries struct { + // Changed, when not nil, is told of each entry that was answered or found missing, so the part starts a new pass at + // once, or writes the finding `entry-not-found`. + Changed func(row EntryRow) + // Now is the clock, time.Now when nil. + Now func() time.Time + + call Call + mu sync.Mutex + rows map[string]*EntryRow +} + +// NewEntries is an Entries asking the seat through call. +func NewEntries(call Call) *Entries { return &Entries{call: call, rows: map[string]*EntryRow{}} } + +func (e *Entries) now() time.Time { + if e.Now != nil { + return e.Now() + } + return time.Now() +} + +// Need reads fingerprint's entry: one kept is answered as it stands (`asked` as pending, a state apart from none); one +// not kept is asked of `entry` once, on its own goroutine, and answered as `asked`. +func (e *Entries) Need(ctx context.Context, fingerprint string) (seat.CatalogEntry, EntryState) { + e.mu.Lock() + row := e.rows[fingerprint] + if row == nil { + row = &EntryRow{Fingerprint: fingerprint, State: EntryAsked, Since: e.now()} + e.rows[fingerprint] = row + e.mu.Unlock() + go e.ask(ctx, fingerprint) + return seat.CatalogEntry{}, EntryAsked + } + defer e.mu.Unlock() + if row.Entry != nil { + return *row.Entry, row.State + } + return seat.CatalogEntry{}, row.State +} + +// Again asks fingerprint's entry now, whatever its state (the part's tool `_entry_again`). It waits for the answer. +func (e *Entries) Again(ctx context.Context, fingerprint string) error { + e.mu.Lock() + row := e.rows[fingerprint] + if row == nil { + e.mu.Unlock() + return fmt.Errorf("meshcatalogclient: no entry of %s is kept", fingerprint) + } + *row = EntryRow{Fingerprint: fingerprint, State: EntryAsked, Since: e.now()} + e.mu.Unlock() + e.ask(ctx, fingerprint) + return nil +} + +// Release removes fingerprint's entry: no `running` or `wanted` the part holds names it any more. +func (e *Entries) Release(fingerprint string) { + e.mu.Lock() + delete(e.rows, fingerprint) + e.mu.Unlock() +} + +// Rows lists the entries kept, by fingerprint (the part's tool `_first_views` lists them with their state and +// since when). +func (e *Entries) Rows() []EntryRow { + e.mu.Lock() + defer e.mu.Unlock() + out := make([]EntryRow, 0, len(e.rows)) + for _, r := range e.rows { + out = append(out, *r) + } + slices.SortFunc(out, func(a, b EntryRow) int { + switch { + case a.Fingerprint < b.Fingerprint: + return -1 + case a.Fingerprint > b.Fingerprint: + return 1 + } + return 0 + }) + return out +} + +// ask reads fingerprint's entry once. A call that failed to reach the holder leaves the entry asked, with its reason; +// the holder's own answer that it holds none is `not-found`. +func (e *Entries) ask(_ context.Context, fingerprint string) { + raw, err := e.call(EntryVerb, seat.EntryInput{Fingerprint: fingerprint}) + var row EntryRow + e.mu.Lock() + cur := e.rows[fingerprint] + if cur == nil { + e.mu.Unlock() + return + } + switch { + case err != nil: + cur.Detail = err.Error() + e.mu.Unlock() + return + default: + answer, said, ok := buscalls.Unwrap(raw) + var a seat.EntryAnswer + switch { + case !ok: + cur.State, cur.Since, cur.Detail = EntryNotFound, e.now(), said + case json.Unmarshal(answer, &a) != nil || a.Entry.Module == "": + cur.Detail = errors.New("answered no entry: " + clipBytes(answer)).Error() + e.mu.Unlock() + return + default: + entry := a.Entry + cur.State, cur.Since, cur.Detail, cur.Entry = EntryAnswered, e.now(), "", &entry + } + } + row = *cur + e.mu.Unlock() + if e.Changed != nil { + e.Changed(row) + } +} diff --git a/contracts/go/clients/meshcatalogclient/branches_test.go b/contracts/go/clients/meshcatalogclient/branches_test.go new file mode 100644 index 00000000..b69ef1d4 --- /dev/null +++ b/contracts/go/clients/meshcatalogclient/branches_test.go @@ -0,0 +1,224 @@ +package meshcatalogclient + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "strings" + "sync" + "testing" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// fakeWiring is a part's reader: it keeps the views adopted, by `.`, and says each drop. +type fakeWiring struct { + mu sync.Mutex + adopted map[string]*views.Follower + hears map[string]func(context.Context, []byte) error + dropped []string +} + +func (w *fakeWiring) Adopt(f *views.Follower, _, _ string, hear func(context.Context, []byte) error) error { + w.mu.Lock() + defer w.mu.Unlock() + name := f.Owner + "." + f.State + if w.adopted[name] != nil { + return fmt.Errorf("%s is followed already", name) + } + w.adopted[name], w.hears[name] = f, hear + return nil +} + +func (w *fakeWiring) Drop(owner, state string) error { + w.mu.Lock() + defer w.mu.Unlock() + name := owner + "." + state + if w.adopted[name] == nil { + return fmt.Errorf("%s is not followed", name) + } + delete(w.adopted, name) + delete(w.hears, name) + w.dropped = append(w.dropped, name) + return nil +} + +func (w *fakeWiring) names() []string { + w.mu.Lock() + defer w.mu.Unlock() + out := []string{} + for n := range w.adopted { + out = append(out, n) + } + slices.Sort(out) + return out +} + +func (w *fakeWiring) get(name string) (*views.Follower, func(context.Context, []byte) error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.adopted[name], w.hears[name] +} + +// fakeSeats answers source, registrations, revision and entry from what it is told. +type fakeSeats struct { + mu sync.Mutex + source []string + regs map[string][]string // module -> branches + version int64 + entries map[string]seat.CatalogEntry + entryAsk int +} + +func (f *fakeSeats) call(key string, body any) (json.RawMessage, error) { + f.mu.Lock() + defer f.mu.Unlock() + switch key { + case SourceVerb: + return json.Marshal(delivery.SourceAnswer{Branches: f.source}) + case RegistrationsVerb: + a := seat.RegistrationsAnswer{Registrations: []seat.Registration{}} + for m, br := range f.regs { + a.Registrations = append(a.Registrations, seat.Registration{Module: m, Version: f.version, Branches: br}) + } + return json.Marshal(a) + case RevisionVerb: + return json.Marshal(seat.RevisionAnswer{NoneYet: true}) + case EntryVerb: + f.entryAsk++ + e, ok := f.entries[body.(seat.EntryInput).Fingerprint] + if !ok { + return json.RawMessage(`{"error":"no such entry"}`), nil + } + return json.Marshal(seat.EntryAnswer{Entry: e}) + } + return nil, fmt.Errorf("asked %s", key) +} + +// CT16 (design 47 G §614, A-5): the branch follower follows the gate's source branches (read at the start, then +// source-moved) before the first registration-changed, adopts a view per branch and drops the view of a branch that goes: +// the first registration-changed ends the gate's source, and the registrations' branches are the views then. +func TestBranchesFollowSourceThenRegistrations(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + said := make(chan string, 64) + waitFor := func(part string) { + t.Helper() + for { + select { + case s := <-said: + if strings.Contains(s, part) { + return + } + case <-ctx.Done(): + t.Fatalf("never said %q", part) + } + } + } + seats := &fakeSeats{source: []string{"main"}, regs: map[string][]string{}, version: 1} + w := &fakeWiring{adopted: map[string]*views.Follower{}, hears: map[string]func(context.Context, []byte) error{}} + b, err := NewBranches(w, seats.call) + if err != nil { + t.Fatal(err) + } + b.Logf = func(format string, args ...any) { said <- fmt.Sprintf(format, args...) } + want := func(names ...string) { + t.Helper() + if got := w.names(); !slices.Equal(got, names) { + t.Fatalf("views = %v, want %v", got, names) + } + } + want("mesh-catalog.registrations", "mesh-delivery.source") + + // before the first registration: the gate's source says main, and source-moved brings feat/x + src, hearSource := w.get("mesh-delivery.source") + if err := src.Start(ctx); err != nil { + t.Fatal(err) + } + want("mesh-catalog.catalog-revisions.main", "mesh-catalog.registrations", "mesh-delivery.source") + seats.mu.Lock() + seats.source = []string{"main", "feat/x"} + seats.mu.Unlock() + src.Wait = time.Nanosecond + if err := hearSource(ctx, []byte(`{"branch":"feat/x","revision":4}`)); err != nil { + t.Fatal(err) + } + waitFor(`"feat/x" is followed`) + if got := b.Followed(); !slices.Equal(got, []string{"main", "feat/x"}) { + t.Fatalf("followed = %v", got) + } + + // the registrations hold one branch, and the first registration-changed ends the source + seats.mu.Lock() + seats.regs["nats"] = []string{"main"} + seats.version = 2 + seats.mu.Unlock() + reg, hearReg := w.get("mesh-catalog.registrations") + if err := reg.Start(ctx); err != nil { + t.Fatal(err) + } + if err := hearReg(ctx, []byte(`{"module":"nats","version":2,"moved":["branches"],"warrant":"a"}`)); err != nil { + t.Fatal(err) + } + waitFor(`"feat/x" is followed no more`) + want("mesh-catalog.catalog-revisions.main", "mesh-catalog.registrations") + if !slices.Contains(w.dropped, "mesh-delivery.source") { + t.Fatalf("the source is still followed: dropped %v", w.dropped) + } + + // a branch the registrations add is adopted, one they remove is dropped + seats.mu.Lock() + seats.regs["nats"] = []string{"dev"} + seats.version = 3 + seats.mu.Unlock() + if err := hearReg(ctx, []byte(`{"module":"nats","version":3,"moved":["branches"],"warrant":"b"}`)); err != nil { + t.Fatal(err) + } + waitFor(`"main" is followed no more`) + want("mesh-catalog.catalog-revisions.dev", "mesh-catalog.registrations") +} + +// CT17 (design 47 G §614, An older build's entry): an entry no branch lists is asked of `entry` once and read as asked, +// then answered (the part is told, to start a pass); one the holder does not hold is not-found; a released one is +// asked again when next needed. +func TestEntriesAreAskedOnceAnsweredOrNotFound(t *testing.T) { + ctx := context.Background() + seats := &fakeSeats{entries: map[string]seat.CatalogEntry{"src2:a": {Module: "m", Fp: "src2:a"}}} + e := NewEntries(seats.call) + told := make(chan EntryRow, 4) + e.Changed = func(r EntryRow) { told <- r } + + if _, st := e.Need(ctx, "src2:a"); st != EntryAsked { + t.Fatalf("state = %s, want asked", st) + } + if _, st := e.Need(ctx, "src2:a"); st != EntryAsked && st != EntryAnswered { + t.Fatalf("state = %s", st) + } + if r := <-told; r.State != EntryAnswered || r.Entry == nil || r.Entry.Module != "m" { + t.Fatalf("told %+v", r) + } + if got, st := e.Need(ctx, "src2:a"); st != EntryAnswered || got.Module != "m" { + t.Fatalf("kept entry = %s %s", st, got.Module) + } + if seats.entryAsk != 1 { + t.Fatalf("entry asked %d times, want once", seats.entryAsk) + } + e.Need(ctx, "src2:none") + if r := <-told; r.State != EntryNotFound || r.Fingerprint != "src2:none" || r.Detail == "" { + t.Fatalf("told %+v", r) + } + if rows := e.Rows(); len(rows) != 2 || rows[1].State != EntryNotFound { + t.Fatalf("rows %+v", rows) + } + e.Release("src2:none") + if rows := e.Rows(); len(rows) != 1 { + t.Fatalf("rows after release %+v", rows) + } + if err := e.Again(ctx, "src2:a"); err != nil || seats.entryAsk != 3 { + t.Fatalf("again: %v, asks %d", err, seats.entryAsk) + } +} diff --git a/contracts/go/clients/meshcatalogclient/go.mod b/contracts/go/clients/meshcatalogclient/go.mod index 8a0a9aa3..285aebe0 100644 --- a/contracts/go/clients/meshcatalogclient/go.mod +++ b/contracts/go/clients/meshcatalogclient/go.mod @@ -5,6 +5,7 @@ go 1.26.0 require ( git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 ) diff --git a/contracts/go/clients/meshcatalogclient/registrations.go b/contracts/go/clients/meshcatalogclient/registrations.go index d58aaf0e..5bb9acc0 100644 --- a/contracts/go/clients/meshcatalogclient/registrations.go +++ b/contracts/go/clients/meshcatalogclient/registrations.go @@ -144,30 +144,8 @@ func (r *Registrations) holds(module string, version int64) bool { 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 - } - } + catchUp(ctx, r.Follower, func() bool { return r.holds(module, version) }, fmt.Sprintf("registration of %s at version %d said", module, version)) } // ask is the Follower's Ask: every registration, whole in one answer.