A-5 (§614): meshcatalogclient follows the branches once for every part: registrations, the gate's source, views adopted and dropped, entries by fingerprint
Builds: design 47 G §614, A-5 (G-20) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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 `<module>_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 `<module>_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)
|
||||
}
|
||||
}
|
||||
@@ -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 `<owner>.<state>`, 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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user